#!/usr/bin/env python3 """Publish the latest Yunyou Libo food/hotel fusion CSVs to FalkorDB and PG schema. Safety and idempotency: - hard-scoped to tenant/project/graph ``yunyou_libo``; - defaults to dry-run; graph and PostgreSQL writes require ``--apply``; - exports the current FoodPlace/Hotel graph slice and PG schema/release rows; - MERGE/upsert by AMap element_id, so reruns do not duplicate entities; - only clears properties managed by this importer before refreshing them; - never deletes unrelated scenic, transit, bus, area, or grid data. """ from __future__ import annotations import argparse import csv import hashlib import json import os import re import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib.parse import urlsplit, urlunsplit import h3 import psycopg from falkordb import FalkorDB from psycopg.rows import dict_row from psycopg.types.json import Jsonb ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.config import settings # noqa: E402 TENANT_ID = "yunyou_libo" PROJECT_ID = "yunyou_libo" GRAPH_NAME = "yunyou_libo" SCHEMA_NAMESPACE = "yunyou_libo" SCHEMA_VERSION = 1 SCHEMA_DISPLAY_NAME = "云游荔波旅游POI融合知识图谱 Schema v1" SCHEMA_DSL_PATH = ROOT / "schema搭建/yunyou_libo/yunyou_libo_schema.current.dsl.md" FOOD_CSV = Path( "/Users/xuexue/Desktop/荔波小七孔源数据/高德/" "云游荔波_高德POI_美食_细网格.csv" ) HOTEL_CSV = Path( "/Users/xuexue/Desktop/荔波小七孔源数据/高德/" "云游荔波_高德POI_酒店_细网格.csv" ) OUTPUT_ROOT = Path( "/Users/xuexue/Documents/云游荔波/outputs/" "knowledge_graph_publish_20260729" ) COMMON_FIELD_MAP: dict[str, tuple[str, str]] = { "高德POI_ID": ("gaode_poi_id", "text"), "POI名称": ("name", "text"), "数据分类": ("type_label", "text"), "Schema类型": ("schema_type", "text"), "业务细分类": ("business_subcategory", "text"), "高德一级分类": ("amap_category_l1", "text"), "高德二级分类": ("amap_category_l2", "text"), "高德三级分类": ("amap_category_l3", "text"), "高德完整分类": ("amap_type", "text"), "高德分类编码": ("typecode", "text"), "省": ("province", "text"), "地级市/自治州": ("city", "text"), "区县": ("district", "text"), "行政区划代码": ("adcode", "text"), "详细地址": ("address", "text"), "商圈": ("business_area", "text"), "经度(GCJ-02)": ("lng", "float"), "纬度(GCJ-02)": ("lat", "float"), "电话": ("tel", "text"), "评分": ("amap_rating", "float"), "人均消费/参考价格": ("amap_cost_text", "text"), "营业时间": ("open_time", "text"), "图片数量": ("amap_image_count", "int"), "首图URL": ("cover_image_url", "text"), "全部图片URL": ("photo_urls", "text"), "高德地图URL": ("amap_url", "text"), "去重键": ("dedup_key", "text"), "采集网格序号": ("collection_grid_no", "text"), "网格中心经度": ("grid_center_lng", "float"), "网格中心纬度": ("grid_center_lat", "float"), "扫描半径(m)": ("scan_radius_m", "float"), "采集时间": ("collected_at", "text"), "数据来源": ("source_name", "text"), "数据质量标记": ("data_quality", "text"), "命中扫描类型": ("scan_hit_types", "text"), "扫描命中次数": ("scan_hit_count", "int"), "采集轮次": ("collection_rounds", "text"), } FOOD_FIELD_MAP: dict[str, tuple[str, str]] = { "融合状态": ("food_fusion_status", "text"), "匹配置信度": ("food_match_confidence", "text"), "匹配规则": ("food_match_rule", "text"), "匹配综合分": ("food_match_score", "float"), "名称完全一致": ("food_name_exact_match", "bool"), "名称相似度": ("food_name_similarity", "float"), "地址相似度": ("food_address_similarity", "float"), "坐标距离(m)": ("food_coordinate_distance_m", "float"), "融合来源说明": ("food_fusion_source_note", "text"), "大众点评源POI_ID": ("dianping_source_poi_id", "text"), "大众点评店铺ID": ("dianping_shop_id", "text"), "大众点评店铺UUID": ("dianping_shop_uuid", "text"), "大众点评店铺名称": ("dianping_name", "text"), "大众点评地址": ("dianping_address", "text"), "大众点评经度": ("dianping_lng", "float"), "大众点评纬度": ("dianping_lat", "float"), "大众点评URL": ("dianping_url", "text"), "大众点评评分": ("dianping_rating", "float"), "大众点评评论数": ("dianping_review_count_text", "text"), "大众点评评论数(数值)": ("dianping_review_count", "int"), "大众点评人均消费": ("dianping_avg_price_text", "text"), "大众点评人均消费(元)": ("dianping_avg_price", "float"), "大众点评区域": ("dianping_area", "text"), "大众点评分类": ("dianping_category", "text"), "大众点评详细评分": ("dianping_score_details", "text"), "大众点评排名": ("dianping_ranking", "text"), "大众点评营业状态": ("dianping_business_status", "text"), "大众点评营业时间": ("dianping_business_hours", "text"), "大众点评特色标签": ("dianping_tags", "text"), "大众点评交通信息": ("dianping_transportation", "text"), "大众点评图片数量": ("dianping_image_count", "int"), "大众点评店铺图片": ("dianping_shop_image", "text"), "大众点评评论标签": ("dianping_review_tags", "text"), "大众点评页面评论数量": ("dianping_page_review_count", "int"), "大众点评更新时间": ("dianping_updated_at", "text"), "推荐菜采集条数": ("recommended_dish_count", "int"), "推荐菜名称摘要": ("recommended_dish_names", "text"), "推荐菜最高推荐人数": ("recommended_dish_max_recommendations", "int"), "推荐菜图片URL样例": ("recommended_dish_image_samples", "text"), "团购采集条数": ("group_buy_count", "int"), "团购最低价": ("group_buy_min_price", "float"), "团购最高价": ("group_buy_max_price", "float"), "团购标题摘要": ("group_buy_titles", "text"), "团购包含菜品摘要": ("group_buy_items", "text"), "评论采集条数": ("review_sample_count", "int"), "评论样本平均星级": ("review_sample_avg_stars", "float"), "评论时间摘要": ("review_sample_times", "text"), "评论内容摘要": ("review_sample_summary", "text"), "评论图片URL样例": ("review_image_samples", "text"), "菜单图片采集条数": ("menu_image_count", "int"), "菜单图片URL样例": ("menu_image_samples", "text"), "候选点评源POI_ID": ("candidate_dianping_source_poi_id", "text"), "候选点评店铺ID": ("candidate_dianping_shop_id", "text"), "候选点评店铺名": ("candidate_dianping_name", "text"), "候选名称相似度": ("food_candidate_name_similarity", "float"), "候选地址相似度": ("food_candidate_address_similarity", "float"), "候选坐标距离(m)": ("food_candidate_coordinate_distance_m", "float"), } HOTEL_FIELD_MAP: dict[str, tuple[str, str]] = { "融合状态": ("hotel_fusion_status", "text"), "匹配置信度": ("hotel_match_confidence", "text"), "匹配规则": ("hotel_match_rule", "text"), "名称相似度": ("hotel_name_similarity", "float"), "地址相似度": ("hotel_address_similarity", "float"), "电话一致": ("hotel_phone_exact_match", "bool"), "融合来源说明": ("hotel_fusion_source_note", "text"), "携程酒店ID": ("ctrip_hotel_id", "text"), "携程酒店中文名": ("ctrip_name_cn", "text"), "携程酒店英文名": ("ctrip_name_en", "text"), "携程地址": ("ctrip_address", "text"), "携程电话": ("ctrip_phone", "text"), "携程星级": ("ctrip_star_level", "text"), "携程钻级": ("ctrip_diamond_level", "text"), "携程开业时间": ("ctrip_opened_year", "text"), "携程客房数": ("ctrip_room_count", "int"), "携程酒店类型": ("ctrip_hotel_type", "text"), "携程榜单/排名": ("ctrip_ranking", "text"), "携程评分": ("ctrip_rating", "float"), "携程评分描述": ("ctrip_rating_description", "text"), "携程点评总数": ("ctrip_review_count", "int"), "携程图片总数": ("ctrip_image_count", "int"), "携程交通信息": ("ctrip_transportation", "text"), "携程特色标签": ("ctrip_tags", "text"), "携程热门设施": ("ctrip_popular_facilities", "text"), "携程评分明细": ("ctrip_score_details", "text"), "携程卫生评分": ("ctrip_cleanliness_score", "float"), "携程设施评分": ("ctrip_facilities_score", "float"), "携程环境评分": ("ctrip_environment_score", "float"), "携程服务评分": ("ctrip_service_score", "float"), "携程酒店简介": ("ctrip_description", "text"), "携程详情URL": ("ctrip_url", "text"), "携程房型数": ("room_type_count", "int"), "房型名称摘要": ("room_type_names", "text"), "房型最低价": ("room_min_price", "float"), "房型最高价": ("room_max_price", "float"), "售卖方案数": ("offer_count", "int"), "方案最低现价": ("offer_min_price", "float"), "方案最高现价": ("offer_max_price", "float"), "设施服务条数": ("facility_service_count", "int"), "设施分类数": ("facility_category_count", "int"), "设施分类统计": ("facility_category_stats", "text"), "设施摘要": ("facility_summary", "text"), "设施分类明细JSON": ("facility_category_items_json", "text"), "酒店政策条数": ("policy_count", "int"), "政策分类统计": ("policy_category_stats", "text"), "政策摘要": ("policy_summary", "text"), "住客点评样本数": ("guest_review_sample_count", "int"), "样本平均评分": ("guest_review_sample_avg_score", "float"), "最新点评时间": ("latest_review_at", "text"), "点评出行类型": ("review_travel_types", "text"), "酒店图片条数": ("hotel_image_record_count", "int"), "图片类型统计": ("hotel_image_type_stats", "text"), "图片URL样例": ("hotel_image_samples", "text"), "周边地点条数": ("nearby_place_count", "int"), "周边地点类型统计": ("nearby_place_type_stats", "text"), "周边地点摘要": ("nearby_place_summary", "text"), "周边分类明细JSON": ("nearby_category_items_json", "text"), "候选携程酒店ID": ("candidate_ctrip_hotel_id", "text"), "候选携程酒店名": ("candidate_ctrip_name", "text"), "候选名称相似度": ("hotel_candidate_name_similarity", "float"), "候选地址相似度": ("hotel_candidate_address_similarity", "float"), } SYSTEM_FIELDS: dict[str, str] = { "element_id": "text", "place_id": "text", "place_type": "text", "source": "text", "h3_r6": "text", "h3_r7": "text", "h3_r8": "text", "h3_r9": "text", "h3_r10": "text", "first_seen_at": "datetime", "updated_at": "datetime", "data_version": "text", "enrichment_source": "text", "enrichment_updated_at": "datetime", "fusion_status": "text", } LEGACY_SHARED_FUSION_FIELDS = { "match_confidence", "match_rule", "match_score", "name_exact_match", "name_similarity", "address_similarity", "coordinate_distance_m", "phone_exact_match", "fusion_source_note", "candidate_name_similarity", "candidate_address_similarity", "candidate_coordinate_distance_m", } def now_iso() -> str: return datetime.now(timezone.utc).isoformat() def clean(value: Any) -> str: return "" if value is None else str(value).strip() def parse_number(value: Any, integer: bool = False) -> int | float | None: match = re.search(r"-?\d+(?:\.\d+)?", clean(value).replace(",", "")) if not match: return None number = float(match.group(0)) return int(round(number)) if integer else number def parse_bool(value: Any) -> bool | None: normalized = clean(value).lower() if normalized in {"是", "true", "1", "yes", "一致"}: return True if normalized in {"否", "false", "0", "no", "不一致"}: return False return None def coerce(value: Any, kind: str) -> str | int | float | bool | None: if kind == "text": normalized = clean(value) return normalized or None if kind == "int": return parse_number(value, integer=True) if kind == "float": return parse_number(value) if kind == "bool": return parse_bool(value) raise ValueError(f"Unsupported kind: {kind}") def normalize_ctrip_diamond_level(value: Any) -> str | None: normalized = clean(value) if not normalized: return None if re.fullmatch(r"\d+(?:\.\d+)?钻", normalized): return normalized matched = re.fullmatch( r"(\d+(?:\.\d+)?)\s+out\s+of\s+5\s+(?:rating|diamonds?)", normalized, flags=re.IGNORECASE, ) return f"{matched.group(1)}钻" if matched else None def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def load_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: with path.open("r", encoding="utf-8-sig", newline="") as stream: reader = csv.DictReader(stream) rows = [dict(row) for row in reader] return list(reader.fieldnames or []), rows def h3_props(lat: float, lng: float) -> dict[str, str]: return {f"h3_r{resolution}": h3.latlng_to_cell(lat, lng, resolution) for resolution in range(6, 11)} def row_props( row: dict[str, str], field_map: dict[str, tuple[str, str]], *, place_type: str, data_version: str, run_at: str, ) -> dict[str, Any]: props: dict[str, Any] = {} for source_field, (target_field, kind) in {**COMMON_FIELD_MAP, **field_map}.items(): value = coerce(row.get(source_field), kind) if target_field == "ctrip_diamond_level": value = normalize_ctrip_diamond_level(value) if value is not None: props[target_field] = value poi_id = clean(row.get("高德POI_ID")) element_id = clean(row.get("去重键")) or f"amap:{poi_id}" props.update( { "element_id": element_id, "place_id": element_id, "place_type": place_type, "source": "amap", "source_cell_id": ( f"amap-js-grid:{clean(row.get('采集网格序号'))}" if clean(row.get("采集网格序号")) else "" ), "source_resolution": 0, "data_version": data_version, "enrichment_source": ( "amap+dianping" if place_type == "eat" else "amap+ctrip" ), "enrichment_updated_at": run_at, "updated_at": run_at, } ) lng = parse_number(row.get("经度(GCJ-02)")) lat = parse_number(row.get("纬度(GCJ-02)")) if lng is None or lat is None: raise ValueError(f"{poi_id} missing coordinates") props.update(h3_props(lat, lng)) return {key: value for key, value in props.items() if value not in (None, "")} def validate_rows( *, kind: str, headers: list[str], rows: list[dict[str, str]], expected_schema_type: str, expected_min_rows: int, ) -> dict[str, Any]: required = { "高德POI_ID", "POI名称", "Schema类型", "区县", "行政区划代码", "经度(GCJ-02)", "纬度(GCJ-02)", "融合状态", } missing = sorted(required - set(headers)) if missing: raise ValueError(f"{kind} missing columns: {missing}") if len(rows) < expected_min_rows: raise ValueError(f"{kind} row count too small: {len(rows)}") ids = [clean(row.get("高德POI_ID")) for row in rows] if any(not poi_id for poi_id in ids): raise ValueError(f"{kind} contains empty AMap POI ID") duplicate_ids = [poi_id for poi_id, count in Counter(ids).items() if count > 1] if duplicate_ids: raise ValueError(f"{kind} duplicate AMap POI IDs: {duplicate_ids[:10]}") foreign_rows = [ row for row in rows if clean(row.get("行政区划代码")) != "522722" or clean(row.get("区县")) != "荔波县" ] if foreign_rows: raise ValueError(f"{kind} contains {len(foreign_rows)} non-Libo rows") wrong_schema = [ row for row in rows if clean(row.get("Schema类型")) != expected_schema_type ] if wrong_schema: raise ValueError(f"{kind} contains {len(wrong_schema)} unexpected schema types") return { "rows": len(rows), "columns": len(headers), "unique_amap_ids": len(set(ids)), "fusion_status": dict(Counter(clean(row.get("融合状态")) for row in rows)), "match_confidence": dict(Counter(clean(row.get("匹配置信度")) for row in rows)), } def schema_fields(field_map: dict[str, tuple[str, str]]) -> tuple[list[str], dict[str, str]]: field_types = { **SYSTEM_FIELDS, **{target: kind for target, kind in COMMON_FIELD_MAP.values()}, **{target: kind for target, kind in field_map.values()}, } fields = list(dict.fromkeys(field_types)) return fields, field_types def build_schema_payload() -> dict[str, Any]: food_fields, food_types = schema_fields(FOOD_FIELD_MAP) hotel_fields, hotel_types = schema_fields(HOTEL_FIELD_MAP) return { "namespace": SCHEMA_NAMESPACE, "version": str(SCHEMA_VERSION), "display_name": SCHEMA_DISPLAY_NAME, "purpose": ( "支持荔波县地图展示、餐饮与住宿检索、游客决策和多来源证据追溯；" "高德POI作为空间锚点，大众点评与携程详情只在同实体确认后融合。" ), "entity_types": { "Place": { "cn": "空间地点", "primary_key": "element_id", "fields": [ "element_id", "place_id", "name", "place_type", "type_label", "business_subcategory", "amap_type", "typecode", "address", "district", "adcode", "lng", "lat", "h3_r9", "source", ], }, "FoodPlace": { "cn": "美食店铺", "primary_key": "element_id", "description": "高德POI锚点及经审核融合的大众点评店铺详情。", "fields": food_fields, "field_types": food_types, }, "Hotel": { "cn": "酒店住宿", "primary_key": "element_id", "description": "高德POI锚点及经审核融合的携程酒店详情。", "fields": hotel_fields, "field_types": hotel_types, }, "ScenicSpot": { "cn": "景点", "primary_key": "element_id", "fields": [ "element_id", "name", "scenic_type", "scenic_level", "parent_scenic", "visitor_value", "audit_result", "lng", "lat", ], }, "ScenicArea": { "cn": "景区", "primary_key": "element_id", "fields": ["element_id", "name", "scenic_grade", "lng", "lat"], }, "TransitFacility": { "cn": "交通设施", "primary_key": "element_id", "fields": ["element_id", "name", "business_subcategory", "address", "lng", "lat"], }, "BusStop": { "cn": "公交站", "primary_key": "element_id", "fields": ["element_id", "name", "lng", "lat"], }, "BusLine": { "cn": "公交线路", "primary_key": "element_id", "fields": ["element_id", "name", "color"], }, "BusRoute": { "cn": "公交行驶方向", "primary_key": "element_id", "fields": ["element_id", "name", "direction", "polyline"], }, "Area": { "cn": "行政区域", "primary_key": "element_id", "fields": ["element_id", "name", "level", "adcode"], }, "GeoCell": { "cn": "片区", "primary_key": "h3_id", "fields": ["h3_id", "resolution"], }, }, "relation_types": { "LOCATED_IN": { "from": "Place|BusStop|GeoCell", "to": "Area", "description": "地点或片区位于荔波县行政区域。", }, "IN_H3_R9": { "from": "Place|BusStop", "to": "GeoCell", "description": "地点落入H3九级网格。", }, "PART_OF_SCENIC_AREA": { "from": "ScenicSpot", "to": "ScenicArea", "description": "景区内部子景点归属。", }, "PART_OF": { "from": "Area", "to": "Area", "description": "行政区层级关系。", }, "STOPS_AT": { "from": "BusRoute", "to": "BusStop", "description": "公交线路方向停靠站点。", }, "NEXT_STOP": { "from": "BusStop", "to": "BusStop", "description": "同一公交方向的站点顺序。", }, }, "fusion_policy": { "anchor": "AMap POI", "food_detail_source": "Dianping public shop detail", "hotel_detail_source": "Ctrip public hotel detail", "confirmed_rows_write_detail": True, "candidate_rows_only_write_candidate_fields": True, "source_tables_immutable": True, }, "data_versions": { "food": "2026-07-28-food-dianping-fusion", "hotel": "2026-07-28-hotel-ctrip-fusion", }, } def replace_port(database_url: str, port: int) -> str: parts = urlsplit(database_url) host = parts.hostname or "localhost" userinfo = "" if parts.username: userinfo = parts.username if parts.password: userinfo += f":{parts.password}" userinfo += "@" netloc = f"{userinfo}{host}:{port}" return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment)) def pg_connect() -> psycopg.Connection: candidates = [] explicit = os.environ.get("YUNYOU_DATABASE_URL", "").strip() if explicit: candidates.append(explicit) candidates.append(settings.database_url) candidates.append(replace_port(settings.database_url, 16433)) errors = [] for url in dict.fromkeys(candidates): try: return psycopg.connect(url, connect_timeout=3, row_factory=dict_row) except psycopg.OperationalError as exc: errors.append(str(exc).splitlines()[0]) raise RuntimeError(f"PostgreSQL unavailable: {errors}") def graph_client(): db = FalkorDB( host=settings.falkordb_host, port=settings.falkordb_port, password=settings.falkordb_password or None, ) return db.select_graph(GRAPH_NAME) def graph_scalar(graph, query: str, params: dict[str, Any] | None = None) -> int: rows = graph.query(query, params or {}).result_set return int(rows[0][0]) if rows else 0 def graph_snapshot(graph) -> dict[str, Any]: nodes = {} for label in ("FoodPlace", "Hotel"): rows = graph.query(f"MATCH (n:{label}) RETURN properties(n)").result_set nodes[label] = [dict(row[0]) for row in rows] outgoing = [ { "source_element_id": row[0], "relation_type": row[1], "relation_properties": dict(row[2]), "target_labels": list(row[3]), "target_properties": dict(row[4]), } for row in graph.query( "MATCH (n)-[r]->(m) " "WHERE n:FoodPlace OR n:Hotel " "RETURN n.element_id, type(r), properties(r), labels(m), properties(m)" ).result_set ] incoming = [ { "source_labels": list(row[0]), "source_properties": dict(row[1]), "relation_type": row[2], "relation_properties": dict(row[3]), "target_element_id": row[4], } for row in graph.query( "MATCH (m)-[r]->(n) " "WHERE n:FoodPlace OR n:Hotel " "RETURN labels(m), properties(m), type(r), properties(r), n.element_id" ).result_set ] return {"nodes": nodes, "outgoing_relations": outgoing, "incoming_relations": incoming} def pg_snapshot(conn: psycopg.Connection) -> dict[str, Any]: schema = settings.db_schema with conn.cursor() as cur: cur.execute( f"SELECT * FROM {schema}.ontology_schemas " "WHERE tenant_id=%s AND project_id=%s ORDER BY version", (TENANT_ID, PROJECT_ID), ) schemas = [dict(row) for row in cur.fetchall()] cur.execute( f"SELECT * FROM {schema}.graph_releases " "WHERE tenant_id=%s AND project_id=%s ORDER BY updated_at DESC", (TENANT_ID, PROJECT_ID), ) releases = [dict(row) for row in cur.fetchall()] cur.execute( f"SELECT * FROM {schema}.projects " "WHERE tenant_id=%s AND project_id=%s", (TENANT_ID, PROJECT_ID), ) projects = [dict(row) for row in cur.fetchall()] return {"schemas": schemas, "graph_releases": releases, "projects": projects} def json_write(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8", ) def batched(values: list[Any], size: int = 100): for index in range(0, len(values), size): yield values[index:index + size] def clear_managed_properties(graph, label: str, fields: set[str]) -> None: if not fields: return remove_clause = ", ".join(f"n.{field}" for field in sorted(fields)) graph.query(f"MATCH (n:{label}) REMOVE {remove_clause}") def upsert_nodes(graph, label: str, rows: list[dict[str, Any]], run_at: str) -> None: query = ( "UNWIND $rows AS row " "MERGE (n:Place {element_id: row.element_id}) " "ON CREATE SET n.first_seen_at = $run_at " f"SET n:{label} " "SET n += row.props" ) for chunk in batched(rows): graph.query(query, {"rows": chunk, "run_at": run_at}) def upsert_spatial_relations(graph, rows: list[dict[str, Any]]) -> None: cell_rows = [ {"h3_id": row["props"]["h3_r9"], "element_id": row["element_id"]} for row in rows ] for chunk in batched(cell_rows): graph.query( "UNWIND $rows AS row " "MERGE (cell:GeoCell {h3_id: row.h3_id}) " "SET cell.resolution = 9", {"rows": chunk}, ) graph.query( "UNWIND $rows AS row " "MATCH (cell:GeoCell {h3_id: row.h3_id}) " "MATCH (area:Area {adcode: '522722'}) " "MERGE (cell)-[:LOCATED_IN]->(area)", {"rows": chunk}, ) graph.query( "UNWIND $rows AS row " "MATCH (n:Place {element_id: row.element_id}) " "MATCH (area:Area {adcode: '522722'}) " "MERGE (n)-[:LOCATED_IN]->(area)", {"rows": chunk}, ) graph.query( "UNWIND $rows AS row " "MATCH (n:Place {element_id: row.element_id}) " "MATCH (cell:GeoCell {h3_id: row.h3_id}) " "MERGE (n)-[:IN_H3_R9]->(cell)", {"rows": chunk}, ) def update_compatibility_fusion_status(graph) -> None: """Keep one readable legacy status while preserving per-business truth.""" graph.query( "MATCH (n:FoodPlace) WHERE NOT n:Hotel " "SET n.fusion_status = n.food_fusion_status" ) graph.query( "MATCH (n:Hotel) WHERE NOT n:FoodPlace " "SET n.fusion_status = n.hotel_fusion_status" ) graph.query( "MATCH (n:FoodPlace:Hotel) " "SET n.fusion_status = " "'美食:' + n.food_fusion_status + '|酒店:' + n.hotel_fusion_status" ) def publish_pg( conn: psycopg.Connection, *, schema_payload: dict[str, Any], food_hash: str, hotel_hash: str, food_rows: int, hotel_rows: int, report_summary: dict[str, Any], ) -> dict[str, Any]: schema = settings.db_schema with conn.cursor() as cur: cur.execute( f"UPDATE {schema}.ontology_schemas " "SET status='archived', updated_at=now() " "WHERE tenant_id=%s AND project_id=%s AND namespace=%s " "AND status='active' AND version<>%s", (TENANT_ID, PROJECT_ID, SCHEMA_NAMESPACE, SCHEMA_VERSION), ) cur.execute( f""" INSERT INTO {schema}.ontology_schemas ( tenant_id, project_id, namespace, version, display_name, description, status, schema_jsonb, created_by, published_by, published_at, updated_at ) VALUES (%s, %s, %s, %s, %s, %s, 'active', %s, 'codex', 'codex', now(), now()) ON CONFLICT (tenant_id, project_id, namespace, version) DO UPDATE SET display_name=EXCLUDED.display_name, description=EXCLUDED.description, status='active', schema_jsonb=EXCLUDED.schema_jsonb, published_by='codex', published_at=now(), updated_at=now() RETURNING id """, ( TENANT_ID, PROJECT_ID, SCHEMA_NAMESPACE, SCHEMA_VERSION, SCHEMA_DISPLAY_NAME, schema_payload["purpose"], Jsonb(schema_payload), ), ) schema_id = int(cur.fetchone()["id"]) dataset_version = "food-dianping+hotel-ctrip-2026-07-28" release_metadata = { "business": "yunyou_libo_tourism_poi", "schema_version": SCHEMA_VERSION, "food_rows": food_rows, "hotel_rows": hotel_rows, "food_sha256": food_hash, "hotel_sha256": hotel_hash, "published_at": now_iso(), "publish_summary": report_summary, } cur.execute( f""" UPDATE {schema}.graph_releases SET schema_id=%s, source_dataset_version=%s, metadata_jsonb=COALESCE(metadata_jsonb, '{{}}'::jsonb) || %s, updated_at=now() WHERE tenant_id=%s AND project_id=%s AND graph_name=%s AND status='active' RETURNING id """, ( schema_id, dataset_version, Jsonb(release_metadata), TENANT_ID, PROJECT_ID, GRAPH_NAME, ), ) release = cur.fetchone() if not release: raise RuntimeError("Active yunyou_libo graph release not found") cur.execute( f""" UPDATE {schema}.projects SET default_namespace=%s, metadata_jsonb=COALESCE(metadata_jsonb, '{{}}'::jsonb) || %s, updated_at=now() WHERE tenant_id=%s AND project_id=%s """, ( SCHEMA_NAMESPACE, Jsonb( { "business": "yunyou_libo_tourism_poi", "graph_name": GRAPH_NAME, "schema_version": SCHEMA_VERSION, "schema_file": str(SCHEMA_DSL_PATH), "latest_food_rows": food_rows, "latest_hotel_rows": hotel_rows, } ), TENANT_ID, PROJECT_ID, ), ) batches = [] for template_id, source_name, file_path, file_hash, row_count in ( ( "yunyou_libo_food_enriched_v1", "高德美食POI+大众点评详情融合", FOOD_CSV, food_hash, food_rows, ), ( "yunyou_libo_hotel_enriched_v1", "高德酒店POI+携程详情融合", HOTEL_CSV, hotel_hash, hotel_rows, ), ): cur.execute( f""" SELECT id FROM {schema}.import_batches WHERE tenant_id=%s AND project_id=%s AND template_id=%s AND file_hash=%s ORDER BY id DESC LIMIT 1 """, (TENANT_ID, PROJECT_ID, template_id, file_hash), ) existing = cur.fetchone() if existing: batch_id = int(existing["id"]) cur.execute( f""" UPDATE {schema}.import_batches SET status='published', total_rows=%s, success_rows=%s, failed_rows=0, updated_at=now() WHERE id=%s """, (row_count, row_count, batch_id), ) else: cur.execute( f""" INSERT INTO {schema}.import_batches ( tenant_id, project_id, graph_name, template_id, source_name, file_name, file_hash, status, total_rows, success_rows, failed_rows, created_by, updated_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, 'published', %s, %s, 0, 'codex', now()) RETURNING id """, ( TENANT_ID, PROJECT_ID, GRAPH_NAME, template_id, source_name, file_path.name, file_hash, row_count, row_count, ), ) batch_id = int(cur.fetchone()["id"]) batches.append(batch_id) conn.commit() return {"schema_id": schema_id, "graph_release_id": int(release["id"]), "batch_ids": batches} def graph_summary(graph) -> dict[str, Any]: return { "nodes": graph_scalar(graph, "MATCH (n) RETURN count(n)"), "relations": graph_scalar(graph, "MATCH ()-[r]->() RETURN count(r)"), "food": graph_scalar(graph, "MATCH (n:FoodPlace) RETURN count(n)"), "hotel": graph_scalar(graph, "MATCH (n:Hotel) RETURN count(n)"), "food_fused": graph_scalar( graph, "MATCH (n:FoodPlace {food_fusion_status:'已融合'}) RETURN count(n)", ), "food_pending": graph_scalar( graph, "MATCH (n:FoodPlace {food_fusion_status:'待人工确认'}) RETURN count(n)", ), "food_with_dianping": graph_scalar( graph, "MATCH (n:FoodPlace) WHERE n.dianping_shop_id IS NOT NULL RETURN count(n)", ), "hotel_fused": graph_scalar( graph, "MATCH (n:Hotel {hotel_fusion_status:'已融合'}) RETURN count(n)", ), "hotel_pending": graph_scalar( graph, "MATCH (n:Hotel {hotel_fusion_status:'待人工确认'}) RETURN count(n)", ), "hotel_with_ctrip": graph_scalar( graph, "MATCH (n:Hotel) WHERE n.ctrip_hotel_id IS NOT NULL RETURN count(n)", ), "food_located_in": graph_scalar( graph, "MATCH (:FoodPlace)-[:LOCATED_IN]->(:Area {adcode:'522722'}) RETURN count(*)", ), "hotel_located_in": graph_scalar( graph, "MATCH (:Hotel)-[:LOCATED_IN]->(:Area {adcode:'522722'}) RETURN count(*)", ), "food_h3": graph_scalar( graph, "MATCH (:FoodPlace)-[:IN_H3_R9]->(:GeoCell) RETURN count(*)", ), "hotel_h3": graph_scalar( graph, "MATCH (:Hotel)-[:IN_H3_R9]->(:GeoCell) RETURN count(*)", ), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument( "--apply", action="store_true", help="Perform scoped FalkorDB and PostgreSQL writes. Without this flag, dry-run only.", ) return parser.parse_args() def main() -> None: args = parse_args() run_at = now_iso() run_slug = datetime.now().strftime("%Y%m%d_%H%M%S") run_dir = OUTPUT_ROOT / run_slug run_dir.mkdir(parents=True, exist_ok=True) food_headers, food_csv_rows = load_csv(FOOD_CSV) hotel_headers, hotel_csv_rows = load_csv(HOTEL_CSV) food_validation = validate_rows( kind="food", headers=food_headers, rows=food_csv_rows, expected_schema_type="FoodPlace", expected_min_rows=1000, ) hotel_validation = validate_rows( kind="hotel", headers=hotel_headers, rows=hotel_csv_rows, expected_schema_type="Hotel", expected_min_rows=1200, ) food_hash = sha256(FOOD_CSV) hotel_hash = sha256(HOTEL_CSV) food_nodes = [ { "element_id": clean(row.get("去重键")) or f"amap:{clean(row.get('高德POI_ID'))}", "props": row_props( row, FOOD_FIELD_MAP, place_type="eat", data_version="2026-07-28-food-dianping-fusion", run_at=run_at, ), } for row in food_csv_rows ] hotel_nodes = [ { "element_id": clean(row.get("去重键")) or f"amap:{clean(row.get('高德POI_ID'))}", "props": row_props( row, HOTEL_FIELD_MAP, place_type="hotel", data_version="2026-07-28-hotel-ctrip-fusion", run_at=run_at, ), } for row in hotel_csv_rows ] graph = graph_client() before = graph_summary(graph) dry_run_report = { "mode": "apply" if args.apply else "dry-run", "run_at": run_at, "scope": { "tenant_id": TENANT_ID, "project_id": PROJECT_ID, "graph_name": GRAPH_NAME, }, "sources": { "food": {"path": str(FOOD_CSV), "sha256": food_hash, **food_validation}, "hotel": {"path": str(HOTEL_CSV), "sha256": hotel_hash, **hotel_validation}, }, "schema": { "namespace": SCHEMA_NAMESPACE, "version": SCHEMA_VERSION, "dsl_path": str(SCHEMA_DSL_PATH), "entity_types": list(build_schema_payload()["entity_types"]), "relation_types": list(build_schema_payload()["relation_types"]), }, "graph_before": before, } json_write(run_dir / "preflight.json", dry_run_report) if not args.apply: print(json.dumps(dry_run_report, ensure_ascii=False, indent=2)) print("DRY-RUN ONLY: rerun with --apply after reviewing preflight.json") return if not SCHEMA_DSL_PATH.exists(): raise FileNotFoundError(f"Schema DSL missing: {SCHEMA_DSL_PATH}") with pg_connect() as conn: json_write(run_dir / "postgres_before.json", pg_snapshot(conn)) json_write(run_dir / "graph_food_hotel_before.json", graph_snapshot(graph)) enrichment_system_fields = { "data_version", "enrichment_source", "enrichment_updated_at", "fusion_status", } clear_managed_properties( graph, "FoodPlace", enrichment_system_fields | LEGACY_SHARED_FUSION_FIELDS | { target for target, _kind in FOOD_FIELD_MAP.values() }, ) clear_managed_properties( graph, "Hotel", enrichment_system_fields | LEGACY_SHARED_FUSION_FIELDS | { target for target, _kind in HOTEL_FIELD_MAP.values() }, ) upsert_nodes(graph, "FoodPlace", food_nodes, run_at) upsert_nodes(graph, "Hotel", hotel_nodes, run_at) update_compatibility_fusion_status(graph) upsert_spatial_relations(graph, food_nodes) upsert_spatial_relations(graph, hotel_nodes) after = graph_summary(graph) expected = { "food": len(food_nodes), "hotel": len(hotel_nodes), "food_fused": food_validation["fusion_status"].get("已融合", 0), "food_pending": food_validation["fusion_status"].get("待人工确认", 0), "food_with_dianping": food_validation["fusion_status"].get("已融合", 0), "hotel_fused": hotel_validation["fusion_status"].get("已融合", 0), "hotel_pending": hotel_validation["fusion_status"].get("待人工确认", 0), "hotel_with_ctrip": hotel_validation["fusion_status"].get("已融合", 0), "food_located_in": len(food_nodes), "hotel_located_in": len(hotel_nodes), "food_h3": len(food_nodes), "hotel_h3": len(hotel_nodes), } mismatches = { key: {"expected": value, "actual": after.get(key)} for key, value in expected.items() if after.get(key) != value } if mismatches: raise RuntimeError(f"Graph verification failed: {mismatches}") pg_result = publish_pg( conn, schema_payload=build_schema_payload(), food_hash=food_hash, hotel_hash=hotel_hash, food_rows=len(food_nodes), hotel_rows=len(hotel_nodes), report_summary=after, ) postgres_after = pg_snapshot(conn) report = { **dry_run_report, "mode": "applied", "graph_after": after, "graph_expected": expected, "verification_mismatches": {}, "postgres_result": pg_result, "postgres_after": postgres_after, "backup_files": { "graph": str(run_dir / "graph_food_hotel_before.json"), "postgres": str(run_dir / "postgres_before.json"), }, } json_write(run_dir / "publish_report.json", report) json_write(OUTPUT_ROOT / "latest_publish_report.json", report) print(json.dumps(report, ensure_ascii=False, indent=2, default=str)) if __name__ == "__main__": main()