#!/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():
            rows = fetch_table(connection, schema, table)
            data[file_stem] = rows
            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()