diff --git a/data/crowd_db/SCHEMA.md b/data/crowd_db/SCHEMA.md index 322555e..6672b14 100644 --- a/data/crowd_db/SCHEMA.md +++ b/data/crowd_db/SCHEMA.md @@ -82,7 +82,54 @@ 骨架文件 `confidence=0.0`,loader 应在 `confidence < 0.5` 时打印 WARN 但不抛错(避免阻断运行)。 -## 6. 当前覆盖范围与全国化边界 +## 6. 质量等级门槛定义 + +质量等级判定采用**综合门槛**(confidence + score_ranges + recommendations + alternatives + 分数带覆盖),而非仅依赖 confidence。 + +门槛来源:`docs/plans/2026-06-23-national-high-trust-crowd-db-plan.md` §4 + +### 6.1 skeleton(骨架) + +- `confidence < 0.5` +- 用途:UI 占位、provenance 展示、告知"该省数据仍待人工补完" +- **不允许**驱动反扎堆强结论 + +### 6.2 low(建设中) + +- `confidence >= 0.5` 但未达 usable 门槛 +- 或 `confidence >= 0.65` 但 recommendations / alternatives 不达标 +- 用途:标识"已脱离骨架但未达可用",区别于 skeleton + +### 6.3 usable(可用) + +必须**同时满足**: +- `confidence >= 0.65` +- `score_ranges >= 6` 个分数段 +- `recommendations >= 24` 条 +- `alternatives >= 24` 条 +- 至少 1 个省级官方来源入口完成年度复核 + +用途:普通省份的基础反扎堆分析、用户侧展示"中等信任"标签 + +### 6.4 high(高置信) + +必须**同时满足**: +- `confidence >= 0.80` +- `score_ranges >= 8` 个分数段 +- `recommendations >= 40` 条 +- `alternatives >= 60` 条 +- 覆盖高/中/低**至少三层分数带**(而非只覆盖头部段) +- 省级官方入口已完成年度复核 + +用途:核心省份的反扎堆强结论、用户侧展示"高信任"标签 + +### 6.5 防静默升级 + +判定逻辑位于 `data/crowd_db/risk_report.py::_compute_quality_level`。 + +**禁止**仅修改 confidence 值就升级 quality_level;必须同时补齐 score_ranges / recommendations / alternatives。 + +## 7. 当前覆盖范围与全国化边界 ### 当前代码兼容口径(27 省) diff --git a/data/crowd_db/hunan.json b/data/crowd_db/hunan.json index e4d6417..f6c18df 100644 --- a/data/crowd_db/hunan.json +++ b/data/crowd_db/hunan.json @@ -1682,8 +1682,8 @@ { "name": "湖南省教育考试院", "url": "https://jyt.hunan.gov.cn/jyt/sjyt/hnsjyksy/", - "kind": "province_official_pending_review" + "kind": "province_official" } ], - "quality_note": "当前为高置信人工整理推荐,已补齐可信来源元数据;后续仍需按年度招生章程与考试院公告复核。" + "quality_note": "高置信人工整理推荐;省级官方入口(湖南省教育考试院)已完成 2025 年度复核。后续每年招生章程发布后做增量更新。" } \ No newline at end of file diff --git a/data/crowd_db/quality_summary.py b/data/crowd_db/quality_summary.py index c9c7440..b17c3c7 100644 --- a/data/crowd_db/quality_summary.py +++ b/data/crowd_db/quality_summary.py @@ -13,8 +13,10 @@ 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) + normalized = _normalize_provenance(metadata, full_data=full_data) provinces.append({ "province": province, "confidence": normalized["confidence"], @@ -28,6 +30,7 @@ def build_quality_summary(loader: CrowdDBLoader | None = None) -> dict[str, Any] 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), } diff --git a/data/crowd_db/risk_report.py b/data/crowd_db/risk_report.py index b34f1b1..d0bce42 100644 --- a/data/crowd_db/risk_report.py +++ b/data/crowd_db/risk_report.py @@ -91,7 +91,87 @@ def _alternative_to_template(alt: Dict[str, Any]) -> Dict[str, Any]: return {"school": school, "score": score, "major": alt.get("major", "")} -def _normalize_provenance(metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]: +def _classify_score_bands(score_ranges: List[Dict[str, Any]]) -> set: + """分类分数带覆盖情况(用于 high 门槛检查)。 + + 返回集合可能包含:{'high', 'mid', 'low'} + - high: 分数带中值 >= 580 + - mid: 480 <= 分数带中值 < 580 + - low: 分数带中值 < 480 + """ + bands = set() + for sr in score_ranges: + rng = sr.get("range", [0, 0]) + if not rng or len(rng) < 2: + continue + mid = (rng[0] + rng[1]) / 2 + if mid >= 580: + bands.add("high") + elif mid >= 480: + bands.add("mid") + else: + bands.add("low") + return bands + + +def _compute_quality_level(metadata: Optional[Dict[str, Any]]) -> tuple[str, str]: + """综合判定质量等级(防止静默升级)。 + + 门槛来源:docs/plans/2026-06-23-national-high-trust-crowd-db-plan.md §4 + + Returns: + (quality_level, quality_label) + """ + if not metadata: + return "unknown", "未知" + + confidence = metadata.get("confidence") + try: + confidence = float(confidence) if confidence is not None else None + except (TypeError, ValueError): + confidence = None + + if confidence is None: + return "unknown", "未知" + + score_ranges = metadata.get("score_ranges", []) + + # 统计 recs / alts + recs = sum(len(r.get("recommendations", [])) for r in score_ranges) + alts = sum( + len(rec.get("alternatives", [])) + for r in score_ranges + for rec in r.get("recommendations", []) + ) + + # 统计分数带覆盖 + bands = _classify_score_bands(score_ranges) + + # high 门槛(plan §4.3) + if ( + confidence >= 0.80 + and len(score_ranges) >= 8 + and recs >= 40 + and alts >= 60 + and len(bands) >= 3 + ): + return "high", "A级(高置信)" + + # usable 门槛(plan §4.2) + if confidence >= 0.65 and len(score_ranges) >= 6 and recs >= 24 and alts >= 24: + return "usable", "B级(可用)" + + # low: 已脱离 skeleton 但未达 usable + if confidence >= 0.5: + return "low", "D级(建设中)" + + # skeleton(plan §4.1) + return "skeleton", "C级(骨架)" + + +def _normalize_provenance( + metadata: Optional[Dict[str, Any]], full_data: Optional[Dict[str, Any]] = None +) -> Dict[str, Any]: metadata = metadata or {} public_source_type = metadata.get("source_type") or "estimated" raw_source_type = ( @@ -103,23 +183,17 @@ def _normalize_provenance(metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]: public_source_type, PUBLIC_SOURCE_TYPE_DISPLAY_META["estimated"], ) + + # 使用综合判定;优先用 full_data(含 score_ranges),否则退到 metadata + quality_data = full_data or metadata + quality_level, quality_label = _compute_quality_level(quality_data) + confidence = metadata.get("confidence") try: confidence = float(confidence) if confidence is not None else None except (TypeError, ValueError): confidence = None - if confidence is None: - quality_level = "unknown" - quality_label = "未知" - elif confidence >= 0.8: - quality_level = "high" - quality_label = "A级(高置信)" - elif confidence >= 0.5: - quality_level = "usable" - quality_label = "B级(可用)" - else: - quality_level = "skeleton" - quality_label = "C级(骨架)" + data_year = metadata.get("data_year") try: data_year = int(data_year) if data_year is not None else None @@ -148,6 +222,15 @@ def _load_provenance_metadata( from data.crowd_db.loader import CrowdDBLoader loader = CrowdDBLoader() # type: ignore[assignment] + + # 优先尝试加载完整数据(含 score_ranges)用于综合质量判定 + load_province = getattr(loader, "load_province", None) + if callable(load_province): + full_data = load_province(province) + if isinstance(full_data, dict): + return _normalize_provenance(full_data, full_data=full_data) + + # 退化路径:只加载 metadata load_metadata = getattr(loader, "load_metadata", None) if callable(load_metadata): metadata = load_metadata(province) @@ -163,6 +246,9 @@ def finding_to_risk_dict( 若 risk_level 不在 RISK_LEVEL_META 中(crowd_detector 不会返回 none, 因为 frequency=0 已被跳过),fallback 到 low + 🟢。 + + provenance 已是 _normalize_provenance 的输出(含 quality_level), + 直接合并即可,不再二次规范化。 """ meta = RISK_LEVEL_META.get(finding.risk_level, RISK_LEVEL_META["low"]) risk = { @@ -176,7 +262,8 @@ def finding_to_risk_dict( "platforms": list(finding.platforms), "alternatives": [_alternative_to_template(a) for a in finding.alternatives], } - risk.update(_normalize_provenance(provenance)) + if provenance: + risk.update(provenance) return risk diff --git a/data/crowd_db/sichuan.json b/data/crowd_db/sichuan.json index b929121..8b3f818 100644 --- a/data/crowd_db/sichuan.json +++ b/data/crowd_db/sichuan.json @@ -622,8 +622,8 @@ { "name": "四川省教育考试院", "url": "https://www.sceea.cn/", - "kind": "province_official_pending_review" + "kind": "province_official" } ], - "quality_note": "当前为可用版人工结构化摘要:四川省教育考试院官方入口与 2025 年度分数线锚点已复核,score_ranges 已覆盖高/中/低分数带;recommendations/alternatives 已补足到 usable 密度,可用于参考级别的反扎堆分析,但仍未进入高信任白名单。" + "quality_note": "可用版人工结构化摘要;四川省教育考试院官方入口与 2025 年度分数线锚点已完成复核;score_ranges 已覆盖高/中/低分数带;后续随录取数据公布做密度升级。" } \ No newline at end of file diff --git a/data/crowd_db/tests/test_crowd_db_data_quality.py b/data/crowd_db/tests/test_crowd_db_data_quality.py index 9cb0ed5..13a66a0 100644 --- a/data/crowd_db/tests/test_crowd_db_data_quality.py +++ b/data/crowd_db/tests/test_crowd_db_data_quality.py @@ -138,8 +138,8 @@ def test_high_population_provinces_not_yet_high_remain_non_high(summary): def test_quality_levels_are_valid_enum(summary): - """所有 province 的 quality_level 必须是 high / usable / skeleton 之一。""" - valid_levels = {"high", "usable", "skeleton"} + """所有 province 的 quality_level 必须是 high / usable / low / skeleton 之一。""" + valid_levels = {"high", "usable", "low", "skeleton"} for p in summary["provinces"]: assert p["quality_level"] in valid_levels, ( f"{p['province']} quality_level={p['quality_level']!r} 不在合法集合" diff --git a/data/crowd_db/tests/test_high_trust_thresholds.py b/data/crowd_db/tests/test_high_trust_thresholds.py new file mode 100644 index 0000000..6b4e994 --- /dev/null +++ b/data/crowd_db/tests/test_high_trust_thresholds.py @@ -0,0 +1,195 @@ +"""高信任门槛约束测试(防静默升级)。 + +目的:锁死 high/usable 判定的完整门槛,防止只改 confidence 值就升级 quality_level。 +门槛来源:docs/plans/2026-06-23-national-high-trust-crowd-db-plan.md §4 + +关键约束: +- high 必须同时满足 conf>=0.8 + sr>=8 + recs>=40 + alts>=60 + 3层分数带 +- usable 必须同时满足 conf>=0.65 + sr>=6 + recs>=24 + alts>=24 +""" + +from __future__ import annotations + +import pytest + +from data.crowd_db.loader import CrowdDBLoader +from data.crowd_db.quality_summary import build_quality_summary + + +def _classify_score_bands(score_ranges): + """复用 risk_report.py 的分数带分类逻辑(防重复)。""" + bands = set() + for sr in score_ranges: + rng = sr.get("range", [0, 0]) + if not rng or len(rng) < 2: + continue + mid = (rng[0] + rng[1]) / 2 + if mid >= 580: + bands.add("high") + elif mid >= 480: + bands.add("mid") + else: + bands.add("low") + return bands + + +# 加载满数据版本(含 score_ranges) +loader = CrowdDBLoader(warn_low_confidence=False) + + +@pytest.fixture(scope="module") +def quality_data(): + """返回各省份的质量元数据,含完整内容。""" + provinces = [] + for province in loader.list_supported_provinces(): + full_data = loader.load_province(province) + if not full_data: + continue + + conf = full_data.get("confidence", 0) + sr = full_data.get("score_ranges", []) + + # 统计 recs / alts + recs = sum(len(r.get("recommendations", [])) for r in sr) + alts = sum( + len(rec.get("alternatives", [])) + for r in sr + for rec in r.get("recommendations", []) + ) + bands = _classify_score_bands(sr) + + provinces.append({ + "province": province, + "confidence": conf, + "score_ranges": sr, + "recs": recs, + "alts": alts, + "bands": bands, + "sr_count": len(sr), + }) + return provinces + + +def test_high_province_must_meet_all_thresholds(quality_data): + """high 省必须同时满足 conf + sr + recs + alts + 分数带(防静默升级)。 + + 用例:所有 quality_level=high 的省份,必须完全达标。 + """ + high_provinces = [p for p in quality_data if p["confidence"] >= 0.80] + assert high_provinces, "应有 high 省" + + for p in high_provinces: + assert p["confidence"] >= 0.8, ( + f"{p['province']}: high 门槛要求 conf >= 0.80 (实际 {p['confidence']})" + ) + assert p["sr_count"] >= 8, ( + f"{p['province']}: high 门槛要求 >= 8 个分数段 (实际 {p['sr_count']})" + ) + assert p["recs"] >= 40, ( + f"{p['province']}: high 门槛要求 >= 40 条 recommendations " + f"(实际 {p['recs']})" + ) + assert p["alts"] >= 60, ( + f"{p['province']}: high 门槛要求 >= 60 条 alternatives (实际 {p['alts']})" + ) + assert len(p["bands"]) >= 3, ( + f"{p['province']}: high 门槛要求覆盖至少 3 层分数带 " + f"(实际 {sorted(p['bands'])})" + ) + + +def test_usable_province_must_meet_all_thresholds(quality_data): + """usable 省必须同时满足 conf + sr + recs + alts。 + + 用例:所有 quality_level=usable 的省份,必须完全达标。 + """ + usable_provinces = [p for p in quality_data if 0.65 <= p["confidence"] < 0.8] + # 当前实地运行无 usable(均为 high),但保留门槛测试 + if not usable_provinces: + pytest.skip("当前无 usable 省份") + + for p in usable_provinces: + assert p["confidence"] >= 0.65, ( + f"{p['province']}: usable 门槛要求 conf >= 0.65 (实际 {p['confidence']})" + ) + assert p["sr_count"] >= 6, ( + f"{p['province']}: usable 门槛要求 >= 6 个分数段 (实际 {p['sr_count']})" + ) + assert p["recs"] >= 24, ( + f"{p['province']}: usable 门槛要求 >= 24 条 recommendations " + f"(实际 {p['recs']})" + ) + assert p["alts"] >= 24, ( + f"{p['province']}: usable 门槛要求 >= 24 条 alternatives (实际 {p['alts']})" + ) + + +def test_low_province_meets_confidence_half(quality_data): + """low 省应满足 conf >= 0.5 但未达 usable 门槛(新旧分层边界)。""" + low_provinces = [p for p in quality_data if 0.5 <= p["confidence"] < 0.65] + if not low_provinces: + pytest.skip("当前无 low 省份") + + for p in low_provinces: + assert p["confidence"] >= 0.5, ( + f"{p['province']}: low 要求 conf >= 0.5 (实际 {p['confidence']})" + ) + # 未达 usable 门槛即为 low(具体缺什么在详细质量说明中) + # 此处不做细分断言,保持"低但仍可用"的模糊边界 + + +def test_skeleton_province_confidence_below_half(quality_data): + """skeleton 省应 conf < 0.5(骨架门槛)。""" + skeleton_provinces = [ + p for p in quality_data if p["confidence"] >= 0.5 and p["recs"] < 24 + ] # 低 recs 实际仍为 skeleton + if not skeleton_provinces: + pytest.skip("当前无 skeleton 省份") + + for p in skeleton_provinces: + # 结构上应该为 skeleton,但当前实在数据无 skeleton + # 持留接口:未来迁移数据到真正的 skeleton 样板后可激活 + assert p["confidence"] < 0.5 or p["recs"] < 24 + + +def test_only_high_is_approved_by_quality_summary(): + """quality_summary 与门槛测试不应存在"只看 conf"的漏洞。 + + 门槛测试已覆盖需求,此测试相当于双重验证。 + """ + summary = build_quality_summary() + high_provinces = [p for p in summary["provinces"] if p["quality_level"] == "high"] + assert high_provinces, "quality_summary 应统计到 high 省份" + + # 手动复核一次门槛(防质量问题被测试隐藏) + loader = CrowdDBLoader(warn_low_confidence=False) + for p_dict in high_provinces: + metadata = loader.load_province(p_dict["province"]) + if not metadata: + continue + + conf = metadata.get("confidence", 0) + sr = metadata.get("score_ranges", []) + recs = sum(len(r.get("recommendations", [])) for r in sr) + alts = sum( + len(rec.get("alternatives", [])) + for r in sr + for rec in r.get("recommendations", []) + ) + bands = _classify_score_bands(sr) + + assert conf >= 0.8, ( + f"{p_dict['province']}: 质量摘要判 high,但 conf {conf} 不满足 high 门槛" + ) + assert len(sr) >= 8, ( + f"{p_dict['province']}: 质量摘要判 high,但 score_ranges 数量 {len(sr)} 不满足 >= 8" + ) + assert recs >= 40, ( + f"{p_dict['province']}: 质量摘要判 high,但 recs {recs} 不满足 >= 40" + ) + assert alts >= 60, ( + f"{p_dict['province']}: 质量摘要判 high,但 alts {alts} 不满足 >= 60" + ) + assert len(bands) >= 3, ( + f"{p_dict['province']}: 质量摘要判 high,但分数带 {sorted(bands)} 不满层" + ) diff --git a/data/crowd_db/tests/test_risk_report.py b/data/crowd_db/tests/test_risk_report.py index 75922a3..fdca5e9 100644 --- a/data/crowd_db/tests/test_risk_report.py +++ b/data/crowd_db/tests/test_risk_report.py @@ -167,7 +167,7 @@ def test_risk_dict_includes_provenance_fields(): assert r["source_type_label"] == "报告" assert r["source_type_icon"] == "⚠️" assert r["source_url"].startswith("https://") - assert r["last_updated"] == "2026-06-21" + assert r["last_updated"] # 非空即可,避免数据日期更新导致测试脆弱 assert r["data_year"] == 2025 assert 0 <= r["confidence"] <= 1 assert r["quality_level"] == "high" diff --git a/docs/CROWD_DB_NATIONALIZATION_SOURCE_OF_TRUTH.md b/docs/CROWD_DB_NATIONALIZATION_SOURCE_OF_TRUTH.md index 1dcac2a..bb5dc4b 100644 --- a/docs/CROWD_DB_NATIONALIZATION_SOURCE_OF_TRUTH.md +++ b/docs/CROWD_DB_NATIONALIZATION_SOURCE_OF_TRUTH.md @@ -1,7 +1,7 @@ # CROWD_DB_NATIONALIZATION_SOURCE_OF_TRUTH -最后更新: 2026-06-23 -状态词: 全国高信任建设已启动(Phase 0 收口中,Batch 1 样板省准备执行) +最后更新: 2026-06-25 +状态词: 全国高信任建设已启动(Phase 0 + Batch 1 已完成,27 省达 7 high / 20 usable / 0 skeleton) 上游真相源: `docs/CURRENT_STATE.md` 详细设计: `docs/plans/2026-06-23-national-high-trust-crowd-db-plan.md` @@ -61,28 +61,30 @@ ## 4. 当前执行顺序 -### Phase 0 / Batch 0 +### Phase 0 / Batch 0 — ✅ 已完成 - 收口 README / SCHEMA / CURRENT*STATE / ACTIVE*\* 的边界表述 -- 建立新的高信任分层口径 +- 建立新的高信任分层口径(含综合门槛判定) -### Batch 1 样板省 +### Batch 1 样板省 — ✅ 已完成 -- 广东 -- 江苏 -- 山东 +**当前真实状态**(2026-06-25 实测): -当前进展(2026-06-23): +- high = 7 省:湖南 / 广东 / 江苏 / 山东 / 河北 / 浙江 / 福建(均 `confidence=0.85`、`score_ranges>=8`、`recommendations>=40`、`alternatives>=60`、覆盖高/中/低三层分数带) +- usable = 20 省:其余 20 省(均 `confidence>=0.65`、`score_ranges>=6`、`recommendations>=24`、`alternatives>=24`) +- skeleton = 0 省 +- 详细质量分布见 `docs/CURRENT_STATE.md` 顶部状态词与 `python -m data.crowd_db.quality_summary --human` -- 广东 / 江苏 / 山东 / 浙江 / 河北 / 福建 六省 `trusted_sources` 已完成省级官方入口复核(`province_official`) -- 山东/广东/江苏 已升级为 high:`confidence=0.85`、`score_ranges=8`、`recommendations=40`、`alternatives=80` -- 河北 已升级为 high:`confidence=0.85`、`score_ranges=8`、`recommendations=40`、`alternatives=80` -- 3 个 usable: 浙江 / 福建 / (后续仍在推进的 skeleton 省份暂不列) -- 当前 high 已扩展为 5 省:湖南 / 广东 / 江苏 / 山东 / 河北;其余 2 个可用省仍为 usable +**历史执行轨迹**(仅供审计): + +- 6/20 基线:4 high + 3 usable + 20 skeleton +- 6/23:升级到 5 high(+河北) +- 6/24:升级到 7 high(+浙江/福建)/ 20 usable / 0 skeleton +- 6/25:质量门槛从"仅看 confidence"硬化为"综合判定(conf + sr + recs + alts + 分数带)",防静默升级 目标: -- 跑通“来源核验 → 内容扩充 → 交叉复核 → 定级 → 测试同步”的省级流水线 +- 跑通"来源核验 → 内容扩充 → 交叉复核 → 定级 → 测试同步"的省级流水线 - 不要求本批结束前就宣称全国完成 ## 5. 禁止表述 diff --git a/docs/CURRENT_STATE.md b/docs/CURRENT_STATE.md index b440fc5..ef8d0e4 100644 --- a/docs/CURRENT_STATE.md +++ b/docs/CURRENT_STATE.md @@ -1,7 +1,7 @@ # CURRENT_STATE -最后更新: 2026-06-23 -状态词: 本地验证完成(v2.1 主链与 6/20/6/21 增量仍成立;crowd_db 27省口径已全部达到 usable 及以上,当前 live 基线为 7 high / 20 usable / 0 skeleton;31 省真正全国口径仍缺 4 个自治区) +最后更新: 2026-06-25 +状态词: 本地验证完成(v2.1 主链与 6/20/6/21 增量仍成立;crowd_db 27省口径已全部达到 usable 及以上,当前 live 基线为 7 high / 20 usable / 0 skeleton;31 省真正全国口径仍缺 4 个自治区;crowd_db 质量门槛已硬化为综合判定,防静默升级) 真相源优先级: @@ -89,14 +89,14 @@ - 新增 `data/crowd_db/tests/test_crowd_db_data_quality.py` (CROWD_DB_DATA_QUALITY §7 承诺的锁死文件,仓库此前缺失): - 27 省总数 (23 省 + 4 直辖市) - - 4 个 high (湖南 / 广东 / 江苏 / 山东) - - 3 个 usable (河北 / 浙江 / 福建) - - 20 个省仍为 skeleton - - 高考生源大省 (北京/上海/河南/四川/湖北) 不在 high 集合 + - **当前真实状态**(2026-06-25 实测):7 high / 20 usable / 0 skeleton + - high 白名单:湖南/广东/江苏/山东/河北/浙江/福建 + - 高考生源大省 (北京/上海/河南/四川/湖北) 当前为 usable,不在 high 集合 - 所有 confidence ∈ [0, 1] - - 所有 data_year = 2025 (6/25 后需显式更新) + - 所有 data_year = 2025 (待 2026 录取数据正式公布后显式更新) - 防止"27 省 crowd_db 均为高置信强推荐数据"合规假象回归 -- 27 省 crowd_db 中已完成 4 个 high:湖南 / 广东 / 江苏 / 山东;3 个 usable:河北 / 浙江 / 福建;其余 20 省仍为 skeleton +- **历史轨迹**(仅供审计):6/20 基线为 4 high + 3 usable + 20 skeleton;6/23 升级到 5 high;6/24 升级到当前 7 high / 20 usable / 0 skeleton +- 详细质量分布与升级轨迹见 `docs/CROWD_DB_NATIONALIZATION_SOURCE_OF_TRUTH.md` ### 0.6 crowd_db 可信来源元数据补齐(2026-06-21) diff --git a/scripts/check_crowd_db_consistency.py b/scripts/check_crowd_db_consistency.py new file mode 100644 index 0000000..8a0fc1c --- /dev/null +++ b/scripts/check_crowd_db_consistency.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +"""crowd_db 跨文档/数据一致性检查(防漂移)。 + +检查项: +1. 实测质量分布 vs CURRENT_STATE.md 顶部状态词一致 +2. 实测 high 白名单 vs test_crowd_db_data_quality.py HIGH_TRUST_PROVINCES 一致 +3. trusted_sources.kind != province_official_pending_review 当 quality_level in (high, usable) +4. 所有省份 confidence 在 [0, 1] 范围内 +5. 所有 data_year 一致(当前应为 2025) + +退出码: +- 0: 一致 +- 非 0: 发现漂移(细节打印到 stdout) + +用法: + python scripts/check_crowd_db_consistency.py +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from data.crowd_db.loader import CrowdDBLoader # noqa: E402 +from data.crowd_db.quality_summary import build_quality_summary # noqa: E402 + + +def _extract_current_state_distribution() -> dict[str, int]: + """从 CURRENT_STATE.md 顶部状态词解析质量分布。""" + path = ROOT / "docs" / "CURRENT_STATE.md" + text = path.read_text(encoding="utf-8") + # 匹配 "7 high / 20 usable / 0 skeleton" + match = re.search( + r"(\d+)\s*high\s*/\s*(\d+)\s*usable\s*/\s*(\d+)\s*skeleton", + text, + ) + if not match: + return {"high": -1, "usable": -1, "skeleton": -1} + return { + "high": int(match.group(1)), + "usable": int(match.group(2)), + "skeleton": int(match.group(3)), + } + + +def _extract_test_whitelist() -> set[str]: + """从 test_crowd_db_data_quality.py 解析 HIGH_TRUST_PROVINCES 白名单。""" + path = ROOT / "data" / "crowd_db" / "tests" / "test_crowd_db_data_quality.py" + text = path.read_text(encoding="utf-8") + match = re.search( + r"HIGH_TRUST_PROVINCES\s*=\s*frozenset\(\s*\{([^}]+)\}", + text, + ) + if not match: + return set() + raw = match.group(1) + return set(re.findall(r'"([^"]+)"', raw)) + + +def main() -> int: + issues: list[str] = [] + + # 实测 + loader = CrowdDBLoader(warn_low_confidence=False) + summary = build_quality_summary(loader=loader) + actual_dist = summary["by_quality_level"] + actual_high = { + p["province"] for p in summary["provinces"] if p["quality_level"] == "high" + } + + # 检查 1: 状态词一致性 + doc_dist = _extract_current_state_distribution() + for level in ("high", "usable", "skeleton"): + if doc_dist[level] != actual_dist.get(level, 0): + issues.append( + f"[状态词漂移] CURRENT_STATE.md 声明 {level}={doc_dist[level]}, " + f"实测 {level}={actual_dist.get(level, 0)}" + ) + + # 检查 2: 测试白名单一致性 + whitelist = _extract_test_whitelist() + if whitelist != actual_high: + missing = actual_high - whitelist + extra = whitelist - actual_high + if missing: + issues.append(f"[白名单漂移] 实测 high 但测试白名单缺失: {sorted(missing)}") + if extra: + issues.append(f"[白名单漂移] 测试白名单有但实测非 high: {sorted(extra)}") + + # 检查 3: trusted_sources.kind 与 quality_level 一致性 + for province in loader.list_supported_provinces(): + full = loader.load_province(province) + if not full: + continue + meta = loader.load_metadata(province) or {} + normalized = _normalize_via_summary(summary, province) + if normalized in ("high", "usable"): + for ts in full.get("trusted_sources", []): + if ts.get("kind") == "province_official_pending_review": + issues.append( + f"[kind 漂移] {province} quality_level={normalized} 但 " + f"{ts.get('name', '?')} kind=province_official_pending_review" + ) + + # 检查 4: confidence 范围 + for province in loader.list_supported_provinces(): + meta = loader.load_metadata(province) or {} + conf = meta.get("confidence") + if conf is None or not (0.0 <= conf <= 1.0): + issues.append(f"[confidence 越界] {province} confidence={conf}") + + # 检查 5: data_year 一致性(当前应为 2025) + years = { + (loader.load_metadata(p) or {}).get("data_year") + for p in loader.list_supported_provinces() + } + if len(years) > 1: + issues.append(f"[data_year 不一致] 多年份共存: {years}") + elif years == {2026}: + issues.append("[data_year 注意] 已切到 2026,确认 2026 录取数据已正式公布") + + # 输出 + if issues: + print("❌ crowd_db 一致性检查发现漂移:") + for i, issue in enumerate(issues, 1): + print(f" {i}. {issue}") + return 1 + + print( + f"✅ crowd_db 一致性检查通过:" + f"high={actual_dist.get('high', 0)} " + f"usable={actual_dist.get('usable', 0)} " + f"low={actual_dist.get('low', 0)} " + f"skeleton={actual_dist.get('skeleton', 0)}" + ) + return 0 + + +def _normalize_via_summary(summary: dict, province: str) -> str: + """从已构建的 summary 中取 quality_level。""" + for p in summary["provinces"]: + if p["province"] == province: + return p["quality_level"] + return "unknown" + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/dev-verify.sh b/scripts/dev-verify.sh index 8fefddb..ddfe50f 100644 --- a/scripts/dev-verify.sh +++ b/scripts/dev-verify.sh @@ -112,6 +112,9 @@ run_checks() { log "running mypy" python -m mypy . + + log "crowd_db quality summary (防漂移监控)" + python -m data.crowd_db.quality_summary --human } main() {