#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""
CM95 / CM93 助航名称候选审计 v1

目的
====
为 CM95→NavSea `navigation_marks.chart_label_text` 找到可审计的 CM95 自有名称来源，
不使用 NEWPEC/reference PBF 名称反灌。

默认检查第一测试 AOI：
    z10/914/385
    lon 141.328125 .. 141.6796875
    lat 40.44694705960049 .. 40.71395582628604
默认重点比例尺：
    1:200000

候选来源：
1. 助航组件自身 OBJNAM / NOBJNM；
2. 同 cell、同 native_scale 的 `_texto._texta.text`；
3. 同 native_scale、邻近 cell 的 `_texto._texta.text`，仅作为低置信参考。

关联原则：
- 助航组件仍然用 `(cell_id, _cm93_relation20_raw)` 合成；
- 本脚本只审计名称，不改变 relation20 合成规则；
- 不自动把最近文字当名称；
- 输出距离和来源，由下一步确定安全规则。

输出：
    cm95_nav_name_candidates_v1.json
    cm95_nav_name_candidates_v1.csv
    cm95_nav_name_candidates_v1.txt

执行：
    python .\tools\cm95_nav_name_candidate_audit_v1.py `
      --host 127.0.0.1 `
      --port 3306 `
      --user root `
      --database cm93_chart `
      --dataset-id 1 `
      --native-scale 200000
