diff --git a/admin-web/src/ProjectLanding.tsx b/admin-web/src/ProjectLanding.tsx index f81a398..15a6ea2 100644 --- a/admin-web/src/ProjectLanding.tsx +++ b/admin-web/src/ProjectLanding.tsx @@ -10,9 +10,11 @@ import { Input, List, Modal, + Select, Space, Spin, Steps, + Switch, Tag, Typography, Upload, @@ -23,6 +25,7 @@ import { CheckCircleOutlined, DeleteOutlined, EditOutlined, + EnvironmentOutlined, FolderOutlined, InboxOutlined, MoreOutlined, @@ -45,6 +48,7 @@ import { type ProjectDeletionImpact, type ProjectSummary, type GraphReleaseSummary, + type SpatialMapScope, } from "./api"; import { mergeProjectCatalogItem, @@ -57,6 +61,8 @@ const { TextArea } = Input; type BasicProjectValues = { project_id: string; display_name: string; + map_enabled: boolean; + map_scope: SpatialMapScope; }; type GraphImportData = { @@ -73,6 +79,11 @@ const SAFE_RESOURCE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$/; const SAFE_SCHEMA_IDENTIFIER_PATTERN = /^[A-Za-z_][A-Za-z0-9_]{0,127}$/; const SAFE_SCHEMA_VERSION_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/; +const MAP_SCOPE_LABELS: Record = { + libo: "荔波县知识地图", + guizhou: "贵州省知识地图", +}; + const FIELD_TYPE_ALIASES: Record = { str: "string", text: "string", @@ -714,6 +725,8 @@ export default function ProjectLanding() { const normalized: BasicProjectValues = { project_id: values.project_id.trim(), display_name: values.display_name.trim(), + map_enabled: Boolean(values.map_enabled), + map_scope: values.map_scope || "libo", }; setCheckingBasics(true); try { @@ -822,6 +835,10 @@ export default function ProjectLanding() { const { data } = await provisionGraphProject({ project_id: basicValues.project_id, display_name: basicValues.display_name, + spatial_map: { + enabled: basicValues.map_enabled, + scope: basicValues.map_scope, + }, schema: schemaResult.value, graph_data: graphResult.value, }); @@ -1054,6 +1071,7 @@ export default function ProjectLanding() { layout="vertical" className="project-basic-form" disabled={checkingBasics} + initialValues={{ map_enabled: false, map_scope: "libo" }} >
+
+
+ +
+ 在数据大屏显示知识地图 + + 适用于包含经纬度的酒店、美食、景点、交通等空间数据；与项目英文标识无关。 + +
+
+ + + +
+ previous.map_enabled !== current.map_enabled}> + {({ getFieldValue }) => getFieldValue("map_enabled") ? ( + + div { + display: flex; + flex-direction: column; + gap: 3px; +} + +.project-map-option-icon { + display: inline-flex; + align-items: center; + justify-content: center; + flex: 0 0 auto; + width: 32px; + height: 32px; + color: #1677ff; + border-radius: 8px; + background: #e6f4ff; +} + +.project-map-scope { + max-width: 360px; +} + .project-json-step { display: flex; flex-direction: column; @@ -2797,6 +2838,10 @@ body { margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Ro grid-template-columns: 1fr; } + .project-map-option { + align-items: flex-start; + } + .project-json-workspace { min-height: 0; } diff --git a/app/api/plaza.py b/app/api/plaza.py index e841eda..6c73ac7 100644 --- a/app/api/plaza.py +++ b/app/api/plaza.py @@ -27,6 +27,14 @@ SPATIAL_GRAPH_NAME = "guiyang_spatial_v1" LIBO_ZONE_SET_ID = "ZS-LB-YUPING-V2" LIBO_BUS_DATASET = "libo_bus_xlsx_v1" +GRAPH_POI_LABELS = { + "FoodPlace": "美食", + "Hotel": "酒店", + "ScenicSpot": "景点", + "TransitFacility": "交通设施", + "BusStop": "公交站", +} + FOOD_ENRICHMENT_KEYS = ( "food_fusion_status", "food_match_confidence", @@ -525,20 +533,42 @@ def _build_libo_bus_route_payload( def _read_libo_bus_routes(graph_name: str) -> dict[str, Any]: - graph = FalkorDB( + db = FalkorDB( host=settings.falkordb_host, port=settings.falkordb_port, - ).select_graph(graph_name) - rows = graph.query( - "MATCH (b:BusLine) WHERE b.dataset=$dataset " - "OPTIONAL MATCH (b)-[r:STOPS_AT]->(s:Place) " - "RETURN b.route_id,b.line_name,b.direction,b.start_stop,b.end_stop," - "b.stop_count,b.fare_yuan,b.first_bus,b.last_bus,b.distance_km," - "b.path_source,s.element_id,s.name,s.lng,s.lat,r.sequence " - "ORDER BY b.line_name,b.route_id,r.sequence", - {"dataset": LIBO_BUS_DATASET}, - ).result_set - return _build_libo_bus_route_payload(rows, graph_name) + ) + try: + graph = db.select_graph(graph_name) + rows = graph.query( + "MATCH (b:BusLine) WHERE b.dataset=$dataset " + "OPTIONAL MATCH (b)-[r:STOPS_AT]->(s:Place) " + "RETURN b.route_id,b.line_name,b.direction,b.start_stop,b.end_stop," + "b.stop_count,b.fare_yuan,b.first_bus,b.last_bus,b.distance_km," + "b.path_source,s.element_id,s.name,s.lng,s.lat,r.sequence " + "ORDER BY b.line_name,b.route_id,r.sequence", + {"dataset": LIBO_BUS_DATASET}, + ).result_set + payload = _build_libo_bus_route_payload(rows, graph_name) + if payload["direction_count"]: + return payload + + # Full JSON v2 stores one BusLine with one or more directional + # BusRoute nodes. Keep the legacy query above for existing graphs and + # fall back to the normalized structure for newly imported projects. + rows = graph.query( + "MATCH (b:BusLine)-[:HAS_ROUTE]->(route:BusRoute) " + "WHERE b.dataset=$dataset " + "OPTIONAL MATCH (route)-[r:STOPS_AT]->(s:BusStop) " + "RETURN route.route_id,coalesce(route.line_name,b.line_name)," + "route.direction,route.start_stop,route.end_stop,route.stop_count," + "route.fare_yuan,route.first_bus,route.last_bus,route.distance_km," + "route.path_source,s.element_id,s.name,s.lng,s.lat,r.sequence " + "ORDER BY route.line_name,route.route_id,r.sequence", + {"dataset": LIBO_BUS_DATASET}, + ).result_set + return _build_libo_bus_route_payload(rows, graph_name) + finally: + db.close() _INTERNAL_ENRICHMENT_KEYS = { @@ -641,35 +671,304 @@ def _graph_poi_enrichment( def _graph_poi_categories(graph_name: str) -> dict[str, list[str]]: - graph = FalkorDB( + db = FalkorDB( host=settings.falkordb_host, port=settings.falkordb_port, - ).select_graph(graph_name) - rows = graph.query( - "MATCH (n) " - "WHERE n.element_id IS NOT NULL " - "AND any(label IN labels(n) WHERE label IN " - "['FoodPlace','Hotel','ScenicSpot','TransitFacility']) " - "RETURN n.element_id, labels(n)" - ).result_set - label_names = { - "FoodPlace": "美食", - "Hotel": "酒店", - "ScenicSpot": "景点", - "TransitFacility": "交通设施", - } - result: dict[str, list[str]] = {} - for element_id, labels in rows: + ) + try: + graph = db.select_graph(graph_name) + rows = graph.query( + "MATCH (n) " + "WHERE n.element_id IS NOT NULL " + "AND any(label IN labels(n) WHERE label IN " + "['FoodPlace','Hotel','ScenicSpot','TransitFacility','BusStop']) " + "RETURN n.element_id, labels(n)" + ).result_set + result: dict[str, list[str]] = {} + for element_id, labels in rows: + categories = [ + GRAPH_POI_LABELS[label] + for label in (labels or []) + if label in GRAPH_POI_LABELS + ] + if element_id and categories: + result[str(element_id)] = categories + return result + finally: + db.close() + + +def _graph_json_value(value: Any) -> Any: + """Decode object/array properties serialized for FalkorDB storage.""" + if not isinstance(value, str): + return value + text = value.strip() + if not text or text[0] not in "[{": + return value + try: + return json.loads(text) + except json.JSONDecodeError: + return value + + +def _graph_map_poi_items(rows: list[list[Any]]) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + for node_id, labels, raw_properties in rows: + properties = raw_properties if isinstance(raw_properties, dict) else {} + lng = properties.get("lng") + lat = properties.get("lat") + if lng is None or lat is None: + continue categories = [ - label_names[label] + GRAPH_POI_LABELS[label] for label in (labels or []) - if label in label_names + if label in GRAPH_POI_LABELS ] - if element_id and categories: - result[str(element_id)] = categories + category = str(properties.get("type_label") or (categories[0] if categories else "其他地点")) + element_id = str( + properties.get("element_id") + or node_id + or properties.get("gaode_poi_id") + or "" + ) + items.append( + { + "id": element_id, + "gaode_poi_id": str(properties.get("gaode_poi_id") or ""), + "name": str(properties.get("name") or "未命名POI"), + "category": category, + "categories": categories or [category], + "place_type": str(properties.get("place_type") or "poi"), + "lng": float(lng), + "lat": float(lat), + "address": str(properties.get("address") or ""), + "district": str(properties.get("district") or ""), + "city": str(properties.get("city") or ""), + "towncode": str(properties.get("towncode") or ""), + "town_name": str(properties.get("town_name") or ""), + "zone_id": str(properties.get("zone_id") or ""), + "zone_name": str(properties.get("zone_name") or ""), + } + ) + return items + + +def _read_graph_map_pois(graph_name: str) -> list[dict[str, Any]]: + db = FalkorDB(host=settings.falkordb_host, port=settings.falkordb_port) + try: + rows = db.select_graph(graph_name).query( + "MATCH (n) " + "WHERE n.lng IS NOT NULL AND n.lat IS NOT NULL " + "AND any(label IN labels(n) WHERE label IN " + "['FoodPlace','Hotel','ScenicSpot','TransitFacility','BusStop']) " + "RETURN n.__kg_node_id,labels(n),properties(n) " + "ORDER BY n.name LIMIT 15000" + ).result_set + return _graph_map_poi_items(rows) + finally: + db.close() + + +def _graph_zone_payload( + town_rows: list[list[Any]], + zone_rows: list[list[Any]], + graph_name: str, +) -> dict[str, Any]: + towns = [] + for raw_properties, poi_count, min_lng, min_lat, max_lng, max_lat in town_rows: + properties = raw_properties if isinstance(raw_properties, dict) else {} + bbox = None + if min_lng is not None: + bbox = [float(min_lng), float(min_lat), float(max_lng), float(max_lat)] + towns.append( + { + "towncode": str(properties.get("towncode") or ""), + "name": str(properties.get("name") or ""), + "unit_type": str(properties.get("unit_type") or ""), + "civil_adcode": str(properties.get("civil_adcode") or ""), + "community_count": int(properties.get("community_count") or 0), + "village_count": int(properties.get("village_count") or 0), + "poi_count": int(poi_count or 0), + "bbox": bbox, + "has_boundary": False, + } + ) + + zones = [] + for raw_properties, poi_count in zone_rows: + properties = raw_properties if isinstance(raw_properties, dict) else {} + anchor_lng = properties.get("anchor_lng") + anchor_lat = properties.get("anchor_lat") + zones.append( + { + "zone_id": str(properties.get("zone_id") or ""), + "zone_name": str(properties.get("zone_name") or ""), + "anchor_name": str(properties.get("anchor_name") or ""), + "anchor": ( + [float(anchor_lng), float(anchor_lat)] + if anchor_lng is not None and anchor_lat is not None + else None + ), + "cap_m": float(properties["cap_m"]) if properties.get("cap_m") is not None else None, + "parent_admin_name": str(properties.get("parent_admin_name") or ""), + "anchor_source": str(properties.get("anchor_source") or ""), + "poi_count": int(poi_count or 0), + "area_km2": float(properties["area_km2"]) if properties.get("area_km2") is not None else None, + "dominant_category": str(properties.get("dominant_category") or ""), + "bank": _graph_json_value(properties.get("bank")), + "food_count": int(properties.get("food_count") or 0), + "hotel_count": int(properties.get("hotel_count") or 0), + "transit_count": int(properties.get("transit_count") or 0), + "scenic_count": int(properties.get("scenic_count") or 0), + "geometry": _graph_json_value(properties.get("geometry")), + "geometry_variant": str(properties.get("geometry_variant") or ""), + } + ) + towns.sort(key=lambda item: (-item["poi_count"], item["towncode"])) + zones.sort(key=lambda item: item["zone_id"]) + return { + "graph_name": graph_name, + "zone_set_id": LIBO_ZONE_SET_ID, + "towns": towns, + "zones": zones, + } + + +def _read_graph_map_zones(graph_name: str) -> dict[str, Any]: + db = FalkorDB(host=settings.falkordb_host, port=settings.falkordb_port) + try: + graph = db.select_graph(graph_name) + town_rows = graph.query( + "MATCH (town:Area) WHERE town.area_level='town' " + "OPTIONAL MATCH (poi)-[:LOCATED_IN]->(town) " + "RETURN properties(town),count(poi),min(poi.lng),min(poi.lat)," + "max(poi.lng),max(poi.lat)" + ).result_set + zone_rows = graph.query( + "MATCH (zone:BusinessZone) " + "OPTIONAL MATCH (poi)-[:IN_BUSINESS_ZONE]->(zone) " + "RETURN properties(zone),count(poi)" + ).result_set + return _graph_zone_payload(town_rows, zone_rows, graph_name) + finally: + db.close() + + +def _graph_enrichment_properties( + properties: dict[str, Any], + keys: tuple[str, ...], +) -> dict[str, Any]: + result: dict[str, Any] = {} + for key in keys: + if key not in properties: + continue + value = _clean_enrichment_display_value(key, properties[key]) + if value in (None, "", "[]", "{}"): + continue + result[key] = _graph_json_value(value) return result +def _graph_poi_detail_payload( + graph_name: str, + labels: list[str], + raw_properties: dict[str, Any], +) -> dict[str, Any]: + properties = {key: _graph_json_value(value) for key, value in raw_properties.items()} + raw = properties.get("spatial_raw_data") or properties.get("raw_jsonb") or {} + if not isinstance(raw, dict): + raw = {} + image_rows = properties.get("images") or [] + image_urls = [ + str(item.get("image_url")) + for item in image_rows + if isinstance(item, dict) and item.get("image_url") + ] + photo_urls = list( + dict.fromkeys( + [ + *_photo_urls(properties.get("photo_urls")), + *_photo_urls(properties.get("image_urls")), + *image_urls, + ] + ) + ) + element_id = str( + properties.get("element_id") + or properties.get("__kg_node_id") + or properties.get("gaode_poi_id") + or "" + ) + return { + "graph_name": graph_name, + "id": element_id, + "gaode_poi_id": str(properties.get("gaode_poi_id") or ""), + "name": str(properties.get("name") or "未命名POI"), + "category": str(properties.get("type_label") or "其他地点"), + "place_type": str(properties.get("place_type") or "poi"), + "business_subcategory": raw.get("business_subcategory") or "", + "scenic_type": raw.get("scenic_type") or properties.get("scenic_type") or "", + "scenic_level": raw.get("scenic_level") or properties.get("scenic_level") or "", + "parent_scenic": raw.get("parent_scenic") or "", + "scenic_grade": raw.get("scenic_grade") or "", + "visitor_value": raw.get("visitor_value") or properties.get("visitor_value_type") or "", + "audit_result": raw.get("audit_result") or "", + "audit_confidence": raw.get("audit_confidence") or "", + "audit_basis": raw.get("audit_basis") or "", + "audit_date": raw.get("audit_date") or "", + "amap_type": str(properties.get("amap_type") or ""), + "typecode": str(properties.get("typecode") or ""), + "scan_hit_count": raw.get("scan_hit_count") or 0, + "matched_scan_types": raw.get("matched_scan_types") or [], + "lng": float(properties.get("lng") or 0), + "lat": float(properties.get("lat") or 0), + "province": str(properties.get("province") or ""), + "city": str(properties.get("city") or ""), + "district": str(properties.get("district") or ""), + "adcode": str(properties.get("adcode") or ""), + "business_area": str(properties.get("business_area") or ""), + "address": str(properties.get("address") or ""), + "tel": str(properties.get("tel") or properties.get("phone") or ""), + "open_time": str(properties.get("open_time") or ""), + "rating": properties.get("rating") or "", + "cost": properties.get("cost") or "", + "level": properties.get("level") or "", + "tags": properties.get("tags") or "", + "photo_urls": photo_urls, + "source": str(properties.get("source") or ""), + "source_cell_id": str(properties.get("source_cell_id") or ""), + "source_resolution": properties.get("source_resolution"), + "source_scope_adcode": str(properties.get("source_scope_adcode") or ""), + "first_fetched_at": str(properties.get("first_fetched_at") or ""), + "last_fetched_at": str(properties.get("last_fetched_at") or ""), + "graph_labels": [str(label) for label in labels], + "food_enrichment": _graph_enrichment_properties(properties, FOOD_ENRICHMENT_KEYS), + "hotel_enrichment": _graph_enrichment_properties(properties, HOTEL_ENRICHMENT_KEYS), + } + + +def _read_graph_poi_detail(graph_name: str, place_id: str) -> dict[str, Any] | None: + db = FalkorDB(host=settings.falkordb_host, port=settings.falkordb_port) + try: + rows = db.select_graph(graph_name).query( + "MATCH (n) WHERE n.element_id=$place_id " + "OR n.gaode_poi_id=$gaode_poi_id OR n.__kg_node_id=$place_id " + "RETURN labels(n),properties(n) LIMIT 1", + { + "place_id": place_id, + "gaode_poi_id": place_id.removeprefix("amap:"), + }, + ).result_set + if not rows: + return None + labels, properties = rows[0] + if not isinstance(properties, dict): + return None + return _graph_poi_detail_payload(graph_name, labels or [], properties) + finally: + db.close() + + @router.get("/plaza/map-pois") async def map_pois( context: ProjectContext = Depends(get_project_context), @@ -696,35 +995,42 @@ async def map_pois( (LIBO_ZONE_SET_ID, graph_name), ) rows = await cur.fetchall() - graph_categories = await asyncio.to_thread( - _graph_poi_categories, - graph_name, - ) - - items = [ - { - "id": row["element_id"], - "gaode_poi_id": row["gaode_poi_id"], - "name": row["name"] or "未命名POI", - "category": row["type_label"] or "其他地点", - "categories": graph_categories.get( - str(row["element_id"]), - [row["type_label"] or "其他地点"], - ), - "place_type": row["place_type"] or "poi", - "lng": float(row["lng"]), - "lat": float(row["lat"]), - "address": row["address"] or "", - "district": row["district"] or "", - "city": row["city"] or "", - "towncode": row["towncode"] or "", - "town_name": row["town_name"] or "", - "zone_id": row["zone_id"] or "", - "zone_name": row["zone_name"] or "", - } - for row in rows - if row.get("lng") is not None and row.get("lat") is not None - ] + if rows: + try: + graph_categories = await asyncio.to_thread( + _graph_poi_categories, + graph_name, + ) + except Exception: # noqa: BLE001 - relational points remain usable + graph_categories = {} + items = [ + { + "id": row["element_id"], + "gaode_poi_id": row["gaode_poi_id"], + "name": row["name"] or "未命名POI", + "category": row["type_label"] or "其他地点", + "categories": graph_categories.get( + str(row["element_id"]), + [row["type_label"] or "其他地点"], + ), + "place_type": row["place_type"] or "poi", + "lng": float(row["lng"]), + "lat": float(row["lat"]), + "address": row["address"] or "", + "district": row["district"] or "", + "city": row["city"] or "", + "towncode": row["towncode"] or "", + "town_name": row["town_name"] or "", + "zone_id": row["zone_id"] or "", + "zone_name": row["zone_name"] or "", + } + for row in rows + if row.get("lng") is not None and row.get("lat") is not None + ] + else: + # A project created entirely from JSON has no project-specific rows in + # the administrative spatial tables. Its map data lives in FalkorDB. + items = await asyncio.to_thread(_read_graph_map_pois, graph_name) category_counts: dict[str, int] = {} for item in items: for category in item["categories"]: @@ -766,12 +1072,14 @@ async def map_zones( (f"{s}.libo_business_zones",), ) if not (await cur.fetchone())["reg"]: - return { - "graph_name": graph_name, - "zone_set_id": LIBO_ZONE_SET_ID, - "towns": [], - "zones": [], - } + return await asyncio.to_thread(_read_graph_map_zones, graph_name) + + await cur.execute( + f"SELECT EXISTS(SELECT 1 FROM {s}.amap_spatial_pois WHERE graph_name=%s) AS found", + (graph_name,), + ) + if not bool((await cur.fetchone())["found"]): + return await asyncio.to_thread(_read_graph_map_zones, graph_name) await cur.execute( f"""SELECT t.towncode, t.name, t.unit_type, t.civil_adcode, @@ -910,7 +1218,14 @@ async def map_poi_detail( ) row = await cur.fetchone() if not row: - raise HTTPException(status_code=404, detail="当前项目中未找到该POI") + graph_detail = await asyncio.to_thread( + _read_graph_poi_detail, + graph_name, + place_id, + ) + if graph_detail is None: + raise HTTPException(status_code=404, detail="当前项目中未找到该POI") + return graph_detail raw = row.get("raw_jsonb") or {} if not isinstance(raw, dict): diff --git a/app/api/projects.py b/app/api/projects.py index f73782d..505dba9 100644 --- a/app/api/projects.py +++ b/app/api/projects.py @@ -1019,7 +1019,12 @@ async def provision_project(body: dict, _user: CurrentUser): payload["display_name"], payload["description"], payload["schema"]["namespace"], - Jsonb({"provision_source": "admin-project-wizard"}), + Jsonb( + { + "provision_source": "admin-project-wizard", + "spatial_map": payload["spatial_map"], + } + ), _user["username"], ), ) @@ -1181,6 +1186,7 @@ async def provision_project(body: dict, _user: CurrentUser): "provision_source": "admin-project-wizard", "schema_id": schema_id, "graph_name": payload["graph_name"], + "spatial_map": payload["spatial_map"], **payload["counts"], } ), diff --git a/app/project_lifecycle.py b/app/project_lifecycle.py index 491ff0b..b7b3cce 100644 --- a/app/project_lifecycle.py +++ b/app/project_lifecycle.py @@ -8,6 +8,7 @@ from __future__ import annotations import json import re +from collections import defaultdict from collections.abc import Mapping from typing import Any @@ -47,6 +48,9 @@ SUPPORTED_FIELD_TYPES = { "any", } +GRAPH_IMPORT_BATCH_THRESHOLD = 500 +GRAPH_IMPORT_BATCH_SIZE = 250 + class ProjectValidationError(ValueError): """Raised when a provision payload is invalid before any write occurs.""" @@ -227,6 +231,22 @@ def normalize_provision_payload(body: Any) -> dict[str, Any]: display_name = str(body.get("display_name") or "").strip() graph_name = explicit_graph_name or project_id + raw_spatial_map = body.get("spatial_map", {}) + if raw_spatial_map is None: + raw_spatial_map = {} + if not isinstance(raw_spatial_map, Mapping): + errors.append("spatial_map 必须是 JSON 对象") + raw_spatial_map = {} + raw_map_enabled = raw_spatial_map.get("enabled", False) + if not isinstance(raw_map_enabled, bool): + errors.append("spatial_map.enabled 必须是布尔值") + map_enabled = False + else: + map_enabled = raw_map_enabled + map_scope = str(raw_spatial_map.get("scope") or "libo").strip().lower() + if map_scope not in {"libo", "guizhou"}: + errors.append("spatial_map.scope 仅支持 libo 或 guizhou") + if not project_id: errors.append("项目英文标识不能为空") elif not SAFE_RESOURCE_ID.fullmatch(project_id): @@ -415,6 +435,7 @@ def normalize_provision_payload(body: Any) -> dict[str, Any]: "display_name": display_name, "description": str(body.get("description") or "").strip() or None, "graph_name": graph_name, + "spatial_map": {"enabled": map_enabled, "scope": map_scope}, "schema": schema, "graph_data": {"nodes": nodes, "relations": relations}, "counts": { @@ -467,6 +488,65 @@ def _graph_safe_properties(properties: Mapping[str, Any]) -> dict[str, Any]: } +def _batches(items: list[Any], size: int = GRAPH_IMPORT_BATCH_SIZE): + for start in range(0, len(items), size): + yield items[start:start + size] + + +def _import_falkor_graph_batched(graph: Any, graph_data: Mapping[str, Any]) -> None: + """Import a validated graph with bounded UNWIND batches and lookup indexes.""" + nodes = list(graph_data.get("nodes", [])) + relations = list(graph_data.get("relations", [])) + node_types = {str(item["id"]): str(item["type"]) for item in nodes} + + nodes_by_type: dict[str, list[dict[str, Any]]] = defaultdict(list) + for item in nodes: + properties = _graph_safe_properties(item.get("properties") or {}) + properties.setdefault("id", str(item["id"])) + properties["__kg_node_id"] = str(item["id"]) + nodes_by_type[str(item["type"])].append({"properties": properties}) + + for label, items in nodes_by_type.items(): + for batch in _batches(items): + graph.query( + f"UNWIND $rows AS row CREATE (n:{label}) SET n += row.properties", + {"rows": batch}, + ) + # Relation creation matches both endpoint labels and this indexed id, + # avoiding repeated full-graph scans for large JSON imports. + graph.query(f"CREATE INDEX FOR (n:{label}) ON (n.__kg_node_id)") + + relations_by_signature: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) + for item in relations: + source = str(item["source"]) + target = str(item["target"]) + signature = ( + str(item["type"]), + node_types[source], + node_types[target], + ) + relations_by_signature[signature].append( + { + "source": source, + "target": target, + "properties": _graph_safe_properties(item.get("properties") or {}), + } + ) + + for (relation_type, source_label, target_label), items in relations_by_signature.items(): + for batch in _batches(items): + graph.query( + f""" + UNWIND $rows AS row + MATCH (source:{source_label} {{__kg_node_id:row.source}}) + MATCH (target:{target_label} {{__kg_node_id:row.target}}) + CREATE (source)-[relation:{relation_type}]->(target) + SET relation += row.properties + """, + {"rows": batch}, + ) + + def import_falkor_graph(graph_name: str, graph_data: Mapping[str, Any]) -> dict[str, int]: """Create a brand-new FalkorDB graph from already validated data.""" @@ -480,31 +560,37 @@ def import_falkor_graph(graph_name: str, graph_data: Mapping[str, Any]) -> dict[ if graph_name in existing: raise FileExistsError(f"FalkorDB 图 {graph_name!r} 已存在") graph = db.select_graph(graph_name) - for node in graph_data.get("nodes", []): - label = str(node["type"]) - properties = _graph_safe_properties(node.get("properties") or {}) - properties.setdefault("id", str(node["id"])) - properties["__kg_node_id"] = str(node["id"]) - # From this point onward the graph may exist even if the client - # raises after the server accepted the command. - created = True - graph.query(f"CREATE (n:{label}) SET n += $props", {"props": properties}) - for relation in graph_data.get("relations", []): - relation_type = str(relation["type"]) - properties = _graph_safe_properties(relation.get("properties") or {}) - graph.query( - f""" - MATCH (source {{__kg_node_id:$source}}), (target {{__kg_node_id:$target}}) - CREATE (source)-[relation:{relation_type}]->(target) - SET relation += $props - """, - { - "source": str(relation["source"]), - "target": str(relation["target"]), - "props": properties, - }, - ) + nodes = list(graph_data.get("nodes", [])) + relations = list(graph_data.get("relations", [])) + if len(nodes) + len(relations) >= GRAPH_IMPORT_BATCH_THRESHOLD: created = True + _import_falkor_graph_batched(graph, {"nodes": nodes, "relations": relations}) + else: + for node in nodes: + label = str(node["type"]) + properties = _graph_safe_properties(node.get("properties") or {}) + properties.setdefault("id", str(node["id"])) + properties["__kg_node_id"] = str(node["id"]) + # From this point onward the graph may exist even if the client + # raises after the server accepted the command. + created = True + graph.query(f"CREATE (n:{label}) SET n += $props", {"props": properties}) + for relation in relations: + relation_type = str(relation["type"]) + properties = _graph_safe_properties(relation.get("properties") or {}) + graph.query( + f""" + MATCH (source {{__kg_node_id:$source}}), (target {{__kg_node_id:$target}}) + CREATE (source)-[relation:{relation_type}]->(target) + SET relation += $props + """, + { + "source": str(relation["source"]), + "target": str(relation["target"]), + "props": properties, + }, + ) + created = True if not created: return {"nodes": 0, "relations": 0} node_count = int(graph.query("MATCH (n) RETURN count(n)").result_set[0][0]) diff --git a/tests/test_plaza_graph_fallback.py b/tests/test_plaza_graph_fallback.py new file mode 100644 index 0000000..9d6b405 --- /dev/null +++ b/tests/test_plaza_graph_fallback.py @@ -0,0 +1,98 @@ +import unittest + +from app.api.plaza import ( + _build_libo_bus_route_payload, + _graph_map_poi_items, + _graph_poi_detail_payload, + _graph_zone_payload, +) + + +class PlazaGraphFallbackTests(unittest.TestCase): + def test_graph_pois_cover_business_places_and_bus_stops(self) -> None: + rows = [ + [ + "poi:hotel-1", + ["Hotel"], + { + "element_id": "amap:H1", + "gaode_poi_id": "H1", + "name": "荔波测试酒店", + "type_label": "酒店", + "place_type": "hotel", + "lng": 107.88, + "lat": 25.40, + "towncode": "522722100000", + "zone_id": "A01", + }, + ], + [ + "poi:stop-1", + ["BusStop"], + { + "element_id": "amap:S1", + "gaode_poi_id": "S1", + "name": "汽车站", + "type_label": "公交站", + "place_type": "transit_stop", + "lng": 107.89, + "lat": 25.41, + }, + ], + ] + + items = _graph_map_poi_items(rows) + + self.assertEqual(len(items), 2) + self.assertEqual(items[0]["categories"], ["酒店"]) + self.assertEqual(items[1]["categories"], ["公交站"]) + self.assertEqual(items[0]["zone_id"], "A01") + + def test_graph_zones_decode_serialized_geometry(self) -> None: + payload = _graph_zone_payload( + [[{"towncode": "522722100000", "name": "玉屏街道"}, 12, 107.8, 25.3, 107.9, 25.5]], + [[{"zone_id": "A01", "zone_name": "古城片区", "geometry": "[[[107.8,25.3],[107.9,25.4]]]"}, 6]], + "yunyou_libo_full_v2", + ) + + self.assertEqual(payload["towns"][0]["poi_count"], 12) + self.assertEqual(payload["towns"][0]["bbox"], [107.8, 25.3, 107.9, 25.5]) + self.assertIsInstance(payload["zones"][0]["geometry"], list) + self.assertEqual(payload["zones"][0]["poi_count"], 6) + + def test_graph_detail_uses_embedded_json_images(self) -> None: + payload = _graph_poi_detail_payload( + "yunyou_libo_full_v2", + ["ScenicSpot"], + { + "element_id": "amap:P1", + "gaode_poi_id": "P1", + "name": "小七孔", + "type_label": "景点", + "lng": 107.72, + "lat": 25.25, + "images": '[{"image_url":"https://example.com/a.jpg"}]', + "spatial_raw_data": '{"visitor_value":"世界遗产"}', + }, + ) + + self.assertEqual(payload["visitor_value"], "世界遗产") + self.assertEqual(payload["photo_urls"], ["https://example.com/a.jpg"]) + self.assertEqual(payload["graph_labels"], ["ScenicSpot"]) + + def test_normalized_bus_rows_build_ordered_route_payload(self) -> None: + rows = [ + ["r1", "荔波1路", "上行", "甲", "乙", 2, 2, "07:00", "18:00", 5, "station_sequence", "s1", "甲", 107.8, 25.3, 1], + ["r1", "荔波1路", "上行", "甲", "乙", 2, 2, "07:00", "18:00", 5, "station_sequence", "s2", "乙", 107.9, 25.4, 2], + ] + + payload = _build_libo_bus_route_payload(rows, "yunyou_libo_full_v2") + + self.assertEqual(payload["line_count"], 1) + self.assertEqual(payload["direction_count"], 1) + self.assertEqual(payload["stop_count"], 2) + self.assertEqual(payload["route_stop_count"], 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_project_lifecycle.py b/tests/test_project_lifecycle.py index 7561c74..52e2a75 100644 --- a/tests/test_project_lifecycle.py +++ b/tests/test_project_lifecycle.py @@ -85,6 +85,30 @@ class ProjectPayloadValidationTests(unittest.TestCase): self.assertEqual(payload["tenant_id"], "lifecycle_test") self.assertEqual(payload["graph_name"], "lifecycle_test") + def test_spatial_map_defaults_to_disabled_libo_scope(self) -> None: + payload = normalize_provision_payload(valid_payload()) + + self.assertEqual(payload["spatial_map"], {"enabled": False, "scope": "libo"}) + + def test_spatial_map_option_is_normalized(self) -> None: + body = valid_payload() + body["spatial_map"] = {"enabled": True, "scope": "guizhou"} + + payload = normalize_provision_payload(body) + + self.assertEqual(payload["spatial_map"], {"enabled": True, "scope": "guizhou"}) + + def test_invalid_spatial_map_is_rejected(self) -> None: + body = valid_payload() + body["spatial_map"] = {"enabled": "yes", "scope": "unknown"} + + with self.assertRaises(ProjectValidationError) as raised: + normalize_provision_payload(body) + + joined = "；".join(raised.exception.errors) + self.assertIn("spatial_map.enabled", joined) + self.assertIn("spatial_map.scope", joined) + def test_blank_resource_ids_also_default_to_project_id(self) -> None: body = valid_payload() body["tenant_id"] = " " @@ -296,10 +320,19 @@ class _Result: class _FakeGraph: - def __init__(self, *, fail_on_query: int | None = None, delete_fails: bool = False): + def __init__( + self, + *, + fail_on_query: int | None = None, + delete_fails: bool = False, + node_count: int = 2, + relation_count: int = 1, + ): self.queries: list[tuple[str, dict | None]] = [] self.fail_on_query = fail_on_query self.delete_fails = delete_fails + self.node_count = node_count + self.relation_count = relation_count self.deleted = False def query(self, query: str, params: dict | None = None): @@ -307,9 +340,9 @@ class _FakeGraph: if self.fail_on_query and len(self.queries) == self.fail_on_query: raise RuntimeError("write failed") if "MATCH (n) RETURN count(n)" in query: - return _Result(2) + return _Result(self.node_count) if "MATCH ()-[r]->() RETURN count(r)" in query: - return _Result(1) + return _Result(self.relation_count) return _Result(0) def delete(self): @@ -384,6 +417,33 @@ class FalkorLifecycleTests(unittest.TestCase): self.assertFalse(graph.queries) self.assertFalse(graph.deleted) + def test_large_import_uses_unwind_batches_and_endpoint_indexes(self) -> None: + graph_data = { + "nodes": [ + {"id": f"p{index}", "type": "Place", "properties": {"name": f"地点{index}"}} + for index in range(500) + ], + "relations": [ + { + "type": "RELATED_TO", + "source": "p0", + "target": "p1", + "properties": {"weight": 1}, + } + ], + } + graph = _FakeGraph(node_count=500, relation_count=1) + database = _FakeDb(graph) + + with patch("app.project_lifecycle._falkor_client", return_value=database): + counts = import_falkor_graph("large_graph", graph_data) + + self.assertEqual(counts, {"nodes": 500, "relations": 1}) + write_queries = [query for query, _ in graph.queries] + self.assertTrue(any("UNWIND $rows AS row CREATE (n:Place)" in query for query in write_queries)) + self.assertTrue(any("CREATE INDEX FOR (n:Place)" in query for query in write_queries)) + self.assertTrue(any("MATCH (source:Place" in query for query in write_queries)) + class LifecycleWriteGuardTests(unittest.IsolatedAsyncioTestCase): async def test_blocking_graph_worker_is_drained_before_cancellation_returns(self) -> None: