mini-openclaw-v2/backend/contracts/prescreen.py

"""兴邦预审 v2 · 首个实施切片（S1）契约：材料集合导入与事实确认。

唯一职责 owner：[docs/v2/03-detailed-design.md](../../../docs/v2/03-detailed-design.md) §1/§9.2 指定的
`backend/contracts/prescreen.py`。本文件是该 owner 的**首版实现**，字段命名以本文件为准；
`claude_use/T00/04-first-slice-contracts.md` 是提案记录，不是第二份契约定义。

S1 范围（T00/04 §4/§5）：
    SourceLocator · ImportItem · ImportBatch · FactConflict + 共用约束 + 错误信封

**不在本文件**（属后续任务包，不得在此提前定义）：CaseTimeline、DecisionEvent、
ExperienceCard、KnowledgeCard、ReviewPlan、DimensionSpec、ScreeningRun、Judgment、
Simulation、Evaluation、TrajectoryStep、ReviewOutcome、ImprovementCandidate、
DatasetManifest、ProbabilityView。

依赖方向：本模块**不导入** `services/`。标准字段白名单（`assistant_materials.FIELDS`）
的校验属 service 层，避免 contracts → services 的反向依赖。
"""

from __future__ import annotations

import re
from datetime import datetime
from enum import StrEnum
from typing import Annotated, Literal

from pydantic import AfterValidator, BaseModel, ConfigDict, Field, model_validator

# ---------------------------------------------------------------- 共用约束
ID_MAX = 128
PATH_MAX = 512
NOTE_MAX = 2000
EVIDENCE_MAX = 100
DESC_MAX = 2000

# P0 初始工程限制（**非业务阈值**，见 docs/v2/03-detailed-design.md §1 末）。
# 单文件上限沿用既有 assistant_materials.LIMIT = 10 MiB。
# 单批总量取保守值：现有实现把整份文件 base64 存进 SQLite 单行，
# 设计文档曾提 100 MiB/批 ≈ 133 MiB base64 入库，可直接造成演示故障
# （claude_use/T00/04 §3.1）。**降低**无需重新验证；**放宽**须重新验证并更新规格。
MAX_ITEM_BYTES = 10 * 1024 * 1024
MAX_BATCH_BYTES = 20 * 1024 * 1024
MAX_BATCH_FILES = 20

_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_WINDOWS_DRIVE = re.compile(r"^[A-Za-z]:")


def _aware(value: datetime) -> datetime:
    """时间必须带时区（docs/v2/03-detailed-design.md §9.1）。"""
    if value.tzinfo is None or value.utcoffset() is None:
        raise ValueError("时间必须带时区（ISO-8601）")
    return value


AwareDatetime = Annotated[datetime, AfterValidator(_aware)]
Identifier = Annotated[str, Field(min_length=1, max_length=ID_MAX)]


class Contract(BaseModel):
    """所有 S1 对象共用：禁止未知字段（对应 E-07 INVALID_REQUEST）。"""

    model_config = ConfigDict(extra="forbid")


class DataDomain(StrEnum):
    demo = "demo"
    production = "production"


class Ref(Contract):
    """通用版本引用（docs/v2/03-detailed-design.md §9.1）。"""

    id: Identifier
    version: str = Field(min_length=1, max_length=ID_MAX)


# ---------------------------------------------------------------- 错误契约
class ErrorCode(StrEnum):
    """S1 涉及的错误码。取值与 docs/v2/01-requirements.md §5 一一对应，不另造。"""

    NOT_FOUND = "NOT_FOUND"
    ACCESS_DENIED = "ACCESS_DENIED"
    INVALID_REQUEST = "INVALID_REQUEST"
    INVALID_EVIDENCE = "INVALID_EVIDENCE"
    REVISION_CONFLICT = "REVISION_CONFLICT"
    IDEMPOTENCY_CONFLICT = "IDEMPOTENCY_CONFLICT"
    MATERIAL_PARSE_FAILED = "MATERIAL_PARSE_FAILED"
    IMPORT_BATCH_INVALID = "IMPORT_BATCH_INVALID"
    FACT_CONFLICT_UNRESOLVED = "FACT_CONFLICT_UNRESOLVED"


# HTTP 语义沿用既有 translate()（backend/api/project_assistant.py:36-46），
# 不另立一套。
# E-17 FACT_CONFLICT_UNRESOLVED 有**两个面**，不要混为一谈：
#   · 确认动作被拒（本映射）：422 —— 客户端必须先裁定冲突，属可立即纠正的请求问题
#   · 运行级结果（不在此映射内）：HTTP 200，冲突标识保存在 degradations/finding，
#     "HTTP 200 仅表示结果可读取，不表示核查通过"（01 §5 末）
S1_ERROR_HTTP: dict[ErrorCode, int] = {
    ErrorCode.NOT_FOUND: 404,
    ErrorCode.ACCESS_DENIED: 403,
    ErrorCode.INVALID_REQUEST: 422,
    ErrorCode.INVALID_EVIDENCE: 422,
    ErrorCode.REVISION_CONFLICT: 409,
    ErrorCode.IDEMPOTENCY_CONFLICT: 409,
    ErrorCode.MATERIAL_PARSE_FAILED: 422,
    ErrorCode.IMPORT_BATCH_INVALID: 422,
    ErrorCode.FACT_CONFLICT_UNRESOLVED: 422,
}

