"""Pydantic request/response schemas for the LegacyHUB API.""" from __future__ import annotations import uuid from datetime import datetime from typing import Any, Literal from pydantic import BaseModel, Field, model_validator # ---------------- Health ---------------- class ComponentHealth(BaseModel): name: str status: Literal["ok", "error", "degraded"] detail: dict[str, Any] = Field(default_factory=dict) class HealthResponse(BaseModel): status: Literal["ok", "error", "degraded"] version: str components: list[ComponentHealth] # ---------------- Ingestion ---------------- class IngestFolderRequest(BaseModel): path: str = Field(..., description="Absolute path inside the API container") recursive: bool = True force: bool = False class IngestFolderResponse(BaseModel): run_id: uuid.UUID discovered: int queued: int skipped_duplicates: int invalid_files: int AssetKind = Literal["document", "image", "audio", "video", "archive", "dataset", "other"] KnowledgeIngestStatus = Literal["accepted", "duplicate", "rejected"] def _looks_like_local_path(value: str) -> bool: stripped = value.strip() if not stripped: return False lower = stripped.lower() if lower.startswith("file://"): return True if stripped.startswith(("/", "\\")): return True if len(stripped) >= 3 and stripped[1] == ":" and stripped[2] in {"\\", "/"}: return stripped[0].isalpha() if "\\" in stripped: return True return False def _iter_string_values(value: Any, prefix: str = ""): if isinstance(value, str): yield prefix, value return if isinstance(value, dict): for key, child in value.items(): child_prefix = f"{prefix}.{key}" if prefix else str(key) yield from _iter_string_values(child, child_prefix) return if isinstance(value, list): for idx, child in enumerate(value): yield from _iter_string_values(child, f"{prefix}[{idx}]") class AssetIdentity(BaseModel): asset_id: str = Field(..., min_length=1) owner_module: str = Field(..., min_length=1) owner_record_type: str = Field(..., min_length=1) owner_record_id: str = Field(..., min_length=1) asset_kind: AssetKind title: str | None = None original_filename: str | None = None content_type: str = Field(..., min_length=1) size_bytes: int = Field(..., ge=0) sha256: str = Field(..., min_length=64, max_length=64, pattern=r"^[0-9a-fA-F]{64}$") created_at: datetime | None = None created_by: str | None = None class AssetStorageRef(BaseModel): provider: str = Field(..., min_length=1) bucket: str = Field(..., min_length=1) object_key: str = Field(..., min_length=1) version_id: str | None = None region: str | None = None kms_key_id: str | None = None class AssetSecurityState(BaseModel): gate_status: str = Field(..., min_length=1) classification: str | None = None av_status: str | None = None content_scan_status: str | None = None pii_status: str | None = None approved_at: datetime | None = None approved_by: str | None = None quarantine_reason: str | None = None class AssetRetentionState(BaseModel): policy_id: str = Field(..., min_length=1) retain_until: datetime | None = None legal_hold: bool legal_hold_reason: str | None = None class AssetManifest(BaseModel): manifest_version: str = Field(..., min_length=1) asset: AssetIdentity storage: AssetStorageRef security: AssetSecurityState retention: AssetRetentionState derivatives: list[dict[str, Any]] = Field(default_factory=list) links: list[dict[str, Any]] = Field(default_factory=list) source: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def reject_local_filesystem_paths(self) -> AssetManifest: dumped = self.model_dump(mode="json") for path, value in _iter_string_values(dumped): if _looks_like_local_path(value): raise ValueError( f"AssetManifest MUST NOT contain local filesystem paths: {path}" ) return self class KnowledgeIngestOptions(BaseModel): generate_derivatives: bool = True index_full_text: bool = True index_vectors: bool = True emit_events: bool = True class AssetManifestEnvelope(BaseModel): manifest: AssetManifest ingest_options: KnowledgeIngestOptions = Field(default_factory=KnowledgeIngestOptions) class KnowledgeIngestResponse(BaseModel): status: KnowledgeIngestStatus ingest_job_id: uuid.UUID | None = None asset_id: str idempotency_key: str | None = None reason_code: str | None = None class DocumentSummary(BaseModel): id: uuid.UUID original_file_name: str source_path: str sha256: str status: str file_size_bytes: int created_at: datetime # ---------------- Search ---------------- SearchMode = Literal["lexical", "semantic", "hybrid"] class SearchFilters(BaseModel): document_id: uuid.UUID | None = None source_path: str | None = None block_type: str | None = None min_ocr_confidence: float | None = Field(None, ge=0.0, le=1.0) class SearchRequest(BaseModel): query: str = Field(..., min_length=1) limit: int = Field(10, ge=1, le=100) filters: SearchFilters = Field(default_factory=SearchFilters) search_mode: SearchMode = "hybrid" class Citation(BaseModel): pdf: str page: int block_id: str | None = None table_id: str | None = None figure_id: str | None = None # Source identity comes from asset_id (10 ยง8.4), never from the filename. asset_id: str | None = None class SearchHit(BaseModel): rank: int score: float document_id: uuid.UUID chunk_id: uuid.UUID original_file_name: str # source_path is an s3:// ref for asset ingests; legacy folder ingest (now # deprecated) produced local paths. Prefer asset_id / owner_module for # identity and cross-module references. source_path: str asset_id: str | None = None owner_module: str | None = None page_number: int block_type: str text: str citation: Citation quality_flags: dict[str, Any] = Field(default_factory=dict) metadata: dict[str, Any] = Field(default_factory=dict) class SearchResponse(BaseModel): query: str mode: SearchMode total_candidates: int reranked: bool results: list[SearchHit] # ---------------- Dispatch (inter-module bus) ---------------- class DispatchEnvelope(BaseModel): message_id: str | None = None event_id: str = Field(..., min_length=1) card_uid: str | None = None message_type: str = Field(..., min_length=1) participant_code: str | None = None created_at: str | None = None body: dict[str, Any] = Field(default_factory=dict) class DispatchConfirm(BaseModel): status: str event_id: str handled: bool | None = None duplicate: bool | None = None