#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""
CM95 全 Dataset → NavSea delivery Mapping Gap 审计 v1

目的
====
在已经完成 production 首轮生成后，对 Dataset 全量 chart_feature 与当前
cm95_to_navsea_v1.yaml 做只读对照，找出“当前 delivery 不会输出”的真实
object_class + geometry_type 组合。

这一步专门用于 production mapping 收口，不修改数据库、不生成 PBF。

默认输入：
  dataset_id = 1
  mapping    = config/mappings/cm95_to_navsea_v1.yaml

默认输出：
  build/production/cm95_global_mapping_gap_audit_v1.json
  build/production/cm95_global_mapping_gap_variants_v1.csv
  build/production/cm95_global_mapping_gap_samples_v1.jsonl
  build/production/cm95_global_mapping_gap_summary_v1.txt

审计内容：
- 全 Dataset object_class + geometry_type + native_scale 数量；
- 当前 YAML 是否存在 exact rule；
- 同 object_class 的其它 geometry rule；
- 已知 hard_gate 是否出现未覆盖 geometry variant；
- 显式 ENGINEERING_ONLY / cm95_unmapped；
- 每个 gap variant 的属性键频率、样本值、relation20 覆盖；
- 每个 gap variant 的代表 Feature 样本。

数据库连接直接读取项目 .env：
  CM93_DB_HOST
  CM93_DB_PORT
  CM93_DB_NAME
  CM93_DB_USER
  CM93_DB_PASSWORD

执行：
  & ".\.venv\Scripts\python.exe" .\tools\cm95_global_mapping_gap_audit_v1.py `
    --dataset-id 1 `
    --mapping .\config\mappings\cm95_to_navsea_v1.yaml
