CM95/cm93_mysql_audit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CM93 -> MySQL Dataset Audit + Geometry Issue Classification
v1.0

目标：
  1. 审计 Dataset ID=1
  2. 分类 geometry issue
  3. 导出 CSV / JSON / TXT

安全：
  - 只读事务
  - 不执行 INSERT / UPDATE / DELETE / DDL
  - 不修 geometry
  - 不涉及 PBF

依赖：
  pip install pymysql

示例：
  python cm93_mysql_audit.py --host 127.0.0.1 --port 3306 --user root --database YOUR_DB --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 dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple

try:
    import pymysql
    from pymysql.cursors import DictCursor
except ImportError:
    print("缺少 pymysql，请执行: pip install pymysql")
    sys.exit(2)


CATEGORY_RULES: List[Tuple[str, List[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"\bwkb\b", r"\bwkt\b", r"parse exception", r"conversion exception",
        r"geometry exception", r"failed to convert", r"cannot convert",
        r"invalid gis data", r"data truncation", r"spatial.*exception"
    ]),
]

STAGE_RULES: List[Tuple[str, List[str]]] = [
    ("MYSQL_WRITE", [r"mysql", r"insert", r"database", r"spatial", r"\bwkb\b", r"\bwkt\b"]),
    ("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"binary", r"record"]),
    ("SOURCE", [r"source", r"raw", r"cm93"]),
]


@dataclass
class TableInfo:
    name: str
    columns: List[str]


def qi(name: str) -> str:
    return "`" + name.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 score_name(name: str, candidates: Sequence[str]) -> int:
    n = name.lower()
    best = 0
    for c in candidates:
        c = c.lower()
        if n == c:
            best = max(best, 100)
        elif n.endswith("_" + c) or n.startswith(c + "_"):
            best = max(best, 45)
        elif c in n:
            best = max(best, 10)
    return best


def best_col(cols: Sequence[str], candidates: Sequence[str]) -> Optional[str]:
    ranked = sorted(((score_name(c, candidates), c) for c in cols), reverse=True)
    return ranked[0][1] if ranked and ranked[0][0] > 0 else None


def best_table(tables: List[TableInfo], candidates: Sequence[str], must_have: Sequence[str] = ()) -> Optional[TableInfo]:
    ranked = []
    for t in tables:
        if must_have and not all(any(x.lower() in c.lower() for c in t.columns) for x in must_have):
            continue
        ranked.append((score_name(t.name, candidates), t.name, t))
    ranked.sort(reverse=True)
    return ranked[0][2] if ranked and ranked[0][0] > 0 else None


def discover_tables(cur, db: str) -> List[TableInfo]:
    cur.execute("""
        SELECT TABLE_NAME, COLUMN_NAME
        FROM information_schema.COLUMNS
        WHERE TABLE_SCHEMA=%s
        ORDER BY TABLE_NAME, ORDINAL_POSITION
    """, (db,))
    grouped: Dict[str, List[str]] = defaultdict(list)
    for r in cur.fetchall():
        grouped[r["TABLE_NAME"]].append(r["COLUMN_NAME"])
    return [TableInfo(k, v) for k, v in grouped.items()]


def resolve_schema(tables: List[TableInfo]) -> Dict[str, Optional[TableInfo]]:
    issue_candidates = []
    for t in tables:
        n = t.name.lower()
        if ("geometry" in n or "geom" in n) and ("issue" in n or "error" in n):
            issue_candidates.append(t)
    issue_candidates.sort(key=lambda t: (len(t.name), t.name))

    return {
        "dataset": best_table(tables, ["datasets", "dataset"]),
        "cell": best_table(tables, ["cells", "cell"]),
        "feature": best_table(tables, ["features", "feature"]),
        "attribute": best_table(tables, ["feature_attributes", "feature_attribute", "attributes", "attribute"]),
        "sounding": best_table(tables, ["sounding_points", "sounding_point", "soundings", "sounding"]),
        "geometry_issue": issue_candidates[0] if issue_candidates else best_table(tables, ["geometry_issues", "geometry_issue", "geom_issue"]),
    }


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


def dataset_col(t: Optional[TableInfo]) -> Optional[str]:
    if not t:
        return None
    return best_col(t.columns, ["dataset_id", "datasetid", "dataset"])


def count_table(cur, t: Optional[TableInfo], dataset_id: int) -> Dict[str, Any]:
    if not t:
        return {"table": None, "count": None, "dataset_column": None, "error": "table not found"}
    ds = dataset_col(t)
    try:
        if ds:
            n = scalar(cur, f"SELECT COUNT(*) AS n FROM {qi(t.name)} WHERE {qi(ds)}=%s", (dataset_id,))
        else:
            n = scalar(cur, f"SELECT COUNT(*) AS n FROM {qi(t.name)}")
        return {"table": t.name, "count": int(n), "dataset_column": ds, "error": None}
    except Exception as e:
        return {"table": t.name, "count": None, "dataset_column": ds, "error": str(e)}


def orphan_count(cur, child: Optional[TableInfo], parent: Optional[TableInfo], child_fk_names, parent_pk_names, dataset_id: int) -> Dict[str, Any]:
    out = {
        "child": child.name if child else None,
        "parent": parent.name if parent else None,
        "child_fk": None,
        "parent_pk": None,
        "count": None,
        "error": None,
    }
    if not child or not parent:
        out["error"] = "table not found"
        return out

    cfk = best_col(child.columns, child_fk_names)
    ppk = best_col(parent.columns, parent_pk_names)
    cds = dataset_col(child)
    out["child_fk"] = cfk
    out["parent_pk"] = ppk
    if not cfk or not ppk:
        out["error"] = "required columns not found"
        return out

    try:
        where = [f"c.{qi(cfk)} IS NOT NULL", f"p.{qi(ppk)} IS NULL"]
        params = []
        if cds:
            where.insert(0, f"c.{qi(cds)}=%s")
            params.append(dataset_id)
        sql = f"""
            SELECT COUNT(*) AS n
            FROM {qi(child.name)} c
            LEFT JOIN {qi(parent.name)} p
              ON c.{qi(cfk)} = p.{qi(ppk)}
            WHERE {' AND '.join(where)}
        """
        out["count"] = int(scalar(cur, sql, params))
    except Exception as e:
        out["error"] = str(e)
    return out


def geometry_audit(cur, feature: Optional[TableInfo], dataset_id: int) -> Dict[str, Any]:
    out: Dict[str, Any] = {
        "table": feature.name if feature else None,
        "geometry_column": None,
        "null_count": None,
        "empty_count": None,
        "valid_count": None,
        "invalid_count": None,
        "types": {},
        "errors": [],
    }
    if not feature:
        out["errors"].append("feature table not found")
        return out

    geom = best_col(feature.columns, ["geometry", "geom", "shape"])
    ds = dataset_col(feature)
    out["geometry_column"] = geom
    if not geom:
        out["errors"].append("geometry column not found")
        return out

    params = [dataset_id] if ds else []
    base = f"{qi(ds)}=%s" if ds else "1=1"

    try:
        out["null_count"] = int(scalar(cur,
            f"SELECT COUNT(*) AS n FROM {qi(feature.name)} WHERE {base} AND {qi(geom)} IS NULL", params))
    except Exception as e:
        out["errors"].append(f"NULL count failed: {e}")

    try:
        cur.execute(f"""
            SELECT ST_GeometryType({qi(geom)}) AS geometry_type, COUNT(*) AS n
            FROM {qi(feature.name)}
            WHERE {base} AND {qi(geom)} IS NOT NULL
            GROUP BY ST_GeometryType({qi(geom)})
            ORDER BY n DESC
        """, params)
        out["types"] = {text(r["geometry_type"] or "NULL"): int(r["n"]) for r in cur.fetchall()}
    except Exception as e:
        out["errors"].append(f"ST_GeometryType failed: {e}")

    try:
        cur.execute(f"""
            SELECT
                SUM(CASE WHEN ST_IsEmpty({qi(geom)}) THEN 1 ELSE 0 END) AS empty_count,
                SUM(CASE WHEN ST_IsValid({qi(geom)}) THEN 1 ELSE 0 END) AS valid_count,
                SUM(CASE WHEN NOT ST_IsValid({qi(geom)}) THEN 1 ELSE 0 END) AS invalid_count
            FROM {qi(feature.name)}
            WHERE {base} AND {qi(geom)} IS NOT NULL
        """, params)
        r = cur.fetchone() or {}
        out["empty_count"] = int(r.get("empty_count") or 0)
        out["valid_count"] = int(r.get("valid_count") or 0)
        out["invalid_count"] = int(r.get("invalid_count") or 0)
    except Exception as e:
        out["errors"].append(f"ST_IsEmpty/ST_IsValid failed: {e}")

    return out


def issue_text(row: Dict[str, Any]) -> str:
    first = []
    rest = []
    for k, v in row.items():
        if v is None:
            continue
        s = text(v).strip()
        if not s:
            continue
        kl = k.lower()
        part = f"{k}={s}"
        if any(x in kl for x in ["issue", "error", "reason", "message", "detail", "exception", "stage", "type", "status"]):
            first.append(part)
        else:
            rest.append(part)
    return " | ".join(first + rest)


def classify_issue(s: str) -> str:
    for category, patterns in CATEGORY_RULES:
        for p in patterns:
            if re.search(p, s, re.I):
                return category
    return "OTHER"


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


def fetch_issues(cur, issue: Optional[TableInfo], feature: Optional[TableInfo], cell: Optional[TableInfo], dataset_id: int, max_rows: int):
    meta = {
        "table": issue.name if issue else None,
        "dataset_column": None,
        "feature_fk": None,
        "joined_feature": False,
        "joined_cell": False,
        "error": None,
    }
    if not issue:
        meta["error"] = "geometry issue table not found"
        return [], meta

    ids = dataset_col(issue)
    ifk = best_col(issue.columns, ["feature_id", "featureid", "feature"])
    meta["dataset_column"] = ids
    meta["feature_fk"] = ifk

    joins = []
    extras = []
    where = []
    params = []

    fpk = best_col(feature.columns, ["id", "feature_id", "featureid"]) if feature else None
    fds = dataset_col(feature)
    fcell = best_col(feature.columns, ["cell_id", "cellid", "cell"]) if feature else None

    if feature and ifk and fpk:
        joins.append(f"LEFT JOIN {qi(feature.name)} f ON i.{qi(ifk)} = f.{qi(fpk)}")
        meta["joined_feature"] = True
        for c in feature.columns:
            cl = c.lower()
            if any(x in cl for x in ["object", "class", "type", "geom", "cell", "feature"]):
                extras.append(f"f.{qi(c)} AS {qi('feature__' + c)}")

        if cell and fcell:
            cpk = best_col(cell.columns, ["id", "cell_id", "cellid"])
            if cpk:
                joins.append(f"LEFT JOIN {qi(cell.name)} c ON f.{qi(fcell)} = c.{qi(cpk)}")
                meta["joined_cell"] = True
                for cc in cell.columns:
                    ccl = cc.lower()
                    if any(x in ccl for x in ["name", "cell", "file", "path"]):
                        extras.append(f"c.{qi(cc)} AS {qi('cell__' + cc)}")

    if ids:
        where.append(f"i.{qi(ids)}=%s")
        params.append(dataset_id)
    elif meta["joined_feature"] and fds:
        where.append(f"f.{qi(fds)}=%s")
        params.append(dataset_id)

    select_extra = (", " + ", ".join(extras)) if extras else ""
    sql = f"""
        SELECT i.*{select_extra}
        FROM {qi(issue.name)} i
        {' '.join(joins)}
        {'WHERE ' + ' AND '.join(where) if where else ''}
        LIMIT {int(max_rows)}
    """
    try:
        cur.execute(sql, params)
        return list(cur.fetchall()), meta
    except Exception as e:
        meta["error"] = str(e)
        return [], meta


def find_object_class(row: Dict[str, Any]) -> str:
    for k, v in row.items():
        kl = k.lower()
        if v is not None and "object" in kl and ("class" in kl or "type" in kl or "code" in kl or "name" in kl):
            return text(v)
    for k, v in row.items():
        if v is not None and "class" in k.lower():
            return text(v)
    return ""


def find_cell(row: Dict[str, Any]) -> str:
    for k, v in row.items():
        kl = k.lower()
        if v is not None and kl.startswith("cell__") and any(x in kl for x in ["name", "file", "path", "cell"]):
            return text(v)
    for k, v in row.items():
        if v is not None and "cell" in k.lower():
            return text(v)
    return ""


def enrich(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    out = []
    for r in rows:
        s = issue_text(r)
        x = dict(r)
        x["_issue_text"] = s
        x["_issue_category"] = classify_issue(s)
        x["_issue_stage"] = classify_stage(s)
        x["_object_class"] = find_object_class(r)
        x["_cell"] = find_cell(r)
        out.append(x)
    return out


def summarize(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
    by_cat = Counter(r["_issue_category"] for r in rows)
    by_stage = Counter(r["_issue_stage"] for r in rows)
    cross = Counter((r["_issue_stage"], r["_issue_category"]) for r in rows)
    obj = Counter(r["_object_class"] for r in rows if r["_object_class"])
    cell = Counter(r["_cell"] for r in rows if r["_cell"])
    return {
        "total": len(rows),
        "by_category": dict(by_cat.most_common()),
        "by_stage": dict(by_stage.most_common()),
        "by_stage_category": [
            {"stage": s, "category": c, "count": n}
            for (s, c), n in cross.most_common()
        ],
        "by_object_class_top20": [
            {"object_class": k, "count": v} for k, v in obj.most_common(20)
        ],
        "by_cell_top30": [
            {"cell": k, "count": v} for k, v in cell.most_common(30)
        ],
    }


def duplicate_audit(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
    groups = defaultdict(int)
    for r in rows:
        parts = []
        for k, v in r.items():
            kl = k.lower()
            if v is not None and any(x in kl for x in ["feature_id", "featureid", "reason", "message", "detail", "exception", "stage", "issue_type", "error_type"]):
                parts.append(f"{k}={text(v)}")
        if parts:
            groups[tuple(parts)] += 1
    dup = [(k, n) for k, n in groups.items() if n > 1]
    dup.sort(key=lambda x: x[1], reverse=True)
    return {
        "duplicate_groups": len(dup),
        "duplicate_rows_beyond_first": sum(n - 1 for _, n in dup),
        "examples": [{"count": n, "identity": " | ".join(k)[:1500]} for k, n in dup[:20]],
    }


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


def make_samples(rows: List[Dict[str, Any]], n=5) -> List[Dict[str, Any]]:
    buckets = defaultdict(list)
    for r in rows:
        cat = r["_issue_category"]
        if len(buckets[cat]) < n:
            buckets[cat].append(r)
    result = []
    for cat in sorted(buckets):
        result.extend(buckets[cat])
    return result


def parse_args():
    p = argparse.ArgumentParser(description="CM93 MySQL Dataset audit + geometry issue classification")
    p.add_argument("--host", default=os.getenv("MYSQL_HOST", "127.0.0.1"))
    p.add_argument("--port", type=int, default=int(os.getenv("MYSQL_PORT", "3306")))
    p.add_argument("--user", default=os.getenv("MYSQL_USER", "root"))
    p.add_argument("--password", default=os.getenv("MYSQL_PASSWORD"))
    p.add_argument("--database", default=os.getenv("MYSQL_DATABASE"))
    p.add_argument("--dataset-id", type=int, default=1)
    p.add_argument("--output", default=None)
    p.add_argument("--max-issues", type=int, default=1000000)
    return p.parse_args()


def main():
    a = parse_args()
    if not a.database:
        a.database = input("MySQL database name: ").strip()
    if not a.password:
        a.password = getpass.getpass("MySQL password: ")

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

    report: Dict[str, Any] = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "database": a.database,
        "dataset_id": a.dataset_id,
        "read_only": True,
        "schema": {},
        "counts": {},
        "relationships": {},
        "geometry": {},
        "issues": {},
        "warnings": [],
    }

    conn = None
    try:
        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=300,
            write_timeout=30,
        )

        with conn.cursor() as cur:
            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")

            print("[1/7] 扫描 schema...")
            tables = discover_tables(cur, a.database)
            schema = resolve_schema(tables)
            report["schema"]["resolved_tables"] = {k: (v.name if v else None) for k, v in schema.items()}
            report["schema"]["all_tables"] = {t.name: t.columns for t in tables}
            for k, v in report["schema"]["resolved_tables"].items():
                print(f"  {k:16s}: {v}")

            print("[2/7] 数量审计...")
            for k in ["dataset", "cell", "feature", "attribute", "sounding", "geometry_issue"]:
                report["counts"][k] = count_table(cur, schema[k], a.dataset_id)

            print("[3/7] 关系完整性审计...")
            report["relationships"]["feature_to_cell"] = orphan_count(
                cur, schema["feature"], schema["cell"],
                ["cell_id", "cellid", "cell"], ["id", "cell_id", "cellid"], a.dataset_id)
            report["relationships"]["attribute_to_feature"] = orphan_count(
                cur, schema["attribute"], schema["feature"],
                ["feature_id", "featureid", "feature"], ["id", "feature_id", "featureid"], a.dataset_id)
            report["relationships"]["sounding_to_feature"] = orphan_count(
                cur, schema["sounding"], schema["feature"],
                ["feature_id", "featureid", "feature"], ["id", "feature_id", "featureid"], a.dataset_id)
            report["relationships"]["issue_to_feature"] = orphan_count(
                cur, schema["geometry_issue"], schema["feature"],
                ["feature_id", "featureid", "feature"], ["id", "feature_id", "featureid"], a.dataset_id)

            print("[4/7] Geometry 基础审计...")
            report["geometry"] = geometry_audit(cur, schema["feature"], a.dataset_id)

            print("[5/7] 读取 geometry issues...")
            raw, meta = fetch_issues(cur, schema["geometry_issue"], schema["feature"], schema["cell"], a.dataset_id, a.max_issues)
            report["issues"]["meta"] = meta
            print(f"  loaded={len(raw)}")

            print("[6/7] 分类 geometry issues...")
            rows = enrich(raw)
            sm = summarize(rows)
            dup = duplicate_audit(rows)
            report["issues"]["summary"] = sm
            report["issues"]["duplicates"] = dup

            write_csv(outdir / "geometry_issues_classified.csv", rows)
            write_csv(outdir / "geometry_issue_samples.csv", make_samples(rows, 5))
            write_csv(outdir / "geometry_issue_stage_category.csv", sm["by_stage_category"], ["stage", "category", "count"])
            write_csv(outdir / "geometry_issue_object_class_top20.csv", sm["by_object_class_top20"], ["object_class", "count"])
            write_csv(outdir / "geometry_issue_cell_top30.csv", sm["by_cell_top30"], ["cell", "count"])

            expected = {
                "cell": 784,
                "feature": 322465,
                "attribute": 405153,
                "sounding": 275700,
                "geometry_issue": 4406,
            }

            lines = []
            lines.append("CM93 -> MySQL Dataset Audit + Geometry Issue Classification")
            lines.append("=" * 72)
            lines.append(f"Generated: {report['generated_at']}")
            lines.append(f"Database: {a.database}")
            lines.append(f"Dataset ID: {a.dataset_id}")
            lines.append("")
            lines.append("Resolved tables:")
            for k, v in report["schema"]["resolved_tables"].items():
                lines.append(f"  {k:16s}: {v}")
            lines.append("")
            lines.append("Counts:")
            for k, v in report["counts"].items():
                lines.append(f"  {k:16s}: count={v['count']} table={v['table']} dataset_col={v['dataset_column']} error={v['error']}")
            lines.append("")
            lines.append("Relationships:")
            for k, v in report["relationships"].items():
                lines.append(f"  {k:24s}: orphan={v['count']} error={v['error']}")
            lines.append("")
            lines.append("Geometry audit:")
            lines.append(json.dumps(report["geometry"], ensure_ascii=False, indent=2, default=str))
            lines.append("")
            lines.append(f"Geometry issues loaded: {sm['total']}")
            lines.append("By category:")
            for k, n in sm["by_category"].items():
                pct = n / sm["total"] * 100 if sm["total"] else 0
                lines.append(f"  {k:28s} {n:8d} {pct:7.3f}%")
            lines.append("")
            lines.append("By stage:")
            for k, n in sm["by_stage"].items():
                pct = n / sm["total"] * 100 if sm["total"] else 0
                lines.append(f"  {k:16s} {n:8d} {pct:7.3f}%")
            lines.append("")
            lines.append(f"Duplicate groups: {dup['duplicate_groups']}")
            lines.append(f"Duplicate rows beyond first: {dup['duplicate_rows_beyond_first']}")
            lines.append("")
            lines.append("Acceptance checks:")
            for k, exp in expected.items():
                actual = report["counts"].get(k, {}).get("count")
                lines.append(f"  {k:16s}: actual={actual} expected={exp} => {'PASS' if actual == exp else 'CHECK'}")
            category_sum = sum(sm["by_category"].values())
            lines.append(f"  issue loaded total: actual={sm['total']} expected=4406 => {'PASS' if sm['total'] == 4406 else 'CHECK'}")
            lines.append(f"  category sum:       actual={category_sum} expected={sm['total']} => {'PASS' if category_sum == sm['total'] else 'FAIL'}")
            lines.append(f"  OTHER count:        {sm['by_category'].get('OTHER', 0)}")
            lines.append("")
            lines.append("说明：")
            lines.append("  1. 本脚本仅根据数据库现有 schema、issue 文本和空间字段进行只读审计。")
            lines.append("  2. OTHER 表示现有文本证据不足，不表示该 issue 可以忽略。")
            lines.append("  3. 本脚本没有修复 geometry，也没有修改任何数据库记录。")
            lines.append("  4. 不包含任何 PBF 逻辑。")

            (outdir / "audit_summary.txt").write_text("\n".join(lines), encoding="utf-8")
            (outdir / "audit_report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2, default=str), encoding="utf-8")

            conn.rollback()

            print("[7/7] 完成")
            print("")
            print("\n".join(lines[-18:]))
            print("")
            print(f"输出目录: {outdir.resolve()}")

    except KeyboardInterrupt:
        if conn:
            try:
                conn.rollback()
            except Exception:
                pass
        print("已中断。")
        sys.exit(130)
    except Exception:
        if conn:
            try:
                conn.rollback()
            except Exception:
                pass
        err = traceback.format_exc()
        print(err)
        try:
            (outdir / "fatal_error.txt").write_text(err, encoding="utf-8")
        except Exception:
            pass
        sys.exit(1)
    finally:
        if conn:
            try:
                conn.close()
            except Exception:
                pass


if __name__ == "__main__":
    main()