476 lines
16 KiB
Python
476 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
CM93 V10.1 结果汇总与完整性验证器
|
||
Version: v10.2-result-digest
|
||
|
||
用途：
|
||
- 不连接 MySQL。
|
||
- 不重新运行 normalization。
|
||
- 直接读取 V10.1 的输出目录或 ZIP。
|
||
- 自动判断 V10.1 是否整体 PASS。
|
||
- 生成一个很小的结果摘要 ZIP，便于上传给 ChatGPT。
|
||
|
||
支持输入：
|
||
1. V10.1 完整输出目录，例如：
|
||
cm93_normalization_canonical_v10_1_dataset_1_20260831_215602
|
||
|
||
2. V10.1 完整 ZIP，例如：
|
||
cm93_normalization_canonical_v10_1_dataset_1_20260831_215602.zip
|
||
|
||
3. 只有部分 CSV 的 ZIP。
|
||
此时脚本会明确显示“INCOMPLETE”，不会误判整体 PASS。
|
||
|
||
执行：
|
||
python cm93_v10_1_result_digest_v10_2.py ^
|
||
--input "D:\...\cm93_normalization_canonical_v10_1_dataset_1_20260831_215602"
|
||
|
||
或：
|
||
|
||
python cm93_v10_1_result_digest_v10_2.py ^
|
||
--input "D:\...\cm93_normalization_canonical_v10_1_dataset_1_20260831_215602.zip"
|
||
|
||
输出：
|
||
- V10_1_结果摘要_v10_2.txt
|
||
- V10_1_结果摘要_v10_2.json
|
||
- cm93_v10_1_result_digest_v10_2.zip
|
||
|
||
只读，不修改原始结果。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import zipfile
|
||
from collections import Counter
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
|
||
EXPECTED_FILES = {
|
||
"summary": "normalization_canonical_summary_v10_1.txt",
|
||
"report": "normalization_canonical_report_v10_1.json",
|
||
"golden_csv": "normalization_golden_vectors_v10_1.csv",
|
||
"golden_json": "normalization_golden_vectors_v10_1.json",
|
||
"variant": "normalization_variant_results_v10_1.csv",
|
||
"face_order": "normalization_face_order_results_v10_1.csv",
|
||
"failures": "normalization_canonical_failures_v10_1.csv",
|
||
"bbox": "normalization_bbox_metadata_impact_v10_1.csv",
|
||
}
|
||
|
||
|
||
def parse_args():
|
||
p = argparse.ArgumentParser(
|
||
description="CM93 V10.1 result digest verifier v10.2"
|
||
)
|
||
p.add_argument("--input", required=True, help="V10.1 输出目录或 ZIP")
|
||
p.add_argument("--output-dir", default=None)
|
||
return p.parse_args()
|
||
|
||
|
||
def find_file(root: Path, filename: str) -> Optional[Path]:
|
||
direct = root / filename
|
||
if direct.exists():
|
||
return direct
|
||
|
||
hits = list(root.rglob(filename))
|
||
if hits:
|
||
return hits[0]
|
||
|
||
return None
|
||
|
||
|
||
def read_csv(path: Path) -> List[Dict[str, str]]:
|
||
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
||
return list(csv.DictReader(f))
|
||
|
||
|
||
def intv(v: Any, default: int = 0) -> int:
|
||
try:
|
||
return int(v)
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def boolv(v: Any) -> bool:
|
||
if isinstance(v, bool):
|
||
return v
|
||
if v is None:
|
||
return False
|
||
s = str(v).strip().lower()
|
||
return s in {"1", "true", "yes", "y", "pass"}
|
||
|
||
|
||
def main():
|
||
args = parse_args()
|
||
src = Path(args.input).expanduser().resolve()
|
||
|
||
if not src.exists():
|
||
print(f"输入不存在：{src}")
|
||
sys.exit(2)
|
||
|
||
temp_dir = None
|
||
|
||
try:
|
||
if src.is_file() and src.suffix.lower() == ".zip":
|
||
temp_dir = Path(tempfile.mkdtemp(prefix="cm93_v10_1_digest_"))
|
||
with zipfile.ZipFile(src, "r") as z:
|
||
z.extractall(temp_dir)
|
||
root = temp_dir
|
||
elif src.is_dir():
|
||
root = src
|
||
else:
|
||
print("输入必须是 V10.1 输出目录或 ZIP。")
|
||
sys.exit(2)
|
||
|
||
found: Dict[str, Optional[Path]] = {
|
||
key: find_file(root, filename)
|
||
for key, filename in EXPECTED_FILES.items()
|
||
}
|
||
|
||
present = {k: bool(v) for k, v in found.items()}
|
||
|
||
digest: Dict[str, Any] = {
|
||
"version": "v10.2-result-digest",
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"input": str(src),
|
||
"files_present": present,
|
||
"complete_core_set": all(
|
||
present[k]
|
||
for k in (
|
||
"report",
|
||
"golden_csv",
|
||
"variant",
|
||
"face_order",
|
||
"failures",
|
||
"bbox",
|
||
)
|
||
),
|
||
"checks": {},
|
||
"overall": "INCOMPLETE",
|
||
}
|
||
|
||
checks = digest["checks"]
|
||
|
||
# --------------------------------------------------------------
|
||
# Report JSON
|
||
# --------------------------------------------------------------
|
||
if found["report"]:
|
||
report = json.loads(
|
||
found["report"].read_text(encoding="utf-8")
|
||
)
|
||
summary = report.get("summary", {})
|
||
|
||
checks["report"] = {
|
||
"overall_pass": summary.get("overall_pass"),
|
||
"total_polygon_records": summary.get("total_polygon_records"),
|
||
"valid_polygon_records": summary.get("valid_polygon_records"),
|
||
"invalid_polygon_records": summary.get("invalid_polygon_records"),
|
||
"valid_policy_changed_count":
|
||
summary.get("valid_policy_changed_count"),
|
||
"valid_hash_changed_count":
|
||
summary.get("valid_hash_changed_count"),
|
||
"golden_vector_count":
|
||
summary.get("golden_vector_count"),
|
||
"golden_record_status_counts":
|
||
summary.get("golden_record_status_counts"),
|
||
"input_variant_case_count":
|
||
summary.get("input_variant_case_count"),
|
||
"input_variant_status_counts":
|
||
summary.get("input_variant_status_counts"),
|
||
"face_order_case_count":
|
||
summary.get("face_order_case_count"),
|
||
"face_order_hash_mismatch_count":
|
||
summary.get("face_order_hash_mismatch_count"),
|
||
"canonical_topology_failure_count":
|
||
summary.get("canonical_topology_failure_count"),
|
||
"repeat_hash_failure_count":
|
||
summary.get("repeat_hash_failure_count"),
|
||
"input_variant_failure_record_count":
|
||
summary.get("input_variant_failure_record_count"),
|
||
"future_area_feature_bbox_update_needed_count":
|
||
summary.get("future_area_feature_bbox_update_needed_count"),
|
||
}
|
||
|
||
# --------------------------------------------------------------
|
||
# Golden vectors
|
||
# --------------------------------------------------------------
|
||
if found["golden_csv"]:
|
||
rows = read_csv(found["golden_csv"])
|
||
statuses = Counter(
|
||
str(r.get("record_status", "")).strip()
|
||
for r in rows
|
||
)
|
||
hashes = [
|
||
r.get("canonical_wkb_sha256", "")
|
||
for r in rows
|
||
if r.get("canonical_wkb_sha256")
|
||
]
|
||
|
||
checks["golden"] = {
|
||
"rows": len(rows),
|
||
"status_counts": dict(statuses),
|
||
"unique_hash_count": len(set(hashes)),
|
||
"missing_hash_count":
|
||
sum(not bool(r.get("canonical_wkb_sha256")) for r in rows),
|
||
"variant_failure_records":
|
||
sum(
|
||
intv(r.get("input_variant_failure_count")) > 0
|
||
for r in rows
|
||
),
|
||
"face_order_unstable_records":
|
||
sum(
|
||
not boolv(r.get("face_order_hash_stable"))
|
||
for r in rows
|
||
),
|
||
"repeat_unstable_records":
|
||
sum(
|
||
not boolv(r.get("repeat_hash_stable"))
|
||
for r in rows
|
||
),
|
||
"canonical_topology_changed_records":
|
||
sum(
|
||
not boolv(r.get("canonical_topology_unchanged"))
|
||
for r in rows
|
||
),
|
||
}
|
||
|
||
# --------------------------------------------------------------
|
||
# Variant
|
||
# --------------------------------------------------------------
|
||
if found["variant"]:
|
||
rows = read_csv(found["variant"])
|
||
statuses = Counter(
|
||
str(r.get("status", "")).strip()
|
||
for r in rows
|
||
)
|
||
|
||
checks["variant"] = {
|
||
"rows": len(rows),
|
||
"status_counts": dict(statuses),
|
||
"fail_rows":
|
||
sum(str(r.get("status", "")).strip() != "PASS" for r in rows),
|
||
"hash_mismatch_rows":
|
||
sum(
|
||
not boolv(r.get("matches_baseline_hash"))
|
||
for r in rows
|
||
if str(r.get("status", "")).strip() != "EXCEPTION"
|
||
),
|
||
"topology_mismatch_rows":
|
||
sum(
|
||
not boolv(r.get("topology_equal_baseline"))
|
||
for r in rows
|
||
if str(r.get("status", "")).strip() != "EXCEPTION"
|
||
),
|
||
}
|
||
|
||
# --------------------------------------------------------------
|
||
# Face order
|
||
# --------------------------------------------------------------
|
||
if found["face_order"]:
|
||
rows = read_csv(found["face_order"])
|
||
checks["face_order"] = {
|
||
"rows": len(rows),
|
||
"hash_mismatch_rows":
|
||
sum(
|
||
not boolv(r.get("matches_baseline_hash"))
|
||
for r in rows
|
||
),
|
||
}
|
||
|
||
# --------------------------------------------------------------
|
||
# Failure CSV
|
||
# --------------------------------------------------------------
|
||
if found["failures"]:
|
||
rows = read_csv(found["failures"])
|
||
# 只有表头时 DictReader 返回 0 rows，正是期望。
|
||
checks["failures"] = {
|
||
"rows": len(rows),
|
||
}
|
||
|
||
# --------------------------------------------------------------
|
||
# BBox metadata impact
|
||
# --------------------------------------------------------------
|
||
if found["bbox"]:
|
||
rows = read_csv(found["bbox"])
|
||
checks["bbox"] = {
|
||
"rows": len(rows),
|
||
"bbox_changed_rows":
|
||
sum(
|
||
boolv(r.get("bbox_changed_over_tolerance"))
|
||
for r in rows
|
||
),
|
||
"future_feature_bbox_update_needed_rows":
|
||
sum(
|
||
boolv(r.get("future_feature_bbox_update_needed"))
|
||
for r in rows
|
||
),
|
||
"rows_with_connector":
|
||
sum(
|
||
float(r.get("discarded_connector_length_deg") or 0) > 0
|
||
for r in rows
|
||
),
|
||
}
|
||
|
||
# --------------------------------------------------------------
|
||
# Independent overall verdict
|
||
# --------------------------------------------------------------
|
||
required_for_pass = all(
|
||
k in checks
|
||
for k in (
|
||
"report",
|
||
"golden",
|
||
"variant",
|
||
"face_order",
|
||
"failures",
|
||
"bbox",
|
||
)
|
||
)
|
||
|
||
if required_for_pass:
|
||
report_check = checks["report"]
|
||
golden = checks["golden"]
|
||
variant = checks["variant"]
|
||
face = checks["face_order"]
|
||
failures = checks["failures"]
|
||
|
||
pass_conditions = [
|
||
boolv(report_check.get("overall_pass")),
|
||
intv(report_check.get("invalid_polygon_records")) == 55,
|
||
intv(report_check.get("valid_policy_changed_count")) == 0,
|
||
intv(report_check.get("valid_hash_changed_count")) == 0,
|
||
|
||
golden["rows"] == 55,
|
||
golden["status_counts"].get("PASS", 0) == 55,
|
||
golden["status_counts"].get("FAIL", 0) == 0,
|
||
golden["missing_hash_count"] == 0,
|
||
golden["variant_failure_records"] == 0,
|
||
golden["face_order_unstable_records"] == 0,
|
||
golden["repeat_unstable_records"] == 0,
|
||
golden["canonical_topology_changed_records"] == 0,
|
||
|
||
variant["rows"] == 55 * 6,
|
||
variant["fail_rows"] == 0,
|
||
variant["hash_mismatch_rows"] == 0,
|
||
variant["topology_mismatch_rows"] == 0,
|
||
|
||
face["rows"] == 55 * 4,
|
||
face["hash_mismatch_rows"] == 0,
|
||
|
||
failures["rows"] == 0,
|
||
]
|
||
|
||
digest["overall"] = (
|
||
"PASS"
|
||
if all(pass_conditions)
|
||
else "FAIL"
|
||
)
|
||
digest["pass_conditions"] = pass_conditions
|
||
else:
|
||
digest["overall"] = "INCOMPLETE"
|
||
|
||
# --------------------------------------------------------------
|
||
# Output
|
||
# --------------------------------------------------------------
|
||
outdir = (
|
||
Path(args.output_dir).expanduser().resolve()
|
||
if args.output_dir
|
||
else src.parent
|
||
)
|
||
outdir.mkdir(parents=True, exist_ok=True)
|
||
|
||
txt_path = outdir / "V10_1_结果摘要_v10_2.txt"
|
||
json_path = outdir / "V10_1_结果摘要_v10_2.json"
|
||
zip_path = outdir / "cm93_v10_1_result_digest_v10_2.zip"
|
||
|
||
json_path.write_text(
|
||
json.dumps(digest, ensure_ascii=False, indent=2),
|
||
encoding="utf-8"
|
||
)
|
||
|
||
lines = []
|
||
lines.append("CM93 V10.1 结果汇总与完整性验证 v10.2")
|
||
lines.append("=" * 78)
|
||
lines.append(f"Input: {src}")
|
||
lines.append(f"Overall: {digest['overall']}")
|
||
lines.append("")
|
||
|
||
lines.append("FILES")
|
||
for key, filename in EXPECTED_FILES.items():
|
||
lines.append(
|
||
f" {key:12s} "
|
||
f"{'YES' if present[key] else 'NO ':3s} "
|
||
f"{filename}"
|
||
)
|
||
|
||
for section, value in checks.items():
|
||
lines.append("")
|
||
lines.append(section.upper())
|
||
for k, v in value.items():
|
||
lines.append(f" {k}: {v}")
|
||
|
||
if digest["overall"] == "INCOMPLETE":
|
||
lines.append("")
|
||
lines.append(
|
||
"INCOMPLETE：当前输入不足以判定 V10.1 整体 PASS。"
|
||
)
|
||
elif digest["overall"] == "PASS":
|
||
lines.append("")
|
||
lines.append(
|
||
"PASS：V10.1 canonical normalization 全部门槛通过。"
|
||
)
|
||
else:
|
||
lines.append("")
|
||
lines.append(
|
||
"FAIL：至少一个 V10.1 canonical validation 门槛未通过。"
|
||
)
|
||
|
||
txt_path.write_text(
|
||
"\n".join(lines),
|
||
encoding="utf-8"
|
||
)
|
||
|
||
with zipfile.ZipFile(
|
||
zip_path,
|
||
"w",
|
||
compression=zipfile.ZIP_DEFLATED
|
||
) as z:
|
||
z.write(txt_path, txt_path.name)
|
||
z.write(json_path, json_path.name)
|
||
|
||
# 把关键小文件一起放进 digest ZIP。
|
||
for key in (
|
||
"summary",
|
||
"report",
|
||
"golden_csv",
|
||
"variant",
|
||
"face_order",
|
||
"failures",
|
||
"bbox",
|
||
):
|
||
p = found.get(key)
|
||
if p and p.exists():
|
||
z.write(p, p.name)
|
||
|
||
print("")
|
||
print(f"Overall: {digest['overall']}")
|
||
print(f"TXT: {txt_path}")
|
||
print(f"JSON: {json_path}")
|
||
print(f"ZIP: {zip_path}")
|
||
print("")
|
||
print("把这个很小的 digest ZIP 上传给 ChatGPT。")
|
||
|
||
finally:
|
||
if temp_dir and temp_dir.exists():
|
||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|