#!/usr/bin/env python3 """显式导入已人工核验的事实,按文档 SHA256 缓存,不改变原文件。""" import argparse import json from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) def main(): parser = argparse.ArgumentParser() parser.add_argument("--document", type=Path, required=True) parser.add_argument("--facts", type=Path, required=True) parser.add_argument("--sha256", help="人工转录时记录的原文档哈希;样例完整清单可自动读取") args = parser.parse_args() from algorithms.main_fan.models import DocumentFacts from extensions.main_fan.documents import fingerprint, _prepare from db import db_conn, main_fan_store import tempfile raw = json.loads(args.facts.read_text(encoding="utf-8")) expected = args.sha256 if "document_facts" in raw: expected = expected or raw["assets"]["performance_pdf"]["sha256"] raw = raw["document_facts"] actual = fingerprint(args.document) if not expected or actual != expected: raise ValueError("原文件哈希与转录记录不一致或缺失,不能导入") facts = DocumentFacts.model_validate(raw) with tempfile.TemporaryDirectory(prefix="main-fan-import-") as folder: page_count = len(_prepare(args.document.resolve(), Path(folder))) for report in facts.reports: for evidence in [report.evidence, *(p.evidence for p in report.points)]: if evidence.page is None or evidence.page > page_count or not evidence.excerpt: raise ValueError("报告事实须包含有效物理页码与原文/单位") evidence.sha256, evidence.method = actual, "manual_transcription" for item in [*facts.devices, *facts.points]: if item.evidence.page is None or item.evidence.page > page_count or not item.evidence.excerpt: raise ValueError("截图事实须包含原页与原文") item.evidence.sha256, item.evidence.method = actual, "manual_transcription" try: db_conn.init_schema() main_fan_store.save_document(actual, facts.model_dump(mode="json"), "manual_transcription") finally: db_conn.close() print(f"已导入哈希绑定的人工转录:{len(facts.reports)} 份报告,{len(facts.points)} 条测点;原件未改变") if __name__ == "__main__": main()