4c732eb836
review 发现数据本身真实达标（7 high / 20 usable / 0 skeleton）， 但代码/文档/测试/元数据 4 层出现严重脱节，存在静默升级风险与 合规假象回归漏洞。 Phase 1: 契约硬化（P0） - risk_report.py 新增 _classify_score_bands + _compute_quality_level - 质量等级判定从"仅看 confidence"升级为综合门槛 (conf + sr + recs + alts + 三层分数带)，对齐 plan §4 - 新增 low 等级区分"已脱离骨架但未达可用" - _load_provenance_metadata 改为优先 load_province 取完整数据 - finding_to_risk_dict 不再二次规范化已规范化数据 - quality_summary.py 增加 low 等级统计 - SCHEMA.md §6 同步完整门槛定义 Phase 2: 测试加固（P0） - 新增 test_high_trust_thresholds.py 锁死 high/usable 完整门槛 (plan §9.2 要求的"防静默升级"测试) - 修复 test_crowd_db_data_quality.py 等级枚举支持 low - 修复 test_risk_report.py 硬编码日期脆弱性 Phase 3: 真相源单一化（P0） - CURRENT_STATE.md §0.5 清除 6/20 旧文案"4 high + 3 usable + 20 skeleton" - 改为引用顶部状态词 + 历史轨迹仅供审计 - NATIONALIZATION §4 清除"当前 high 已扩展为 5 省"矛盾文案 - 顶部状态词从"Phase 0 收口中"升级为"已完成" Phase 4: 元数据状态对齐（P1） - hunan.json / sichuan.json trusted_sources.kind province_official_pending_review -> province_official - 同步更新 quality_note 说明已完成 2025 年度复核 - 消除"状态 high/usable 但 kind=pending_review"的矛盾 Phase 5: 防漂移机制（P1） - 新增 scripts/check_crowd_db_consistency.py 跨文档+数据+测试白名单一致性检查（5 项检查） - dev-verify.sh 接入 crowd_db quality summary 打印 验证: - ruff: All checks passed - mypy: Success, no issues in 16 source files - pytest crowd_db/: 148 passed, 2 skipped - pytest 全量: 1283 passed, 2 skipped (无回归) - consistency check: high=7 usable=20 low=0 skeleton=0 Refs: docs/plans/2026-06-23-national-high-trust-crowd-db-plan.md §4/§9
75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
from collections import Counter
|
||
from typing import Any
|
||
|
||
from .loader import CrowdDBLoader
|
||
from .risk_report import _normalize_provenance
|
||
|
||
|
||
def build_quality_summary(loader: CrowdDBLoader | None = None) -> dict[str, Any]:
|
||
loader = loader or CrowdDBLoader(warn_low_confidence=False)
|
||
provinces: list[dict[str, Any]] = []
|
||
for province in loader.list_supported_provinces():
|
||
# 加载完整数据（含 score_ranges）用于质量判定
|
||
full_data = loader.load_province(province)
|
||
metadata = loader.load_metadata(province) or {"province": province}
|
||
normalized = _normalize_provenance(metadata, full_data=full_data)
|
||
provinces.append({
|
||
"province": province,
|
||
"confidence": normalized["confidence"],
|
||
"quality_level": normalized["quality_level"],
|
||
"quality_label": normalized["quality_label"],
|
||
"data_year": normalized["data_year"],
|
||
"source_type": normalized["source_type"],
|
||
})
|
||
provinces.sort(key=lambda item: str(item["province"]))
|
||
counts = Counter(item["quality_level"] for item in provinces)
|
||
by_quality_level = {
|
||
"high": counts.get("high", 0),
|
||
"usable": counts.get("usable", 0),
|
||
"low": counts.get("low", 0),
|
||
"skeleton": counts.get("skeleton", 0),
|
||
"unknown": counts.get("unknown", 0),
|
||
}
|
||
return {
|
||
"total_provinces": len(provinces),
|
||
"by_quality_level": by_quality_level,
|
||
"provinces": provinces,
|
||
}
|
||
|
||
|
||
def _emit_human(summary: dict[str, Any]) -> None:
|
||
print(f"province_count: {summary['total_provinces']}")
|
||
print("quality_counts:")
|
||
for level, count in sorted(summary["by_quality_level"].items()):
|
||
print(f" - {level}: {count}")
|
||
print("provinces:")
|
||
for item in summary["provinces"]:
|
||
print(
|
||
f" - {item['province']}: {item['quality_level']} ({item['quality_label']}), confidence={item['confidence']}, year={item['data_year']}"
|
||
)
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="crowd_db province quality summary")
|
||
parser.add_argument("--human", action="store_true")
|
||
return parser
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
summary = build_quality_summary()
|
||
if args.human:
|
||
_emit_human(summary)
|
||
else:
|
||
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|