"""

from __future__ import annotations

import argparse
import csv
import json
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any

import pymysql
import yaml

from cm93db.config import load_config


DELIVERY_MAIN_STATUSES = {
    "MAPPED",
    "MAPPED_CONDITIONAL",
    "COMPOSITE",
    "MAPPED_DERIVED_POINTS",
}
DELIVERY_DERIVED_STATUSES = {
    "MAPPED",
    "MAPPED_CONDITIONAL",
    "DERIVED_LABEL_ONLY",
}


def parse_args():
    p = argparse.ArgumentParser(
        description="CM95 全 Dataset → NavSea delivery Mapping Gap 审计 v1"
    )
    p.add_argument("--dataset-id", type=int, default=1)
    p.add_argument(
        "--mapping",
        default=r"config\mappings\cm95_to_navsea_v1.yaml",
    )
    p.add_argument(
        "--output",
        default=r"build\production\cm95_global_mapping_gap_audit_v1.json",
    )
    p.add_argument("--sample-per-variant", type=int, default=6)
    p.add_argument("--attribute-value-limit", type=int, default=12)
    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", errors="replace")
    try:
        obj = json.loads(str(value))
        return obj if isinstance(obj, dict) else {}
    except Exception:
        return {}


def scalar_text(value: Any) -> str:
    if isinstance(value, (dict, list)):
        return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return str(value)


def load_rules(path: Path):
    obj = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    rules = obj.get("rules") or obj.get("mappings") or []
    if not isinstance(rules, list):
        raise RuntimeError("mapping YAML 中找不到 rules/mappings list")

    exact: dict[tuple[str, str], dict[str, Any]] = {}
    by_class: dict[str, list[dict[str, Any]]] = defaultdict(list)

    for rule in rules:
        if not isinstance(rule, dict):
            continue
        cls = str(rule.get("cm95_object_class") or "").strip()
        geom = str(rule.get("geometry_type") or "").strip().upper()
        if not cls or not geom:
            continue
        exact[(cls, geom)] = rule
        by_class[cls].append(rule)

    return obj, exact, by_class


def has_delivery_output(rule: dict[str, Any]) -> bool:
    status = str(rule.get("mapping_status") or "")
    target = str(rule.get("target_source_layer") or "")
    if target and target != "cm95_unmapped" and status in DELIVERY_MAIN_STATUSES:
        return True

    derived = rule.get("derived_outputs") or []
    if derived and status in DELIVERY_DERIVED_STATUSES:
        for item in derived:
            if not isinstance(item, dict):
                continue
            layer = str(item.get("source_layer") or "")
            if layer and layer != "cm95_unmapped":
                return True
    return False


def gap_reason(
    cls: str,
    geom: str,
    exact_rules: dict[tuple[str, str], dict[str, Any]],
    by_class: dict[str, list[dict[str, Any]]],
):
    rule = exact_rules.get((cls, geom))
    class_rules = by_class.get(cls, [])

    if rule is None:
        inherited_hard_gate = any(
            str(r.get("safety_class") or "") == "hard_gate"
            for r in class_rules
        )
        return {
            "is_gap": True,
            "reason": (
                "NO_EXACT_RULE_KNOWN_HARD_GATE_VARIANT"
                if inherited_hard_gate
                else "NO_EXACT_RULE"
            ),
            "rule": None,
            "inherited_hard_gate": inherited_hard_gate,
        }

    if has_delivery_output(rule):
        return {
            "is_gap": False,
            "reason": "DELIVERED",
            "rule": rule,
            "inherited_hard_gate": False,
        }

    target = str(rule.get("target_source_layer") or "")
    status = str(rule.get("mapping_status") or "")
    if target == "cm95_unmapped":
        reason = "EXPLICIT_ENGINEERING_ONLY"
    else:
        reason = f"NOT_DELIVERED_STATUS_{status or 'EMPTY'}"

    return {
        "is_gap": True,
        "reason": reason,
        "rule": rule,
        "inherited_hard_gate": (
            str(rule.get("safety_class") or "") == "hard_gate"
        ),
    }


def connect():
    cfg = load_config()
    return pymysql.connect(
        host=cfg.host,
        port=cfg.port,
        user=cfg.user,
        password=cfg.password,
        database=cfg.database,
        charset=cfg.charset,
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=False,
        connect_timeout=20,
        read_timeout=900,
        write_timeout=30,
    )


def main():
    a = parse_args()
    mapping_path = Path(a.mapping).resolve()
    output_path = Path(a.output).resolve()
    outdir = output_path.parent
    outdir.mkdir(parents=True, exist_ok=True)

    _mapping_obj, exact_rules, by_class = load_rules(mapping_path)

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

        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    f.object_class_id,
                    f.object_class,
                    f.geometry_type,
                    f.native_scale,
                    COUNT(*) AS feature_count,
                    COUNT(DISTINCT f.cell_id) AS cell_count
                FROM chart_feature f
                WHERE f.dataset_id=%s
                  AND f.geometry_type IN ('POINT','LINE','AREA')
                GROUP BY
                    f.object_class_id,
                    f.object_class,
                    f.geometry_type,
                    f.native_scale
                ORDER BY
                    f.object_class,
                    f.geometry_type,
                    f.native_scale
                """,
                (a.dataset_id,),
            )
            grouped = list(cur.fetchall())

            cur.execute(
                """
                SELECT
                    object_class_id,
                    object_class,
                    geometry_code,
                    display_code,
                    description
                FROM chart_dictionary_object
                WHERE dataset_id=%s
                """,
                (a.dataset_id,),
            )
            dictionary_rows = list(cur.fetchall())

        conn.rollback()

    dictionary = {
        int(r["object_class_id"]): r
        for r in dictionary_rows
        if r.get("object_class_id") is not None
    }

    variants: dict[tuple[str, str], dict[str, Any]] = {}
    for row in grouped:
        cls = str(row["object_class"])
        geom = str(row["geometry_type"]).upper()
        key = (cls, geom)
        v = variants.setdefault(
            key,
            {
                "object_class_id": (
                    int(row["object_class_id"])
                    if row.get("object_class_id") is not None
                    else None
                ),
                "object_class": cls,
                "geometry_type": geom,
                "feature_count": 0,
                "cell_count_max": 0,
                "scale_counts": Counter(),
            },
        )
        cnt = int(row["feature_count"])
        v["feature_count"] += cnt
        v["cell_count_max"] = max(v["cell_count_max"], int(row["cell_count"]))
        v["scale_counts"][str(int(row["native_scale"]))] += cnt

    gap_variants = []
    for key, v in sorted(
        variants.items(),
        key=lambda kv: (-kv[1]["feature_count"], kv[0][0], kv[0][1]),
    ):
        cls, geom = key
        state = gap_reason(cls, geom, exact_rules, by_class)
        if not state["is_gap"]:
            continue

        oid = v["object_class_id"]
        d = dictionary.get(oid) if oid is not None else None
        rule = state["rule"]
        class_rules = by_class.get(cls, [])

        item = {
            "object_class_id": oid,
            "object_class": cls,
            "geometry_type": geom,
            "feature_count": int(v["feature_count"]),
            "cell_count": int(v["cell_count_max"]),
            "scale_counts": dict(sorted(v["scale_counts"].items(), key=lambda x: int(x[0]))),
            "gap_reason": state["reason"],
            "dictionary_description": d.get("description") if d else None,
            "dictionary_geometry_code": d.get("geometry_code") if d else None,
            "dictionary_display_code": d.get("display_code") if d else None,
            "exact_rule_id": str(rule.get("rule_id")) if rule else None,
            "exact_rule_status": str(rule.get("mapping_status")) if rule else None,
            "exact_rule_target": str(rule.get("target_source_layer")) if rule else None,
            "exact_rule_safety": str(rule.get("safety_class")) if rule else None,
            "class_known_geometry_rules": [
                {
                    "geometry_type": str(r.get("geometry_type") or ""),
                    "rule_id": str(r.get("rule_id") or ""),
                    "target_source_layer": str(r.get("target_source_layer") or ""),
                    "mapping_status": str(r.get("mapping_status") or ""),
                    "safety_class": str(r.get("safety_class") or ""),
                    "canonical_object_type": str(r.get("canonical_object_type") or ""),
                }
                for r in class_rules
            ],
            "inherited_hard_gate": bool(state["inherited_hard_gate"]),
            "attribute_key_counts": {},
            "attribute_value_samples": {},
            "relation20_present": 0,
            "relation20_missing": 0,
            "sample_features": [],
        }
        gap_variants.append(item)

    gap_keys = {(x["object_class"], x["geometry_type"]) for x in gap_variants}
    gap_classes = sorted({x[0] for x in gap_keys})

    # Gap rows are normally a small subset of Dataset. Read them once and
    # build attribute/sample evidence in Python, avoiding dozens of SQL round trips.
    if gap_classes:
        placeholders = ",".join(["%s"] * len(gap_classes))
        with connect() as conn:
            try:
                conn.cursor().execute("SET SESSION TRANSACTION READ ONLY")
            except Exception:
                pass
            with conn.cursor() as cur:
                cur.execute(
                    f"""
                    SELECT
                        f.id AS feature_id,
                        f.cell_id,
                        c.source_relative_path,
                        f.source_record_index,
                        f.object_class_id,
                        f.object_class,
                        f.geometry_type,
                        f.native_scale,
                        f.raw_geotype,
                        f.coverage_id,
                        f.bbox_min_lon,
                        f.bbox_min_lat,
                        f.bbox_max_lon,
                        f.bbox_max_lat,
                        f.attributes_json,
                        f.geometry_repaired,
                        f.geometry_issue
                    FROM chart_feature f
                    JOIN chart_cell c ON c.id=f.cell_id
                    WHERE f.dataset_id=%s
                      AND f.object_class IN ({placeholders})
                      AND f.geometry_type IN ('POINT','LINE','AREA')
                    ORDER BY
                        f.object_class,
                        f.geometry_type,
                        f.native_scale,
                        c.source_relative_path,
                        f.source_record_index
                    """,
                    (a.dataset_id, *gap_classes),
                )
                source_rows = list(cur.fetchall())
            conn.rollback()
    else:
        source_rows = []

    item_by_key = {
        (x["object_class"], x["geometry_type"]): x
        for x in gap_variants
    }
    key_counters: dict[tuple[str, str], Counter] = defaultdict(Counter)
    value_counters: dict[tuple[str, str, str], Counter] = defaultdict(Counter)
    sample_counts: Counter = Counter()

    sample_jsonl_path = outdir / "cm95_global_mapping_gap_samples_v1.jsonl"
    with sample_jsonl_path.open("w", encoding="utf-8") as sample_file:
        for row in source_rows:
            key = (
                str(row["object_class"]),
                str(row["geometry_type"]).upper(),
            )
            item = item_by_key.get(key)
            if item is None:
                continue

            attrs = json_dict(row.get("attributes_json"))
            for attr_key, attr_value in attrs.items():
                key_counters[key][str(attr_key)] += 1
                value_counters[(key[0], key[1], str(attr_key))][scalar_text(attr_value)] += 1

            rel = attrs.get("_cm93_relation20_raw")
            if rel is None or rel == "":
                item["relation20_missing"] += 1
            else:
                item["relation20_present"] += 1

            if sample_counts[key] < max(1, int(a.sample_per_variant)):
                sample_counts[key] += 1
                sample = {
                    "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"]),
                    "object_class_id": (
                        int(row["object_class_id"])
                        if row.get("object_class_id") is not None
                        else None
                    ),
                    "object_class": key[0],
                    "geometry_type": key[1],
                    "native_scale": int(row["native_scale"]),
                    "raw_geotype": row.get("raw_geotype"),
                    "coverage_id": row.get("coverage_id"),
                    "bbox": [
                        row.get("bbox_min_lon"),
                        row.get("bbox_min_lat"),
                        row.get("bbox_max_lon"),
                        row.get("bbox_max_lat"),
                    ],
                    "attributes": attrs,
                    "geometry_repaired": bool(row.get("geometry_repaired")),
                    "geometry_issue": row.get("geometry_issue"),
                }
                item["sample_features"].append(sample)
                sample_file.write(
                    json.dumps(sample, ensure_ascii=False, separators=(",", ":"), default=str)
                    + "\n"
                )

    for item in gap_variants:
        key = (item["object_class"], item["geometry_type"])
        item["attribute_key_counts"] = dict(
            key_counters[key].most_common()
        )
        values = {}
        for attr_key, _count in key_counters[key].most_common():
            values[attr_key] = [
                {"value": value, "count": count}
                for value, count in value_counters[
                    (key[0], key[1], attr_key)
                ].most_common(max(1, int(a.attribute_value_limit)))
            ]
        item["attribute_value_samples"] = values

    reason_counts = Counter()
    gap_feature_total = 0
    no_exact_total = 0
    explicit_total = 0
    inherited_hard_gate_total = 0
    for item in gap_variants:
        reason_counts[item["gap_reason"]] += 1
        gap_feature_total += int(item["feature_count"])
        if item["gap_reason"].startswith("NO_EXACT_RULE"):
            no_exact_total += int(item["feature_count"])
        else:
            explicit_total += int(item["feature_count"])
        if item["inherited_hard_gate"]:
            inherited_hard_gate_total += int(item["feature_count"])

    report = {
        "schema": "navsea.cm95.global-mapping-gap-audit/v1",
        "dataset_id": int(a.dataset_id),
        "mapping_path": str(mapping_path),
        "source_variant_count": len(variants),
        "gap_variant_count": len(gap_variants),
        "gap_object_class_count": len({x["object_class"] for x in gap_variants}),
        "gap_feature_total": gap_feature_total,
        "no_exact_rule_feature_total": no_exact_total,
        "explicit_engineering_or_status_gap_feature_total": explicit_total,
        "inherited_hard_gate_feature_total": inherited_hard_gate_total,
        "gap_reason_counts": dict(reason_counts),
        "gap_variants": gap_variants,
    }

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

    csv_path = outdir / "cm95_global_mapping_gap_variants_v1.csv"
    with csv_path.open("w", encoding="utf-8-sig", newline="") as f:
        fields = [
            "object_class_id",
            "object_class",
            "geometry_type",
            "feature_count",
            "cell_count",
            "gap_reason",
            "inherited_hard_gate",
            "dictionary_description",
            "dictionary_geometry_code",
            "dictionary_display_code",
            "exact_rule_id",
            "exact_rule_status",
            "exact_rule_target",
            "exact_rule_safety",
            "scale_counts",
            "known_geometry_rules",
            "attribute_keys",
            "relation20_present",
            "relation20_missing",
        ]
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        for item in gap_variants:
            w.writerow(
                {
                    "object_class_id": item["object_class_id"],
                    "object_class": item["object_class"],
                    "geometry_type": item["geometry_type"],
                    "feature_count": item["feature_count"],
                    "cell_count": item["cell_count"],
                    "gap_reason": item["gap_reason"],
                    "inherited_hard_gate": item["inherited_hard_gate"],
                    "dictionary_description": item["dictionary_description"],
                    "dictionary_geometry_code": item["dictionary_geometry_code"],
                    "dictionary_display_code": item["dictionary_display_code"],
                    "exact_rule_id": item["exact_rule_id"],
                    "exact_rule_status": item["exact_rule_status"],
                    "exact_rule_target": item["exact_rule_target"],
                    "exact_rule_safety": item["exact_rule_safety"],
                    "scale_counts": json.dumps(item["scale_counts"], ensure_ascii=False, separators=(",", ":")),
                    "known_geometry_rules": json.dumps(item["class_known_geometry_rules"], ensure_ascii=False, separators=(",", ":")),
                    "attribute_keys": json.dumps(item["attribute_key_counts"], ensure_ascii=False, separators=(",", ":")),
                    "relation20_present": item["relation20_present"],
                    "relation20_missing": item["relation20_missing"],
                }
            )

    txt_path = outdir / "cm95_global_mapping_gap_summary_v1.txt"
    lines = [
        "CM95 全 Dataset → NavSea delivery Mapping Gap 审计 v1",
        "=" * 76,
        f"Dataset ID              : {a.dataset_id}",
        f"Source variants         : {len(variants)}",
        f"Gap variants            : {len(gap_variants)}",
        f"Gap object classes      : {report['gap_object_class_count']}",
        f"Gap feature rows        : {gap_feature_total}",
        f"No exact rule rows      : {no_exact_total}",
        f"Explicit/status gap rows: {explicit_total}",
        f"Inherited hard-gate rows: {inherited_hard_gate_total}",
        "",
        "Top gap variants:",
    ]
    for item in gap_variants[:80]:
        scales = ",".join(
            f"{k}:{v}" for k, v in item["scale_counts"].items()
        )
        lines.append(
            f"  {item['object_class']:<10} {item['geometry_type']:<5} "
            f"rows={item['feature_count']:<6} "
            f"reason={item['gap_reason']:<34} "
            f"hard={str(item['inherited_hard_gate']):<5} "
            f"scales=[{scales}]"
        )
    lines += [
        "",
        f"JSON   : {output_path}",
        f"CSV    : {csv_path}",
        f"SAMPLES: {sample_jsonl_path}",
    ]
    txt_path.write_text("\n".join(lines), encoding="utf-8")

    print("\n".join(lines))


if __name__ == "__main__":
    main()