"""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]