diff --git a/backend/scripts/audit_mesh_major.py b/backend/scripts/audit_mesh_major.py new file mode 100644 index 0000000..537933c --- /dev/null +++ b/backend/scripts/audit_mesh_major.py @@ -0,0 +1,188 @@ +"""MeSH Major Topic 完整性抽检 + +从已有 mesh_headings JSONB 数据做内部一致性审计： +- 对有 mesh_headings 的文献随机抽样 +- 将 JSONB 中的 major=true 条目与 global_literature_tag 交叉验证 +- 报告漏标率（false negative）和误标率（false positive） + +用法： + cd backend && python scripts/audit_mesh_major.py [--sample 1000] +""" + +import os, sys +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://scilit:scilit_dev@localhost:5432/scilit") +os.environ.setdefault("JWT_SECRET", "dev-secret-change-in-production") +os.environ.setdefault("DEBUG", "true") +os.environ.setdefault("SPECIALTY", "oncology") + +import asyncio +import argparse +from collections import Counter + +from sqlalchemy import select, func, text +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + +from app.models.literature import GlobalLiterature, GlobalLiteratureTag, GlobalTag +from app.config import settings + + +def parse_mesh_json(headings: list | None) -> tuple[set[str], set[str]]: + """从 mesh_headings JSONB 解析 major/non-major UI。 + 返回 (major_uis, non_major_uis) + """ + major_uis: set[str] = set() + non_major_uis: set[str] = set() + if not headings: + return major_uis, non_major_uis + for h in headings: + ui = h.get("ui", "") + if not ui: + continue + if h.get("major"): + major_uis.add(ui) + else: + non_major_uis.add(ui) + return major_uis, non_major_uis + + +async def audit(sample_size: int = 1000): + engine = create_async_engine(settings.DATABASE_URL) + async with AsyncSession(engine) as db: + # 1. 总计数 + total_with_mesh = await db.scalar( + select(func.count(GlobalLiterature.id)).where( + GlobalLiterature.mesh_headings.isnot(None), + ) + ) + total_lit_tags = await db.scalar( + select(func.count(GlobalLiteratureTag.literature_id)) + ) + total_major_tags = await db.scalar( + select(func.count(GlobalLiteratureTag.literature_id)).where( + GlobalLiteratureTag.is_major == True + ) + ) + print(f"有 mesh_headings 的文献: {total_with_mesh}") + print(f"global_literature_tag 记录: {total_lit_tags}") + print(f" └─ is_major=True: {total_major_tags}") + print() + + # 2. 抽样 + count = await db.scalar(select(func.count(GlobalLiterature.id)).where( + GlobalLiterature.mesh_headings.isnot(None), + )) + if not count: + print("没有数据可审计") + return + + # 随机抽样 + ids = ( + await db.execute( + select(GlobalLiterature.id) + .where(GlobalLiterature.mesh_headings.isnot(None)) + .order_by(text("RANDOM()")) + .limit(sample_size) + ) + ).scalars().all() + + print(f"抽样 {len(ids)} 篇...") + print() + + fn_count = 0 # false negative: JSONB major=true 但没有 is_major=True tag + fp_count = 0 # false positive: is_major=True tag 但 JSONB 该 UI 非 major + total_major_from_jsonb = 0 + total_major_from_tag = 0 + lit_with_gap = 0 # 至少有一条漏标的文献数 + lit_with_fp = 0 + lit_with_any_mesh = 0 # 样本文献数 + + for lid in ids: + # 获取 mesh_headings JSONB + row = await db.execute( + select(GlobalLiterature.mesh_headings).where(GlobalLiterature.id == lid) + ) + mesh_heading = row.scalar_one_or_none() + major_uis, _ = parse_mesh_json(mesh_heading) + + # 获取 global_literature_tag 记录（通过 GlobalTag 拿到 mesh_ui） + tag_rows = await db.execute( + select(GlobalTag.mesh_ui, GlobalLiteratureTag.is_major) + .join(GlobalTag, GlobalTag.id == GlobalLiteratureTag.tag_id) + .where(GlobalLiteratureTag.literature_id == lid) + .where(GlobalTag.mesh_ui.isnot(None)) + ) + tag_map: dict[str, bool] = {} + for ui, is_major in tag_rows: + # 同一个 UI 可能有多个 tag（一对多），任一为 True 就算 major + if ui in tag_map: + tag_map[ui] = tag_map[ui] or is_major + else: + tag_map[ui] = is_major + + if not major_uis and not tag_map: + continue + + lit_with_any_mesh += 1 + + # 检查在 JSONB 中 major=true 的 UI 是否在 tag 中有 is_major=True + for ui in major_uis: + total_major_from_jsonb += 1 + tag_major = tag_map.get(ui, None) + if tag_major is None: + fn_count += 1 + elif not tag_major: + fn_count += 1 + + # 检查 tag 中 is_major=True 的 UI 在 JSONB 中是否也是 major + for ui, tag_major in tag_map.items(): + if tag_major: + total_major_from_tag += 1 + if ui not in major_uis: + fp_count += 1 + + # 按文献统计 + if major_uis: + found_major = any(tag_map.get(ui) for ui in major_uis) + if not found_major: + lit_with_gap += 1 + + print("═══ 审计结果 ═══") + print(f"实际参与审计的文献数（有 mesh_headings 或 tag）: {lit_with_any_mesh}") + print() + + fn_rate = fp_rate = 0.0 + if total_major_from_jsonb: + fn_rate = fn_count / total_major_from_jsonb * 100 + print(f"JSONB 中 major=true 的 UI 总数: {total_major_from_jsonb}") + print(f"漏标（JSONB major 但在 tag 中非 major/缺失）: {fn_count} ({fn_rate:.1f}%)") + else: + print("JSONB 中无 major=true 条目") + + if total_major_from_tag: + fp_rate = fp_count / total_major_from_tag * 100 + print(f"Tag 中 is_major=True 的 UI 总数: {total_major_from_tag}") + print(f"误标（tag 是 major 但 JSONB 该 UI 非 major）: {fp_count} ({fp_rate:.1f}%)") + else: + print("Tag 中无 is_major=True") + + print() + print(f"至少有一条漏标的文献数: {lit_with_gap} / {lit_with_any_mesh} ({lit_with_gap/lit_with_any_mesh*100:.1f}%)") + print() + + # 结论 + if fn_rate < 1 and fp_rate < 1: + print("[OK] 漏标率和误标率均 < 1%，Major Topic 数据质量良好") + elif fn_rate < 5: + print(f"[WARN] 漏标率 {fn_rate:.1f}%，在可接受范围，无需立即修复") + else: + print(f"[ALERT] 漏标率 {fn_rate:.1f}%，建议检查 pipeline MeSH 解析逻辑") + + await engine.dispose() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="MeSH Major Topic 完整性抽检") + parser.add_argument("--sample", type=int, default=1000, help="抽样数量（默认 1000）") + args = parser.parse_args() + asyncio.run(audit(sample_size=args.sample))