#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CM93 -> MySQL 非 Geometry 数据完整性全库最终验收 v1

目标
====
在 Geometry normalization 已通过后，对同一 Dataset 的“非 Geometry”数据链做最终只读验收。

检查：
1. Dataset / Cell 完整性
2. 每个 Cell 的 feature_count / attribute_count 与实际落库一致
3. Feature 的 dataset/cell/source_record_index 可追溯性
4. Feature source_record_index 在 Cell 内连续、唯一
5. Attribute 外键、ordinal、dictionary 映射、类型值通道
6. Coverage 与 _m_sor Feature 一一对应且同 Cell
7. Sounding 外键、point_index 连续性、depth 公式
8. raw_feature_blob / raw_attribute_blob 保留
9. chart_raw_record 分块索引、chunk_count、total_bytes、JSON metadata
10. Object / Attribute dictionary 映射完整性
11. attributes_json 可解析；_cm93_related_indices 若存在，必须指向同 Cell Feature
12. 可选：源 CM93 文件存在、source_size、SHA-256 与数据库一致
13. orphan / duplicate 统计必须为 0

不检查 Geometry validity（该项已经由独立 Geometry audit 验收）。
不修改数据库。
不依赖历史 V 系列测试输出。
不涉及 PBF。

当前 CM93JAPAN 固定基线默认：
- Cells      784
- Features   322465
- Attributes 405153
- Soundings  275700

