ZNJJ-api-server/src/agent/fields.py

"""事故字段候选与校验模型（作业 4 关键代码）。

来源字段清单对照 ``src/api/endpoints.py`` 的 get_info 分组：
- 事故字段（acd_keys）与两方人员字段（human1/human2_keys）；
- 每个字段区分 未知 / 候选 / 已确认 三态（计划作业 4-5）：
  已明确提供的有效信息直接记为已确认并跳过已知项；
  候选值需经核实节点确认后才覆盖正式字段。
"""

from enum import StrEnum
from typing import Any

from pydantic import BaseModel, Field

from .clock import Clock, SystemClock, validate_accident_time

ACCIDENT_FIELD_KEYS = (
    "ywrysw", "ywfjdc", "ywmtc", "bjrjs",
    "sgfssj", "sfsgxc", "jdcsl", "sgyy", "accident_cause",
)
PERSON1_FIELD_KEYS = (
    "xm1", "hpzl1", "hphm1", "sfzmhm1", "sfzmwh1", "sjhm1", "sjwh1", "csbw1",
)
PERSON2_FIELD_KEYS = (
    "xm2", "hpzl2", "hphm2", "sfzmhm2", "sfzmwh2", "sjhm2", "sjwh2", "csbw2",
)
KNOWN_FIELD_KEYS = frozenset(
    ACCIDENT_FIELD_KEYS + PERSON1_FIELD_KEYS + PERSON2_FIELD_KEYS
)

# 事故表单字段（来源：workflow@20261021「文本内容提取事故信息」白名单 +
# 「更新表单」code 节点的差量合并语义）
FORM_FIELDS: dict[str, type] = {
    "has_injury": bool,
    "has_non_motor_vehicle_hit": bool,
    "is_on_scene": bool,
    "vehicle_count": int,
    "accident_time": str,
    "accident_cause": str,
}


def validate_form_update(data: dict) -> dict:
    """按白名单与类型校验提取输出；非法键/类型丢弃（对应更新表单的忽略语义）。"""
    validated = {}
    for key, value in (data or {}).items():
        spec = FORM_FIELDS.get(key)
        if spec is None:
            continue
        if spec is bool and not isinstance(value, bool):
            continue  # 布尔字段严格 bool，不接受 0/1/字符串
        if spec is int and (isinstance(value, bool) or not isinstance(value, int)):
            continue
        if spec is str and not isinstance(value, str):
            continue
        validated[key] = value
    return validated


TIME_FIELDS = frozenset({"sgfssj"})  # 事故发生时间：需时间校验

# 布尔语义字段：明确肯定/否定之外（模糊、ASR 破碎）不得猜测。
_BOOL_FIELD_KEYS = frozenset({"ywrysw", "ywfjdc", "ywmtc", "sfsgxc", "bjrjs"})
_BOOL_TRUE_WORDS = frozenset({"是", "是的", "有", "有人受伤", "true", "1"})
_BOOL_FALSE_WORDS = frozenset({"否", "不是", "没有", "没人受伤", "false", "0"})


class FieldStatus(StrEnum):
    UNKNOWN = "unknown"
    CANDIDATE = "candidate"
    CONFIRMED = "confirmed"


class FieldCandidate(BaseModel):
    """一个待提交/核实的字段值。"""

    key: str
    value: Any
    status: FieldStatus = FieldStatus.CANDIDATE
    source_turn: int = 0  # 产生候选的对话轮数


def _parse_count(value) -> int | None:
    """把数量解析为正整数；无法解析或小于 1 返回 None。

    注意：不得从“追尾”“剐蹭”等描述性词语推断数量——
    那些输入走 NLP 理解后才可能成为候选，本函数只做字面解析。
    """
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value if value >= 1 else None
    if isinstance(value, str) and value.strip().isdigit():
        n = int(value.strip())
        return n if n >= 1 else None
    return None


def _parse_bool(value) -> bool | None:
    """解析明确肯定/否定；模糊或无法判定返回 None（保持候选，不猜）。"""
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        v = value.strip().lower()
        if v in _BOOL_TRUE_WORDS:
            return True
        if v in _BOOL_FALSE_WORDS:
            return False
    return None


def validate_candidate(
    candidate: FieldCandidate,
    clock: Clock | None = None,
) -> FieldCandidate:
    """校验候选字段，返回带最终状态的候选。

    三态语义（计划作业 4-5）：
    - 语义明确且校验通过 → CONFIRMED（可直接写入正式字段并跳过已知项）；
    - 语义模糊/格式不合规/未来时间 → 保持 CANDIDATE（待核实节点确认）；
    - 空值 → UNKNOWN。
    未知 key 直接拒绝（ValueError），不得写入正式字段。
    """
    if candidate.key not in KNOWN_FIELD_KEYS:
        raise ValueError(f"未知字段 key: {candidate.key!r}")
    if candidate.value in (None, ""):
        return candidate.model_copy(update={"status": FieldStatus.UNKNOWN})

    if candidate.key == "jdcsl":
        count = _parse_count(candidate.value)
        if count is not None:
            return candidate.model_copy(
                update={"value": count, "status": FieldStatus.CONFIRMED}
            )
        return candidate  # CANDIDATE：数量未知，不得推断

    if candidate.key in TIME_FIELDS:
        text = str(candidate.value)
        checked = validate_accident_time(text, (clock or SystemClock()).now())
        status = FieldStatus.CONFIRMED if checked else FieldStatus.CANDIDATE
        return candidate.model_copy(update={"value": text, "status": status})

    if candidate.key in _BOOL_FIELD_KEYS:
        parsed = _parse_bool(candidate.value)
        if parsed is not None:
            return candidate.model_copy(
                update={"value": parsed, "status": FieldStatus.CONFIRMED}
            )
        return candidate  # CANDIDATE：模糊/ASR 破碎，不猜

    return candidate.model_copy(update={"status": FieldStatus.CONFIRMED})


__all__ = [
    "ACCIDENT_FIELD_KEYS",
    "PERSON1_FIELD_KEYS",
    "PERSON2_FIELD_KEYS",
    "KNOWN_FIELD_KEYS",
    "FieldCandidate",
    "FieldStatus",
    "validate_candidate",
    "Clock",
    "SystemClock",
    "validate_accident_time",
]