# 403/422 不自动重试；409 需客户端改载荷或刷新。见 03 §5。
S1_ERROR_RETRYABLE: dict[ErrorCode, bool] = {code: False for code in ErrorCode}


class ErrorBody(Contract):
    code: ErrorCode
    message: str = Field(min_length=1, max_length=DESC_MAX)
    retryable: bool
    request_id: str = Field(min_length=1, max_length=ID_MAX)


class ErrorResponse(Contract):
    """新 prescreen 路由的**独立**错误信封（T00/04 §2 决策 D2）。

    既有路由仍是 FastAPI 默认 `{"detail": ...}`，本次不改，避免影响
    backend/api/project_assistant.py 的 translate() 及其全部消费方。
    """

    error: ErrorBody


# ---------------------------------------------------------------- 来源定位
class SourceLocator(Contract):
    """证据在原件中的定位。定位**不等于**真实性核验（01 §1）。

    kind 决定必填字段，且三种寻址方式互斥：
        pdf    → page
        table  → sheet + cell_range
        manual → note
    """

    kind: Literal["pdf", "table", "manual"]
    document_id: Identifier
    document_version: int = Field(ge=1)
    page: int | None = Field(default=None, ge=1)
    sheet: str | None = Field(default=None, min_length=1, max_length=200)
    cell_range: str | None = Field(default=None, min_length=1, max_length=200)
    note: str | None = Field(default=None, min_length=1, max_length=PATH_MAX)
    excerpt: str | None = Field(default=None, max_length=4000)

    _REQUIRED: dict[str, tuple[str, ...]] = {
        "pdf": ("page",),
        "table": ("sheet", "cell_range"),
        "manual": ("note",),
    }
    _ADDRESS_FIELDS = ("page", "sheet", "cell_range", "note")

    @model_validator(mode="after")
    def _check_kind_fields(self) -> SourceLocator:
        required = self._REQUIRED[self.kind]
        missing = [name for name in required if getattr(self, name) is None]
        if missing:
            raise ValueError(f"{self.kind} 定位缺少必填字段：{', '.join(missing)}")
        extra = [name for name in self._ADDRESS_FIELDS
                 if name not in required and getattr(self, name) is not None]
        if extra:
            raise ValueError(f"{self.kind} 定位不允许字段：{', '.join(extra)}")
        return self


# ---------------------------------------------------------------- 导入清单
class ImportItem(Contract):
    """集合中的一个文件条目。`relative_path` 仅表达用户选择的文件集合，
    绝不作服务器写入路径（docs/v2/14-experience-loop.md §3）。
    """

    id: Identifier
    document_id: Identifier | None = None  # 解析成功后由既有 upload 链回填
    relative_path: str = Field(min_length=1, max_length=PATH_MAX)
    sha256: str = Field(pattern=_SHA256.pattern)
    size_bytes: int = Field(ge=0, le=MAX_ITEM_BYTES)
    mime_type: str = Field(min_length=1, max_length=200)
    status: Literal["queued", "parsing", "parsed", "failed"]
    error_code: ErrorCode | None = None
    observed_at: AwareDatetime | None = None  # 材料**发生**时间；未知为 null
    received_at: AwareDatetime

    @model_validator(mode="after")
    def _check_path_and_status(self) -> ImportItem:
        # 路径安全：不允许绝对路径、盘符或上级引用。校验用规范化副本，
        # 但**保留原值**，使客户端提交的内容与落库内容一致。
        normalized = self.relative_path.replace("\\", "/")
        if normalized.startswith("/") or _WINDOWS_DRIVE.match(normalized):
            raise ValueError("relative_path 不得为绝对路径")
        if any(part == ".." for part in normalized.split("/")):
            raise ValueError("relative_path 不得包含上级引用")

        if self.status == "failed" and self.error_code is None:
            raise ValueError("failed 的条目必须有 error_code")
        if self.status == "parsed" and self.error_code is not None:
            raise ValueError("parsed 的条目不得带 error_code")
        if self.status == "parsed" and self.document_id is None:
            raise ValueError("parsed 的条目必须已关联 document_id")
        return self


