YoVinchen
ecdb10c37a
Some checks failed
verify / verify (push) Has been cancelled
Separate local parsing from model indexing, bind review decisions to immutable manifests, persist vectors behind active profiles, and expose retrieval, chat, evaluation, and document workflows through the React workbench. Constraint: Live Bailian authentication currently fails for all three configured capabilities Rejected: Direct upload-to-embedding flow | bypasses local review and manifest binding Confidence: high Scope-risk: broad Directive: Keep private-data deployment blocked until authentication, RBAC, and separate database roles land Tested: make verify; fresh and replay Docker document smoke; worker recovery smoke; frozen synthetic evaluation; migration 0003-0004 roundtrip Not-tested: Successful live Bailian calls, OCR, real multi-user authorization
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""PARSE_DOCUMENT business handler with storage verification and DB fencing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import uuid
|
|
from collections.abc import Awaitable, Callable, Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Protocol
|
|
|
|
from app.adapters.local_storage import LocalUploadStorage
|
|
from app.core.config import Settings
|
|
from app.persistence.document_workflows import (
|
|
DocumentWorkflowRepository,
|
|
PostgresDocumentWorkflowRepository,
|
|
)
|
|
from app.persistence.job_queue import BackgroundJob
|
|
from app.services.document_ingestion import (
|
|
ChunkingConfig,
|
|
CloudTextPolicy,
|
|
DocumentIngestionError,
|
|
ingest_document,
|
|
)
|
|
|
|
type DocumentJobHandler = Callable[[BackgroundJob], Awaitable[None]]
|
|
|
|
|
|
class VerifiedUploadStorage(Protocol):
|
|
async def read_verified(
|
|
self,
|
|
*,
|
|
storage_key: uuid.UUID,
|
|
expected_size: int,
|
|
expected_sha256: str,
|
|
) -> bytes: ...
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ParseDocumentHandler:
|
|
repository: DocumentWorkflowRepository
|
|
storage: VerifiedUploadStorage
|
|
max_upload_bytes: int
|
|
chunking: ChunkingConfig
|
|
cloud_policy: CloudTextPolicy
|
|
embedding_model: str
|
|
embedding_dimension: int
|
|
|
|
async def __call__(self, job: BackgroundJob) -> None:
|
|
source = await asyncio.to_thread(self.repository.load_source, job)
|
|
content = await self.storage.read_verified(
|
|
storage_key=source.storage_key,
|
|
expected_size=source.byte_size,
|
|
expected_sha256=source.raw_sha256,
|
|
)
|
|
try:
|
|
artifact = await asyncio.to_thread(
|
|
ingest_document,
|
|
filename=source.filename,
|
|
declared_mime_type=source.mime_type,
|
|
content=content,
|
|
max_upload_bytes=self.max_upload_bytes,
|
|
chunking=self.chunking,
|
|
cloud_policy=self.cloud_policy,
|
|
)
|
|
except DocumentIngestionError as exc:
|
|
await asyncio.to_thread(
|
|
self.repository.record_terminal_parse_failure,
|
|
lease=job.lease,
|
|
source=source,
|
|
error_code=exc.code.value,
|
|
)
|
|
return
|
|
await asyncio.to_thread(
|
|
self.repository.persist_artifact,
|
|
lease=job.lease,
|
|
source=source,
|
|
artifact=artifact,
|
|
cloud_policy=self.cloud_policy,
|
|
embedding_model=self.embedding_model,
|
|
embedding_dimension=self.embedding_dimension,
|
|
)
|
|
|
|
|
|
def build_document_handlers(settings: Settings) -> Mapping[str, DocumentJobHandler]:
|
|
"""Build the handler mapping consumed by the generic worker runtime."""
|
|
|
|
maximum_bytes = settings.max_upload_mb * 1024 * 1024
|
|
handler = ParseDocumentHandler(
|
|
repository=PostgresDocumentWorkflowRepository(settings),
|
|
storage=LocalUploadStorage(settings.upload_root, max_bytes=maximum_bytes),
|
|
max_upload_bytes=maximum_bytes,
|
|
chunking=ChunkingConfig(
|
|
target_tokens=settings.chunk_target_tokens,
|
|
max_tokens=settings.chunk_max_tokens,
|
|
overlap_tokens=settings.chunk_overlap_tokens,
|
|
),
|
|
cloud_policy=CloudTextPolicy(),
|
|
embedding_model=settings.embedding_model,
|
|
embedding_dimension=settings.embedding_dimension,
|
|
)
|
|
return {"PARSE_DOCUMENT": handler}
|