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

"""S1 材料导入服务：文件集合清单的生命周期。

职责 owner（docs/v2/03-detailed-design.md §9.2「材料导入服务」）：
    集合清单、逐项状态与重试、批次状态派生。

边界（不得越界）：
- **不建立平行上传 API**：文件字节仍走既有
  `POST /projects/{project_id}/documents` → `ProjectAssistantService.upload()`
  → `assistant_materials.material()`。本模块只登记清单与回写解析结果。
- **不创建平行持久化入口**：复用 `assistant_objects`（kind=`import_batch`）。
  items 是批次内的值对象（契约中 `ImportItem` 无 `revision`），故不另立
  `import_item` kind；批次 revision 由 `put()` 维护。
- **不做真实性核验**：`SourceLocator` 能定位不等于材料真实可信（01 §1）。
"""

from __future__ import annotations

from services import assistant_store as store
from contracts.prescreen import ErrorCode, MAX_BATCH_BYTES, MAX_BATCH_FILES

KIND = "import_batch"
IDEMPOTENCY_OPERATION = "import_batch"

# 解析实现版本：冻结进批次，使"用哪版解析器得到这个结果"可追溯（BR-22）。
# 解析逻辑在 assistant_materials.parse()，其行为变更时必须同步提升此版本。
PARSER_VERSION = "assistant_materials.parse@1"

_STATUS_ORDER = ("queued", "parsing", "parsed", "failed")

# 客户端在清单里**只能**声明这 4 个字段（docs/v2/03-detailed-design.md §9.4）。
# id / status / received_at / document_id / error_code 一律由服务端决定——
# 让客户端自报 id 或 status 等于把状态机的写权交给调用方。
MANIFEST_FIELDS = frozenset({"relative_path", "mime_type", "size_bytes", "sha256"})


def _derive_status(items: list[dict]) -> str:
    """由逐项状态派生批次状态。

    ready   = 全部 parsed
    failed  = 全部 failed
    partial = 既有 parsed 又有 failed（含 queued/parsing 混入的中间态）
    """
    statuses = {item["status"] for item in items}
    if statuses == {"parsed"}:
        return "ready"
    if statuses == {"failed"}:
        return "failed"
    if "failed" in statuses and "parsed" in statuses:
        return "partial"
    return "parsing"


def _validate_manifest(manifest: list[dict]) -> list[dict]:
    from pydantic import ValidationError
    from contracts.prescreen import ImportItem

    if not manifest:
        raise ValueError("材料集合为空；请选择至少一个文件")
    if len(manifest) > MAX_BATCH_FILES:
        raise ValueError(f"单批最多 {MAX_BATCH_FILES} 个文件；请拆分后重试")

    items: list[dict] = []
    seen_sha: set[str] = set()
    for index, entry in enumerate(manifest):
        if not isinstance(entry, dict):
            raise ValueError(f"第 {index + 1} 个文件不是对象")
        extra = sorted(set(entry) - MANIFEST_FIELDS)
        if extra:
            raise ValueError(f"第 {index + 1} 个文件包含不允许由客户端声明的字段：{extra}")

        # id / status / received_at 由服务端决定；observed_at 留空——
        # 导入时间不冒充材料发生时间（14 §3），未知保持 null（BR-02）
        draft = dict(entry, id=store.uid("item"), status="queued",
                     received_at=store.now(), document_id=None,
                     error_code=None, observed_at=None)
        try:
            parsed = ImportItem(**draft)
        except ValidationError as exc:
            # 逐项错误沿用 E-01/E-16 的文案口径，指明是第几项
            first = exc.errors()[0]
            where = ".".join(str(p) for p in first["loc"]) or "输入"
            raise ValueError(f"第 {index + 1} 个文件不合法（{where}）：{first['msg']}") from exc
        if parsed.sha256 in seen_sha:
            raise ValueError(f"第 {index + 1} 个文件与同批其他条目哈希重复：{parsed.relative_path}")
        seen_sha.add(parsed.sha256)
        items.append(parsed.model_dump(mode="json"))

    total = sum(item["size_bytes"] for item in items)
    if total > MAX_BATCH_BYTES:
        raise ValueError(f"单批总量 {total} 超过上限 {MAX_BATCH_BYTES} 字节；请拆分后重试")
    return items