class ImportBatch(Contract):
    """一次「文件集合」导入的容器对象。

    状态机（T00/04 §6.1）：
        queued → parsing → ready | partial | failed
        queued | parsing → cancelled
        partial | failed → queued（retry，仅重置失败项，revision++）
    """

    id: Identifier
    project_id: Identifier
    data_domain: DataDomain
    revision: int = Field(ge=1)
    status: Literal["queued", "parsing", "ready", "partial", "failed", "cancelled"]
    created_at: AwareDatetime
    created_by: Identifier
    parser_version: str = Field(min_length=1, max_length=64)
    items: list[ImportItem] = Field(min_length=1, max_length=MAX_BATCH_FILES)
    # 不在此处加 le=MAX_BATCH_BYTES：字段级约束会先于模型级校验触发，
    # 使用户看到 pydantic 的英文默认文案而非下面的中文提示。
    total_bytes: int = Field(ge=0)
    warnings: list[str] = Field(default_factory=list, max_length=EVIDENCE_MAX)

    @model_validator(mode="after")
    def _check_consistency(self) -> ImportBatch:
        if len({item.id for item in self.items}) != len(self.items):
            raise ValueError("同一批次内 item id 必须唯一")
        # 单项上限已由 ImportItem.size_bytes 的 le 约束保证，此处不重复检查。
        total = sum(item.size_bytes for item in self.items)
        if total != self.total_bytes:
            raise ValueError("total_bytes 必须等于各 item size_bytes 之和")
        if total > MAX_BATCH_BYTES:
            raise ValueError(
                f"单批总量 {total} 超过上限 {MAX_BATCH_BYTES} 字节；请拆分后重试"
            )
        return self


# ---------------------------------------------------------------- 事实冲突
class ConflictCandidate(Contract):
    """并列展示的候选值之一。未知用 null，**不用 0 或空串代替**（03 §9.1）。"""

    evidence_id: Identifier
    value: str | None = None
    source: str = Field(min_length=1, max_length=PATH_MAX)
    locator: str = Field(min_length=1, max_length=PATH_MAX)
    unit: str | None = Field(default=None, max_length=64)
    currency: str | None = Field(default=None, max_length=16)
    period: str | None = Field(default=None, max_length=64)


class FactConflict(Contract):
    """同一字段在多来源上给出不同值时的待处理对象（BR-23 / E-17 / AC-24）。

    只允许 unresolved → resolved。再次冲突时**新建**对象并用 supersedes_id
    指向旧对象，不覆写历史。
    """

    id: Identifier
    project_id: Identifier
    data_domain: DataDomain
    revision: int = Field(ge=1)
    # 标准字段名（assistant_materials.FIELDS.values()）。白名单校验在 service 层，
    # 见本模块顶部「依赖方向」说明。
    field: str = Field(min_length=1, max_length=ID_MAX)
    candidate_evidence_ids: list[Identifier] = Field(min_length=2, max_length=EVIDENCE_MAX)
    candidates: list[ConflictCandidate] = Field(min_length=2, max_length=EVIDENCE_MAX)
    status: Literal["unresolved", "resolved"]
    resolution_note: str | None = Field(default=None, min_length=1, max_length=NOTE_MAX)
    selected_evidence_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX)
    resolved_by: Identifier | None = None
    resolved_at: AwareDatetime | None = None
    created_at: AwareDatetime
    supersedes_id: Identifier | None = None

    @model_validator(mode="after")
    def _check_candidates_and_resolution(self) -> FactConflict:
        ids = self.candidate_evidence_ids
        if len(set(ids)) != len(ids):
            raise ValueError("candidate_evidence_ids 不得重复")
        if [c.evidence_id for c in self.candidates] != ids:
            raise ValueError("candidates 必须与 candidate_evidence_ids 同序同集合")
        # 「至少两个不同值」：全部候选值相同说明没有分歧，不应建冲突对象。
        if len({c.value for c in self.candidates}) < 2:
            raise ValueError("候选值全部相同，不构成事实冲突")

        if self.status == "resolved":
            missing = [name for name, value in (
                ("resolution_note", self.resolution_note),
                ("resolved_by", self.resolved_by),
                ("resolved_at", self.resolved_at),
            ) if value is None]
            if missing:
                raise ValueError(f"resolved 必须提供：{', '.join(missing)}")
            if not self.selected_evidence_ids:
                raise ValueError("resolved 必须提供 selected_evidence_ids")
            unknown = set(self.selected_evidence_ids) - set(ids)
            if unknown:
                raise ValueError(f"选择的值不属于该冲突的候选：{sorted(unknown)}")
        else:
            stray = [name for name, value in (
                ("resolution_note", self.resolution_note),
                ("resolved_by", self.resolved_by),
                ("resolved_at", self.resolved_at),
            ) if value is not None]
            if stray or self.selected_evidence_ids:
                raise ValueError("unresolved 不得带解决字段")
        return self