1273 lines
45 KiB
Python
1273 lines
45 KiB
Python
#!/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()
|