from __future__ import annotations
|
|
from typing import Any, Mapping
|
|
from .models import ContractError, ErrorCode, HASH_RE, ItemStatus, PackageState, QuotaState
|
|
|
ITEM_KEYS = (
|
"item_id", "slot_id", "candidate_id", "report_identity", "state", "status",
|
"stop_code", "trigger_attempted", "triggered", "quota_state",
|
"quota_reservation_id", "quota_terminal_event_id", "quota_artifact_event_id",
|
"title", "institution", "report_date", "analysts", "page_count", "bytes",
|
"sha256", "source_cache_path", "source_file_name", "final_path", "manifest_row_id",
|
"reused_without_new_trigger", "error_or_note",
|
)
|
|
|
def terminal_item(values: Mapping[str, Any]) -> dict[str, Any]:
|
"""Build and validate the fixed public 26-key item wire."""
|
row = {key: values.get(key) for key in ITEM_KEYS}
|
if tuple(row) != ITEM_KEYS:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.keys", "26-key order mismatch")
|
if row["status"] not in {item.value for item in ItemStatus}:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.status", "enum")
|
if row["state"] not in {item.value for item in PackageState}:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.state", "enum")
|
if row["quota_state"] not in {item.value for item in QuotaState}:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.quota_state", "enum")
|
for key in ("trigger_attempted", "triggered", "reused_without_new_trigger"):
|
if type(row[key]) is not bool:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"item.{key}", "BOOL required")
|
if row["stop_code"] is not None and row["stop_code"] not in {item.value for item in ErrorCode}:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.stop_code", "enum")
|
if row["sha256"] is not None and not HASH_RE.fullmatch(row["sha256"]):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.sha256", "hash")
|
if row["analysts"] is not None and not (
|
isinstance(row["analysts"], (list, tuple)) and all(isinstance(x, str) for x in row["analysts"])
|
):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.analysts", "string array or null")
|
if row["status"] == ItemStatus.SUCCESS.value:
|
required = (
|
"candidate_id", "report_identity", "title", "institution", "report_date",
|
"analysts", "page_count", "bytes", "sha256", "source_cache_path",
|
"source_file_name", "final_path", "manifest_row_id",
|
)
|
if any(row[key] is None for key in required) or row["stop_code"] is not None:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.success", "evidence incomplete")
|
if row["status"] == ItemStatus.DUPLICATE.value:
|
if row["stop_code"] != ErrorCode.DUPLICATE_EXISTING_ARTIFACT.value:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.duplicate", "stop code")
|
if row["status"] in {ItemStatus.FAILED.value, ItemStatus.STOPPED.value}:
|
if row["stop_code"] in {None, ErrorCode.NONE.value} or not row["error_or_note"]:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item.failure", "code/note required")
|
return row
|
