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

"""S1 预审接口层：材料集合导入与事实确认。

命名空间（claude_use/T00/04-first-slice-contracts.md §1 定案 P1）：
    /api/xingbang/risk-agent/project-assistant/prescreen/...

挂在 project-assistant 前缀之下，因此**复用现有鉴权与代理，零改动**：
- `authorize()` 的角色→路径表已有 `path.startswith("project-assistant/")` → 客户经理放行；
  reviewer 的写豁免同样覆盖该前缀（backend/api/risk_agent_access.py:49,53）。
- 同源代理 `/api/risk-agent/[...path]` 已转发 Authorization / X-Xingbang-Demo-Role /
  Idempotency-Key（src/app/api/risk-agent/[...path]/route.ts:7）。
**不**使用设计草案原提的 `/api/prescreen` + `/api/core/prescreen`：那条路要求 admin
角色，且需改三处鉴权/代理代码。

错误信封（T00/04 §2 决策 D2）：**只在本模块内**输出
`{error:{code,message,retryable,request_id}}`。不改 `api/project_assistant.translate()`，
也不注册 app 级处理器——既有端点的 `{"detail": ...}` 形状完全不变。

404 统一口径（03 §2）：不存在与无访问权一律对外 404，不泄露对象是否存在。
既有 `check_project` 对无权项目抛 403 且被 5 个模块共用，故**不改它**，由本模块的
`scope()` 把访问失败包装成 404；演示域策略仍按 403（E-03 语义）。
"""

from __future__ import annotations

import hashlib

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field

from api.project_assistant import enabled
from api.risk_agent_access import authorize, check_project
from contracts.prescreen import (
    EVIDENCE_MAX,
    MAX_BATCH_FILES,
    NOTE_MAX,
    ErrorCode,
    S1_ERROR_HTTP,
    S1_ERROR_RETRYABLE,
)
from services import assistant_store as store
from services import fact_conflict, material_import
from services.assistant_settings import flags
from services.fact_conflict import FactConflictUnresolved
from services.project_assistant import ProjectAssistantService

router = APIRouter(
    prefix="/api/xingbang/risk-agent/project-assistant/prescreen",
    dependencies=[Depends(authorize), Depends(enabled)],
    tags=["兴邦预审 S1"],
)

DEMO_PROJECTS = ("DEMO-A", "DEMO-B", "DEMO-C")
NOT_FOUND_MESSAGE = "对象不可访问或不存在"


# ---------------------------------------------------------------- 请求模型
class Input(BaseModel):
    """与既有 api/project_assistant.Input 同口径：禁止未知字段（E-07）。"""

    model_config = ConfigDict(extra="forbid")


class ImportBatchInput(Input):
    # 清单条目只有 4 个字段；id/status/received_at 由服务端生成（03 §9.4）。
    # data_domain 不在请求里：从项目读取，避免客户端与项目数据域不一致。
    manifest: list[dict] = Field(min_length=1, max_length=MAX_BATCH_FILES)


class RetryInput(Input):
    item_ids: list[str] = Field(min_length=1, max_length=MAX_BATCH_FILES)
    expected_revision: int = Field(ge=1)


class CancelInput(Input):
    expected_revision: int = Field(ge=1)


class ResolveInput(Input):
    selected_evidence_ids: list[str] = Field(min_length=1, max_length=EVIDENCE_MAX)
    resolution_note: str = Field(min_length=1, max_length=NOTE_MAX)
    expected_revision: int = Field(ge=1)


# ---------------------------------------------------------------- 错误信封
def _envelope(request: Request, code: ErrorCode, message: str) -> JSONResponse:
    request_id = request.headers.get("x-request-id") or store.uid("req")
    return JSONResponse(
        status_code=S1_ERROR_HTTP[code],
        content={"error": {"code": code.value, "message": message,
                           "retryable": S1_ERROR_RETRYABLE[code],
                           "request_id": request_id}},
    )


def scope(request: Request, project_id: str):
    """项目范围检查。通过返回 None；不通过返回 404/403 信封。

    无权访问的项目按 404 处理（03 §2 不泄露对象存在性）；演示域被禁用是策略性
    拒绝，按 403（E-03）。
    """
    try:
        check_project(request, project_id)
    except HTTPException:
        return _envelope(request, ErrorCode.NOT_FOUND, NOT_FOUND_MESSAGE)
    if project_id in DEMO_PROJECTS and not flags()["demo"]:
        return _envelope(request, ErrorCode.ACCESS_DENIED, "演示数据域已禁用")
    return None


