lyf
c146beba9e
- 刷新讲解静态数据包并保留完整 outline 树 - 按后台顶层业务单元口径归并讲解点统计 - 更新讲解静态数据刷新操作文档 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
562 lines
19 KiB
Python
562 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Export the museum guide/explain static data package from Dameng DM8.
|
||
|
||
The script reads the backend datasource config, connects read-only through ODBC,
|
||
and writes JSON files consumed by frontend-miniapp. Passwords are only kept in
|
||
memory and are never printed.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import decimal
|
||
import json
|
||
import math
|
||
import re
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pyodbc
|
||
import yaml
|
||
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
DEFAULT_CONFIG = (
|
||
PROJECT_ROOT.parent.parent.parent
|
||
/ "智慧导览"
|
||
/ "smart-navigation-system"
|
||
/ "yudao-server"
|
||
/ "src"
|
||
/ "main"
|
||
/ "resources"
|
||
/ "application-local.yaml"
|
||
)
|
||
DEFAULT_OUTPUT = PROJECT_ROOT / "static" / "guide-data"
|
||
DEFAULT_NAV_ASSETS = (
|
||
PROJECT_ROOT
|
||
/ "static"
|
||
/ "nav-assets"
|
||
/ "codex_nav_20260621_175310_museum_default_centerline_finalxy"
|
||
)
|
||
SCHEMA_VERSION = "sgs-guide-static/v1"
|
||
BRIDGE_SCHEMA_VERSION = "sgs-guide-poi-bridge/v1"
|
||
TABLES = {
|
||
"SGS_EXHIBITION_HALL": "halls",
|
||
"SGS_EXHIBIT_OUTLINE": "outlines",
|
||
"SGS_GUIDE_STOP": "guide-stops",
|
||
"SGS_EXHIBIT_ITEM": "exhibits",
|
||
"SGS_GUIDE_CONTENT": "guide-contents",
|
||
"SGS_POI": "pois",
|
||
}
|
||
|
||
FLOOR_Z_LEVELS = [
|
||
("L-2", -13.447996),
|
||
("L-1", -8.941772),
|
||
("L1", 0.258651),
|
||
("L1.5", 5.329),
|
||
("L2", 10.706019),
|
||
("L3", 16.540436),
|
||
("L4", 22.92407),
|
||
("L5", 27.164251),
|
||
]
|
||
|
||
|
||
def snake_to_camel(value: str) -> str:
|
||
lowered = value.lower()
|
||
parts = lowered.split("_")
|
||
return parts[0] + "".join(part.capitalize() for part in parts[1:])
|
||
|
||
|
||
def normalize_bridge_name(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
|
||
text = str(value).lower().replace("（", "(").replace("）", ")")
|
||
text = re.sub(r"^(l-?\d(?:\.5)?[_\s-]*)+", "", text, flags=re.I)
|
||
text = re.sub(r"展厅\s*_?\s*\d+", "", text)
|
||
text = re.sub(r"展厅\s*\d+", "", text)
|
||
text = re.sub(r"(?<=卫生间)0+(?=\d)", "", text)
|
||
return re.sub(r"[\s_\-·、，,。:：;；/\\()（）\[\]【】\"“”']+", "", text)
|
||
|
||
|
||
def floor_from_name(value: Any) -> str | None:
|
||
if value is None:
|
||
return None
|
||
|
||
match = re.match(r"^(L-?\d+(?:\.5)?)_", str(value), re.I)
|
||
return match.group(1).upper() if match else None
|
||
|
||
|
||
def floor_from_z(value: Any) -> str | None:
|
||
if value is None:
|
||
return None
|
||
|
||
try:
|
||
z_value = float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
return min(FLOOR_Z_LEVELS, key=lambda item: abs(item[1] - z_value))[0]
|
||
|
||
|
||
def distance_sgs_to_nav(source: dict[str, Any], nav_poi: dict[str, Any]) -> float | None:
|
||
position = nav_poi.get("positionGltf")
|
||
if (
|
||
not isinstance(position, list)
|
||
or len(position) < 3
|
||
or source.get("x") is None
|
||
or source.get("y") is None
|
||
or source.get("z") is None
|
||
):
|
||
return None
|
||
|
||
try:
|
||
# SGS POI coordinates align to current nav positionGltf as [x, z, y].
|
||
return math.dist(
|
||
[float(source["x"]), float(source["z"]), float(source["y"])],
|
||
[float(position[0]), float(position[1]), float(position[2])],
|
||
)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def bridge_entry(
|
||
nav_poi: dict[str, Any],
|
||
*,
|
||
method: str,
|
||
confidence: str,
|
||
distance_meters: float | None = None,
|
||
source_poi: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
entry = {
|
||
"navPoiId": nav_poi.get("id"),
|
||
"navPoiName": nav_poi.get("name"),
|
||
"navFloorId": nav_poi.get("floorId"),
|
||
"method": method,
|
||
"confidence": confidence,
|
||
}
|
||
if distance_meters is not None:
|
||
entry["distanceMeters"] = round(distance_meters, 6)
|
||
if source_poi:
|
||
entry["sourcePoiId"] = source_poi.get("id")
|
||
entry["sourcePoiName"] = source_poi.get("name")
|
||
return entry
|
||
|
||
|
||
def is_id_column(column: str) -> bool:
|
||
upper_column = column.upper()
|
||
return upper_column == "ID" or upper_column.endswith("_ID")
|
||
|
||
|
||
def json_value(column: str, value: Any) -> Any:
|
||
if value is not None and is_id_column(column):
|
||
if isinstance(value, decimal.Decimal) and value == value.to_integral_value():
|
||
return str(int(value))
|
||
if isinstance(value, int):
|
||
return str(value)
|
||
return str(value)
|
||
|
||
if isinstance(value, (dt.datetime, dt.date, dt.time)):
|
||
return value.isoformat()
|
||
if isinstance(value, decimal.Decimal):
|
||
if value == value.to_integral_value():
|
||
return int(value)
|
||
return float(value)
|
||
if isinstance(value, bytes):
|
||
return value.hex()
|
||
return value
|
||
|
||
|
||
def normalize_record(columns: list[str], row: pyodbc.Row) -> dict[str, Any]:
|
||
return {
|
||
snake_to_camel(column): json_value(column, value)
|
||
for column, value in zip(columns, row)
|
||
}
|
||
|
||
|
||
def read_master_datasource(config_path: Path) -> dict[str, str]:
|
||
documents = yaml.safe_load_all(config_path.read_text(encoding="utf-8"))
|
||
for document in documents:
|
||
try:
|
||
return document["spring"]["datasource"]["dynamic"]["datasource"]["master"]
|
||
except Exception:
|
||
continue
|
||
raise RuntimeError(f"master datasource not found in {config_path}")
|
||
|
||
|
||
def parse_dm_jdbc_url(url: str) -> tuple[str, str, str]:
|
||
host_match = re.search(r"jdbc:dm://([^:/?]+)", url)
|
||
port_match = re.search(r"jdbc:dm://[^:/?]+:(\d+)", url)
|
||
schema_match = re.search(r"[?&]schema=([^&]+)", url)
|
||
if not host_match or not port_match or not schema_match:
|
||
raise RuntimeError(f"unsupported DM JDBC url: {url}")
|
||
return host_match.group(1), port_match.group(1), schema_match.group(1).upper()
|
||
|
||
|
||
def connect(config_path: Path) -> tuple[pyodbc.Connection, str, str, str]:
|
||
master = read_master_datasource(config_path)
|
||
host, port, schema = parse_dm_jdbc_url(str(master["url"]))
|
||
username = str(master["username"])
|
||
password = str(master["password"])
|
||
connection_string = (
|
||
"DRIVER={DM8 ODBC DRIVER};"
|
||
f"SERVER={host};"
|
||
f"TCP_PORT={port};"
|
||
f"UID={username};"
|
||
f"PWD={password};"
|
||
)
|
||
connection = pyodbc.connect(connection_string, timeout=15, autocommit=True)
|
||
return connection, host, port, schema
|
||
|
||
|
||
def fetch_table(connection: pyodbc.Connection, schema: str, table: str) -> list[dict[str, Any]]:
|
||
cursor = connection.cursor()
|
||
cursor.execute(f'SELECT * FROM "{schema}"."{table}" ORDER BY "ID"')
|
||
columns = [description[0] for description in cursor.description]
|
||
return [normalize_record(columns, row) for row in cursor.fetchall()]
|
||
|
||
|
||
def append_index(index: dict[str, list[Any]], key: Any, value: Any) -> None:
|
||
if key is None:
|
||
return
|
||
index[str(key)].append(value)
|
||
|
||
|
||
def build_outline_hall_resolver(data: dict[str, list[dict[str, Any]]]):
|
||
hall_ids = {str(hall.get("id")) for hall in data["halls"] if hall.get("id") is not None}
|
||
outline_by_id = {
|
||
str(outline.get("id")): outline
|
||
for outline in data["outlines"]
|
||
if outline.get("id") is not None
|
||
}
|
||
resolved_cache: dict[str, str | None] = {}
|
||
|
||
def resolve(outline_id: Any, visiting: set[str] | None = None) -> str | None:
|
||
if outline_id is None:
|
||
return None
|
||
|
||
key = str(outline_id)
|
||
if key in resolved_cache:
|
||
return resolved_cache[key]
|
||
|
||
visiting = visiting or set()
|
||
if key in visiting:
|
||
resolved_cache[key] = None
|
||
return None
|
||
|
||
outline = outline_by_id.get(key)
|
||
parent_id = outline.get("parentId") if outline else key
|
||
if parent_id is None:
|
||
resolved_cache[key] = None
|
||
return None
|
||
|
||
parent_key = str(parent_id)
|
||
if parent_key in hall_ids:
|
||
resolved_cache[key] = parent_key
|
||
return parent_key
|
||
|
||
visiting.add(key)
|
||
resolved_cache[key] = resolve(parent_key, visiting)
|
||
visiting.remove(key)
|
||
return resolved_cache[key]
|
||
|
||
return resolve
|
||
|
||
|
||
def build_indexes(data: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
|
||
outlines_by_hall_id: dict[str, list[Any]] = defaultdict(list)
|
||
stops_by_outline_id: dict[str, list[Any]] = defaultdict(list)
|
||
exhibits_by_hall_id: dict[str, list[Any]] = defaultdict(list)
|
||
exhibits_by_outline_id: dict[str, list[Any]] = defaultdict(list)
|
||
exhibits_by_stop_id: dict[str, list[Any]] = defaultdict(list)
|
||
guides_by_exhibit_id: dict[str, list[Any]] = defaultdict(list)
|
||
guides_by_target: dict[str, list[Any]] = defaultdict(list)
|
||
resolve_outline_hall_id = build_outline_hall_resolver(data)
|
||
|
||
for outline in data["outlines"]:
|
||
append_index(outlines_by_hall_id, resolve_outline_hall_id(outline.get("id")), outline.get("id"))
|
||
|
||
for stop in data["guide-stops"]:
|
||
append_index(stops_by_outline_id, stop.get("outlineId"), stop.get("id"))
|
||
|
||
for exhibit in data["exhibits"]:
|
||
exhibit_id = exhibit.get("id")
|
||
append_index(exhibits_by_hall_id, exhibit.get("hallId"), exhibit_id)
|
||
append_index(exhibits_by_outline_id, exhibit.get("outlineId"), exhibit_id)
|
||
append_index(exhibits_by_stop_id, exhibit.get("stopId"), exhibit_id)
|
||
|
||
for guide in data["guide-contents"]:
|
||
guide_id = guide.get("id")
|
||
target_type = guide.get("targetType")
|
||
target_id = guide.get("targetId")
|
||
if guide.get("exhibitId") is not None:
|
||
append_index(guides_by_exhibit_id, guide.get("exhibitId"), guide_id)
|
||
if target_type and target_id is not None:
|
||
append_index(guides_by_target, f"{target_type}:{target_id}", guide_id)
|
||
if str(target_type).upper() == "ITEM":
|
||
append_index(guides_by_exhibit_id, target_id, guide_id)
|
||
|
||
return {
|
||
"schemaVersion": SCHEMA_VERSION,
|
||
"outlinesByHallId": dict(outlines_by_hall_id),
|
||
"stopsByOutlineId": dict(stops_by_outline_id),
|
||
"exhibitsByHallId": dict(exhibits_by_hall_id),
|
||
"exhibitsByOutlineId": dict(exhibits_by_outline_id),
|
||
"exhibitsByStopId": dict(exhibits_by_stop_id),
|
||
"guidesByExhibitId": dict(guides_by_exhibit_id),
|
||
"guidesByTarget": dict(guides_by_target),
|
||
"poiById": {
|
||
str(poi.get("id")): poi
|
||
for poi in data["pois"]
|
||
if poi.get("id") is not None
|
||
},
|
||
}
|
||
|
||
|
||
def load_nav_pois(nav_assets_dir: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
||
manifest_path = nav_assets_dir / "app_nav_manifest.json"
|
||
poi_path = nav_assets_dir / "data" / "poi_all.json"
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
poi_payload = json.loads(poi_path.read_text(encoding="utf-8"))
|
||
return manifest, poi_payload.get("pois", [])
|
||
|
||
|
||
def infer_sgs_floor_id_map(sgs_pois: list[dict[str, Any]]) -> dict[str, str]:
|
||
votes: dict[str, Counter[str]] = defaultdict(Counter)
|
||
for poi in sgs_pois:
|
||
floor_id = poi.get("floorId")
|
||
if floor_id is None:
|
||
continue
|
||
|
||
inferred_floor = floor_from_name(poi.get("name")) or floor_from_z(poi.get("z"))
|
||
if inferred_floor:
|
||
votes[str(floor_id)][inferred_floor] += 1
|
||
|
||
return {
|
||
floor_id: counter.most_common(1)[0][0]
|
||
for floor_id, counter in votes.items()
|
||
if counter
|
||
}
|
||
|
||
|
||
def resolve_sgs_poi_floor(poi: dict[str, Any], floor_id_map: dict[str, str]) -> str | None:
|
||
return (
|
||
floor_from_name(poi.get("name"))
|
||
or floor_id_map.get(str(poi.get("floorId")))
|
||
or floor_from_z(poi.get("z"))
|
||
)
|
||
|
||
|
||
def build_hall_bridge(
|
||
halls: list[dict[str, Any]],
|
||
nav_pois: list[dict[str, Any]],
|
||
) -> dict[str, dict[str, Any]]:
|
||
result: dict[str, dict[str, Any]] = {}
|
||
touring_pois = [
|
||
poi
|
||
for poi in nav_pois
|
||
if poi.get("primaryCategory") == "touring_poi"
|
||
]
|
||
|
||
for hall in halls:
|
||
hall_id = hall.get("id")
|
||
hall_name = normalize_bridge_name(hall.get("name"))
|
||
hall_floor = hall.get("floorCode")
|
||
if hall_id is None or not hall_name:
|
||
continue
|
||
|
||
candidates = []
|
||
for nav_poi in touring_pois:
|
||
nav_name = normalize_bridge_name(nav_poi.get("name"))
|
||
if hall_name and (hall_name == nav_name or hall_name in nav_name or nav_name in hall_name):
|
||
candidates.append(nav_poi)
|
||
|
||
same_floor = [poi for poi in candidates if poi.get("floorId") == hall_floor]
|
||
selected = (same_floor or candidates)[0] if candidates else None
|
||
if selected:
|
||
result[str(hall_id)] = bridge_entry(
|
||
selected,
|
||
method="hall_name_floor" if same_floor else "hall_name",
|
||
confidence="high" if same_floor else "medium",
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
def build_sgs_poi_bridge(
|
||
sgs_pois: list[dict[str, Any]],
|
||
nav_pois: list[dict[str, Any]],
|
||
) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], dict[str, str]]:
|
||
floor_id_map = infer_sgs_floor_id_map(sgs_pois)
|
||
nav_by_key: dict[tuple[str | None, str], list[dict[str, Any]]] = defaultdict(list)
|
||
for nav_poi in nav_pois:
|
||
nav_by_key[(nav_poi.get("floorId"), normalize_bridge_name(nav_poi.get("name")))].append(nav_poi)
|
||
|
||
confirmed: dict[str, dict[str, Any]] = {}
|
||
candidates: dict[str, dict[str, Any]] = {}
|
||
|
||
for source_poi in sgs_pois:
|
||
source_poi_id = source_poi.get("id")
|
||
if source_poi_id is None:
|
||
continue
|
||
|
||
floor_id = resolve_sgs_poi_floor(source_poi, floor_id_map)
|
||
key = (floor_id, normalize_bridge_name(source_poi.get("name")))
|
||
nav_candidates = nav_by_key.get(key, [])
|
||
if not nav_candidates:
|
||
continue
|
||
|
||
scored = [
|
||
(distance_sgs_to_nav(source_poi, nav_poi), nav_poi)
|
||
for nav_poi in nav_candidates
|
||
]
|
||
with_distance = [item for item in scored if item[0] is not None]
|
||
|
||
if with_distance:
|
||
distance, selected = min(with_distance, key=lambda item: item[0] or 0)
|
||
if distance is not None and distance <= 5:
|
||
confirmed[str(source_poi_id)] = bridge_entry(
|
||
selected,
|
||
method="name_floor_coordinate",
|
||
confidence="high",
|
||
distance_meters=distance,
|
||
source_poi=source_poi,
|
||
)
|
||
else:
|
||
candidates[str(source_poi_id)] = bridge_entry(
|
||
selected,
|
||
method="name_floor_coordinate_candidate",
|
||
confidence="low",
|
||
distance_meters=distance,
|
||
source_poi=source_poi,
|
||
)
|
||
continue
|
||
|
||
if len(nav_candidates) == 1:
|
||
confirmed[str(source_poi_id)] = bridge_entry(
|
||
nav_candidates[0],
|
||
method="name_floor",
|
||
confidence="medium",
|
||
source_poi=source_poi,
|
||
)
|
||
else:
|
||
candidates[str(source_poi_id)] = bridge_entry(
|
||
nav_candidates[0],
|
||
method="name_floor_ambiguous",
|
||
confidence="low",
|
||
source_poi=source_poi,
|
||
)
|
||
|
||
return confirmed, candidates, floor_id_map
|
||
|
||
|
||
def build_poi_bridge(
|
||
data: dict[str, list[dict[str, Any]]],
|
||
nav_assets_dir: Path,
|
||
) -> dict[str, Any]:
|
||
nav_manifest, nav_pois = load_nav_pois(nav_assets_dir)
|
||
hall_bridge = build_hall_bridge(data["halls"], nav_pois)
|
||
sgs_bridge, sgs_candidates, floor_id_map = build_sgs_poi_bridge(data["pois"], nav_pois)
|
||
|
||
return {
|
||
"schemaVersion": BRIDGE_SCHEMA_VERSION,
|
||
"generatedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
|
||
"source": {
|
||
"guideStaticSchemaVersion": SCHEMA_VERSION,
|
||
"navAssetRunId": nav_manifest.get("runId"),
|
||
"navAssetPath": str(nav_assets_dir),
|
||
},
|
||
"stats": {
|
||
"hallBridgeCount": len(hall_bridge),
|
||
"sgsPoiBridgeCount": len(sgs_bridge),
|
||
"sgsPoiCandidateCount": len(sgs_candidates),
|
||
"navPoiCount": len(nav_pois),
|
||
},
|
||
"sgsFloorIdToNavFloorId": floor_id_map,
|
||
"hallToNavPoiId": hall_bridge,
|
||
"sgsPoiToNavPoiId": sgs_bridge,
|
||
"sgsPoiCandidates": sgs_candidates,
|
||
}
|
||
|
||
|
||
def write_json(path: Path, payload: Any) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def export_package(config_path: Path, output_dir: Path, nav_assets_dir: Path) -> dict[str, Any]:
|
||
connection, host, port, schema = connect(config_path)
|
||
try:
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
data: dict[str, list[dict[str, Any]]] = {}
|
||
for table, file_stem in TABLES.items():
|
||
data[file_stem] = fetch_table(connection, schema, table)
|
||
|
||
for table, file_stem in TABLES.items():
|
||
rows = data[file_stem]
|
||
write_json(
|
||
output_dir / f"{file_stem}.json",
|
||
{
|
||
"schemaVersion": SCHEMA_VERSION,
|
||
"sourceTable": table,
|
||
"rowCount": len(rows),
|
||
"rows": rows,
|
||
},
|
||
)
|
||
|
||
indexes = build_indexes(data)
|
||
write_json(output_dir / "indexes.json", indexes)
|
||
poi_bridge = build_poi_bridge(data, nav_assets_dir)
|
||
write_json(output_dir / "poi-bridge.json", poi_bridge)
|
||
|
||
manifest = {
|
||
"schemaVersion": SCHEMA_VERSION,
|
||
"generatedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
|
||
"source": {
|
||
"type": "dameng",
|
||
"host": host,
|
||
"port": port,
|
||
"schema": schema,
|
||
"configPath": str(config_path),
|
||
},
|
||
"files": {
|
||
"halls": "halls.json",
|
||
"outlines": "outlines.json",
|
||
"guideStops": "guide-stops.json",
|
||
"exhibits": "exhibits.json",
|
||
"guideContents": "guide-contents.json",
|
||
"pois": "pois.json",
|
||
"indexes": "indexes.json",
|
||
"poiBridge": "poi-bridge.json",
|
||
},
|
||
"counts": {key: len(value) for key, value in data.items()},
|
||
"bridgeStats": poi_bridge["stats"],
|
||
}
|
||
write_json(output_dir / "manifest.json", manifest)
|
||
return manifest
|
||
finally:
|
||
connection.close()
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="Export frontend guide static data from DM8.")
|
||
parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
|
||
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
||
parser.add_argument("--nav-assets", type=Path, default=DEFAULT_NAV_ASSETS)
|
||
args = parser.parse_args()
|
||
|
||
manifest = export_package(args.config.resolve(), args.output.resolve(), args.nav_assets.resolve())
|
||
print(f"Exported {manifest['schemaVersion']} to {args.output.resolve()}")
|
||
print("Counts:", json.dumps(manifest["counts"], ensure_ascii=False, sort_keys=True))
|
||
print("Bridge:", json.dumps(manifest["bridgeStats"], ensure_ascii=False, sort_keys=True))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|