from datetime import datetime, timezone from typing import Literal from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from app.schemas.catalog import Input class BuildIndex(Input): chunk_size: int = Field(500, ge=100, le=2000) overlap: int = Field(80, ge=0, le=1000) @model_validator(mode="after") def bounded_overlap(self): if self.overlap > self.chunk_size // 2: raise ValueError("重叠不得超过切分长度的一半") return self class PublishIndex(Input): expected_generation: int = Field(ge=0) class IndexConfig(BuildIndex): splitter: Literal["unicode-window-v1"] class ChunkProvenance(BaseModel): knowledge_id: int knowledge_version: int document_id: int | None document_version: int | None source_id: int | None source_version: int | None source_name: str | None source_url: str title: str category: str class TimestampRead(BaseModel): model_config = ConfigDict(from_attributes=True) @field_validator("created_at", "published_at", check_fields=False) @classmethod def utc_timestamp(cls, value): return value.astimezone(timezone.utc) if value is not None else None class IndexRead(TimestampRead): id: UUID status: Literal["ready", "published", "superseded"] config: IndexConfig corpus_fingerprint: str catalog_revision: int item_count: int chunk_count: int created_by: int created_at: datetime published_at: datetime | None class ActiveIndex(BaseModel): generation: int catalog_revision: int is_current: bool active_version: IndexRead | None class ChunkRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: UUID index_version_id: UUID knowledge_id: int ordinal: int start_offset: int end_offset: int text: str text_sha256: str provenance: ChunkProvenance class PublicationRead(TimestampRead): generation: int previous_version_id: UUID | None version_id: UUID actor_id: int published_at: datetime