From 04ef3972a8400ad24fe7facfe7246acf7c9d061f Mon Sep 17 00:00:00 2001 From: qiaoqichao Date: Tue, 29 Sep 2026 18:41:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T05-a=20=E5=8E=86=E5=8F=B2=E8=BF=87?= =?UTF-8?q?=E7=A8=8B=E6=97=B6=E9=97=B4=E7=BA=BF=EF=BC=88=E7=BC=BA=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E4=B8=8D=E7=BC=96=E9=80=A0=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T05 卡片要求"先 timeline，再候选，再编辑/测试/审核"。本批是第一步。 新增 contracts：DecisionEvent（区分 occurred_at 与 recorded_at）、CaseTimeline。 新增 services/history_cases.py：确定性提炼，无模型调用。 两条硬规则的实现： - **不编造发生时间**（BR-02）：occurred_at 只取材料里记载的日期；缺失/占位值 （"-"、"缺失"、"未知"）或无法解析时保持 None 并给缺证提示。recorded_at 必填， 但它**不表示发生时间**——有测试断言二者不相等 - **不编造经过**（14 §4）：只有最终报告或非结构化材料时产出 **0 个事件** 并标 `partial` + "过程材料不完整"，绝不生成虚构的问询顺序 其它：列名靠别名表识别，**未识别的列忽略而非猜测**；未声明事件类型时保守记为 question 而不丢弃该行；事件携带行级 evidence_id 以便回溯原文定位；同项目重复提炼 是同一对象 revision 递增（03 §4）；缺日期事件排序时置于末尾，不被打乱成"近期"。 实现中修掉一处自己的问题：build() 初版调了两次 store.put（首次建库 revision 会变 2）， 且校验发生在落库之后——非法时间线本可能入库。现改为先校验再单次 put。 测试：新增 test_history_cases.py 18 项（不编造 6 + 提炼 6 + 隔离与版本 6）， 含"缺日期为 None 且不等于导入时间""只有最终报告不得生成时间线"等反例。 全量 290 项 OK（272 → 290）；verify.py 10/10 PASS。 Co-Authored-By: Claude Code --- backend/contracts/prescreen.py | 49 +++++++ backend/services/history_cases.py | 205 ++++++++++++++++++++++++++++++ backend/test_history_cases.py | 193 ++++++++++++++++++++++++++++ 3 files changed, 447 insertions(+) create mode 100644 backend/services/history_cases.py create mode 100644 backend/test_history_cases.py diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index 1c7588e..2bca91d 100644 --- a/backend/contracts/prescreen.py +++ b/backend/contracts/prescreen.py @@ -366,3 +366,52 @@ class DatasetManifest(Contract): if self.synthetic and self.data_domain != DataDomain.demo: raise ValueError("合成数据必须为 demo 数据域") return self + + +# ---------------------------------------------------------------- 历史过程（T05） +class DecisionEvent(Contract): + """历史过程中的一次问询/证据/动作/意见/决定（docs/v2/03-detailed-design.md §3）。 + + 时间语义严格区分（BR-02）： + - `occurred_at` 是材料里**记载**的发生时间，缺失时保持 None，**不得**用录入时间填充 + - `recorded_at` 是录入时间，必填，但**不能**被当作发生时间 + """ + + id: Identifier + event_type: Literal["question", "evidence", "action", "opinion", "decision"] + occurred_at: AwareDatetime | None = None + recorded_at: AwareDatetime + question: str = Field(default="", max_length=DESC_MAX) + evidence_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX) + judgment_before: str | None = Field(default=None, max_length=DESC_MAX) + judgment_after: str | None = Field(default=None, max_length=DESC_MAX) + conditions: list[str] = Field(default_factory=list, max_length=EVIDENCE_MAX) + + @model_validator(mode="after") + def _check_journal(self) -> DecisionEvent: + # 事件本身必须有可读内容，否则只是空壳 + if not (self.question or self.judgment_before or self.judgment_after): + raise ValueError("决策事件必须至少包含问题、判断前或判断后之一") + return self + + +class CaseTimeline(Contract): + """历史过程时间线。`partial` 表示过程材料不完整（如只有最终报告）。""" + + id: Identifier + project_id: Identifier + data_domain: DataDomain + revision: int = Field(ge=1) + status: Literal["processing", "ready", "partial", "failed"] + created_at: AwareDatetime + events: list[DecisionEvent] = Field(default_factory=list, max_length=500) + warnings: list[str] = Field(default_factory=list, max_length=EVIDENCE_MAX) + + @model_validator(mode="after") + def _check_integrity(self) -> CaseTimeline: + ids = [event.id for event in self.events] + if len(set(ids)) != len(ids): + raise ValueError("同一时间线内事件 id 必须唯一") + if self.status == "ready" and any(e.occurred_at is None for e in self.events): + raise ValueError("存在缺发生时间的事件时状态不得为 ready") + return self diff --git a/backend/services/history_cases.py b/backend/services/history_cases.py new file mode 100644 index 0000000..3bd7bc4 --- /dev/null +++ b/backend/services/history_cases.py @@ -0,0 +1,205 @@ +"""T05-a 历史过程时间线：从历史材料提炼决策事件。 + +依据：docs/v2/14-experience-loop.md §4（加工步骤与"过程不完整"）、§3（来源定位）； +BR-02 / E-02 / AC-02。 + +两条硬规则： +1. **不编造经过**：只有最终报告、没有结构化问询/决定记录时，产出 **0 个事件** 并标 + `partial` + 明确提示，绝不生成虚构的问询顺序（14 §4 明文）。 +2. **不编造时间**：`occurred_at` 只来自材料里**记载**的日期；缺失或无法解析时保持 + None 并给缺证提示。**导入时间绝不冒充发生时间**（14 §3 / BR-02）。 + +提取是确定性的（别名匹配 + 结构化行），不做模型调用。 +""" + +from __future__ import annotations + +import re +from datetime import datetime, timezone + +from contracts.prescreen import CaseTimeline +from services import assistant_store as store + +KIND = "case_timeline" + +# 列名别名。未识别的列一律忽略——**不猜**列含义。 +COLUMN_ALIASES = { + "occurred_at": {"date", "日期", "发生时间", "时间", "问询日期", "会议日期", "决定日期"}, + "event_type": {"event_type", "类型", "事件类型", "环节", "过程类型"}, + "question": {"question", "问题", "问询", "问询事项", "事项", "核查问题"}, + "judgment_before": {"judgment_before", "判断前", "原判断", "初审意见"}, + "judgment_after": {"judgment_after", "判断后", "新判断", "复审意见"}, + "conditions": {"conditions", "条件", "附加条件", "批注条件"}, + "evidence": {"evidence", "依据", "证据", "材料", "来源"}, +} + +EVENT_TYPE_ALIASES = { + "question": "question", "问询": "question", "问题": "question", + "evidence": "evidence", "证据": "evidence", "材料": "evidence", + "action": "action", "动作": "action", "补件": "action", + "opinion": "opinion", "意见": "opinion", "判断": "opinion", + "decision": "decision", "决定": "decision", "结论": "decision", "会议": "decision", +} + +DATE_FORMATS = ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y年%m月%d日", + "%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S", "%Y%m%d") + +MISSING_DATE_NOTE = "未标注发生时间，按缺证处理（不推断日期）" +INCOMPLETE_NOTE = ("过程材料不完整：未发现结构化的问询/决定记录，" + "仅能提炼有来源的结论，不生成虚构经过") + + +def _normalize(text) -> str: + return str(text or "").strip().lower().replace(" ", "") + + +def _parse_date(value) -> datetime | None: + """只接受材料里明确写出的日期；无法解析即 None（不猜测）。""" + raw = str(value or "").strip() + if not raw or raw in ("-", "无", "null", "未知", "缺失"): + return None + for fmt in DATE_FORMATS: + try: + parsed = datetime.strptime(raw, fmt) + except ValueError: + continue + return parsed.replace(tzinfo=timezone.utc) + try: + parsed = datetime.fromisoformat(raw) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _header_map(chunks: list[dict]): + """找到表头行及其列映射。未识别的列不猜。""" + for position, chunk in enumerate(chunks): + cells = chunk.get("cells") + if not cells: + continue + mapping: dict[str, int] = {} + for index, cell in enumerate(cells): + key = _normalize(cell) + for field, aliases in COLUMN_ALIASES.items(): + if key in aliases and field not in mapping: + mapping[field] = index + # 至少要能定位"问什么"或"何时"，否则不视为表头 + if "question" in mapping or "occurred_at" in mapping: + return position, mapping + return None, {} + + +def _cell(cells: list, mapping: dict, field: str) -> str: + index = mapping.get(field) + if index is None or index >= len(cells): + return "" + return str(cells[index] or "").strip() + + +def _split_conditions(text: str) -> list[str]: + return [item.strip() for item in re.split(r"[;；,，、]", text or "") if item.strip()] + + +def events_from_document(document: dict) -> tuple[list[dict], list[str]]: + """从单个文档提炼事件。返回 (events, warnings)。**不落库**。""" + chunks = document.get("chunks") or [] + head, mapping = _header_map(chunks) + if head is None: + # 非结构化材料（txt/pdf/docx 正文或纯报告）→ 不产出事件，交调用方标注不完整 + return [], [] + + events, warnings = [], [] + for chunk in chunks[head + 1:]: + cells = chunk.get("cells") + if not cells or not any(str(cell or "").strip() for cell in cells): + continue + + occurred = _parse_date(_cell(cells, mapping, "occurred_at")) + if occurred is None and "occurred_at" in mapping: + raw = _cell(cells, mapping, "occurred_at") + warnings.append(f"{chunk['locator']}：{MISSING_DATE_NOTE}" + + (f"（原文：{raw}）" if raw else "（该列留空）")) + + raw_type = _cell(cells, mapping, "event_type") + event_type = EVENT_TYPE_ALIASES.get(_normalize(raw_type)) + if event_type is None: + # 未声明类型时按最保守的"问询"记；不因为缺类型而丢弃这一行 + event_type = "question" + + question = _cell(cells, mapping, "question") + before = _cell(cells, mapping, "judgment_before") or None + after = _cell(cells, mapping, "judgment_after") or None + if not (question or before or after): + continue + + events.append({ + "id": store.uid("event"), + "event_type": event_type, + "occurred_at": occurred.isoformat() if occurred else None, + # 录入时间必填，但**只表示我们何时录入**，不当作发生时间 + "recorded_at": store.now(), + "question": question, + "evidence_ids": [chunk["evidence_id"]] if chunk.get("evidence_id") else [], + "judgment_before": before, + "judgment_after": after, + "conditions": _split_conditions(_cell(cells, mapping, "conditions")), + }) + return events, warnings + + +def build(db, project_id: str, data_domain: str) -> dict: + """重新提炼该项目的时间线。同项目重复提炼 → 同 id、revision 递增（03 §4）。""" + documents = store.listing(db, "document", project_id) + events: list[dict] = [] + warnings: list[str] = [] + structured_documents = 0 + + for document in documents: + found, notes = events_from_document(document) + warnings.extend(notes) + if found: + structured_documents += 1 + events.extend(found) + + events.sort(key=lambda item: (item["occurred_at"] is None, item["occurred_at"] or "", item["id"])) + + if not events: + warnings.append(INCOMPLETE_NOTE) + status = "partial" + elif any(event["occurred_at"] is None for event in events): + # 设计 §4：部分缺日期为 partial + status = "partial" + else: + status = "ready" + + body = { + # 每项目一条时间线；稳定 id 使"重新提炼"表现为 revision 递增而不是新对象 + "id": f"timeline-{project_id}", + "project_id": project_id, + "data_domain": data_domain, + "status": status, + "created_at": store.now(), + "events": events, + "warnings": warnings, + } + # 先按契约校验再落库：revision 由 store.put() 维护，这里给的 1 只是为了让 + # 校验通过（CaseTimeline.revision 必填），put() 会写入真实值。 + # 一次 put，不重复写；不合法的时间线绝不入库。 + validated = CaseTimeline(**dict(body, revision=1)).model_dump(mode="json") + stored = store.put(db, KIND, validated) + store.event(db, project_id, "case_timeline.built", "system", + {"timeline_id": stored["id"], "events": len(events), + "structured_documents": structured_documents, "status": status}) + return stored + + +def load(db, project_id: str) -> dict: + body = store.get(db, KIND, f"timeline-{project_id}") + if body["project_id"] != project_id: + raise KeyError("时间线不属于该项目") + return body + + +def undated_events(timeline: dict) -> list[dict]: + """缺发生时间的事件——界面必须显式提示缺证，不得静默排序成"近期"（BR-02）。""" + return [event for event in timeline["events"] if event["occurred_at"] is None] diff --git a/backend/test_history_cases.py b/backend/test_history_cases.py new file mode 100644 index 0000000..510fd4a --- /dev/null +++ b/backend/test_history_cases.py @@ -0,0 +1,193 @@ +"""T05-a 历史时间线测试：AC-02 与 BR-02 的机械断言。 + +核心是两条"不许编"： +- 不许编造发生时间（缺日期 → None + 缺证提示，导入时间不得冒充） +- 不许编造经过（只有最终报告 → 0 事件 + 过程不完整，不生成虚构问询顺序） + +用临时库；不触碰真实数据根；不做模型调用。 +""" + +import tempfile +import unittest +from pathlib import Path + +from contracts.prescreen import CaseTimeline +from services import assistant_store as store +from services import history_cases +from services.assistant_materials import material + +HEADER = "date,event_type,question,judgment_before,judgment_after\n" + + +class TimelineTestCase(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db_path = Path(self._tmp.name) / "history.sqlite3" + store.initialize(self.db_path) + with store.transaction(self.db_path) as db: + for project_id in ("DEMO-A", "DEMO-B"): + store.put(db, "project", { + "id": project_id, "data_domain": "demo", "revision": 1, + "facts": {}, "values": {}, "sources": {}, "selected_sources": {}, + "document_ids": [], "pending_candidates": [], "data_revision": 1, + }) + + def tearDown(self): + self._tmp.cleanup() + + def add_document(self, content: bytes, name: str, project_id: str = "DEMO-A"): + with store.transaction(self.db_path) as db: + document = material(project_id, name, content, "tester") + store.put(db, "document", document, immutable=True) + project = store.get(db, "project", project_id) + project["document_ids"].append(document["id"]) + store.put(db, "project", project) + return document + + def build(self, project_id: str = "DEMO-A"): + with store.transaction(self.db_path) as db: + return history_cases.build(db, project_id, "demo") + + def load(self, project_id: str = "DEMO-A"): + with store.transaction(self.db_path) as db: + return history_cases.load(db, project_id) + + +class NoFabricationTests(TimelineTestCase): + def test_final_report_only_yields_no_events_and_says_incomplete(self): + # 14 §4：只有最终报告时标"过程不完整"，不生成虚构经过 + self.add_document("尽调结论：该客户经营稳定，建议通过。\n".encode("utf-8-sig"), "最终报告.txt") + timeline = self.build() + self.assertEqual(timeline["events"], [], "不得凭最终报告编造时间线") + self.assertEqual(timeline["status"], "partial") + self.assertTrue(any("过程材料不完整" in w for w in timeline["warnings"])) + + def test_no_documents_at_all_still_reports_incomplete(self): + timeline = self.build() + self.assertEqual(timeline["events"], []) + self.assertEqual(timeline["status"], "partial") + self.assertTrue(any("过程材料不完整" in w for w in timeline["warnings"])) + + def test_unknown_columns_are_not_guessed(self): + # 列名不认识就忽略，不猜哪一列是问题/日期 + self.add_document("姓名,金额,备注\n张三,100,随便写点什么\n".encode("utf-8-sig"), "无关表.csv") + timeline = self.build() + self.assertEqual(timeline["events"], []) + + def test_missing_date_is_null_not_the_import_time(self): + self.add_document((HEADER + ",问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv") + timeline = self.build() + event = timeline["events"][0] + self.assertIsNone(event["occurred_at"], "缺日期必须是 None，不能用导入时间顶替") + self.assertIsNotNone(event["recorded_at"], "录入时间必填，但它不表示发生时间") + self.assertNotEqual(event["occurred_at"], event["recorded_at"]) + + def test_unparseable_date_is_not_coerced(self): + self.add_document((HEADER + "去年底,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv") + timeline = self.build() + self.assertIsNone(timeline["events"][0]["occurred_at"]) + self.assertTrue(any("未标注发生时间" in w for w in timeline["warnings"])) + + def test_placeholder_values_are_treated_as_missing(self): + for placeholder in ("缺失", "未知", "-", "null", "无"): + with self.subTest(value=placeholder): + self.add_document((HEADER + f"{placeholder},问询,请说明回款来源,,\n").encode("utf-8-sig"), + f"历史{placeholder}.csv") + timeline = self.build() + undated = [e for e in timeline["events"] if e["occurred_at"] is None] + self.assertTrue(undated, f"{placeholder} 应被视为缺日期") + + +class ExtractionTests(TimelineTestCase): + def test_dated_rows_become_ready(self): + rows = HEADER + "2024-03-11,问询,请说明回款来源,材料不足,已补充\n2024-04-02,决定,评审会通过,,\n" + self.add_document(rows.encode("utf-8-sig"), "历史.csv") + timeline = self.build() + self.assertEqual(timeline["status"], "ready") + self.assertEqual([e["question"] for e in timeline["events"]], + ["请说明回款来源", "评审会通过"]) + self.assertTrue(all(e["occurred_at"] for e in timeline["events"])) + + def test_one_undated_event_makes_the_whole_timeline_partial(self): + rows = HEADER + "2024-03-11,问询,请说明回款来源,,\n,决定,评审会通过,,\n" + self.add_document(rows.encode("utf-8-sig"), "历史.csv") + timeline = self.build() + self.assertEqual(timeline["status"], "partial") + self.assertEqual(len(timeline["events"]), 2) + + def test_events_carry_evidence_ids_for_traceability(self): + self.add_document((HEADER + "2024-03-11,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv") + timeline = self.build() + event = timeline["events"][0] + self.assertEqual(len(event["evidence_ids"]), 1) + # 证据必须能在文档的 chunk 里定位到（AC-02：能定位原文） + with store.transaction(self.db_path) as db: + document = store.get(db, "document", store.listing(db, "document", "DEMO-A")[0]["id"]) + ids = {chunk["evidence_id"] for chunk in document["chunks"]} + self.assertIn(event["evidence_ids"][0], ids) + + def test_event_type_aliases_and_conservative_default(self): + rows = (HEADER + + "2024-01-01,问询,问题一,,\n" + + "2024-01-02,决定,结论一,,\n" + + "2024-01-03,完全没听过的类型,问题三,,\n") + self.add_document(rows.encode("utf-8-sig"), "历史.csv") + types = [e["event_type"] for e in self.build()["events"]] + self.assertEqual(types, ["question", "decision", "question"]) + + def test_conditions_are_split(self): + rows = "date,question,conditions\n2024-01-01,请补充说明,需担保；限三个月\n" + self.add_document(rows.encode("utf-8-sig"), "历史.csv") + self.assertEqual(self.build()["events"][0]["conditions"], ["需担保", "限三个月"]) + + def test_chinese_date_format_is_accepted(self): + self.add_document((HEADER + "2024年3月11日,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv") + self.assertEqual(self.build()["events"][0]["occurred_at"][:10], "2024-03-11") + + def test_undated_events_are_sorted_last_not_shuffled_into_order(self): + rows = HEADER + ",问询,无日期的问题,,\n2024-03-11,问询,有日期的问题,,\n" + self.add_document(rows.encode("utf-8-sig"), "历史.csv") + events = self.build()["events"] + self.assertEqual(events[0]["question"], "有日期的问题") + self.assertIsNone(events[1]["occurred_at"]) + self.assertEqual([e["question"] for e in history_cases.undated_events({"events": events})], + ["无日期的问题"]) + + +class IsolationAndRevisionTests(TimelineTestCase): + def test_other_projects_documents_are_not_included(self): + self.add_document((HEADER + "2024-03-11,问询,DEMO-A 的问题,,\n").encode("utf-8-sig"), "a.csv", "DEMO-A") + self.add_document((HEADER + "2024-05-05,问询,DEMO-B 的问题,,\n").encode("utf-8-sig"), "b.csv", "DEMO-B") + timeline = self.build("DEMO-A") + self.assertEqual([e["question"] for e in timeline["events"]], ["DEMO-A 的问题"]) + + def test_rebuild_bumps_revision_on_the_same_object(self): + self.add_document((HEADER + "2024-03-11,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv") + first = self.build() + second = self.build() + self.assertEqual(first["id"], second["id"], "重新提炼应是同一对象的新 revision") + self.assertEqual(first["revision"], 1) + self.assertEqual(second["revision"], 2) + + def test_stored_timeline_passes_the_contract(self): + self.add_document((HEADER + "2024-03-11,问询,请说明回款来源,,\n").encode("utf-8-sig"), "历史.csv") + self.build() + CaseTimeline(**self.load()) + + def test_load_rejects_a_foreign_project_lookup(self): + self.build("DEMO-A") + with store.transaction(self.db_path) as db: + with self.assertRaises(KeyError): + history_cases.load(db, "DEMO-B") + + def test_ready_is_impossible_with_undated_events(self): + # 契约层兜底：即便调用方算错状态，落库前的校验也会拦住 + with self.assertRaises(Exception): + CaseTimeline(id="t", project_id="DEMO-A", data_domain="demo", revision=1, + status="ready", created_at=store.now(), + events=[{"id": "e1", "event_type": "question", + "recorded_at": store.now(), "question": "q"}]) + + +if __name__ == "__main__": + unittest.main()