def run(request: Request, call, invalid_code: ErrorCode = ErrorCode.INVALID_REQUEST):
    """执行并把异常映射为统一信封。

    顺序重要：HTTPException 必须原样抛出，否则依赖注入阶段的 401/403 会被吞掉。
    """
    try:
        return call()
    except HTTPException:
        raise
    except FactConflictUnresolved as exc:
        # E-17 的「确认动作被拒」面；运行级结果面是 HTTP 200 且不走此信封
        return _envelope(request, ErrorCode.FACT_CONFLICT_UNRESOLVED, str(exc))
    except store.IdempotencyConflict as exc:
        return _envelope(request, ErrorCode.IDEMPOTENCY_CONFLICT, str(exc))
    except store.Conflict as exc:
        return _envelope(request, ErrorCode.REVISION_CONFLICT, str(exc))
    except PermissionError as exc:
        return _envelope(request, ErrorCode.ACCESS_DENIED, str(exc))
    except KeyError:
        return _envelope(request, ErrorCode.NOT_FOUND, NOT_FOUND_MESSAGE)
    except ValueError as exc:
        return _envelope(request, invalid_code, str(exc))


# ---------------------------------------------------------------- 内部工具
def _service() -> ProjectAssistantService:
    return ProjectAssistantService()


def _db():
    return _service().connect()


def _project(db, project_id: str) -> dict:
    project = store.get(db, "project", project_id)  # 不存在 → KeyError → 404
    if project.get("data_domain") not in ("demo", "production"):
        raise ValueError("项目数据域异常，无法导入材料")
    return project


def _actor(request: Request) -> str:
    return request.state.identity["id"]


def _key(request: Request) -> str:
    key = request.headers.get("idempotency-key", "")
    if len(key) > 160:
        raise ValueError("幂等键过长")
    return key


def _limit(value: int) -> int:
    return max(1, min(value, 100))


# ------------------------------------------------- 既有上传入口的批量摄取
def _pick_item(batch: dict, filename: str, item_id: str | None) -> dict:
    items = batch["items"]
    if item_id:
        item = next((i for i in items if i["id"] == item_id), None)
        if item is None:
            raise KeyError("导入条目不存在")
        return item
    base = filename.replace("\\", "/").rsplit("/", 1)[-1]
    matches = [i for i in items
               if i["relative_path"].replace("\\", "/").rsplit("/", 1)[-1] == base]
    if not matches:
        raise KeyError("该文件不在导入批次清单中")
    if len(matches) > 1:
        raise ValueError("批次中有多个同名文件，请指定 item_id")
    return matches[0]


def ingest_into_batch(project_id: str, filename: str, content: bytes, actor: str,
                      batch_id: str, item_id: str | None = None):
    """把文件字节摄入**既有**上传链，并回写批次条目（T00/04 §4.2）。

    不建立平行上传 API：仍走 `ProjectAssistantService.upload()`，因此解析、
    去重（同哈希复用原件）、pending_candidates、data_revision、mark_stale 等
    既有行为完全不变。本函数只做"清单核对 + 状态回写"。

    服务端**重算**哈希与大小，不信任客户端清单（B-22 的来源可信度）。
    """
    with _db() as db:
        batch = material_import.load(db, batch_id, project_id)
    if batch["status"] == "cancelled":
        raise store.Conflict("批次已取消，不能继续摄入文件")

    item = _pick_item(batch, filename, item_id)

    def fail(code: ErrorCode, message: str):
        with _db() as db:
            db.execute("BEGIN IMMEDIATE")
            material_import.mark_item(db, batch_id, item["id"], "failed", error_code=code)
        raise ValueError(message)

    if item["status"] == "parsed":
        raise store.Conflict("该条目已解析完成，不能重复摄入")
    digest = hashlib.sha256(content).hexdigest()
    if digest != item["sha256"]:
        fail(ErrorCode.INVALID_EVIDENCE, "文件内容与清单声明的 sha256 不一致，请重新选择该文件")
    if len(content) != item["size_bytes"]:
        fail(ErrorCode.INVALID_EVIDENCE, "文件大小与清单声明不一致，请重新选择该文件")

    # 走既有上传链（可能因同哈希而复用既有 document，这是 14 §3 的期望行为）
    document = _service().upload(project_id, filename, content, actor)

    with _db() as db:
        db.execute("BEGIN IMMEDIATE")
        batch = material_import.mark_item(db, batch_id, item["id"], "parsed",
                                          document_id=document["id"])
    # 返回体保持**document 形状**（既有调用方读 id 与 candidates），
    # 批次状态作为附加字段给出，不替换既有契约。
    return dict(document, import_batch=batch)