"""

from __future__ import annotations

import argparse
import csv
import getpass
import json
import math
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any

import pymysql
from pymysql.cursors import DictCursor


NAV_CLASSES = {
    "LIGHTS",
    "BCNLAT",
    "BCNSPP",
    "BOYSPP",
    "RTPBCN",
    "TOWERS",
    "_extgn",
    "SISTAT",
    "SISTAW",
}

DIRECT_NAME_KEYS = (
    "OBJNAM",
    "NOBJNM",
)

DEFAULT_BBOX = (
    141.328125,
    40.44694705960049,
    141.6796875,
    40.71395582628604,
)


def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=3306)
    p.add_argument("--user", default="root")
    p.add_argument("--password", default=None)
    p.add_argument("--database", default="cm93_chart")
    p.add_argument("--dataset-id", type=int, default=1)
    p.add_argument("--native-scale", type=int, default=200000)
    p.add_argument("--min-lon", type=float, default=DEFAULT_BBOX[0])
    p.add_argument("--min-lat", type=float, default=DEFAULT_BBOX[1])
    p.add_argument("--max-lon", type=float, default=DEFAULT_BBOX[2])
    p.add_argument("--max-lat", type=float, default=DEFAULT_BBOX[3])
    p.add_argument(
        "--radius-m",
        type=float,
        default=800.0,
        help="每个助航对象保存的文字候选搜索半径",
    )
    p.add_argument(
        "--candidate-limit",
        type=int,
        default=12,
    )
    p.add_argument("--output-dir", default=".")
    return p.parse_args()


def json_dict(value: Any) -> dict[str, Any]:
    if value is None:
        return {}
    if isinstance(value, dict):
        return value
    if isinstance(value, (bytes, bytearray, memoryview)):
        value = bytes(value).decode("utf-8", "replace")
    try:
        obj = json.loads(value)
    except Exception:
        return {}
    return obj if isinstance(obj, dict) else {}


def relation20(attrs: dict[str, Any]):
    value = attrs.get("_cm93_relation20_raw")
    if value is None:
        return None
    try:
        return int(value)
    except Exception:
        return str(value)


def texta(attrs: dict[str, Any]):
    value = attrs.get("_texta")
    if not isinstance(value, dict):
        return None, None
    text = value.get("text")
    prefix = value.get("prefix")
    if text is None:
        return None, prefix
    text = str(text).strip()
    return (text or None), prefix


def direct_names(attrs: dict[str, Any]):
    out = []
    for key in DIRECT_NAME_KEYS:
        value = attrs.get(key)
        if isinstance(value, str) and value.strip():
            out.append({
                "field": key,
                "text": value.strip(),
            })
    return out


def haversine_m(lon1, lat1, lon2, lat2):
    r = 6371008.8
    p1 = math.radians(lat1)
    p2 = math.radians(lat2)
    dp = p2 - p1
    dl = math.radians(lon2 - lon1)
    a = (
        math.sin(dp / 2) ** 2
        + math.cos(p1) * math.cos(p2)
        * math.sin(dl / 2) ** 2
    )
    return 2 * r * math.asin(min(1.0, math.sqrt(a)))


def write_csv(path: Path, rows):
    rows = list(rows)
    fields = [
        "cell_id",
        "source_relative_path",
        "native_scale",
        "relation20",
        "lon",
        "lat",
        "component_classes",
        "component_source_record_indices",
        "direct_names",
        "nearest_same_cell_text",
        "nearest_same_cell_distance_m",
        "nearest_any_cell_text",
        "nearest_any_cell_distance_m",
        "same_cell_candidate_count",
        "all_candidate_count",
        "candidate_summary",
    ]
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        for row in rows:
            out = dict(row)
            for key in (
                "component_classes",
                "component_source_record_indices",
                "direct_names",
                "candidate_summary",
            ):
                out[key] = json.dumps(
                    out.get(key),
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
            w.writerow(out)


def main():
    a = parse_args()
    if a.password is None:
        a.password = getpass.getpass("MySQL password: ")

    outdir = Path(a.output_dir).resolve()
    outdir.mkdir(parents=True, exist_ok=True)

    conn = pymysql.connect(
        host=a.host,
        port=a.port,
        user=a.user,
        password=a.password,
        database=a.database,
        charset="utf8mb4",
        cursorclass=DictCursor,
        autocommit=False,
        connect_timeout=20,
        read_timeout=900,
    )

    nav_rows = []
    text_rows = []

    try:
        with conn.cursor() as cur:
            try:
                cur.execute("SET SESSION TRANSACTION READ ONLY")
            except Exception:
                pass
            try:
                cur.execute("START TRANSACTION READ ONLY")
            except Exception:
                cur.execute("START TRANSACTION")

            placeholders = ",".join(["%s"] * len(NAV_CLASSES))
            cur.execute(
                f"""
                SELECT
                    f.id AS feature_id,
                    f.cell_id,
                    c.source_relative_path,
                    f.source_record_index,
                    f.object_class,
                    f.native_scale,
                    f.bbox_min_lon AS lon,
                    f.bbox_min_lat AS lat,
                    f.attributes_json
                FROM chart_feature f
                JOIN chart_cell c ON c.id=f.cell_id
                WHERE f.dataset_id=%s
                  AND f.native_scale=%s
                  AND f.geometry_type='POINT'
                  AND f.object_class IN ({placeholders})
                  AND f.bbox_min_lon BETWEEN %s AND %s
                  AND f.bbox_min_lat BETWEEN %s AND %s
                ORDER BY
                    f.cell_id,
                    f.source_record_index
                """,
                (
                    a.dataset_id,
                    a.native_scale,
                    *sorted(NAV_CLASSES),
                    a.min_lon,
                    a.max_lon,
                    a.min_lat,
                    a.max_lat,
                ),
            )
            nav_rows = list(cur.fetchall())

            # 扩大少量边界，避免 tile 边缘的文字候选被切掉。
            lat_pad = a.radius_m / 111320.0
            mid_lat = (a.min_lat + a.max_lat) / 2.0
            lon_pad = (
                a.radius_m
                / max(1.0, 111320.0 * math.cos(math.radians(mid_lat)))
            )

            cur.execute(
                """
                SELECT
                    f.id AS feature_id,
                    f.cell_id,
                    c.source_relative_path,
                    f.source_record_index,
                    f.native_scale,
                    f.bbox_min_lon AS lon,
                    f.bbox_min_lat AS lat,
                    f.attributes_json
                FROM chart_feature f
                JOIN chart_cell c ON c.id=f.cell_id
                WHERE f.dataset_id=%s
                  AND f.native_scale=%s
                  AND f.geometry_type='POINT'
                  AND f.object_class='_texto'
                  AND f.bbox_min_lon BETWEEN %s AND %s
                  AND f.bbox_min_lat BETWEEN %s AND %s
                ORDER BY
                    f.cell_id,
                    f.source_record_index
                """,
                (
                    a.dataset_id,
                    a.native_scale,
                    a.min_lon - lon_pad,
                    a.max_lon + lon_pad,
                    a.min_lat - lat_pad,
                    a.max_lat + lat_pad,
                ),
            )
            text_rows = list(cur.fetchall())

            conn.rollback()
    finally:
        conn.close()

    for row in nav_rows:
        row["attributes"] = json_dict(row.pop("attributes_json"))
        row["relation20"] = relation20(row["attributes"])
        row["lon"] = float(row["lon"])
        row["lat"] = float(row["lat"])

    parsed_text = []
    for row in text_rows:
        attrs = json_dict(row.pop("attributes_json"))
        text, prefix = texta(attrs)
        if not text:
            continue
        parsed_text.append({
            "feature_id": int(row["feature_id"]),
            "cell_id": int(row["cell_id"]),
            "source_relative_path": str(row["source_relative_path"]),
            "source_record_index": int(row["source_record_index"]),
            "native_scale": int(row["native_scale"]),
            "lon": float(row["lon"]),
            "lat": float(row["lat"]),
            "text": text,
            "prefix": prefix,
        })

    groups = defaultdict(list)
    standalone = []

    for row in nav_rows:
        rel = row["relation20"]
        if rel is None:
            standalone.append([row])
        else:
            groups[(int(row["cell_id"]), rel)].append(row)

    nav_groups = list(groups.values()) + standalone

    results = []

    for members in nav_groups:
        # relation20 组成员已由上一步证明同址；
        # standalone 则只有一条。
        lon = sum(m["lon"] for m in members) / len(members)
        lat = sum(m["lat"] for m in members) / len(members)
        cell_id = int(members[0]["cell_id"])
        source_path = str(members[0]["source_relative_path"])
        rel = members[0]["relation20"]

        strong_names = []
        for m in members:
            for n in direct_names(m["attributes"]):
                strong_names.append({
                    "object_class": m["object_class"],
                    "source_record_index": m["source_record_index"],
                    **n,
                })

        candidates = []
        for t in parsed_text:
            distance = haversine_m(
                lon,
                lat,
                t["lon"],
                t["lat"],
            )
            if distance > a.radius_m:
                continue
            candidates.append({
                **t,
                "same_cell": t["cell_id"] == cell_id,
                "distance_m": round(distance, 2),
            })

        candidates.sort(
            key=lambda c: (
                not c["same_cell"],
                c["distance_m"],
                c["text"],
            )
        )

        same_cell = [
            c for c in candidates if c["same_cell"]
        ]

        nearest_same = same_cell[0] if same_cell else None
        nearest_any = (
            min(candidates, key=lambda c: c["distance_m"])
            if candidates
            else None
        )

        results.append({
            "cell_id": cell_id,
            "source_relative_path": source_path,
            "native_scale": a.native_scale,
            "relation20": rel,
            "lon": lon,
            "lat": lat,
            "component_classes": [
                m["object_class"] for m in members
            ],
            "component_source_record_indices": [
                int(m["source_record_index"]) for m in members
            ],
            "direct_names": strong_names,
            "nearest_same_cell_text": (
                nearest_same["text"] if nearest_same else None
            ),
            "nearest_same_cell_distance_m": (
                nearest_same["distance_m"] if nearest_same else None
            ),
            "nearest_any_cell_text": (
                nearest_any["text"] if nearest_any else None
            ),
            "nearest_any_cell_distance_m": (
                nearest_any["distance_m"] if nearest_any else None
            ),
            "same_cell_candidate_count": len(same_cell),
            "all_candidate_count": len(candidates),
            "candidate_summary": candidates[: a.candidate_limit],
        })

    results.sort(
        key=lambda r: (
            r["cell_id"],
            -1 if r["relation20"] is None else r["relation20"],
            r["lon"],
            r["lat"],
        )
    )

    direct_named = sum(
        1 for r in results if r["direct_names"]
    )
    same_100 = sum(
        1
        for r in results
        if r["nearest_same_cell_distance_m"] is not None
        and r["nearest_same_cell_distance_m"] <= 100.0
    )
    same_250 = sum(
        1
        for r in results
        if r["nearest_same_cell_distance_m"] is not None
        and r["nearest_same_cell_distance_m"] <= 250.0
    )

    report = {
        "schema": "navsea.cm95.nav-name-candidate-audit/v1",
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "dataset_id": a.dataset_id,
        "native_scale": a.native_scale,
        "aoi_bbox": [
            a.min_lon,
            a.min_lat,
            a.max_lon,
            a.max_lat,
        ],
        "search_radius_m": a.radius_m,
        "counts": {
            "nav_component_rows": len(nav_rows),
            "nav_groups": len(results),
            "texto_rows_with_text": len(parsed_text),
            "groups_with_direct_OBJNAM_or_NOBJNM": direct_named,
            "groups_with_same_cell_text_within_100m": same_100,
            "groups_with_same_cell_text_within_250m": same_250,
        },
        "groups": results,
        "decision": (
            "DATA_READY_FOR_NAME_ASSOCIATION_REVIEW"
            if results
            else "NO_NAV_GROUPS"
        ),
    }

    json_path = outdir / "cm95_nav_name_candidates_v1.json"
    csv_path = outdir / "cm95_nav_name_candidates_v1.csv"
    txt_path = outdir / "cm95_nav_name_candidates_v1.txt"

    json_path.write_text(
        json.dumps(
            report,
            ensure_ascii=False,
            indent=2,
            default=str,
        ),
        encoding="utf-8",
    )
    write_csv(csv_path, results)

    lines = [
        "CM95 / CM93 助航名称候选审计 v1",
        "=" * 76,
        f"Native scale                         : {a.native_scale}",
        f"Nav component rows                   : {len(nav_rows)}",
        f"Nav groups                           : {len(results)}",
        f"_texto rows with text                : {len(parsed_text)}",
        f"Groups with direct OBJNAM/NOBJNM     : {direct_named}",
        f"Same-cell text <=100m                : {same_100}",
        f"Same-cell text <=250m                : {same_250}",
        "",
        "注意：最近文字只是候选，不代表自动确定为助航名称。",
        "下一步由映射审计决定哪些关系可安全写入 chart_label_text。",
    ]
    txt_path.write_text("\n".join(lines), encoding="utf-8")

    print("\n".join(lines))
    print("")
    print(json_path)
    print(csv_path)
    print(txt_path)


if __name__ == "__main__":
    main()