def create(db, project_id, data_domain, manifest, actor, idempotency_key="", parser_version=PARSER_VERSION):
    """登记一次文件集合导入。返回 ImportBatch（status=queued）。

    幂等：同一 actor+project+key 且载荷相同 → 返回原批次；载荷不同 → E-09。
    """
    request_hash = store.digest({"project": project_id, "domain": data_domain, "manifest": manifest})
    if idempotency_key:
        prior = store.idempotent_lookup(db, actor, IDEMPOTENCY_OPERATION, idempotency_key, request_hash)
        if prior:
            return store.get(db, KIND, prior)

    items = _validate_manifest(manifest)
    batch = {
        "id": store.uid("import"),
        "project_id": project_id,
        "data_domain": data_domain,
        "status": "queued",
        "created_at": store.now(),
        "created_by": actor,
        "parser_version": parser_version,
        "items": items,
        "total_bytes": sum(item["size_bytes"] for item in items),
        "warnings": [],
    }
    batch = store.put(db, KIND, batch)  # revision=1
    if idempotency_key:
        store.idempotent_remember(db, actor, IDEMPOTENCY_OPERATION, idempotency_key, request_hash, batch["id"])
    store.event(db, project_id, "import_batch.created", actor,
                {"batch_id": batch["id"], "files": len(items), "total_bytes": batch["total_bytes"]})
    return batch


def load(db, batch_id, project_id=None):
    batch = store.get(db, KIND, batch_id)
    if project_id is not None and batch["project_id"] != project_id:
        # 不存在与无访问权统一对外 404（03 §2），故这里也走 KeyError
        raise KeyError("导入批次不存在")
    return batch


def listing(db, project_id, limit=20):
    rows = store.listing(db, KIND, project_id)
    return {"items": rows[:limit], "next_cursor": None if len(rows) <= limit else str(limit)}


def mark_item(db, batch_id, item_id, status, document_id=None, error_code=None, warning=None):
    """由既有上传链回写单条解析结果。

    `parsed` 必须带 document_id；`failed` 必须带 error_code —— 与契约同规则，
    避免出现"状态说成功但无法定位原件"的条目（AC-23）。
    """
    if status not in _STATUS_ORDER:
        raise ValueError(f"未知条目状态：{status}")
    if status == "failed" and not error_code:
        raise ValueError("failed 的条目必须给出 error_code")
    batch = store.get(db, KIND, batch_id)
    item = next((i for i in batch["items"] if i["id"] == item_id), None)
    if item is None:
        raise KeyError("导入条目不存在")
    if batch["status"] == "cancelled":
        # 取消后忽略迟到结果（03 §9.5），不写回
        return batch
    item["status"] = status
    if document_id is not None:
        item["document_id"] = document_id
    item["error_code"] = error_code if status == "failed" else None
    if warning:
        batch["warnings"].append(f"{item['relative_path']}: {warning}")
    batch["status"] = _derive_status(batch["items"])
    return store.put(db, KIND, batch)


def retry(db, batch_id, item_ids, actor, expected):
    """仅把失败项重置为 queued；已 parsed 的条目不重解析、不重写（AC-23 反例）。"""
    batch = store.get(db, KIND, batch_id)
    if batch["status"] == "cancelled":
        raise store.Conflict("批次已取消，不能重试")
    if batch["status"] not in ("partial", "failed"):
        raise ValueError("只有部分失败或全部失败的批次可以重试")

    targets = set(item_ids)
    unknown = targets - {i["id"] for i in batch["items"]}
    if unknown:
        raise KeyError(f"导入条目不存在：{sorted(unknown)}")

    reset = []
    for item in batch["items"]:
        if item["id"] not in targets:
            continue
        if item["status"] != "failed":
            raise ValueError(f"条目 {item['relative_path']} 不是失败项，不能重试")
        item["status"] = "queued"
        item["error_code"] = None
        reset.append(item["id"])

    batch["status"] = _derive_status(batch["items"])
    batch = store.put(db, KIND, batch, expected=expected)
    store.event(db, batch["project_id"], "import_batch.retried", actor,
                {"batch_id": batch_id, "items": reset})
    return batch


def cancel(db, batch_id, actor, expected):
    batch = store.get(db, KIND, batch_id)
    if batch["status"] in ("ready", "partial", "failed", "cancelled"):
        raise store.Conflict("批次已结束，不能取消")
    batch["status"] = "cancelled"
    batch = store.put(db, KIND, batch, expected=expected)
    store.event(db, batch["project_id"], "import_batch.cancelled", actor, {"batch_id": batch_id})
    return batch


# 供 S1c 接口层把服务端异常映射为错误码（03 §2 的统一信封）
ERROR_BY_EXCEPTION = {
    "KeyError": ErrorCode.NOT_FOUND,
    "Conflict": ErrorCode.REVISION_CONFLICT,
    "ValueError": ErrorCode.IMPORT_BATCH_INVALID,
}