mini-openclaw-v2/backend/services/history_cases.py

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