如果以后换 Dataset，可以传 --no-fixed-baseline。
"""

from __future__ import annotations

import argparse
import csv
import getpass
import hashlib
import json
import math
import sys
import zipfile
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable

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


EXPECTED_COUNTS = {
    "cells": 784,
    "features": 322465,
    "attributes": 405153,
    "soundings": 275700,
}


def parse_args():
    p = argparse.ArgumentParser(
        description="CM93 -> MySQL 非 Geometry 数据完整性全库最终验收 v1"
    )
    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(
        "--source-root",
        default=None,
        help="可选：CM93 根目录，例如 D:\\Ship\\CM93V2\\CM93JAPAN；提供后校验文件存在/大小/SHA256",
    )
    p.add_argument(
        "--no-fixed-baseline",
        action="store_true",
        help="不要求当前 CM93JAPAN 的 784/322465/405153/275700 固定数量",
    )
    p.add_argument(
        "--output-dir",
        default=None,
    )
    return p.parse_args()


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


def fetchall(cur, sql: str, params=()):
    cur.execute(sql, params)
    return list(cur.fetchall())


def add_failure(
    failures: list[dict[str, Any]],
    check: str,
    detail: Any = None,
    **kwargs,
):
    row = {"check": check}
    if detail is not None:
        row["detail"] = detail
    row.update(kwargs)
    failures.append(row)


def json_load(value):
    if value is None:
        return None
    if isinstance(value, (dict, list)):
        return value
    if isinstance(value, (bytes, bytearray, memoryview)):
        value = bytes(value).decode("utf-8")
    return json.loads(value)


def file_sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def write_csv(path: Path, rows: Iterable[dict[str, Any]]):
    rows = list(rows)
    if not rows:
        path.write_text("", encoding="utf-8-sig")
        return

    fields = []
    seen = set()
    for row in rows:
        for key in row:
            if key not in seen:
                seen.add(key)
                fields.append(key)

    with path.open("w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        w.writeheader()
        for row in rows:
            out = {}
            for key in fields:
                v = row.get(key)
                if isinstance(v, (dict, list, tuple)):
                    v = json.dumps(v, ensure_ascii=False, default=str)
                out[key] = v
            w.writerow(out)


def table_exists(cur, database: str, table: str) -> bool:
    return bool(scalar(
        cur,
        """
        SELECT COUNT(*)
        FROM information_schema.tables
        WHERE table_schema=%s AND table_name=%s
        """,
        (database, table),
    ))


def begin_read_only(cur, warnings: list[str]):
    try:
        cur.execute("SET SESSION TRANSACTION READ ONLY")
    except Exception as e:
        warnings.append(f"SET SESSION TRANSACTION READ ONLY: {e}")

    try:
        cur.execute("START TRANSACTION READ ONLY")
    except Exception as e:
        warnings.append(f"START TRANSACTION READ ONLY: {e}")
        cur.execute("START TRANSACTION")


def main():
    args = parse_args()

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

    source_root = (
        Path(args.source_root).expanduser().resolve()
        if args.source_root
        else None
    )

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

    failures: list[dict[str, Any]] = []
    warnings: list[str] = []

    report: dict[str, Any] = {
        "version": "v1.0",
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "database": args.database,
        "dataset_id": args.dataset_id,
        "read_only": True,
        "pbf": "OUT_OF_SCOPE",
        "source_root": str(source_root) if source_root else None,
    }

    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=900,
        write_timeout=30,
    )

    try:
        with conn.cursor() as cur:
            begin_read_only(cur, warnings)

            print("[1/10] 基础表 / Dataset / 总数 ...")

            required_tables = [
                "chart_dataset",
                "chart_cell",
                "chart_feature",
                "chart_feature_attribute",
                "chart_coverage",
                "chart_sounding_point",
                "chart_raw_record",
                "chart_dictionary_object",
                "chart_dictionary_attribute",
            ]
            missing_tables = [
                t for t in required_tables
                if not table_exists(cur, args.database, t)
            ]
            report["missing_tables"] = missing_tables
            if missing_tables:
                add_failure(
                    failures,
                    "missing_tables",
                    missing_tables,
                )
                raise RuntimeError(
                    "缺少必要表：" + ", ".join(missing_tables)
                )

            dataset_count = int(scalar(
                cur,
                "SELECT COUNT(*) FROM chart_dataset WHERE id=%s",
                (args.dataset_id,),
            ) or 0)

            if dataset_count != 1:
                add_failure(
                    failures,
                    "dataset_exists",
                    expected=1,
                    actual=dataset_count,
                )

            counts = {
                "cells": int(scalar(
                    cur,
                    "SELECT COUNT(*) FROM chart_cell WHERE dataset_id=%s",
                    (args.dataset_id,),
                ) or 0),
                "features": int(scalar(
                    cur,
                    "SELECT COUNT(*) FROM chart_feature WHERE dataset_id=%s",
                    (args.dataset_id,),
                ) or 0),
                "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,),
                ) or 0),
                "soundings": 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,),
                ) or 0),
                "coverages": int(scalar(
                    cur,
                    "SELECT COUNT(*) FROM chart_coverage WHERE dataset_id=%s",
                    (args.dataset_id,),
                ) or 0),
                "raw_records": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_raw_record r
                    JOIN chart_cell c ON c.id=r.cell_id
                    WHERE c.dataset_id=%s
                    """,
                    (args.dataset_id,),
                ) or 0),
            }
            report["counts"] = counts

            if not args.no_fixed_baseline:
                for key, expected in EXPECTED_COUNTS.items():
                    if counts[key] != expected:
                        add_failure(
                            failures,
                            f"baseline_{key}",
                            expected=expected,
                            actual=counts[key],
                        )

            done_cells = int(scalar(
                cur,
                """
                SELECT COUNT(*)
                FROM chart_cell
                WHERE dataset_id=%s AND import_status='DONE'
                """,
                (args.dataset_id,),
            ) or 0)
            non_done_cells = counts["cells"] - done_cells

            report["cell_status"] = {
                "done": done_cells,
                "non_done": non_done_cells,
            }

            if non_done_cells:
                add_failure(
                    failures,
                    "cell_status_non_done",
                    actual=non_done_cells,
                )

            print("[2/10] Cell header counts vs 实际落库 ...")

            cell_mismatch_rows = fetchall(
                cur,
                """
                SELECT
                    c.id AS cell_id,
                    c.source_relative_path,
                    c.feature_count AS expected_features,
                    COALESCE(f.actual_features,0) AS actual_features,
                    c.attribute_count AS expected_attributes,
                    COALESCE(a.actual_attributes,0) AS actual_attributes
                FROM chart_cell c
                LEFT JOIN (
                    SELECT cell_id,COUNT(*) AS actual_features
                    FROM chart_feature
                    WHERE dataset_id=%s
                    GROUP BY cell_id
                ) f ON f.cell_id=c.id
                LEFT JOIN (
                    SELECT f.cell_id,COUNT(*) AS actual_attributes
                    FROM chart_feature_attribute a
                    JOIN chart_feature f ON f.id=a.feature_id
                    WHERE f.dataset_id=%s
                    GROUP BY f.cell_id
                ) a ON a.cell_id=c.id
                WHERE c.dataset_id=%s
                  AND (
                      c.feature_count<>COALESCE(f.actual_features,0)
                      OR c.attribute_count<>COALESCE(a.actual_attributes,0)
                  )
                ORDER BY c.id
                """,
                (args.dataset_id, args.dataset_id, args.dataset_id),
            )

            report["cell_count_mismatch"] = len(cell_mismatch_rows)
            for row in cell_mismatch_rows:
                add_failure(
                    failures,
                    "cell_header_count_mismatch",
                    **row,
                )

            print("[3/10] Feature traceability / source_record_index ...")

            feature_checks = {
                "feature_cell_dataset_mismatch": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    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
                    """,
                    (args.dataset_id,),
                ) or 0),
                "duplicate_source_record_index": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT cell_id,source_record_index
                        FROM chart_feature
                        WHERE dataset_id=%s
                        GROUP BY cell_id,source_record_index
                        HAVING COUNT(*)>1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "source_record_index_not_contiguous": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT
                            cell_id,
                            COUNT(*) AS n,
                            COUNT(DISTINCT source_record_index) AS nd,
                            MIN(source_record_index) AS min_i,
                            MAX(source_record_index) AS max_i
                        FROM chart_feature
                        WHERE dataset_id=%s
                        GROUP BY cell_id
                        HAVING
                            nd<>n
                            OR min_i<>0
                            OR max_i<>n-1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "raw_feature_blob_null": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature
                    WHERE dataset_id=%s
                      AND raw_feature_blob IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
                "raw_attribute_blob_null": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature
                    WHERE dataset_id=%s
                      AND raw_attribute_blob IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
                "attributes_json_null": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature
                    WHERE dataset_id=%s
                      AND attributes_json IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
            }
            report["feature_checks"] = feature_checks

            for key, value in feature_checks.items():
                if value:
                    add_failure(
                        failures,
                        key,
                        expected=0,
                        actual=value,
                    )

            print("[4/10] Object / Attribute dictionary 映射 ...")

            dictionary_checks = {
                "feature_unknown_object_id": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature f
                    LEFT JOIN chart_dictionary_object d
                      ON d.dataset_id=f.dataset_id
                     AND d.object_class_id=f.object_class_id
                    WHERE f.dataset_id=%s
                      AND d.object_class_id IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
                "feature_object_name_mismatch": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature f
                    JOIN chart_dictionary_object d
                      ON d.dataset_id=f.dataset_id
                     AND d.object_class_id=f.object_class_id
                    WHERE f.dataset_id=%s
                      AND f.object_class<>d.object_class
                    """,
                    (args.dataset_id,),
                ) or 0),
                "attribute_unknown_id": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature_attribute a
                    JOIN chart_feature f ON f.id=a.feature_id
                    LEFT JOIN chart_dictionary_attribute d
                      ON d.dataset_id=f.dataset_id
                     AND d.attribute_id=a.attribute_id
                    WHERE f.dataset_id=%s
                      AND d.attribute_id IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
                "attribute_name_mismatch": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature_attribute a
                    JOIN chart_feature f ON f.id=a.feature_id
                    JOIN chart_dictionary_attribute d
                      ON d.dataset_id=f.dataset_id
                     AND d.attribute_id=a.attribute_id
                    WHERE f.dataset_id=%s
                      AND a.attribute_name<>d.attribute_name
                    """,
                    (args.dataset_id,),
                ) or 0),
            }
            report["dictionary_checks"] = dictionary_checks

            for key, value in dictionary_checks.items():
                if value:
                    add_failure(
                        failures,
                        key,
                        expected=0,
                        actual=value,
                    )

            print("[5/10] Attribute orphan / ordinal / value_type ...")

            attribute_checks = {
                "attribute_orphan": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature_attribute a
                    LEFT JOIN chart_feature f ON f.id=a.feature_id
                    WHERE f.id IS NULL
                    """,
                ) or 0),
                "attribute_duplicate_ordinal": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT a.feature_id,a.ordinal_no
                        FROM chart_feature_attribute a
                        JOIN chart_feature f ON f.id=a.feature_id
                        WHERE f.dataset_id=%s
                        GROUP BY a.feature_id,a.ordinal_no
                        HAVING COUNT(*)>1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "attribute_ordinal_not_contiguous": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT
                            a.feature_id,
                            COUNT(*) AS n,
                            COUNT(DISTINCT a.ordinal_no) AS nd,
                            MIN(a.ordinal_no) AS min_o,
                            MAX(a.ordinal_no) AS max_o
                        FROM chart_feature_attribute a
                        JOIN chart_feature f ON f.id=a.feature_id
                        WHERE f.dataset_id=%s
                        GROUP BY a.feature_id
                        HAVING
                            nd<>n
                            OR min_o<>0
                            OR max_o<>n-1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "attribute_unknown_value_type": 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.value_type NOT IN
                          ('BYTE','INT','LONG','REAL','WORD10','STRING','LIST','COMPLEX','UNKNOWN')
                    """,
                    (args.dataset_id,),
                ) or 0),
            }
            report["attribute_checks"] = attribute_checks

            for key, value in attribute_checks.items():
                if value:
                    add_failure(
                        failures,
                        key,
                        expected=0,
                        actual=value,
                    )

            print("[6/10] Coverage <-> _m_sor Feature 映射 ...")

            coverage_checks = {
                "coverage_duplicate_source_record_index": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT cell_id,source_record_index
                        FROM chart_coverage
                        WHERE dataset_id=%s
                        GROUP BY cell_id,source_record_index
                        HAVING COUNT(*)>1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "coverage_without_m_sor_feature": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_coverage c
                    LEFT JOIN chart_feature f
                      ON f.dataset_id=c.dataset_id
                     AND f.cell_id=c.cell_id
                     AND f.source_record_index=c.source_record_index
                     AND f.object_class='_m_sor'
                    WHERE c.dataset_id=%s
                      AND f.id IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
                "m_sor_without_coverage": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature f
                    LEFT JOIN chart_coverage c
                      ON c.dataset_id=f.dataset_id
                     AND c.cell_id=f.cell_id
                     AND c.source_record_index=f.source_record_index
                    WHERE f.dataset_id=%s
                      AND f.object_class='_m_sor'
                      AND c.id IS NULL
                    """,
                    (args.dataset_id,),
                ) or 0),
                "feature_coverage_cross_cell": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_feature f
                    JOIN chart_coverage c ON c.id=f.coverage_id
                    WHERE f.dataset_id=%s
                      AND (
                          c.dataset_id<>f.dataset_id
                          OR c.cell_id<>f.cell_id
                      )
                    """,
                    (args.dataset_id,),
                ) or 0),
            }
            report["coverage_checks"] = coverage_checks

            for key, value in coverage_checks.items():
                if value:
                    add_failure(
                        failures,
                        key,
                        expected=0,
                        actual=value,
                    )

            print("[7/10] Sounding 数据链 ...")

            sounding_checks = {
                "sounding_orphan": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM chart_sounding_point s
                    LEFT JOIN chart_feature f ON f.id=s.feature_id
                    WHERE f.id IS NULL
                    """,
                ) or 0),
                "sounding_wrong_feature_type": 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 f.geometry_type<>'SOUNDING'
                    """,
                    (args.dataset_id,),
                ) or 0),
                "sounding_duplicate_point_index": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT s.feature_id,s.point_index
                        FROM chart_sounding_point s
                        JOIN chart_feature f ON f.id=s.feature_id
                        WHERE f.dataset_id=%s
                        GROUP BY s.feature_id,s.point_index
                        HAVING COUNT(*)>1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "sounding_index_not_contiguous": int(scalar(
                    cur,
                    """
                    SELECT COUNT(*)
                    FROM (
                        SELECT
                            s.feature_id,
                            COUNT(*) AS n,
                            COUNT(DISTINCT s.point_index) AS nd,
                            MIN(s.point_index) AS min_i,
                            MAX(s.point_index) AS max_i
                        FROM chart_sounding_point s
                        JOIN chart_feature f ON f.id=s.feature_id
                        WHERE f.dataset_id=%s
                        GROUP BY s.feature_id
                        HAVING
                            nd<>n
                            OR min_i<>0
                            OR max_i<>n-1
                    ) x
                    """,
                    (args.dataset_id,),
                ) or 0),
                "sounding_depth_formula_mismatch": 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 ABS(
                          s.depth -
                          CASE
                              WHEN s.raw_z>=12000 THEN s.raw_z-12000
                              ELSE s.raw_z/10.0
                          END
                      ) > 0.000001
                    """,
                    (args.dataset_id,),
                ) or 0),
            }
            report["sounding_checks"] = sounding_checks

            for key, value in sounding_checks.items():
                if value:
                    add_failure(
                        failures,
                        key,
                        expected=0,
                        actual=value,
                    )

            print("[8/10] attributes_json / relation indices 全量解析 ...")

            relation_bad = 0
            attributes_json_bad = 0
            attributes_name_missing = 0

            cur.execute(
                """
                SELECT
                    id,
                    cell_id,
                    source_record_index,
                    attributes_json
                FROM chart_feature
                WHERE dataset_id=%s
                ORDER BY cell_id,source_record_index
                """,
                (args.dataset_id,),
            )
            feature_json_rows = list(cur.fetchall())

            indices_by_cell: dict[int, set[int]] = defaultdict(set)
            parsed_by_feature: dict[int, Any] = {}

            for row in feature_json_rows:
                cell_id = int(row["cell_id"])
                src_idx = int(row["source_record_index"])
                indices_by_cell[cell_id].add(src_idx)

                try:
                    obj = json_load(row["attributes_json"])
                    if not isinstance(obj, dict):
                        raise ValueError("attributes_json 不是 JSON object")
                    parsed_by_feature[int(row["id"])] = obj
                except Exception as e:
                    attributes_json_bad += 1
                    add_failure(
                        failures,
                        "attributes_json_parse",
                        feature_id=row["id"],
                        cell_id=cell_id,
                        source_record_index=src_idx,
                        detail=f"{type(e).__name__}: {e}",
                    )

            for row in feature_json_rows:
                fid = int(row["id"])
                obj = parsed_by_feature.get(fid)
                if obj is None:
                    continue

                related = obj.get("_cm93_related_indices")
                if related is not None:
                    if not isinstance(related, list):
                        relation_bad += 1
                        add_failure(
                            failures,
                            "related_indices_not_list",
                            feature_id=fid,
                            actual_type=type(related).__name__,
                        )
                    else:
                        cell_indices = indices_by_cell[int(row["cell_id"])]
                        for target in related:
                            try:
                                ti = int(target)
                            except Exception:
                                relation_bad += 1
                                add_failure(
                                    failures,
                                    "related_index_not_int",
                                    feature_id=fid,
                                    target=repr(target),
                                )
                                continue

                            if ti not in cell_indices:
                                relation_bad += 1
                                add_failure(
                                    failures,
                                    "related_index_missing_target",
                                    feature_id=fid,
                                    cell_id=row["cell_id"],
                                    source_record_index=row["source_record_index"],
                                    target_source_record_index=ti,
                                )

            # Attribute row 的 attribute_name 应在对应 attributes_json 顶层出现。
            cur.execute(
                """
                SELECT
                    a.feature_id,
                    a.attribute_name
                FROM chart_feature_attribute a
                JOIN chart_feature f ON f.id=a.feature_id
                WHERE f.dataset_id=%s
                """,
                (args.dataset_id,),
            )

            for row in cur:
                obj = parsed_by_feature.get(int(row["feature_id"]))
                if obj is None:
                    continue
                if row["attribute_name"] not in obj:
                    attributes_name_missing += 1
                    if attributes_name_missing <= 1000:
                        add_failure(
                            failures,
                            "attribute_name_missing_from_attributes_json",
                            feature_id=row["feature_id"],
                            attribute_name=row["attribute_name"],
                        )

            report["json_relation_checks"] = {
                "attributes_json_bad": attributes_json_bad,
                "related_indices_bad": relation_bad,
                "attribute_name_missing_from_attributes_json": attributes_name_missing,
            }

            if attributes_name_missing > 1000:
                add_failure(
                    failures,
                    "attribute_name_missing_from_attributes_json_overflow",
                    detail=f"总数={attributes_name_missing}，明细只记录前 1000 条",
                )

            print("[9/10] chart_raw_record 分块完整性 ...")

            cur.execute(
                """
                SELECT
                    r.cell_id,
                    r.record_kind,
                    r.source_record_index,
                    OCTET_LENGTH(r.raw_blob) AS raw_bytes,
                    r.raw_metadata_json
                FROM chart_raw_record r
                JOIN chart_cell c ON c.id=r.cell_id
                WHERE c.dataset_id=%s
                ORDER BY r.cell_id,r.record_kind,r.source_record_index
                """,
                (args.dataset_id,),
            )

            raw_groups: dict[tuple[int, str], list[dict[str, Any]]] = defaultdict(list)
            raw_json_bad = 0
            raw_empty_blob = 0

            for row in cur:
                if int(row["raw_bytes"] or 0) <= 0:
                    raw_empty_blob += 1
                    add_failure(
                        failures,
                        "raw_record_empty_blob",
                        cell_id=row["cell_id"],
                        record_kind=row["record_kind"],
                        source_record_index=row["source_record_index"],
                    )

                try:
                    meta = json_load(row["raw_metadata_json"])
                    if not isinstance(meta, dict):
                        raise ValueError("raw_metadata_json 不是 JSON object")
                except Exception as e:
                    raw_json_bad += 1
                    add_failure(
                        failures,
                        "raw_metadata_json_parse",
                        cell_id=row["cell_id"],
                        record_kind=row["record_kind"],
                        source_record_index=row["source_record_index"],
                        detail=f"{type(e).__name__}: {e}",
                    )
                    meta = None

                raw_groups[
                    (int(row["cell_id"]), str(row["record_kind"]))
                ].append({
                    "index": int(row["source_record_index"]),
                    "raw_bytes": int(row["raw_bytes"] or 0),
                    "meta": meta,
                })

            raw_group_bad = 0

            for (cell_id, kind), rows in raw_groups.items():
                indices = [r["index"] for r in rows]
                expected_indices = list(range(len(rows)))

                if indices != expected_indices:
                    raw_group_bad += 1
                    add_failure(
                        failures,
                        "raw_chunk_index_not_contiguous",
                        cell_id=cell_id,
                        record_kind=kind,
                        indices=indices[:50],
                        count=len(indices),
                    )

                metas = [r["meta"] for r in rows if isinstance(r["meta"], dict)]
                if len(metas) != len(rows):
                    continue

                chunk_counts = {m.get("chunk_count") for m in metas}
                total_bytes_set = {m.get("total_bytes") for m in metas}
                chunk_indices_meta = [m.get("chunk_index") for m in metas]
                actual_bytes = sum(r["raw_bytes"] for r in rows)

                if chunk_counts != {len(rows)}:
                    raw_group_bad += 1
                    add_failure(
                        failures,
                        "raw_chunk_count_mismatch",
                        cell_id=cell_id,
                        record_kind=kind,
                        metadata_chunk_counts=list(chunk_counts),
                        actual_chunk_count=len(rows),
                    )

                if chunk_indices_meta != expected_indices:
                    raw_group_bad += 1
                    add_failure(
                        failures,
                        "raw_metadata_chunk_index_mismatch",
                        cell_id=cell_id,
                        record_kind=kind,
                        metadata_indices=chunk_indices_meta[:50],
                    )

                if len(total_bytes_set) != 1:
                    raw_group_bad += 1
                    add_failure(
                        failures,
                        "raw_total_bytes_inconsistent",
                        cell_id=cell_id,
                        record_kind=kind,
                        values=list(total_bytes_set),
                    )
                else:
                    expected_total = next(iter(total_bytes_set))
                    try:
                        expected_total = int(expected_total)
                    except Exception:
                        expected_total = -1

                    if expected_total != actual_bytes:
                        raw_group_bad += 1
                        add_failure(
                            failures,
                            "raw_total_bytes_mismatch",
                            cell_id=cell_id,
                            record_kind=kind,
                            expected=expected_total,
                            actual=actual_bytes,
                        )

            # 每个 Cell 至少应有 HEADER 和 FEATURE 原始表。
            raw_required_missing = int(scalar(
                cur,
                """
                SELECT COUNT(*)
                FROM (
                    SELECT c.id
                    FROM chart_cell c
                    LEFT JOIN chart_raw_record h
                      ON h.cell_id=c.id AND h.record_kind='HEADER'
                    LEFT JOIN chart_raw_record f
                      ON f.cell_id=c.id AND f.record_kind='FEATURE'
                    WHERE c.dataset_id=%s
                    GROUP BY c.id
                    HAVING COUNT(DISTINCT h.record_kind)=0
                        OR COUNT(DISTINCT f.record_kind)=0
                ) x
                """,
                (args.dataset_id,),
            ) or 0)

            report["raw_checks"] = {
                "raw_metadata_json_bad": raw_json_bad,
                "raw_empty_blob": raw_empty_blob,
                "raw_group_bad": raw_group_bad,
                "required_header_or_feature_missing_cells": raw_required_missing,
                "raw_group_count": len(raw_groups),
            }

            if raw_required_missing:
                add_failure(
                    failures,
                    "raw_required_kind_missing",
                    expected=0,
                    actual=raw_required_missing,
                )

            print("[10/10] 可选源文件 SHA256 + 最终判定 ...")

            source_checks = {
                "enabled": source_root is not None,
                "files_checked": 0,
                "files_missing": 0,
                "size_mismatch": 0,
                "sha256_mismatch": 0,
            }

            if source_root is not None:
                if not source_root.is_dir():
                    add_failure(
                        failures,
                        "source_root_not_found",
                        path=str(source_root),
                    )
                else:
                    cells = fetchall(
                        cur,
                        """
                        SELECT
                            id,
                            source_relative_path,
                            source_size,
                            source_sha256
                        FROM chart_cell
                        WHERE dataset_id=%s
                        ORDER BY id
                        """,
                        (args.dataset_id,),
                    )

                    for cell in cells:
                        rel = str(cell["source_relative_path"]).replace("\\", "/")
                        path = source_root / Path(rel)
                        source_checks["files_checked"] += 1

                        if not path.is_file():
                            source_checks["files_missing"] += 1
                            add_failure(
                                failures,
                                "source_file_missing",
                                cell_id=cell["id"],
                                path=str(path),
                            )
                            continue

                        actual_size = path.stat().st_size
                        expected_size = int(cell["source_size"])

                        if actual_size != expected_size:
                            source_checks["size_mismatch"] += 1
                            add_failure(
                                failures,
                                "source_size_mismatch",
                                cell_id=cell["id"],
                                path=str(path),
                                expected=expected_size,
                                actual=actual_size,
                            )

                        expected_hash = str(cell["source_sha256"] or "").lower()
                        actual_hash = file_sha256(path).lower()

                        if expected_hash != actual_hash:
                            source_checks["sha256_mismatch"] += 1
                            add_failure(
                                failures,
                                "source_sha256_mismatch",
                                cell_id=cell["id"],
                                path=str(path),
                                expected=expected_hash,
                                actual=actual_hash,
                            )

            report["source_checks"] = source_checks

            conn.rollback()

    finally:
        conn.close()

    report["warnings"] = warnings
    report["failure_count"] = len(failures)
    report["overall"] = "PASS" if not failures else "FAIL"

    write_csv(outdir / "失败明细_v1.csv", failures)
    write_csv(outdir / "Cell计数不一致_v1.csv", cell_mismatch_rows)

    (outdir / "最终验收报告_v1.json").write_text(
        json.dumps(
            report,
            ensure_ascii=False,
            indent=2,
            default=str,
        ),
        encoding="utf-8",
    )

    c = report["counts"]
    lines = [
        "CM93 -> MySQL 非 Geometry 数据完整性全库最终验收 v1",
        "=" * 82,
        f"Dataset ID              : {args.dataset_id}",
        "",
        f"Cells                   : {c['cells']}",
        f"Features                : {c['features']}",
        f"Attributes              : {c['attributes']}",
        f"Soundings               : {c['soundings']}",
        f"Coverages               : {c['coverages']}",
        f"Raw records             : {c['raw_records']}",
        "",
        f"Cell count mismatch     : {report['cell_count_mismatch']}",
        f"Feature check failures  : {sum(report['feature_checks'].values())}",
        f"Dictionary failures     : {sum(report['dictionary_checks'].values())}",
        f"Attribute failures      : {sum(report['attribute_checks'].values())}",
        f"Coverage failures       : {sum(report['coverage_checks'].values())}",
        f"Sounding failures       : {sum(report['sounding_checks'].values())}",
        f"JSON/relation failures  : {sum(report['json_relation_checks'].values())}",
        f"Raw failures            : "
        f"{report['raw_checks']['raw_metadata_json_bad'] + report['raw_checks']['raw_empty_blob'] + report['raw_checks']['raw_group_bad'] + report['raw_checks']['required_header_or_feature_missing_cells']}",
        "",
    ]

    if source_root is not None:
        s = report["source_checks"]
        lines += [
            f"Source files checked    : {s['files_checked']}",
            f"Source files missing    : {s['files_missing']}",
            f"Source size mismatch    : {s['size_mismatch']}",
            f"Source SHA256 mismatch  : {s['sha256_mismatch']}",
            "",
        ]

    lines += [
        f"Failure count           : {report['failure_count']}",
        f"OVERALL: {report['overall']}",
    ]

    summary = "\n".join(lines)

    (outdir / "最终验收摘要_v1.txt").write_text(
        summary,
        encoding="utf-8",
    )

    zip_path = outdir.with_suffix(".zip")
    with zipfile.ZipFile(
        zip_path,
        "w",
        compression=zipfile.ZIP_DEFLATED,
    ) as z:
        for p in sorted(outdir.rglob("*")):
            if p.is_file():
                z.write(p, p.relative_to(outdir.parent))

    print("")
    print(summary)
    print("")
    print(f"报告目录: {outdir}")
    print(f"报告 ZIP : {zip_path}")
    print("把 ZIP 上传给 ChatGPT。")


if __name__ == "__main__":
    main()