async def validation_handler(request: Request, exc: RequestValidationError):
    """请求体/查询参数校验失败也走统一信封。

    为什么需要它：`Input(extra="forbid")` 之类由 FastAPI 的校验层在**进入端点之前**
    抛错，不经过本模块的 `run()`，若不处理就会在同一命名空间里出现两种错误形状
    （信封 + FastAPI 默认的 `{"detail": ...}`）。这是 smoke_s1.py 实测发现的。

    **非 prescreen 路径一律交回 FastAPI 默认处理器**，因此既有端点的错误形状
    完全不变（有测试锁定这一隔离）。
    """
    if not request.url.path.startswith(router.prefix):
        return await request_validation_exception_handler(request, exc)
    errors = exc.errors() if callable(exc.errors) else exc.errors
    first = errors[0] if errors else {}
    where = ".".join(str(part) for part in first.get("loc", ()) if part != "body") or "请求"
    return _envelope(request, ErrorCode.INVALID_REQUEST,
                     f"{where}：{first.get('msg', '校验失败')}")


# ---------------------------------------------------------------- 导入批次
@router.post("/projects/{project_id}/import-batches", status_code=202)
def create_import_batch(project_id: str, payload: ImportBatchInput, request: Request):
    blocked = scope(request, project_id)
    if blocked is not None:
        return blocked

    def action():
        with _db() as db:
            db.execute("BEGIN IMMEDIATE")
            project = _project(db, project_id)
            batch = material_import.create(db, project_id, project["data_domain"],
                                           payload.manifest, _actor(request), _key(request))
        return {"batch": batch}

    return run(request, action, ErrorCode.IMPORT_BATCH_INVALID)


@router.get("/projects/{project_id}/import-batches")
def list_import_batches(project_id: str, request: Request, limit: int = 20):
    blocked = scope(request, project_id)
    if blocked is not None:
        return blocked

    def action():
        with _db() as db:
            return material_import.listing(db, project_id, _limit(limit))

    return run(request, action)


@router.get("/import-batches/{batch_id}")
def get_import_batch(batch_id: str, request: Request):
    def action():
        with _db() as db:
            batch = material_import.load(db, batch_id)  # KeyError → 404
        blocked = scope(request, batch["project_id"])
        return blocked if blocked is not None else batch

    return run(request, action)


@router.post("/import-batches/{batch_id}/retry", status_code=202)
def retry_import_batch(batch_id: str, payload: RetryInput, request: Request):
    def action():
        with _db() as db:
            probe = material_import.load(db, batch_id)
        blocked = scope(request, probe["project_id"])
        if blocked is not None:
            return blocked
        with _db() as db:
            db.execute("BEGIN IMMEDIATE")
            return material_import.retry(db, batch_id, payload.item_ids,
                                         _actor(request), payload.expected_revision)

    return run(request, action, ErrorCode.IMPORT_BATCH_INVALID)


@router.post("/import-batches/{batch_id}/cancel")
def cancel_import_batch(batch_id: str, payload: CancelInput, request: Request):
    def action():
        with _db() as db:
            probe = material_import.load(db, batch_id)
        blocked = scope(request, probe["project_id"])
        if blocked is not None:
            return blocked
        with _db() as db:
            db.execute("BEGIN IMMEDIATE")
            return material_import.cancel(db, batch_id, _actor(request), payload.expected_revision)

    return run(request, action, ErrorCode.IMPORT_BATCH_INVALID)


# ---------------------------------------------------------------- 事实冲突
@router.get("/projects/{project_id}/fact-conflicts")
def list_fact_conflicts(project_id: str, request: Request, status: str | None = None,
                        limit: int = 20):
    blocked = scope(request, project_id)
    if blocked is not None:
        return blocked
    if status not in (None, "unresolved", "resolved"):
        return _envelope(request, ErrorCode.INVALID_REQUEST,
                         "状态取值须为 unresolved 或 resolved")

    def action():
        with _db() as db:
            return fact_conflict.listing(db, project_id, status, _limit(limit))

    return run(request, action)


@router.get("/fact-conflicts/{conflict_id}")
def get_fact_conflict(conflict_id: str, request: Request):
    def action():
        with _db() as db:
            row = fact_conflict.load(db, conflict_id)
        blocked = scope(request, row["project_id"])
        return blocked if blocked is not None else row

    return run(request, action)


@router.post("/fact-conflicts/{conflict_id}/resolve")
def resolve_fact_conflict(conflict_id: str, payload: ResolveInput, request: Request):
    def action():
        with _db() as db:
            probe = fact_conflict.load(db, conflict_id)
        blocked = scope(request, probe["project_id"])
        if blocked is not None:
            return blocked
        with _db() as db:
            db.execute("BEGIN IMMEDIATE")
            project = _project(db, probe["project_id"])
            # resolve 就地写 project 的 values/selected_sources/facts，因此必须保存它；
            # 不要在此重复计算 facts 的列名——resolve 已按 FIELDS 反查过。
            resolved = fact_conflict.resolve(db, conflict_id, project,
                                             payload.selected_evidence_ids,
                                             payload.resolution_note,
                                             _actor(request), payload.expected_revision)
            store.put(db, "project", project)
        return resolved

    return run(request, action)