#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CM93 -> MySQL Dataset ID audit + geometry issue classification v2
Target schema discovered from Dataset 1 audit:
  chart_dataset
  chart_cell
  chart_feature
  chart_feature_attribute
  chart_sounding_point
  chart_point_geometry
  chart_line_geometry
  chart_area_geometry
  chart_coverage
  chart_import_job
  chart_import_event

READ ONLY: SELECT only; transaction is opened READ ONLY.
No geometry repair, no re-import, no PBF.

Dependency:
  pip install pymysql

Example:
  python .\\cm93_mysql_audit_v2.py --host 127.0.0.1 --port 3306 --user root --database cm93_chart --dataset-id 1
"""
from __future__ import annotations

import argparse
import csv
import getpass
import json
import os
import re
import sys
import traceback
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

try:
    import pymysql
    from pymysql.cursors import DictCursor
except ImportError:
    print("Missing dependency: pymysql")
    print("Install with: pip install pymysql")
    sys.exit(2)

EXPECTED = {
    "cells": 784,
    "features": 322465,
    "attributes": 405153,
    "sounding_points": 275700,
    "geometry_issues": 4406,
}

# Conservative text classification. Exact raw issue text is always preserved.
CATEGORY_RULES: List[Tuple[str, Sequence[str]]] = [
    ("INVALID_COORDINATE", [
        r"\bnan\b", r"\binf(?:inity)?\b", r"invalid coordinate", r"coordinate.*out of range",
        r"latitude.*out of range", r"longitude.*out of range", r"non[- ]?finite",
    ]),
    ("INSUFFICIENT_POINTS", [
        r"insufficient point", r"too few point", r"not enough point", r"minimum.*point",
        r"less than .* points?", r"requires? .* points?",
    ]),
    ("RING_NOT_CLOSED", [
        r"ring.*not closed", r"not closed.*ring", r"unclosed ring", r"closed linestring",
    ]),
    ("SELF_INTERSECTION", [
        r"self[- ]?intersection", r"self intersect", r"ring self", r"bow[- ]?tie",
    ]),
    ("INVALID_RING", [
        r"invalid ring", r"ring.*invalid", r"illegal ring", r"shell.*invalid", r"hole.*invalid",
        r"interior ring", r"exterior ring",
    ]),
    ("DEGENERATE_GEOMETRY", [
        r"degenerate", r"zero[- ]?length", r"zero[- ]?area", r"collapsed", r"duplicate point",
        r"all points.*same", r"too few distinct",
    ]),
    ("MULTIPART_ASSEMBLY", [
        r"multipart", r"multi[- ]?part", r"cannot.*assemble", r"failed.*assemble", r"assembly failed",
        r"cannot.*merge", r"failed.*merge", r"stitch",
    ]),
    ("TOPOLOGY_MISMATCH", [
        r"topolog", r"missing edge", r"edge.*missing", r"missing node", r"node.*missing",
        r"edge reference", r"node reference", r"primitive reference", r"dangling edge", r"broken chain",
        r"disconnected",
    ]),
    ("GEOMETRY_TYPE_MISMATCH", [
        r"type mismatch", r"geometry type", r"expected.*(?:point|line|polygon|area)",
        r"unexpected.*(?:point|line|polygon|area)", r"wrong.*geometry",
    ]),
    ("EMPTY_GEOMETRY", [
        r"empty geometry", r"geometry.*empty", r"no geometry", r"null geometry", r"geometry is null",
        r"no coordinates?", r"empty point", r"empty line", r"empty polygon",
    ]),
    ("CONVERSION_EXCEPTION", [
        r"wkb", r"wkt", r"parse exception", r"conversion exception", r"geometry exception",
        r"failed to convert", r"cannot convert", r"invalid gis data", r"spatial.*exception",
    ]),
]

STAGE_RULES: List[Tuple[str, Sequence[str]]] = [
    ("MYSQL_WRITE", [r"mysql", r"database", r"insert", r"spatial", r"wkb", r"wkt"]),
    ("VALIDATION", [r"valid", r"self[- ]?intersect", r"ring", r"geometry.*invalid"]),
    ("ASSEMBLY", [r"assemble", r"assembly", r"merge", r"stitch", r"multipart", r"chain"]),
    ("TOPOLOGY", [r"topolog", r"edge", r"node", r"primitive"]),
    ("DECODE", [r"decode", r"parse", r"read", r"binary", r"record"]),
    ("SOURCE", [r"source", r"raw", r"cm93"]),
]


def qi(s: str) -> str:
    return "`" + s.replace("`", "``") + "`"


def text(v: Any) -> str:
    if v is None:
        return ""
    if isinstance(v, (bytes, bytearray)):
        return v.decode("utf-8", errors="replace")
    return str(v)


def classify_category(raw: str) -> str:
    for category, pats in CATEGORY_RULES:
        if any(re.search(p, raw, re.I) for p in pats):
            return category
    return "OTHER"


def classify_stage(raw: str) -> str:
    for stage, pats in STAGE_RULES:
        if any(re.search(p, raw, re.I) for p in pats):
            return stage
    return "UNKNOWN"


def scalar(cur, sql: str, params: Sequence[Any] = ()) -> Any:
    cur.execute(sql, params)
    row = cur.fetchone()
    return next(iter(row.values())) if row else None


def rows(cur, sql: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]:
    cur.execute(sql, params)
    return list(cur.fetchall())


def table_exists(cur, db: str, table: str) -> bool:
    return bool(scalar(cur, "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s", (db, table)))


def columns(cur, db: str, table: str) -> Dict[str, Dict[str, Any]]:
    out = {}
    for r in rows(cur, """
        SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, IS_NULLABLE, COLUMN_KEY
        FROM information_schema.COLUMNS
        WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s
        ORDER BY ORDINAL_POSITION
    """, (db, table)):
        out[r["COLUMN_NAME"]] = r
    return out


def write_csv(path: Path, data: Iterable[Dict[str, Any]], fieldnames: Optional[List[str]] = None) -> None:
    data = list(data)
    if fieldnames is None:
        fieldnames = []
        seen = set()
        for r in data:
            for k in r:
                if k not in seen:
                    fieldnames.append(k)
                    seen.add(k)
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames or [], extrasaction="ignore")
        w.writeheader()
        for r in data:
            w.writerow({k: text(r.get(k)) for k in (fieldnames or [])})


def safe(cur, sql: str, params: Sequence[Any] = ()) -> Dict[str, Any]:
    try:
        return {"ok": True, "rows": rows(cur, sql, params)}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def normalize_issue_value(v: Any) -> str:
    s = text(v).strip()
    if not s:
        return ""
    # If JSON, preserve a stable compact textual form to improve grouping.
    try:
        obj = json.loads(s)
        return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    except Exception:
        return s


def issue_is_present_sql(col: str) -> str:
    # Works for text/varchar/json-like columns; CAST keeps this generic.
    return f"{qi(col)} IS NOT NULL AND TRIM(CAST({qi(col)} AS CHAR)) <> '' AND LOWER(TRIM(CAST({qi(col)} AS CHAR))) NOT IN ('0','false','none','null','[]','{{}}')"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default=os.getenv("MYSQL_HOST", "127.0.0.1"))
    ap.add_argument("--port", type=int, default=int(os.getenv("MYSQL_PORT", "3306")))
    ap.add_argument("--user", default=os.getenv("MYSQL_USER", "root"))
    ap.add_argument("--password", default=os.getenv("MYSQL_PASSWORD"))
    ap.add_argument("--database", default=os.getenv("MYSQL_DATABASE", "cm93_chart"))
    ap.add_argument("--dataset-id", type=int, default=1)
    ap.add_argument("--output", default=None)
    args = ap.parse_args()

    if args.password is None:
        args.password = getpass.getpass("MySQL password: ")

    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    out = Path(args.output or f"cm93_audit_v2_dataset_{args.dataset_id}_{stamp}")
    out.mkdir(parents=True, exist_ok=True)

    report: Dict[str, Any] = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "database": args.database,
        "dataset_id": args.dataset_id,
        "read_only": True,
        "expected": EXPECTED,
        "schema": {},
        "counts": {},
        "relationships": {},
        "import_jobs": [],
        "cell_status": [],
        "geometry_issue_reconciliation": {},
        "geometry_storage": {},
        "issue_summary": {},
        "warnings": [],
    }

    conn = None
    try:
        conn = pymysql.connect(
            host=args.host, port=args.port, user=args.user, password=args.password,
            database=args.database, charset="utf8mb4", cursorclass=DictCursor,
            autocommit=False, connect_timeout=20, read_timeout=600, write_timeout=30,
        )
        cur = conn.cursor()
        try:
            cur.execute("SET SESSION TRANSACTION READ ONLY")
        except Exception as e:
            report["warnings"].append(f"SET SESSION TRANSACTION READ ONLY failed: {e}")
        cur.execute("START TRANSACTION READ ONLY")

        required = [
            "chart_dataset", "chart_cell", "chart_feature", "chart_feature_attribute",
            "chart_sounding_point", "chart_point_geometry", "chart_line_geometry",
            "chart_area_geometry", "chart_coverage", "chart_import_job", "chart_import_event",
        ]
        existing = {t: table_exists(cur, args.database, t) for t in required}
        report["schema"]["tables"] = existing
        report["schema"]["columns"] = {t: columns(cur, args.database, t) for t, ok in existing.items() if ok}

        # ------------------------------------------------------------------
        # Basic counts, scoped correctly through joins when table has no dataset_id.
        # ------------------------------------------------------------------
        print("[1/8] Basic counts")
        report["counts"]["datasets"] = int(scalar(cur, "SELECT COUNT(*) FROM chart_dataset WHERE id=%s", (args.dataset_id,)))
        report["counts"]["cells"] = int(scalar(cur, "SELECT COUNT(*) FROM chart_cell WHERE dataset_id=%s", (args.dataset_id,)))
        report["counts"]["features"] = int(scalar(cur, "SELECT COUNT(*) FROM chart_feature WHERE dataset_id=%s", (args.dataset_id,)))
        report["counts"]["attributes"] = int(scalar(cur, """
            SELECT COUNT(*) FROM chart_feature_attribute a
            JOIN chart_feature f ON f.id=a.feature_id
            WHERE f.dataset_id=%s
        """, (args.dataset_id,)))
        report["counts"]["sounding_points"] = int(scalar(cur, """
            SELECT COUNT(*) FROM chart_sounding_point s
            JOIN chart_feature f ON f.id=s.feature_id
            WHERE f.dataset_id=%s
        """, (args.dataset_id,)))

        # Unknown object/attr dictionary mapping checks.
        report["counts"]["feature_object_class_null"] = int(scalar(cur, """
            SELECT COUNT(*) FROM chart_feature
            WHERE dataset_id=%s AND (object_class IS NULL OR TRIM(object_class)='')
        """, (args.dataset_id,)))
        report["counts"]["attribute_name_null"] = int(scalar(cur, """
            SELECT COUNT(*) FROM chart_feature_attribute a
            JOIN chart_feature f ON f.id=a.feature_id
            WHERE f.dataset_id=%s AND (a.attribute_name IS NULL OR TRIM(a.attribute_name)='')
        """, (args.dataset_id,)))

        # ------------------------------------------------------------------
        # Relational integrity.
        # ------------------------------------------------------------------
        print("[2/8] Relational integrity")
        rel_queries = {
            "feature_to_cell_orphan": """
                SELECT COUNT(*) n FROM chart_feature f
                LEFT JOIN chart_cell c ON c.id=f.cell_id
                WHERE f.dataset_id=%s AND c.id IS NULL
            """,
            "feature_cell_wrong_dataset": """
                SELECT COUNT(*) n FROM chart_feature f
                JOIN chart_cell c ON c.id=f.cell_id
                WHERE f.dataset_id=%s AND c.dataset_id<>f.dataset_id
            """,
            "attribute_to_feature_orphan_global": """
                SELECT COUNT(*) n FROM chart_feature_attribute a
                LEFT JOIN chart_feature f ON f.id=a.feature_id
                WHERE f.id IS NULL
            """,
            "sounding_to_feature_orphan_global": """
                SELECT COUNT(*) n FROM chart_sounding_point s
                LEFT JOIN chart_feature f ON f.id=s.feature_id
                WHERE f.id IS NULL
            """,
            "point_geometry_orphan_global": """
                SELECT COUNT(*) n FROM chart_point_geometry g
                LEFT JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.id IS NULL
            """,
            "line_geometry_orphan_global": """
                SELECT COUNT(*) n FROM chart_line_geometry g
                LEFT JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.id IS NULL
            """,
            "area_geometry_orphan_global": """
                SELECT COUNT(*) n FROM chart_area_geometry g
                LEFT JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.id IS NULL
            """,
        }
        for k, sql in rel_queries.items():
            params = (args.dataset_id,) if "%s" in sql else ()
            report["relationships"][k] = int(scalar(cur, sql, params))

        # Duplicate geometry rows by feature_id should not exist.
        for table in ["chart_point_geometry", "chart_line_geometry", "chart_area_geometry"]:
            report["relationships"][f"{table}_duplicate_feature_ids"] = int(scalar(cur, f"""
                SELECT COUNT(*) FROM (
                    SELECT g.feature_id FROM {table} g
                    JOIN chart_feature f ON f.id=g.feature_id
                    WHERE f.dataset_id=%s
                    GROUP BY g.feature_id HAVING COUNT(*)>1
                ) x
            """, (args.dataset_id,)))

        # ------------------------------------------------------------------
        # Import jobs and cell status: this is likely where 4406 comes from.
        # ------------------------------------------------------------------
        print("[3/8] Import job / cell status")
        report["import_jobs"] = rows(cur, """
            SELECT id, dataset_id, status, started_at, finished_at,
                   cells_discovered, cells_done, cells_failed,
                   features_source, features_imported,
                   attributes_source, attributes_imported,
                   unknown_object_count, unknown_attribute_count,
                   geometry_invalid_count, geometry_repaired_count, summary_json
            FROM chart_import_job
            WHERE dataset_id=%s
            ORDER BY id
        """, (args.dataset_id,))
        report["cell_status"] = rows(cur, """
            SELECT import_status, COUNT(*) AS cell_count,
                   SUM(feature_count) AS feature_count,
                   SUM(attribute_count) AS attribute_count,
                   SUM(CASE WHEN error_message IS NOT NULL AND TRIM(error_message)<>'' THEN 1 ELSE 0 END) AS cells_with_error_message
            FROM chart_cell
            WHERE dataset_id=%s
            GROUP BY import_status
            ORDER BY cell_count DESC
        """, (args.dataset_id,))
        write_csv(out / "import_jobs.csv", report["import_jobs"])
        write_csv(out / "cell_status.csv", report["cell_status"])

        # ------------------------------------------------------------------
        # Geometry issue storage in chart_feature and chart_coverage.
        # ------------------------------------------------------------------
        print("[4/8] Geometry issue reconciliation")
        feature_cols = report["schema"]["columns"].get("chart_feature", {})
        coverage_cols = report["schema"]["columns"].get("chart_coverage", {})

        if "geometry_issue" not in feature_cols:
            raise RuntimeError("chart_feature.geometry_issue not found")

        issue_pred = issue_is_present_sql("geometry_issue")
        feature_issue_count = int(scalar(cur, f"SELECT COUNT(*) FROM chart_feature WHERE dataset_id=%s AND {issue_pred}", (args.dataset_id,)))
        coverage_issue_count = 0
        if "geometry_issue" in coverage_cols:
            coverage_issue_count = int(scalar(cur, f"SELECT COUNT(*) FROM chart_coverage WHERE dataset_id=%s AND {issue_pred}", (args.dataset_id,)))

        feature_repaired_count = int(scalar(cur, "SELECT COUNT(*) FROM chart_feature WHERE dataset_id=%s AND COALESCE(geometry_repaired,0)<>0", (args.dataset_id,)))
        coverage_repaired_count = 0
        if "geometry_repaired" in coverage_cols:
            coverage_repaired_count = int(scalar(cur, "SELECT COUNT(*) FROM chart_coverage WHERE dataset_id=%s AND COALESCE(geometry_repaired,0)<>0", (args.dataset_id,)))

        job_invalid_total = sum(int(r.get("geometry_invalid_count") or 0) for r in report["import_jobs"])
        job_repaired_total = sum(int(r.get("geometry_repaired_count") or 0) for r in report["import_jobs"])
        report["geometry_issue_reconciliation"] = {
            "feature_geometry_issue_rows": feature_issue_count,
            "coverage_geometry_issue_rows": coverage_issue_count,
            "feature_geometry_repaired_rows": feature_repaired_count,
            "coverage_geometry_repaired_rows": coverage_repaired_count,
            "sum_import_job_geometry_invalid_count": job_invalid_total,
            "sum_import_job_geometry_repaired_count": job_repaired_total,
            "expected_geometry_issues": EXPECTED["geometry_issues"],
        }

        # Distribution of raw feature geometry_issue values.
        issue_value_counts = rows(cur, f"""
            SELECT CAST(geometry_issue AS CHAR) AS geometry_issue, COUNT(*) AS n
            FROM chart_feature
            WHERE dataset_id=%s AND {issue_pred}
            GROUP BY CAST(geometry_issue AS CHAR)
            ORDER BY n DESC, geometry_issue
        """, (args.dataset_id,))
        write_csv(out / "geometry_issue_raw_value_counts.csv", issue_value_counts)

        # Import events related to geometry, including raw event_code/message patterns.
        geom_events = rows(cur, """
            SELECT e.id, e.job_id, e.cell_id, e.level, e.event_code, e.message, e.details_json, e.created_at,
                   c.cell_name, c.source_relative_path
            FROM chart_import_event e
            LEFT JOIN chart_import_job j ON j.id=e.job_id
            LEFT JOIN chart_cell c ON c.id=e.cell_id
            WHERE j.dataset_id=%s
              AND (
                  LOWER(COALESCE(e.event_code,'')) LIKE '%%geom%%'
                  OR LOWER(COALESCE(e.message,'')) LIKE '%%geom%%'
                  OR LOWER(COALESCE(e.message,'')) LIKE '%%ring%%'
                  OR LOWER(COALESCE(e.message,'')) LIKE '%%polygon%%'
                  OR LOWER(COALESCE(e.message,'')) LIKE '%%line%%'
                  OR LOWER(COALESCE(e.message,'')) LIKE '%%point%%'
                  OR LOWER(COALESCE(e.message,'')) LIKE '%%topolog%%'
              )
            ORDER BY e.id
        """, (args.dataset_id,))
        write_csv(out / "geometry_related_import_events.csv", geom_events)

        # ------------------------------------------------------------------
        # Pull exact issue feature rows. Preserve raw geometry_issue and enough context.
        # ------------------------------------------------------------------
        print("[5/8] Classifying feature.geometry_issue rows")
        issue_rows = rows(cur, f"""
            SELECT
                f.id AS feature_id,
                f.dataset_id,
                f.cell_id,
                c.cell_name,
                c.source_relative_path,
                f.coverage_id,
                f.source_record_index,
                f.object_class_id,
                f.object_class,
                f.geometry_type,
                f.raw_geotype,
                f.scale_code,
                f.native_scale,
                f.geometry_repaired,
                CAST(f.geometry_issue AS CHAR) AS geometry_issue,
                f.bbox_min_lon, f.bbox_min_lat, f.bbox_max_lon, f.bbox_max_lat,
                CASE WHEN pg.feature_id IS NOT NULL THEN 1 ELSE 0 END AS has_point_geometry,
                CASE WHEN lg.feature_id IS NOT NULL THEN 1 ELSE 0 END AS has_line_geometry,
                CASE WHEN ag.feature_id IS NOT NULL THEN 1 ELSE 0 END AS has_area_geometry,
                (SELECT COUNT(*) FROM chart_sounding_point sp WHERE sp.feature_id=f.id) AS sounding_point_count
            FROM chart_feature f
            JOIN chart_cell c ON c.id=f.cell_id
            LEFT JOIN chart_point_geometry pg ON pg.feature_id=f.id
            LEFT JOIN chart_line_geometry lg ON lg.feature_id=f.id
            LEFT JOIN chart_area_geometry ag ON ag.feature_id=f.id
            WHERE f.dataset_id=%s AND {issue_pred.replace(qi('geometry_issue'), 'f.'+qi('geometry_issue'))}
            ORDER BY f.id
        """, (args.dataset_id,))

        for r in issue_rows:
            raw = normalize_issue_value(r.get("geometry_issue"))
            r["issue_normalized"] = raw
            r["issue_category"] = classify_category(raw)
            r["issue_stage"] = classify_stage(raw)
            has_geom = bool(r.get("has_point_geometry") or r.get("has_line_geometry") or r.get("has_area_geometry") or int(r.get("sounding_point_count") or 0) > 0)
            r["has_any_geometry_storage"] = 1 if has_geom else 0

            # Conservative evidence label. Do not claim importer bug from text alone.
            if r["issue_category"] == "OTHER":
                r["assessment"] = "NEEDS_TEXT_REVIEW"
            elif r["issue_category"] in {"TOPOLOGY_MISMATCH", "MULTIPART_ASSEMBLY", "CONVERSION_EXCEPTION"}:
                r["assessment"] = "IMPORTER_OR_SOURCE_CANDIDATE"
            else:
                r["assessment"] = "GEOMETRY_DATA_OR_ASSEMBLY_ISSUE"

        write_csv(out / "geometry_issues_classified_v2.csv", issue_rows)

        # Counts by category/stage/object/cell/raw issue text.
        by_cat = Counter(r["issue_category"] for r in issue_rows)
        by_stage = Counter(r["issue_stage"] for r in issue_rows)
        by_cross = Counter((r["issue_stage"], r["issue_category"]) for r in issue_rows)
        by_obj = Counter((text(r.get("object_class_id")), text(r.get("object_class"))) for r in issue_rows)
        by_cell = Counter((text(r.get("cell_name")), text(r.get("source_relative_path"))) for r in issue_rows)
        by_raw = Counter(r["issue_normalized"] for r in issue_rows)

        issue_summary = {
            "feature_issue_rows_loaded": len(issue_rows),
            "category_sum": sum(by_cat.values()),
            "other_count": by_cat.get("OTHER", 0),
            "by_category": [{"category": k, "count": v} for k, v in by_cat.most_common()],
            "by_stage": [{"stage": k, "count": v} for k, v in by_stage.most_common()],
            "by_stage_category": [{"stage": k[0], "category": k[1], "count": v} for k, v in by_cross.most_common()],
            "by_object_class_top50": [{"object_class_id": k[0], "object_class": k[1], "count": v} for k, v in by_obj.most_common(50)],
            "by_cell_top50": [{"cell_name": k[0], "source_relative_path": k[1], "count": v} for k, v in by_cell.most_common(50)],
            "raw_issue_top100": [{"geometry_issue": k, "count": v} for k, v in by_raw.most_common(100)],
        }
        report["issue_summary"] = issue_summary
        write_csv(out / "geometry_issue_by_category.csv", issue_summary["by_category"])
        write_csv(out / "geometry_issue_by_stage.csv", issue_summary["by_stage"])
        write_csv(out / "geometry_issue_stage_category.csv", issue_summary["by_stage_category"])
        write_csv(out / "geometry_issue_object_class_top50.csv", issue_summary["by_object_class_top50"])
        write_csv(out / "geometry_issue_cell_top50.csv", issue_summary["by_cell_top50"])
        write_csv(out / "geometry_issue_raw_top100.csv", issue_summary["raw_issue_top100"])

        # Samples: up to 10 per category, prioritizing different object/cell/raw combinations.
        samples = []
        seen_per_cat: Dict[str, set] = defaultdict(set)
        for r in issue_rows:
            cat = r["issue_category"]
            sig = (r.get("object_class_id"), r.get("cell_id"), r.get("issue_normalized"))
            if len(seen_per_cat[cat]) < 10 and sig not in seen_per_cat[cat]:
                samples.append(r)
                seen_per_cat[cat].add(sig)
        write_csv(out / "geometry_issue_samples_v2.csv", samples)

        # ------------------------------------------------------------------
        # Geometry storage audit: true spatial columns live in separate tables.
        # ------------------------------------------------------------------
        print("[6/8] Spatial geometry storage audit")
        storage = {}
        for name, table in [("POINT", "chart_point_geometry"), ("LINE", "chart_line_geometry"), ("AREA", "chart_area_geometry")]:
            item: Dict[str, Any] = {}
            item["row_count"] = int(scalar(cur, f"""
                SELECT COUNT(*) FROM {table} g
                JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.dataset_id=%s
            """, (args.dataset_id,)))
            item["feature_count"] = int(scalar(cur, f"""
                SELECT COUNT(DISTINCT g.feature_id) FROM {table} g
                JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.dataset_id=%s
            """, (args.dataset_id,)))
            item["null_geom"] = int(scalar(cur, f"""
                SELECT COUNT(*) FROM {table} g
                JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.dataset_id=%s AND g.geom IS NULL
            """, (args.dataset_id,)))

            # Spatial functions may fail on a bad row. Capture rather than abort entire audit.
            st = safe(cur, f"""
                SELECT
                  ST_GeometryType(g.geom) AS geometry_type,
                  COUNT(*) AS n,
                  SUM(CASE WHEN ST_IsEmpty(g.geom) THEN 1 ELSE 0 END) AS empty_count,
                  SUM(CASE WHEN ST_IsValid(g.geom) THEN 1 ELSE 0 END) AS valid_count,
                  SUM(CASE WHEN NOT ST_IsValid(g.geom) THEN 1 ELSE 0 END) AS invalid_count
                FROM {table} g
                JOIN chart_feature f ON f.id=g.feature_id
                WHERE f.dataset_id=%s AND g.geom IS NOT NULL
                GROUP BY ST_GeometryType(g.geom)
                ORDER BY n DESC
            """, (args.dataset_id,))
            item["spatial_function_result"] = st
            storage[name] = item

        # Sounding points have their own geom.
        sounding_item: Dict[str, Any] = {
            "row_count": report["counts"]["sounding_points"],
            "null_geom": int(scalar(cur, """
                SELECT COUNT(*) FROM chart_sounding_point s
                JOIN chart_feature f ON f.id=s.feature_id
                WHERE f.dataset_id=%s AND s.geom IS NULL
            """, (args.dataset_id,))),
        }
        sounding_item["spatial_function_result"] = safe(cur, """
            SELECT ST_GeometryType(s.geom) AS geometry_type, COUNT(*) AS n,
                   SUM(CASE WHEN ST_IsEmpty(s.geom) THEN 1 ELSE 0 END) AS empty_count,
                   SUM(CASE WHEN ST_IsValid(s.geom) THEN 1 ELSE 0 END) AS valid_count,
                   SUM(CASE WHEN NOT ST_IsValid(s.geom) THEN 1 ELSE 0 END) AS invalid_count
            FROM chart_sounding_point s
            JOIN chart_feature f ON f.id=s.feature_id
            WHERE f.dataset_id=%s AND s.geom IS NOT NULL
            GROUP BY ST_GeometryType(s.geom)
        """, (args.dataset_id,))
        storage["SOUNDING"] = sounding_item
        report["geometry_storage"] = storage

        # Geometry type distribution from feature semantic field, separate from spatial columns.
        geom_type_dist = rows(cur, """
            SELECT geometry_type, raw_geotype, COUNT(*) AS n,
                   SUM(CASE WHEN geometry_repaired<>0 THEN 1 ELSE 0 END) AS repaired_n,
                   SUM(CASE WHEN geometry_issue IS NOT NULL AND TRIM(CAST(geometry_issue AS CHAR))<>'' THEN 1 ELSE 0 END) AS issue_n
            FROM chart_feature
            WHERE dataset_id=%s
            GROUP BY geometry_type, raw_geotype
            ORDER BY n DESC
        """, (args.dataset_id,))
        write_csv(out / "feature_geometry_type_distribution.csv", geom_type_dist)

        # ------------------------------------------------------------------
        # Cell-level issue rates.
        # ------------------------------------------------------------------
        print("[7/8] Cell issue rates")
        cell_issue_rates = rows(cur, f"""
            SELECT c.id AS cell_id, c.cell_name, c.source_relative_path, c.feature_count,
                   COUNT(f.id) AS feature_rows,
                   SUM(CASE WHEN {issue_pred.replace(qi('geometry_issue'), 'f.'+qi('geometry_issue'))} THEN 1 ELSE 0 END) AS issue_count,
                   ROUND(100.0 * SUM(CASE WHEN {issue_pred.replace(qi('geometry_issue'), 'f.'+qi('geometry_issue'))} THEN 1 ELSE 0 END) / NULLIF(COUNT(f.id),0), 6) AS issue_rate_pct
            FROM chart_cell c
            LEFT JOIN chart_feature f ON f.cell_id=c.id AND f.dataset_id=c.dataset_id
            WHERE c.dataset_id=%s
            GROUP BY c.id, c.cell_name, c.source_relative_path, c.feature_count
            ORDER BY issue_count DESC, c.id
        """, (args.dataset_id,))
        write_csv(out / "geometry_issue_cell_rates_all.csv", cell_issue_rates)
        write_csv(out / "geometry_issue_cell_top30_count.csv", cell_issue_rates[:30])
        write_csv(out / "geometry_issue_cell_top30_rate.csv", sorted(cell_issue_rates, key=lambda r: float(r.get("issue_rate_pct") or 0), reverse=True)[:30])

        # ------------------------------------------------------------------
        # Final report and acceptance checks.
        # ------------------------------------------------------------------
        print("[8/8] Writing report")
        report["acceptance"] = {
            "cells_match": report["counts"]["cells"] == EXPECTED["cells"],
            "features_match": report["counts"]["features"] == EXPECTED["features"],
            "attributes_match": report["counts"]["attributes"] == EXPECTED["attributes"],
            "sounding_points_match": report["counts"]["sounding_points"] == EXPECTED["sounding_points"],
            "feature_issue_rows_match_4406": len(issue_rows) == EXPECTED["geometry_issues"],
            "category_sum_equals_issue_rows": sum(by_cat.values()) == len(issue_rows),
            "no_feature_cell_orphans": report["relationships"]["feature_to_cell_orphan"] == 0,
            "no_attribute_orphans": report["relationships"]["attribute_to_feature_orphan_global"] == 0,
            "no_sounding_orphans": report["relationships"]["sounding_to_feature_orphan_global"] == 0,
        }

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

        lines = []
        lines.append("CM93 -> MySQL Dataset ID=1 Audit v2")
        lines.append("=" * 78)
        lines.append(f"Generated: {report['generated_at']}")
        lines.append(f"Database: {args.database}")
        lines.append(f"Dataset ID: {args.dataset_id}")
        lines.append("Mode: READ ONLY / NO PBF")
        lines.append("")
        lines.append("COUNTS")
        for k, v in report["counts"].items():
            lines.append(f"  {k:30s} {v}")
        lines.append("")
        lines.append("RELATIONSHIPS")
        for k, v in report["relationships"].items():
            lines.append(f"  {k:40s} {v}")
        lines.append("")
        lines.append("GEOMETRY ISSUE RECONCILIATION")
        for k, v in report["geometry_issue_reconciliation"].items():
            lines.append(f"  {k:42s} {v}")
        lines.append("")
        lines.append("ISSUE CATEGORY")
        for r in issue_summary["by_category"]:
            pct = (100.0 * r["count"] / len(issue_rows)) if issue_rows else 0
            lines.append(f"  {r['category']:30s} {r['count']:7d}  {pct:8.3f}%")
        lines.append("")
        lines.append("ISSUE STAGE")
        for r in issue_summary["by_stage"]:
            pct = (100.0 * r["count"] / len(issue_rows)) if issue_rows else 0
            lines.append(f"  {r['stage']:20s} {r['count']:7d}  {pct:8.3f}%")
        lines.append("")
        lines.append("TOP RAW ISSUE VALUES")
        for r in issue_summary["raw_issue_top100"][:30]:
            s = r["geometry_issue"].replace("\n", " ")
            lines.append(f"  {r['count']:7d}  {s[:180]}")
        lines.append("")
        lines.append("ACCEPTANCE")
        for k, v in report["acceptance"].items():
            lines.append(f"  {k:42s} {'PASS' if v else 'CHECK'}")
        lines.append("")
        lines.append("IMPORTANT")
        lines.append("  - category_sum must equal feature_issue_rows_loaded.")
        lines.append("  - OTHER is not considered fully understood; inspect raw top values/samples.")
        lines.append("  - No repair or database mutation was performed.")
        lines.append("  - PBF remains out of scope.")
        (out / "audit_summary_v2.txt").write_text("\n".join(lines), encoding="utf-8")

        conn.rollback()
        print("\n".join(lines))
        print(f"\nOutput: {out.resolve()}")
        return 0

    except Exception:
        if conn:
            try:
                conn.rollback()
            except Exception:
                pass
        err = traceback.format_exc()
        print(err)
        try:
            (out / "fatal_error.txt").write_text(err, encoding="utf-8")
        except Exception:
            pass
        return 1
    finally:
        if conn:
            try:
                conn.close()
            except Exception:
                pass


if __name__ == "__main__":
    sys.exit(main())