#!/usr/bin/env python3 """Export the authoritative province POI store as an importable map project. The legacy city project keeps its semantic graph in ``guiyang_new2`` and its map businesses in PostgreSQL ``amap_spatial_pois``. This exporter snapshots the latter exactly, so an imported project preserves the 80,609-POI business count and can use the same knowledge-map template without a hidden graph-name redirect. """ from __future__ import annotations import argparse import json import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any import psycopg from psycopg.rows import dict_row ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.config import settings from app.project_lifecycle import normalize_provision_payload from scripts.export_yunyou_libo_full_graph_bundle import ( SCHEMA_VERSION, build_schema, dump_json, exported_counts, jsonable, safe_prefix, sha256_file, ) SOURCE_GRAPH_NAME = "guiyang_spatial_v1" DEFAULT_PROJECT_ID = "city_map_export_v3" DEFAULT_DISPLAY_NAME = "城市图谱" # ``raw_jsonb`` and ``photo_urls`` duplicate data that has already been # normalized into the columns below. Keeping them is useful for an immutable # archive, but makes an 80,609-POI browser import several hundred megabytes. # The portable profile keeps every POI while retaining all fields used by the # shared map template, search and detail drawer. PORTABLE_PROPERTY_KEYS = { "element_id", "gaode_poi_id", "name", "type_label", "place_type", "amap_type", "typecode", "lng", "lat", "province", "city", "district", "adcode", "business_area", "address", "tel", "open_time", "rating", "cost", "level", "tags", "source", "towncode", "town_name", } CATEGORY_LABELS = { "景点": "ScenicSpot", "美食": "FoodPlace", "酒店": "Hotel", "商场": "Mall", "医疗保健": "MedicalPlace", "交通设施": "TransitFacility", "生活服务": "LifeServicePlace", "科教文化": "EducationPlace", "政府机构": "GovernmentPlace", "公共设施": "Facility", "体育休闲": "RecreationPlace", "商务住宅": "ResidentialPlace", "公司企业": "EnterprisePlace", "金融保险": "FinancePlace", "汽车服务": "AutoServicePlace", "汽车维修": "AutoRepairPlace", "汽车销售": "AutoSalesPlace", "摩托车服务": "MotorcycleServicePlace", "地名地址": "NamedPlace", "道路附属": "RoadFacility", } def read_nodes( source_graph_name: str, *, profile: str, ) -> tuple[list[dict[str, Any]], Counter[str]]: nodes: list[dict[str, Any]] = [] categories: Counter[str] = Counter() with psycopg.connect(settings.database_url, row_factory=dict_row) as conn: with conn.cursor(name="city_map_export") as cur: cur.execute( f"""SELECT * FROM {settings.db_schema}.amap_spatial_pois WHERE graph_name=%s ORDER BY element_id""", (source_graph_name,), ) for row in cur: element_id = str(row["element_id"]) category = str(row.get("type_label") or "其他地点") business_label = CATEGORY_LABELS.get(category, "BusinessPlace") properties = { str(key): jsonable(value) for key, value in row.items() if key != "graph_name" and (profile == "full" or key in PORTABLE_PROPERTY_KEYS) and ( profile == "full" or value is not None and value != "" and value != [] and value != {} ) } if profile == "full": properties["source_graph_name"] = source_graph_name nodes.append( { "id": element_id, "type": business_label, "labels": ["Place", business_label], "properties": properties, } ) categories[category] += 1 return nodes, categories def export_city_map( output_dir: Path, *, project_id: str, display_name: str, source_graph_name: str = SOURCE_GRAPH_NAME, profile: str = "full", ) -> dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) generated_at = datetime.now(timezone.utc).isoformat() nodes, category_counts = read_nodes(source_graph_name, profile=profile) if len(nodes) != len({item["id"] for item in nodes}): raise RuntimeError("省域 POI 数据存在重复 element_id") relations: list[dict[str, Any]] = [] graph_name = project_id spatial_map = { "enabled": True, "scope": "guizhou", "region_name": "贵阳市", "region_adcode": "520100", "region_level": "city", } counts = exported_counts(nodes, relations) schema = build_schema( nodes, relations, project_id=project_id, graph_name=graph_name, display_name=display_name, ) source_counts = { "nodes": len(nodes), "relations": 0, "coordinate_nodes": sum( 1 for item in nodes if item["properties"].get("lng") is not None and item["properties"].get("lat") is not None ), "category_counts": dict(sorted(category_counts.items())), } graph_data = { "_bundle": { "format": "znkg-city-map-snapshot-v3", "source": "postgresql.amap_spatial_pois", "source_graph_name": source_graph_name, "profile": profile, "project_id": project_id, "graph_name": graph_name, "generated_at": generated_at, "spatial_map": spatial_map, "source_snapshot_counts": source_counts, }, "nodes": nodes, "relations": relations, } bundle = { "format": "znkg-project-bundle-v3", "project_id": project_id, "display_name": display_name, "spatial_map": spatial_map, "schema": schema, "graph_data": graph_data, } normalized = normalize_provision_payload(bundle) if ( normalized["counts"]["nodes"] != len(nodes) or normalized["counts"]["relations"] != 0 ): raise RuntimeError("后端校验后的省域 POI 数量不一致") prefix = safe_prefix(project_id) suffix = "full" if profile == "full" else "portable" schema_path = output_dir / f"{prefix}_{suffix}_schema.v3.json" graph_path = output_dir / f"{prefix}_{suffix}_graph_data.v3.json" bundle_path = output_dir / f"{prefix}_{suffix}_bundle.v3.json" manifest_path = output_dir / f"{prefix}_{suffix}_manifest.v3.json" dump_json(schema_path, schema, pretty=True) dump_json(graph_path, graph_data) dump_json(bundle_path, bundle) files = { path.name: {"bytes": path.stat().st_size, "sha256": sha256_file(path)} for path in (schema_path, graph_path, bundle_path) } manifest = { "format": "znkg-city-map-manifest-v3", "project_id": project_id, "graph_name": graph_name, "source_graph_name": source_graph_name, "profile": profile, "schema_version": SCHEMA_VERSION, "generated_at": generated_at, "spatial_map": spatial_map, "source_snapshot_counts": source_counts, "export_counts": counts, "validation": "passed", "validation_checks": { "node_count_matches_postgresql": counts["nodes"] == source_counts["nodes"], "coordinate_count_matches_postgresql": ( counts["coordinate_nodes"] == source_counts["coordinate_nodes"] ), "node_ids_are_unique": True, "backend_bundle_validation_passed": True, }, "files": files, } dump_json(manifest_path, manifest, pretty=True) manifest["manifest_file"] = str(manifest_path) return manifest def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--project-id", default=DEFAULT_PROJECT_ID) parser.add_argument("--display-name", default=DEFAULT_DISPLAY_NAME) parser.add_argument("--source-graph-name", default=SOURCE_GRAPH_NAME) parser.add_argument( "--profile", choices=("full", "portable"), default="full", help=( "full keeps every source column; portable keeps every POI but " "omits duplicated raw payloads for browser import." ), ) args = parser.parse_args() manifest = export_city_map( args.output_dir.expanduser().resolve(), project_id=args.project_id, display_name=args.display_name, source_graph_name=args.source_graph_name, profile=args.profile, ) print(json.dumps(manifest, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()