From 8d5aec430115153c4c837fed94ab8f604d117add Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Tue, 11 Aug 2026 16:02:19 -0500 Subject: [PATCH] Github Copilot service realignment & cleanup --- docs/ver4/architecture_v4.md | 20 +- docs/ver4/implementation_plan_v4.md | 3 + src/transcription/api/v4_documents.py | 24 +- src/transcription/services/__init__.py | 8 +- src/transcription/services/documents.py | 383 +---------- src/transcription/services/people.py | 353 ++++++++++ src/transcription/services/sources.py | 660 ++++++++++++++++++ src/transcription/services/store.py | 174 +++-- src/transcription/services/transcription.py | 678 ++----------------- src/transcription/services/workflows.py | 22 +- src/transcription/ui/pages/documents_page.py | 195 ++++-- src/transcription/ui/pages/jobs_page.py | 50 +- src/transcription/ui/pages/people_page.py | 128 +++- src/transcription/ui/pages/sources_page.py | 64 +- src/transcription/worker.py | 8 +- tests/api/test_v4_documents.py | 14 +- tests/integration/test_pipeline_flow.py | 37 +- tests/services/test_document_service.py | 41 +- tests/services/test_store.py | 52 +- tests/services/test_transcription_service.py | 26 +- tests/services/test_v2_crud.py | 41 +- tests/services/test_workflows_reliability.py | 4 +- 22 files changed, 1641 insertions(+), 1344 deletions(-) create mode 100644 src/transcription/services/people.py create mode 100644 src/transcription/services/sources.py diff --git a/docs/ver4/architecture_v4.md b/docs/ver4/architecture_v4.md index 0b6c0f4..9b82a44 100644 --- a/docs/ver4/architecture_v4.md +++ b/docs/ver4/architecture_v4.md @@ -68,15 +68,31 @@ Responsibilities: ### Domain and Service Layer - `src/transcription/db/models.py` -- `src/transcription/services/*.py` +- `src/transcription/services/documents.py` +- `src/transcription/services/sources.py` +- `src/transcription/services/jobs.py` +- `src/transcription/services/people.py` +- `src/transcription/services/workflows.py` Responsibilities: -- Manage transactional operations for documents, people, types, links, sources, jobs, and job sources. +- Keep one primary service boundary per aggregate: Documents, Sources, Jobs, and People. +- Documents own document records and the document-type registry. +- Sources own source records, revisions, source media formats, MIME resolution, and page execution evidence. +- Jobs own job lifecycle state and transitions. +- People own person records, relationship roles, document-person links, and portrait media. - Apply deterministic conflict handling for relationship-role writes. - Use set-based synchronization for many-to-many relationship updates. - Resolve and validate registry-backed document types. +### Source Media Policy + +- `services/sources.py` is the single authority for accepted Source extensions and canonical MIME types. +- Storage and provider payload loading must call the same Source validation functions. +- Supported Source formats are JPEG, PNG, TIFF, and PDF. +- Upload is an interface action, not a domain aggregate. Service names, errors, and workflow variables use + `Source` terminology; compatibility aliases may remain temporarily at old import boundaries. + ### Infrastructure Layer - `src/transcription/db/**` diff --git a/docs/ver4/implementation_plan_v4.md b/docs/ver4/implementation_plan_v4.md index 84e8bab..8d5bf9c 100644 --- a/docs/ver4/implementation_plan_v4.md +++ b/docs/ver4/implementation_plan_v4.md @@ -36,6 +36,9 @@ Implement the Version 4 project definition from the current repository state whi ### 3. Update Services and Write Semantics +- Organize service ownership around Documents, Sources, Jobs, and People. +- Centralize Source extension and MIME policy in the Sources service. +- Treat upload as an interface action and remove it from domain service naming where compatibility permits. - Implement set-based synchronization for document-person updates. - Implement deterministic uniqueness and relationship-write conflict checks. - Remove suggestion-related service behavior. diff --git a/src/transcription/api/v4_documents.py b/src/transcription/api/v4_documents.py index a4d5d14..778b1e5 100644 --- a/src/transcription/api/v4_documents.py +++ b/src/transcription/api/v4_documents.py @@ -18,6 +18,7 @@ from transcription.db.models import DocumentPerson from transcription.db.models import DocumentType from transcription.db.models import PersonRole from transcription.services import DocumentService +from transcription.services import PeopleService router = APIRouter(prefix="/api/v4", tags=["v4-documents"]) @@ -114,10 +115,7 @@ def _person_role_to_read(item: PersonRole) -> PersonRoleRead: def _document_person_to_read(item: DocumentPerson) -> DocumentPersonRead: - if item.role_ref is not None: - role_code = item.role_ref.code - else: - role_code = item.role.value + role_code = item.role_ref.code if item.role_ref is not None else item.role.value person_name = item.person.full_name if item.person is not None else None return DocumentPersonRead( @@ -146,6 +144,14 @@ def get_document_service(request: Request) -> DocumentService: return DocumentService() +def get_people_service(request: Request) -> PeopleService: + """Resolve the People service from app lifespan state when available.""" + services = getattr(request.app.state, "services", None) + if services is not None: + return services.people + return PeopleService() + + @router.get("/document-types", response_model=list[DocumentTypeRead]) async def list_document_types( active_only: bool = True, @@ -158,7 +164,7 @@ async def list_document_types( @router.get("/person-roles", response_model=list[PersonRoleRead]) async def list_person_roles( active_only: bool = True, - service: DocumentService = Depends(get_document_service), + service: PeopleService = Depends(get_people_service), ) -> list[PersonRoleRead]: items = await service.list_person_roles(active_only=active_only) return [_person_role_to_read(item) for item in items] @@ -181,7 +187,7 @@ async def set_document_type( @router.get("/documents/{document_id}/people", response_model=DocumentPeopleResponse) async def list_document_people( document_id: UUID, - service: DocumentService = Depends(get_document_service), + service: PeopleService = Depends(get_people_service), ) -> DocumentPeopleResponse: links = await service.list_document_people(document_id=document_id) return DocumentPeopleResponse(document_id=document_id, links=[_document_person_to_read(item) for item in links]) @@ -191,7 +197,7 @@ async def list_document_people( async def add_document_person_link( document_id: UUID, payload: DocumentPersonWriteRequest, - service: DocumentService = Depends(get_document_service), + service: PeopleService = Depends(get_people_service), ) -> DocumentPersonRead: link = await service.add_document_person_link( document_id=document_id, @@ -206,7 +212,7 @@ async def add_document_person_link( async def set_document_person_role( document_person_id: UUID, payload: DocumentPersonRoleUpdateRequest, - service: DocumentService = Depends(get_document_service), + service: PeopleService = Depends(get_people_service), ) -> DocumentPersonRead: link = await service.set_document_person_role( document_person_id=document_person_id, @@ -219,7 +225,7 @@ async def set_document_person_role( @router.delete("/document-people/{document_person_id}", status_code=204) async def delete_document_person_link( document_person_id: UUID, - service: DocumentService = Depends(get_document_service), + service: PeopleService = Depends(get_people_service), ) -> Response: await service.remove_document_person_link(document_person_id=document_person_id) return Response(status_code=204) diff --git a/src/transcription/services/__init__.py b/src/transcription/services/__init__.py index 0635376..27cc6e2 100644 --- a/src/transcription/services/__init__.py +++ b/src/transcription/services/__init__.py @@ -5,9 +5,10 @@ from dataclasses import field from .documents import DocumentService from .jobs import JobService -from .transcription import TranscriptionService +from .people import PeopleService +from .sources import SourceService -__all__ = ["DocumentService", "JobService", "ServiceBundle", "TranscriptionService"] +__all__ = ["DocumentService", "JobService", "PeopleService", "ServiceBundle", "SourceService"] @dataclass(frozen=True, slots=True) @@ -15,5 +16,6 @@ class ServiceBundle: """Container for all service instances.""" documents: DocumentService = field(default_factory=DocumentService) + sources: SourceService = field(default_factory=SourceService) jobs: JobService = field(default_factory=JobService) - transcriptions: TranscriptionService = field(default_factory=TranscriptionService) + people: PeopleService = field(default_factory=PeopleService) diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index 661b9e6..91261b6 100644 --- a/src/transcription/services/documents.py +++ b/src/transcription/services/documents.py @@ -1,10 +1,8 @@ import logging +import shutil from collections.abc import Sequence -from dataclasses import dataclass from datetime import UTC from datetime import datetime -from pathlib import Path -import shutil from uuid import UUID from sqlalchemy.exc import IntegrityError @@ -14,10 +12,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession from ..db.models import Document from ..db.models import DocumentPerson -from ..db.models import DocumentPersonRole from ..db.models import DocumentType -from ..db.models import Person -from ..db.models import PersonRole from ..errors import AppError from ..errors import ErrorCategory from .base import ServiceBase @@ -33,10 +28,6 @@ class MissingSourceError(DocumentError): """Raised when a document has no associated sources.""" -class UploadError(DocumentError): - """Raised when uploaded content cannot be persisted safely.""" - - class DocumentAlreadyExistsError(DocumentError): """Raised when a document with the same name already exists in the database.""" @@ -45,20 +36,6 @@ class DocumentDeleteBlockedError(DocumentError): """Raised when a document delete is blocked by dependent records.""" -class PersonDeleteBlockedError(DocumentError): - """Raised when a person delete is blocked by linked documents.""" - - -@dataclass(frozen=True) -class UploadJobResult: - """Summary of created upload records.""" - - document_id: UUID - job_id: UUID - stored_path: Path - original_filename: str - - class DocumentService(ServiceBase): """Thin service class for managing documents in the database.""" @@ -96,37 +73,6 @@ class DocumentService(ServiceBase): await session.flush() return created - async def _resolve_or_create_person_role( - self, - *, - session: AsyncSession, - role_id: UUID | None, - role_code: str | None, - ) -> PersonRole: - """Resolve canonical person role by id/code with compatibility fallback creation.""" - if role_id is not None: - found = await session.get(PersonRole, role_id) - if found is None: - raise DocumentError( - f"Person role with id {role_id} not found", - category=ErrorCategory.VALIDATION, - suggestion="Select a valid relationship role and retry.", - ) - return found - - normalized_code = (role_code or "").strip().lower() - if not normalized_code: - normalized_code = DocumentPersonRole.AUTHOR.value - - existing = (await session.exec(select(PersonRole).where(PersonRole.code == normalized_code))).first() - if existing is not None: - return existing - - created = PersonRole(code=normalized_code, label=normalized_code.replace("_", " ").title()) - session.add(created) - await session.flush() - return created - async def _sync_document_type_fields(self, *, session: AsyncSession, document: Document) -> None: """Synchronize legacy and canonical document type fields.""" resolved = await self._resolve_or_create_document_type( @@ -142,24 +88,6 @@ class DocumentService(ServiceBase): document.document_type_id = resolved.id document.document_type = resolved.code - async def _sync_document_person_role_fields(self, *, session: AsyncSession, link: DocumentPerson) -> None: - """Synchronize legacy and canonical relationship role fields.""" - role_code = link.role.value if isinstance(link.role, DocumentPersonRole) else str(link.role) - resolved = await self._resolve_or_create_person_role( - session=session, - role_id=link.role_id, - role_code=role_code, - ) - link.role_id = resolved.id - try: - link.role = DocumentPersonRole(resolved.code) - except ValueError as exc: - raise DocumentError( - f"Unsupported role code {resolved.code!r} for legacy compatibility", - category=ErrorCategory.VALIDATION, - suggestion="Use author, recipient, or mentioned for now.", - ) from exc - async def _get_document_or_raise(self, *, session: AsyncSession, document_id: UUID) -> Document: """Get a document by id or raise a not-found service error.""" document = await session.get(Document, document_id) @@ -171,17 +99,6 @@ class DocumentService(ServiceBase): ) return document - async def _get_person_or_raise(self, *, session: AsyncSession, person_id: UUID) -> Person: - """Get a person by id or raise a not-found service error.""" - person = await session.get(Person, person_id) - if person is None: - raise DocumentError( - f"Person with id {person_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the person id and retry.", - ) - return person - # # CRUD Operations # @@ -297,147 +214,6 @@ class DocumentService(ServiceBase): except OSError: logger.warning("Failed to delete document storage folder: %s", document_dir) - async def create_person(self, person: Person, *, session: AsyncSession | None = None) -> Person: - """Create a new person in the database.""" - async with self._session_scope(session) as _session: - _session.add(person) - await self._finalize(session=_session, caller_session=session, refresh=(person,)) - return person - - async def read_person(self, person_id: UUID, *, session: AsyncSession | None = None) -> Person: - """Read an existing person from the database.""" - async with self._session_scope(session) as _session: - person = await _session.get(Person, person_id) - if person is None: - raise DocumentError( - f"Person with id {person_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the person id and retry.", - ) - return person - - async def read_person_detail(self, person_id: UUID, *, session: AsyncSession | None = None) -> Person: - """Read a person with eagerly loaded document links for UI detail rendering.""" - async with self._session_scope(session) as _session: - query = ( - select(Person) - .options( - selectinload(Person.document_people).selectinload(DocumentPerson.document), # pyright: ignore[reportArgumentType] - selectinload(Person.document_people).selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType] - ) - .where(Person.id == person_id) - .execution_options(populate_existing=True) - ) - person = (await _session.exec(query)).first() - - if person is None: - raise DocumentError( - f"Person with id {person_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the person id and retry.", - ) - return person - - async def update_person(self, person: Person, *, session: AsyncSession | None = None) -> Person: - """Update an existing person in the database.""" - async with self._session_scope(session) as _session: - person.updated_at = datetime.now(UTC) - merged = await _session.merge(person) - await self._finalize(session=_session, caller_session=session, refresh=(merged,)) - return merged - - async def delete_person(self, person: Person, *, session: AsyncSession | None = None) -> None: - """Delete a person from the database.""" - async with self._session_scope(session) as _session: - existing = await _session.get( - Person, - person.id, - options=( - selectinload(Person.document_people), # pyright: ignore[reportArgumentType] - ), - ) - if existing is None: - raise DocumentError( - f"Person with id {person.id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the person id and retry.", - ) - - for link in list(existing.document_people): - await _session.delete(link) - - await _session.delete(existing) - await self._finalize(session=_session, caller_session=session) - - async def create_document_person( - self, - document_person: DocumentPerson, - *, - session: AsyncSession | None = None, - ) -> DocumentPerson: - """Create a document-person association in the database.""" - async with self._session_scope(session) as _session: - await self._sync_document_person_role_fields(session=_session, link=document_person) - _session.add(document_person) - try: - await self._finalize(session=_session, caller_session=session, refresh=(document_person,)) - except IntegrityError as exc: - raise DocumentError( - "Duplicate relationship link for document/person/role", - category=ErrorCategory.CONFLICT, - suggestion="Remove the existing relationship link or choose a different role.", - ) from exc - return document_person - - async def read_document_person( - self, - document_person_id: UUID, - *, - session: AsyncSession | None = None, - ) -> DocumentPerson: - """Read an existing document-person association from the database.""" - async with self._session_scope(session) as _session: - document_person = await _session.get(DocumentPerson, document_person_id) - if document_person is None: - raise DocumentError( - f"DocumentPerson with id {document_person_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the document-person id and retry.", - ) - return document_person - - async def update_document_person( - self, - document_person: DocumentPerson, - *, - session: AsyncSession | None = None, - ) -> DocumentPerson: - """Update an existing document-person association in the database.""" - async with self._session_scope(session) as _session: - await self._sync_document_person_role_fields(session=_session, link=document_person) - document_person.updated_at = datetime.now(UTC) - merged = await _session.merge(document_person) - try: - await self._finalize(session=_session, caller_session=session, refresh=(merged,)) - except IntegrityError as exc: - raise DocumentError( - "Duplicate relationship link for document/person/role", - category=ErrorCategory.CONFLICT, - suggestion="Remove the existing relationship link or choose a different role.", - ) from exc - return merged - - async def delete_document_person( - self, - document_person: DocumentPerson, - *, - session: AsyncSession | None = None, - ) -> None: - """Delete a document-person association from the database.""" - async with self._session_scope(session) as _session: - await _session.delete(document_person) - await self._finalize(session=_session, caller_session=session) - # Query Operations async def query_documents( @@ -481,33 +257,6 @@ class DocumentService(ServiceBase): ) return document - async def list_people(self, *, session: AsyncSession | None = None) -> Sequence[Person]: - """List all people in the database.""" - async with self._session_scope(session) as _session: - result = await _session.exec(select(Person)) - return result.all() - - async def list_document_people( - self, - *, - document_id: UUID | None = None, - person_id: UUID | None = None, - session: AsyncSession | None = None, - ) -> Sequence[DocumentPerson]: - """List document-person associations, optionally filtered by document or person.""" - async with self._session_scope(session) as _session: - query = select(DocumentPerson).options( - selectinload(DocumentPerson.document), # pyright: ignore[reportArgumentType] - selectinload(DocumentPerson.person), # pyright: ignore[reportArgumentType] - selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType] - ) - if document_id is not None: - query = query.where(DocumentPerson.document_id == document_id) - if person_id is not None: - query = query.where(DocumentPerson.person_id == person_id) - result = await _session.exec(query) - return result.all() - async def list_document_types( self, *, @@ -523,21 +272,6 @@ class DocumentService(ServiceBase): result = await _session.exec(query) return result.all() - async def list_person_roles( - self, - *, - active_only: bool = True, - session: AsyncSession | None = None, - ) -> Sequence[PersonRole]: - """List configured relationship roles.""" - async with self._session_scope(session) as _session: - query = select(PersonRole) - if active_only: - query = query.where(PersonRole.is_active.is_(True)) - query = query.order_by(PersonRole.code) - result = await _session.exec(query) - return result.all() - async def set_document_type( self, *, @@ -569,118 +303,3 @@ class DocumentService(ServiceBase): document.updated_at = datetime.now(UTC) await self._finalize(session=_session, caller_session=session, refresh=(document,)) return document - - async def add_document_person_link( - self, - *, - document_id: UUID, - person_id: UUID, - role_id: UUID | None = None, - role_code: str | None = None, - session: AsyncSession | None = None, - ) -> DocumentPerson: - """Create a document-person link with role selected by id or code.""" - async with self._session_scope(session) as _session: - await self._get_document_or_raise(session=_session, document_id=document_id) - await self._get_person_or_raise(session=_session, person_id=person_id) - - link = DocumentPerson( - document_id=document_id, - person_id=person_id, - role=DocumentPersonRole.AUTHOR, - role_id=role_id, - ) - if role_code and role_code.strip(): - try: - link.role = DocumentPersonRole(role_code.strip().lower()) - except ValueError as exc: - raise DocumentError( - f"Unsupported role code {role_code!r}", - category=ErrorCategory.VALIDATION, - suggestion="Use author, recipient, or mentioned.", - ) from exc - - await self._sync_document_person_role_fields(session=_session, link=link) - _session.add(link) - try: - await self._finalize(session=_session, caller_session=session, refresh=(link,)) - except IntegrityError as exc: - raise DocumentError( - "Duplicate relationship link for document/person/role", - category=ErrorCategory.CONFLICT, - suggestion="Remove the existing relationship link or choose a different role.", - ) from exc - return link - - async def set_document_person_role( - self, - *, - document_person_id: UUID, - role_id: UUID | None = None, - role_code: str | None = None, - session: AsyncSession | None = None, - ) -> DocumentPerson: - """Update an existing relationship link role by id or code.""" - async with self._session_scope(session) as _session: - link = await _session.get(DocumentPerson, document_person_id) - if link is None: - raise DocumentError( - f"DocumentPerson with id {document_person_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the relationship link id and retry.", - ) - - if role_id is None and (role_code is None or not role_code.strip()): - raise DocumentError( - "Either role_id or role_code is required", - category=ErrorCategory.VALIDATION, - suggestion="Provide a valid role id or code and retry.", - ) - - if role_id is not None and role_code and role_code.strip(): - raise DocumentError( - "Provide role_id or role_code, not both", - category=ErrorCategory.VALIDATION, - suggestion="Send only one role selector and retry.", - ) - - link.role_id = role_id - if role_code and role_code.strip(): - try: - link.role = DocumentPersonRole(role_code.strip().lower()) - except ValueError as exc: - raise DocumentError( - f"Unsupported role code {role_code!r}", - category=ErrorCategory.VALIDATION, - suggestion="Use author, recipient, or mentioned.", - ) from exc - - await self._sync_document_person_role_fields(session=_session, link=link) - link.updated_at = datetime.now(UTC) - try: - await self._finalize(session=_session, caller_session=session, refresh=(link,)) - except IntegrityError as exc: - raise DocumentError( - "Duplicate relationship link for document/person/role", - category=ErrorCategory.CONFLICT, - suggestion="Remove the existing relationship link or choose a different role.", - ) from exc - return link - - async def remove_document_person_link( - self, - *, - document_person_id: UUID, - session: AsyncSession | None = None, - ) -> None: - """Delete a relationship link by id.""" - async with self._session_scope(session) as _session: - link = await _session.get(DocumentPerson, document_person_id) - if link is None: - raise DocumentError( - f"DocumentPerson with id {document_person_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the relationship link id and retry.", - ) - await _session.delete(link) - await self._finalize(session=_session, caller_session=session) diff --git a/src/transcription/services/people.py b/src/transcription/services/people.py new file mode 100644 index 0000000..e5f59cb --- /dev/null +++ b/src/transcription/services/people.py @@ -0,0 +1,353 @@ +"""People, relationship role, and document-person link services.""" + +from __future__ import annotations + +import logging +from collections.abc import Sequence +from datetime import UTC +from datetime import datetime +from pathlib import Path +from uuid import UUID +from uuid import uuid4 + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import selectinload +from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession + +from ..config import Settings +from ..config import get_settings +from ..db.models import Document +from ..db.models import DocumentPerson +from ..db.models import DocumentPersonRole +from ..db.models import Person +from ..db.models import PersonRole +from ..errors import AppError +from ..errors import ErrorCategory +from .base import ServiceBase + +logger = logging.getLogger(__name__) + +PORTRAIT_EXTENSIONS = frozenset({".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tif", ".tiff"}) + + +class PeopleError(AppError): + """Raised when a Person or document-person relationship operation fails.""" + + +class PersonMediaError(PeopleError): + """Raised when Person portrait media cannot be validated or persisted.""" + + +class PeopleService(ServiceBase): + """Manage People, relationship roles, and document-person links.""" + + async def create_person(self, person: Person, *, session: AsyncSession | None = None) -> Person: + async with self._session_scope(session) as _session: + _session.add(person) + await self._finalize(session=_session, caller_session=session, refresh=(person,)) + return person + + async def read_person(self, person_id: UUID, *, session: AsyncSession | None = None) -> Person: + async with self._session_scope(session) as _session: + person = await _session.get(Person, person_id) + if person is None: + raise self._not_found(f"Person with id {person_id} not found") + return person + + async def update_person(self, person: Person, *, session: AsyncSession | None = None) -> Person: + async with self._session_scope(session) as _session: + person.updated_at = datetime.now(UTC) + merged = await _session.merge(person) + await self._finalize(session=_session, caller_session=session, refresh=(merged,)) + return merged + + async def delete_person(self, person: Person, *, session: AsyncSession | None = None) -> None: + async with self._session_scope(session) as _session: + existing = await _session.get( + Person, + person.id, + options=(selectinload(Person.document_people),), # pyright: ignore[reportArgumentType] + ) + if existing is None: + raise self._not_found(f"Person with id {person.id} not found") + for link in list(existing.document_people): + await _session.delete(link) + await _session.delete(existing) + await self._finalize(session=_session, caller_session=session) + + async def create_document_person( + self, + document_person: DocumentPerson, + *, + session: AsyncSession | None = None, + ) -> DocumentPerson: + async with self._session_scope(session) as _session: + await self._sync_role_fields(session=_session, link=document_person) + _session.add(document_person) + return await self._finalize_link(session=_session, caller_session=session, link=document_person) + + async def read_document_person( + self, + document_person_id: UUID, + *, + session: AsyncSession | None = None, + ) -> DocumentPerson: + async with self._session_scope(session) as _session: + link = await _session.get(DocumentPerson, document_person_id) + if link is None: + raise self._not_found(f"DocumentPerson with id {document_person_id} not found") + return link + + async def update_document_person( + self, + document_person: DocumentPerson, + *, + session: AsyncSession | None = None, + ) -> DocumentPerson: + async with self._session_scope(session) as _session: + await self._sync_role_fields(session=_session, link=document_person) + document_person.updated_at = datetime.now(UTC) + merged = await _session.merge(document_person) + return await self._finalize_link(session=_session, caller_session=session, link=merged) + + async def delete_document_person( + self, + document_person: DocumentPerson, + *, + session: AsyncSession | None = None, + ) -> None: + async with self._session_scope(session) as _session: + await _session.delete(document_person) + await self._finalize(session=_session, caller_session=session) + + async def read_person_detail(self, person_id: UUID, *, session: AsyncSession | None = None) -> Person: + async with self._session_scope(session) as _session: + query = ( + select(Person) + .options( + selectinload(Person.document_people).selectinload(DocumentPerson.document), # pyright: ignore[reportArgumentType] + selectinload(Person.document_people).selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType] + ) + .where(Person.id == person_id) + .execution_options(populate_existing=True) + ) + person = (await _session.exec(query)).first() + if person is None: + raise self._not_found(f"Person with id {person_id} not found") + return person + + async def list_people(self, *, session: AsyncSession | None = None) -> Sequence[Person]: + async with self._session_scope(session) as _session: + return (await _session.exec(select(Person))).all() + + async def list_person_roles( + self, + *, + active_only: bool = True, + session: AsyncSession | None = None, + ) -> Sequence[PersonRole]: + async with self._session_scope(session) as _session: + query = select(PersonRole) + if active_only: + query = query.where(PersonRole.is_active.is_(True)) + return (await _session.exec(query.order_by(PersonRole.code))).all() + + async def list_document_people( + self, + *, + document_id: UUID | None = None, + person_id: UUID | None = None, + session: AsyncSession | None = None, + ) -> Sequence[DocumentPerson]: + async with self._session_scope(session) as _session: + query = select(DocumentPerson).options( + selectinload(DocumentPerson.document), # pyright: ignore[reportArgumentType] + selectinload(DocumentPerson.person), # pyright: ignore[reportArgumentType] + selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType] + ) + if document_id is not None: + query = query.where(DocumentPerson.document_id == document_id) + if person_id is not None: + query = query.where(DocumentPerson.person_id == person_id) + return (await _session.exec(query)).all() + + async def add_document_person_link( + self, + *, + document_id: UUID, + person_id: UUID, + role_id: UUID | None = None, + role_code: str | None = None, + session: AsyncSession | None = None, + ) -> DocumentPerson: + async with self._session_scope(session) as _session: + await self._require_document(session=_session, document_id=document_id) + if await _session.get(Person, person_id) is None: + raise self._not_found(f"Person with id {person_id} not found") + + link = DocumentPerson( + document_id=document_id, + person_id=person_id, + role=DocumentPersonRole.AUTHOR, + role_id=role_id, + ) + if role_code and role_code.strip(): + link.role = self._legacy_role(role_code) + await self._sync_role_fields(session=_session, link=link) + _session.add(link) + return await self._finalize_link(session=_session, caller_session=session, link=link) + + async def set_document_person_role( + self, + *, + document_person_id: UUID, + role_id: UUID | None = None, + role_code: str | None = None, + session: AsyncSession | None = None, + ) -> DocumentPerson: + async with self._session_scope(session) as _session: + link = await _session.get(DocumentPerson, document_person_id) + if link is None: + raise self._not_found(f"DocumentPerson with id {document_person_id} not found") + if role_id is None and not (role_code or "").strip(): + raise PeopleError( + "Either role_id or role_code is required", + category=ErrorCategory.VALIDATION, + suggestion="Provide a valid role id or code and retry.", + ) + if role_id is not None and (role_code or "").strip(): + raise PeopleError( + "Provide role_id or role_code, not both", + category=ErrorCategory.VALIDATION, + suggestion="Send only one relationship role selector and retry.", + ) + + link.role_id = role_id + if role_code and role_code.strip(): + link.role = self._legacy_role(role_code) + await self._sync_role_fields(session=_session, link=link) + link.updated_at = datetime.now(UTC) + return await self._finalize_link(session=_session, caller_session=session, link=link) + + async def remove_document_person_link( + self, + *, + document_person_id: UUID, + session: AsyncSession | None = None, + ) -> None: + async with self._session_scope(session) as _session: + link = await _session.get(DocumentPerson, document_person_id) + if link is None: + raise self._not_found(f"DocumentPerson with id {document_person_id} not found") + await _session.delete(link) + await self._finalize(session=_session, caller_session=session) + + async def _resolve_or_create_role( + self, + *, + session: AsyncSession, + role_id: UUID | None, + role_code: str, + ) -> PersonRole: + if role_id is not None: + role = await session.get(PersonRole, role_id) + if role is None: + raise PeopleError( + f"Person role with id {role_id} not found", + category=ErrorCategory.VALIDATION, + suggestion="Select a valid relationship role and retry.", + ) + return role + + normalized_code = role_code.strip().lower() or DocumentPersonRole.AUTHOR.value + role = (await session.exec(select(PersonRole).where(PersonRole.code == normalized_code))).first() + if role is None: + role = PersonRole(code=normalized_code, label=normalized_code.replace("_", " ").title()) + session.add(role) + await session.flush() + return role + + async def _sync_role_fields(self, *, session: AsyncSession, link: DocumentPerson) -> None: + role_code = link.role.value if isinstance(link.role, DocumentPersonRole) else str(link.role) + role = await self._resolve_or_create_role(session=session, role_id=link.role_id, role_code=role_code) + link.role_id = role.id + link.role = self._legacy_role(role.code) + + async def _finalize_link( + self, + *, + session: AsyncSession, + caller_session: AsyncSession | None, + link: DocumentPerson, + ) -> DocumentPerson: + try: + await self._finalize(session=session, caller_session=caller_session, refresh=(link,)) + except IntegrityError as exc: + raise PeopleError( + "Duplicate relationship link for document/person/role", + category=ErrorCategory.CONFLICT, + suggestion="Remove the existing relationship link or choose a different role.", + ) from exc + return link + + async def _require_document(self, *, session: AsyncSession, document_id: UUID) -> None: + if await session.get(Document, document_id) is None: + raise self._not_found(f"Document with id {document_id} not found") + + @staticmethod + def _legacy_role(role_code: str) -> DocumentPersonRole: + try: + return DocumentPersonRole(role_code.strip().lower()) + except ValueError as exc: + raise PeopleError( + f"Unsupported role code {role_code!r}", + category=ErrorCategory.VALIDATION, + suggestion="Use author, recipient, or mentioned.", + ) from exc + + @staticmethod + def _not_found(message: str) -> PeopleError: + return PeopleError( + message, + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the requested Person or relationship id and retry.", + ) + + +def store_person_portrait( + *, + person_id: UUID, + filename: str, + file_bytes: bytes, + settings: Settings | None = None, +) -> Path: + """Persist Person portrait media under persons/.""" + if not file_bytes: + raise PersonMediaError( + "Person portrait content is empty", + category=ErrorCategory.VALIDATION, + suggestion="Select a non-empty portrait file and retry.", + ) + suffix = Path(filename).suffix.lower() + if suffix not in PORTRAIT_EXTENSIONS: + raise PersonMediaError( + f"Unsupported portrait format: {suffix or ''}", + category=ErrorCategory.USER_INPUT, + suggestion="Use JPG, JPEG, PNG, GIF, WEBP, BMP, or TIFF portrait media.", + ) + + runtime_settings = settings or get_settings() + target_dir = runtime_settings.upload_dir / "persons" / str(person_id) + target_dir.mkdir(parents=True, exist_ok=True) + stored_path = target_dir / f"{uuid4()}{suffix}" + try: + stored_path.write_bytes(file_bytes) + except OSError as exc: + raise PersonMediaError( + "Failed to persist Person portrait media", + category=ErrorCategory.INFRA_PERSISTENT, + suggestion="Check media directory permissions and available disk space, then retry.", + ) from exc + logger.info("Stored Person portrait media: %s", stored_path) + return stored_path diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py new file mode 100644 index 0000000..503fadc --- /dev/null +++ b/src/transcription/services/sources.py @@ -0,0 +1,660 @@ +"""Source persistence, media policy, revisions, and transcription execution.""" + +from __future__ import annotations + +import hashlib +import logging +from collections.abc import Sequence +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import UTC +from datetime import datetime +from pathlib import Path +from uuid import UUID + +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.orm import selectinload +from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession + +from transcription.config import Settings +from transcription.config import get_settings +from transcription.db.models import Job +from transcription.db.models import JobSource +from transcription.db.models import JobSourceStatus +from transcription.db.models import Source +from transcription.errors import AppError +from transcription.errors import ErrorCategory +from transcription.providers import ProviderAuthError +from transcription.providers import ProviderError +from transcription.providers import ProviderResponseError +from transcription.providers import TranscriptionProvider +from transcription.providers import TranscriptionResult +from transcription.providers import get_transcription_provider + +from .base import ServiceBase + +logger = logging.getLogger(__name__) + +DEFAULT_PROMPT_FILE = "transcribe_document.md" +SOURCE_MIME_TYPES = { + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".tif": "image/tiff", + ".tiff": "image/tiff", + ".pdf": "application/pdf", +} +SOURCE_EXTENSIONS = frozenset(SOURCE_MIME_TYPES) + + +@dataclass(frozen=True) +class PromptExecution: + """Resolved prompt inputs captured for one page execution.""" + + prompt_name: str + prompt_hash: str + system_prompt: str | None + user_prompt: str + temperature: float | None + top_p: float | None + + +class PromptLoadError(AppError): + """Raised when prompt artifacts cannot be loaded safely.""" + + +class TranscriptionError(AppError): + """Raised when transcription execution fails.""" + + +class TranscriptionNotFoundError(TranscriptionError): + """Raised when a transcription-related resource is not found.""" + + +class SourceDeleteBlockedError(TranscriptionError): + """Raised when source deletion is blocked by dependency policy.""" + + +class SourceService(ServiceBase): + """Manage source records, media payloads, revisions, and page execution output.""" + + provider: TranscriptionProvider + + def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None): + super().__init__(session_factory=session_factory) + self.provider = get_transcription_provider(settings=self.settings) + + async def create_source(self, source: Source, *, session: AsyncSession | None = None) -> Source: + """Create a new source page record in the database.""" + async with self._session_scope(session) as _session: + _session.add(source) + await self._finalize(session=_session, caller_session=session, refresh=(source,)) + return source + + async def read_source(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source: + """Read an existing source page record.""" + async with self._session_scope(session) as _session: + source = await _session.get(Source, source_id) + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + return source + + async def read_source_detail(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source: + """Read a source page record with job-source context for UI detail rendering.""" + async with self._session_scope(session) as _session: + query = ( + select(Source) + .options( + selectinload(Source.job_sources).selectinload(JobSource.job), # pyright: ignore[reportArgumentType] + ) + .where(Source.id == source_id) + .execution_options(populate_existing=True) + ) + source = (await _session.exec(query)).first() + + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + return source + + async def update_source(self, source: Source, *, session: AsyncSession | None = None) -> Source: + """Update an existing source page record.""" + async with self._session_scope(session) as _session: + merged = await _session.merge(source) + await self._finalize(session=_session, caller_session=session, refresh=(merged,)) + return merged + + async def delete_source(self, source: Source, *, session: AsyncSession | None = None) -> None: + """Delete a source page record.""" + source_file_path = source.file_path + async with self._session_scope(session) as _session: + await _session.delete(source) + await self._finalize(session=_session, caller_session=session) + + self._delete_source_file(source_file_path=source_file_path) + + async def delete_unlinked_source(self, *, source_id: UUID, session: AsyncSession | None = None) -> None: + """Delete a source only when no JobSource links exist.""" + async with self._session_scope(session) as _session: + source = await _session.get( + Source, + source_id, + options=( + selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] + ), + ) + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + + if source.job_sources: + raise SourceDeleteBlockedError( + "Source delete blocked because it is linked to one or more jobs", + category=ErrorCategory.VALIDATION, + suggestion="Remove JobSource links first, then retry deletion.", + ) + + source_file_path = source.file_path + await _session.delete(source) + await self._finalize(session=_session, caller_session=session) + + self._delete_source_file(source_file_path=source_file_path) + + async def list_sources( + self, + *, + document_id: UUID | None = None, + session: AsyncSession | None = None, + ) -> Sequence[Source]: + """List source pages, optionally filtered by document.""" + async with self._session_scope(session) as _session: + query = select(Source) + if document_id is not None: + query = query.where(Source.document_id == document_id) + result = await _session.exec(query) + return result.all() + + async def query_sources( + self, + *, + document_id: UUID | None = None, + page_number: int | None = None, + session: AsyncSession | None = None, + ) -> Sequence[Source]: + """Query source pages using the provided filters.""" + async with self._session_scope(session) as _session: + query = select(Source) + if document_id is not None: + query = query.where(Source.document_id == document_id) + if page_number is not None: + query = query.where(Source.page_number == page_number) + result = await _session.exec(query) + return result.all() + + async def list_sources_detail( + self, + *, + document_id: UUID | None = None, + job_id: UUID | None = None, + session: AsyncSession | None = None, + ) -> Sequence[Source]: + """List source pages with document/job link context for UI rendering.""" + async with self._session_scope(session) as _session: + query = select(Source).options( + selectinload(Source.document), + selectinload(Source.job_sources), + ) + if document_id is not None: + query = query.where(Source.document_id == document_id) + + result = await _session.exec(query) + sources = list(result.all()) + + if job_id is not None: + sources = [ + source + for source in sources + if any(job_source.job_id == job_id for job_source in source.job_sources) + ] + + return sources + + async def create_job_source( + self, + job_source: JobSource, + *, + session: AsyncSession | None = None, + ) -> JobSource: + """Create a new job_source execution record in the database.""" + async with self._session_scope(session) as _session: + _session.add(job_source) + await self._finalize(session=_session, caller_session=session, refresh=(job_source,)) + return job_source + + async def read_job_source(self, job_source_id: UUID, *, session: AsyncSession | None = None) -> JobSource: + """Read an existing job_source record.""" + async with self._session_scope(session) as _session: + job_source = await _session.get( + JobSource, + job_source_id, + options=(selectinload(JobSource.source),), # pyright: ignore[reportArgumentType] + ) + if job_source is None: + raise TranscriptionNotFoundError( + f"JobSource with id {job_source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the job source id and retry.", + ) + return job_source + + async def update_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> JobSource: + """Update an existing job_source record.""" + async with self._session_scope(session) as _session: + merged = await _session.merge(job_source) + await self._finalize(session=_session, caller_session=session, refresh=(merged,)) + return merged + + async def delete_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> None: + """Delete a job_source record.""" + async with self._session_scope(session) as _session: + await _session.delete(job_source) + await self._finalize(session=_session, caller_session=session) + + async def delete_source_from_job_context( + self, + *, + job_id: UUID, + source_id: UUID, + session: AsyncSession | None = None, + ) -> None: + """Delete a source from an active job context with dependency guardrails. + + Policy: + - Allowed when exactly one JobSource link exists and it points at ``job_id``. + - Blocked when additional JobSource links exist (history/shared dependencies). + """ + async with self._session_scope(session) as _session: + source = await _session.get( + Source, + source_id, + options=( + selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] + ), + ) + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + + linked_job_sources = list(source.job_sources) + matching_links = [job_source for job_source in linked_job_sources if job_source.job_id == job_id] + if not matching_links: + raise TranscriptionNotFoundError( + f"Source {source_id} is not linked to job {job_id}", + category=ErrorCategory.NOT_FOUND, + suggestion="Open the source from its linked job context and retry.", + ) + + if len(linked_job_sources) > len(matching_links): + raise SourceDeleteBlockedError( + "Source delete blocked by related job history", + category=ErrorCategory.VALIDATION, + suggestion="Remove additional JobSource links first, then retry deletion.", + ) + + for job_source in matching_links: + await _session.delete(job_source) + + source_file_path = source.file_path + await _session.delete(source) + await self._finalize(session=_session, caller_session=session) + + self._delete_source_file(source_file_path=source_file_path) + + def _delete_source_file(self, *, source_file_path: str) -> None: + """Best-effort cleanup for source media files.""" + candidate_path = Path(source_file_path) + resolved_path = candidate_path if candidate_path.is_absolute() else self.settings.upload_dir / candidate_path + + if not resolved_path.exists(): + return + + try: + resolved_path.unlink() + logger.info("Deleted source file: %s", resolved_path) + except OSError: + logger.warning("Failed to delete source file: %s", resolved_path) + + async def list_job_sources( + self, + *, + job_id: UUID | None = None, + session: AsyncSession | None = None, + ) -> Sequence[JobSource]: + """List job-source records, optionally filtered by job.""" + async with self._session_scope(session) as _session: + query = select(JobSource).options( + selectinload(JobSource.job), # pyright: ignore[reportArgumentType] + selectinload(JobSource.source), # pyright: ignore[reportArgumentType] + ) + if job_id is not None: + query = query.where(JobSource.job_id == job_id) + result = await _session.exec(query) + return result.all() + + async def update_job_source_transcription( + self, + *, + job_id: UUID, + source_id: UUID, + text: str | None, + error_detail: str | None = None, + ai_metadata: dict[str, object] | None = None, + raw_api_response: dict[str, object] | None = None, + provider: str | None = None, + model: str | None = None, + session: AsyncSession | None = None, + ) -> JobSource: + """Persist transcription fields for one source within a specific job.""" + async with self._session_scope(session) as _session: + job = await _session.get(Job, job_id) + if job is None: + raise TranscriptionNotFoundError( + f"Job with id {job_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the job id and retry.", + ) + + source = await _session.get(Source, source_id) + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + + if source.document_id != job.document_id: + raise TranscriptionError( + f"Source {source_id} does not belong to job {job_id}", + category=ErrorCategory.VALIDATION, + suggestion="Link the source to the same document as the job and retry.", + ) + + job.provider = provider or job.provider or self.settings.provider.value + job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings) + job.date_updated = datetime.now(UTC) + + if text is not None: + source.raw_transcription = text + + existing_job_source = await _session.exec( + select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id) + ) + job_source = existing_job_source.first() + if job_source is None: + job_source = JobSource( + job_id=job_id, + source_id=source_id, + status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED, + raw_transcription=text, + ai_metadata=ai_metadata, + raw_api_response=raw_api_response, + error_detail=error_detail, + ) + _session.add(job_source) + else: + job_source.raw_transcription = text + job_source.ai_metadata = ai_metadata + job_source.raw_api_response = raw_api_response + job_source.error_detail = error_detail + job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED + job_source.executed_at = datetime.now(UTC) + + await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source)) + return job_source + + async def upsert_revision_for_source( + self, + *, + source_id: UUID, + text: str, + session: AsyncSession | None = None, + ) -> Source: + """Persist a human revision on a source page.""" + async with self._session_scope(session) as _session: + source = await _session.get(Source, source_id) + if source is None: + raise TranscriptionNotFoundError( + f"Source with id {source_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the source id and retry.", + ) + + source.revised_text = text + source.date_revised = datetime.now(UTC) + await self._finalize(session=_session, caller_session=session, refresh=(source,)) + return source + + async def read_revision_by_source( + self, + source_id: UUID, + *, + session: AsyncSession | None = None, + ) -> Source | None: + """Read the source record for a given page, including any revision text.""" + async with self._session_scope(session) as _session: + return await _session.get(Source, source_id) + + async def list_revisions_by_job( + self, + job_id: UUID, + *, + session: AsyncSession | None = None, + ) -> Sequence[Source]: + """List source pages for a job that carry revision text.""" + async with self._session_scope(session) as _session: + query = ( + select(Source) + .join(JobSource, JobSource.source_id == Source.id) + .where(JobSource.job_id == job_id) + .where(Source.revised_text.is_not(None)) + .order_by(Source.date_revised) # pyright: ignore[reportArgumentType] + ) + result = await _session.exec(query) + return result.all() + + +def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str: + provider_model = getattr(provider, "model", None) + if isinstance(provider_model, str) and provider_model.strip(): + return provider_model + + if settings.provider_model and settings.provider_model.strip(): + return settings.provider_model + + return "unknown" + + +async def transcribe_document_image( + image_path: str | Path, + *, + prompt_name: str | None = None, + prompt_text: str | None = None, + temperature: float | None = None, + top_p: float | None = None, + settings: Settings | None = None, + provider: TranscriptionProvider | None = None, +) -> TranscriptionResult: + """Transcribe a local image using the configured prompt and provider.""" + runtime_settings = settings or get_settings() + if prompt_text is None: + prompt_execution = build_prompt_execution(prompt_name=prompt_name, settings=runtime_settings) + else: + effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip() + prompt_execution = PromptExecution( + prompt_name=effective_prompt_name, + prompt_hash=hashlib.sha256(prompt_text.encode("utf-8")).hexdigest(), + system_prompt=None, + user_prompt=prompt_text, + temperature=temperature if temperature is not None else runtime_settings.transcription_temperature, + top_p=top_p if top_p is not None else runtime_settings.transcription_top_p, + ) + image_bytes, mime_type = load_source_payload(image_path) + + adapter = provider or get_transcription_provider(settings=runtime_settings) + logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type) + + with handle_transcription_errors(): + result = await adapter.transcribe( + prompt_text=prompt_execution.user_prompt, + image_bytes=image_bytes, + mime_type=mime_type, + temperature=prompt_execution.temperature, + top_p=prompt_execution.top_p, + ) + logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider) + return TranscriptionResult( + text=result.text, + provider=result.provider, + prompt_name=prompt_execution.prompt_name, + prompt_hash=prompt_execution.prompt_hash, + system_prompt=prompt_execution.system_prompt, + user_prompt=result.user_prompt or prompt_execution.user_prompt, + temperature=result.temperature if result.temperature is not None else prompt_execution.temperature, + top_p=result.top_p if result.top_p is not None else prompt_execution.top_p, + model=result.model, + finish_reason=result.finish_reason, + usage_input_tokens=result.usage_input_tokens, + usage_output_tokens=result.usage_output_tokens, + usage_total_tokens=result.usage_total_tokens, + ai_metadata=result.ai_metadata, + raw_api_response=result.raw_api_response, + ) + + +def build_prompt_execution(*, prompt_name: str | None = None, settings: Settings | None = None) -> PromptExecution: + """Resolve the exact prompt payload and provenance for one execution.""" + runtime_settings = settings or get_settings() + effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip() + user_prompt = load_prompt_text(prompt_name=effective_prompt_name, settings=runtime_settings) + return PromptExecution( + prompt_name=effective_prompt_name, + prompt_hash=hashlib.sha256(user_prompt.encode("utf-8")).hexdigest(), + system_prompt=None, + user_prompt=user_prompt, + temperature=runtime_settings.transcription_temperature, + top_p=runtime_settings.transcription_top_p, + ) + + +def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str: + """Load and validate prompt text from PROMPT_DIR.""" + runtime_settings = settings or get_settings() + prompt_path = runtime_settings.prompt_dir / prompt_name + + if not prompt_path.exists() or not prompt_path.is_file(): + raise PromptLoadError( + f"Prompt file not found: {prompt_path}", + category=ErrorCategory.INFRA_PERSISTENT, + suggestion="Verify PROMPT_DIR and prompt file configuration, then retry.", + ) + + prompt_text = prompt_path.read_text(encoding="utf-8").strip() + if not prompt_text: + raise PromptLoadError( + f"Prompt file is empty: {prompt_path}", + category=ErrorCategory.VALIDATION, + suggestion="Populate the prompt file with valid instructions and retry.", + ) + + logger.info("Loaded prompt artifact: %s", prompt_path) + return prompt_text + + +def source_mime_type(filename: str | Path) -> str: + """Return the canonical MIME type for a supported Source filename.""" + path = Path(filename) + suffix = path.suffix.lower() + mime_type = SOURCE_MIME_TYPES.get(suffix) + if mime_type is None: + supported = ", ".join(sorted(SOURCE_EXTENSIONS)) + raise TranscriptionError( + f"Unsupported Source format: {suffix or ''}", + category=ErrorCategory.USER_INPUT, + suggestion=f"Use one of the supported Source formats: {supported}.", + ) + return mime_type + + +def validate_source_content(*, filename: str | Path, content: bytes) -> str: + """Validate Source content and return its canonical MIME type.""" + if not content: + raise TranscriptionError( + "Source content is empty", + category=ErrorCategory.VALIDATION, + suggestion="Select a non-empty Source file and try again.", + ) + + safe_name = Path(filename).name + if not safe_name: + raise TranscriptionError( + "Source filename is required", + category=ErrorCategory.VALIDATION, + suggestion="Choose a Source file with a valid filename and retry.", + ) + return source_mime_type(safe_name) + + +def load_source_payload(source_path: str | Path) -> tuple[bytes, str]: + """Read Source bytes and resolve MIME type from the canonical format policy.""" + path = Path(source_path) + + if not path.exists() or not path.is_file(): + raise TranscriptionError( + f"Source file not found: {path}", + category=ErrorCategory.NOT_FOUND, + suggestion="Verify the Source file exists and retry from the jobs page.", + ) + + content = path.read_bytes() + return content, validate_source_content(filename=path.name, content=content) + + +@contextmanager +def handle_transcription_errors(): + """Context manager to handle transcription errors.""" + try: + yield + except ProviderAuthError as exc: + raise TranscriptionError( + "Provider authentication failed", + category=ErrorCategory.INFRA_PERSISTENT, + suggestion="Verify provider API credentials and retry.", + ) from exc + except ProviderResponseError as exc: + raise TranscriptionError( + "Provider returned an invalid response", + category=ErrorCategory.EXTERNAL_PROVIDER, + suggestion="Retry once. If it persists, try a different provider model or inspect provider status.", + retriable=True, + ) from exc + except ProviderError as exc: + raise TranscriptionError( + "Provider transcription failed", + category=ErrorCategory.EXTERNAL_PROVIDER, + suggestion="Retry the transcription from jobs. If repeated, check provider availability.", + retriable=True, + ) from exc diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py index bff8c65..47dec89 100644 --- a/src/transcription/services/store.py +++ b/src/transcription/services/store.py @@ -1,9 +1,9 @@ from __future__ import annotations -from collections.abc import Sequence -from dataclasses import dataclass import hashlib import logging +from collections.abc import Sequence +from dataclasses import dataclass from pathlib import Path from uuid import UUID from uuid import uuid4 @@ -21,17 +21,18 @@ from ..db.models import Job from ..db.models import JobSource from ..db.models import JobSourceStatus from ..db.models import Source -from .documents import UploadJobResult +from .sources import TranscriptionError +from .sources import validate_source_content from .transcription import build_prompt_execution logger = logging.getLogger(__name__) -SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"} -SUPPORTED_PORTRAIT_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tif", ".tiff"} + +class SourceStorageError(AppError): + """Raised when Source content cannot be validated or persisted safely.""" -class UploadError(AppError): - """Raised when uploaded content cannot be persisted safely.""" +UploadError = SourceStorageError @dataclass(frozen=True) @@ -44,8 +45,18 @@ class JobCreateResult: @dataclass(frozen=True) -class PendingStoredUpload: - """Pre-staged upload artifact tied to a source id.""" +class DocumentJobResult: + """Summary of a Document, Source, and Job created together.""" + + document_id: UUID + job_id: UUID + stored_path: Path + original_filename: str + + +@dataclass(frozen=True) +class PendingStoredSource: + """Pre-staged Source artifact tied to a Source id.""" source_id: UUID original_filename: str @@ -54,19 +65,19 @@ class PendingStoredUpload: file_size_bytes: int -async def create_upload_job( +async def create_document_job( *, filename: str, file_bytes: bytes, session: AsyncSession, settings: Settings | None = None, -) -> UploadJobResult: - """Create upload-backed document and queued job records.""" +) -> DocumentJobResult: + """Create a Document, its first Source, and a queued Job.""" runtime_settings = settings or get_settings() prompt_execution = build_prompt_execution(settings=runtime_settings) document_id = uuid4() source_id = uuid4() - stored_path = store_file( + stored_path = store_source_file( filename=filename, file_bytes=file_bytes, settings=runtime_settings, @@ -75,7 +86,7 @@ async def create_upload_job( ) file_hash, file_size_bytes = _compute_file_metadata(file_bytes) try: - document, job = await _create_upload_records( + document, job = await _create_document_job_records( session=session, document_id=document_id, source_id=source_id, @@ -87,15 +98,15 @@ async def create_upload_job( ) except Exception as exc: _best_effort_delete(stored_path) - raise UploadError( - "Failed to create upload database records", + raise SourceStorageError( + "Failed to create Document, Source, and Job records", category=ErrorCategory.INFRA_TRANSIENT, suggestion="Retry upload. If this keeps happening, verify database availability.", retriable=True, ) from exc - logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id) - return UploadJobResult( + logger.info("Created document job document_id=%s job_id=%s", document.id, job.id) + return DocumentJobResult( document_id=document.id, job_id=job.id, stored_path=stored_path, @@ -106,31 +117,31 @@ async def create_upload_job( async def create_job_for_document( *, document_id: UUID, - uploads: Sequence[tuple[str, bytes]], + source_files: Sequence[tuple[str, bytes]], session: AsyncSession, provider: str | None = None, model: str | None = None, settings: Settings | None = None, ) -> JobCreateResult: - """Create a queued job for an existing document with one or more uploaded sources.""" - if not uploads: - raise UploadError( - "At least one upload is required to create a job", + """Create a queued Job for an existing Document with one or more Sources.""" + if not source_files: + raise SourceStorageError( + "At least one Source file is required to create a Job", category=ErrorCategory.VALIDATION, suggestion="Upload one or more files and try again.", ) runtime_settings = settings or get_settings() prompt_execution = build_prompt_execution(settings=runtime_settings) - sorted_uploads = sorted(uploads, key=lambda item: Path(item[0]).name.casefold()) - stored_uploads: list[PendingStoredUpload] = [] - for filename, file_bytes in sorted_uploads: + sorted_source_files = sorted(source_files, key=lambda item: Path(item[0]).name.casefold()) + stored_sources: list[PendingStoredSource] = [] + for filename, file_bytes in sorted_source_files: source_id = uuid4() - stored_uploads.append( - PendingStoredUpload( + stored_sources.append( + PendingStoredSource( source_id=source_id, original_filename=filename, - stored_path=store_file( + stored_path=store_source_file( filename=filename, file_bytes=file_bytes, settings=runtime_settings, @@ -146,16 +157,16 @@ async def create_job_for_document( job, source_ids = await _create_job_for_document_records( session=session, document_id=document_id, - stored_uploads=stored_uploads, + stored_sources=stored_sources, provider=provider, model=model, prompt_execution=prompt_execution, ) except Exception as exc: - for upload in stored_uploads: - _best_effort_delete(upload.stored_path) - raise UploadError( - "Failed to create job records from uploads", + for source in stored_sources: + _best_effort_delete(source.stored_path) + raise SourceStorageError( + "Failed to create Job records from Source files", category=ErrorCategory.INFRA_TRANSIENT, suggestion="Retry creation. If this keeps happening, verify database availability.", retriable=True, @@ -169,7 +180,7 @@ async def create_job_for_document( ) -async def _create_upload_records( +async def _create_document_job_records( *, session: AsyncSession, document_id: UUID, @@ -230,23 +241,21 @@ async def _create_job_for_document_records( *, session: AsyncSession, document_id: UUID, - stored_uploads: Sequence[PendingStoredUpload], + stored_sources: Sequence[PendingStoredSource], provider: str | None, model: str | None, prompt_execution, ) -> tuple[Job, list[UUID]]: document = await session.get(Document, document_id) if document is None: - raise UploadError( + raise SourceStorageError( f"Document with id {document_id} not found", category=ErrorCategory.NOT_FOUND, suggestion="Select an existing document and retry.", ) - existing_sources = ( - await session.exec(select(Source).where(Source.document_id == document_id)) - ).all() - next_page_number = (max((source.page_number for source in existing_sources), default=0) + 1) + existing_sources = (await session.exec(select(Source).where(Source.document_id == document_id))).all() + next_page_number = max((source.page_number for source in existing_sources), default=0) + 1 job = Job( document_id=document_id, @@ -263,16 +272,16 @@ async def _create_job_for_document_records( await session.flush() source_ids: list[UUID] = [] - for page_offset, upload in enumerate(stored_uploads): + for page_offset, stored_source in enumerate(stored_sources): source = Source( - id=upload.source_id, + id=stored_source.source_id, document_id=document_id, page_number=next_page_number + page_offset, - upload_name=Path(upload.original_filename).name, - filename=upload.stored_path.name, - file_path=str(upload.stored_path), - file_hash=upload.file_hash, - file_size_bytes=upload.file_size_bytes, + upload_name=Path(stored_source.original_filename).name, + filename=stored_source.stored_path.name, + file_path=str(stored_source.stored_path), + file_hash=stored_source.file_hash, + file_size_bytes=stored_source.file_size_bytes, ) session.add(source) await session.flush() @@ -296,7 +305,7 @@ def _best_effort_delete(path: Path) -> None: if path.exists(): path.unlink() except OSError: - logger.warning("Failed to clean up upload file after DB error: %s", path) + logger.warning("Failed to clean up Source file after database error: %s", path) def _compute_file_hash(file_bytes: bytes) -> str: @@ -307,7 +316,7 @@ def _compute_file_metadata(file_bytes: bytes) -> tuple[str, int]: return _compute_file_hash(file_bytes), len(file_bytes) -def store_file( +def store_source_file( *, filename: str, file_bytes: bytes, @@ -315,9 +324,17 @@ def store_file( relative_directory: Path | None = None, filename_stem: str | None = None, ) -> Path: - """Persist an uploaded file to the configured upload directory.""" + """Validate and persist a Source file to configured media storage.""" runtime_settings = settings or get_settings() - _validate_upload(filename=filename, file_bytes=file_bytes, supported_extensions=SUPPORTED_UPLOAD_EXTENSIONS) + try: + validate_source_content(filename=filename, content=file_bytes) + except TranscriptionError as exc: + raise SourceStorageError( + exc.message, + category=exc.category, + suggestion=exc.suggestion, + retriable=exc.retriable, + ) from exc return _store_file_bytes( filename=filename, file_bytes=file_bytes, @@ -327,24 +344,6 @@ def store_file( ) -def store_person_portrait( - *, - person_id: UUID, - filename: str, - file_bytes: bytes, - settings: Settings | None = None, -) -> Path: - """Persist a portrait upload under persons/.""" - runtime_settings = settings or get_settings() - _validate_upload(filename=filename, file_bytes=file_bytes, supported_extensions=SUPPORTED_PORTRAIT_EXTENSIONS) - return _store_file_bytes( - filename=filename, - file_bytes=file_bytes, - settings=runtime_settings, - relative_directory=Path("persons") / str(person_id), - ) - - def _store_file_bytes( *, filename: str, @@ -363,43 +362,22 @@ def _store_file_bytes( try: stored_path.write_bytes(file_bytes) except OSError as exc: - raise UploadError( - "Failed to persist upload file", + raise SourceStorageError( + "Failed to persist Source file", category=ErrorCategory.INFRA_PERSISTENT, suggestion="Check upload directory permissions and available disk space, then retry.", ) from exc - logger.info("Stored uploaded file: %s", stored_path) + logger.info("Stored Source file: %s", stored_path) return stored_path -def _validate_upload(*, filename: str, file_bytes: bytes, supported_extensions: set[str]) -> None: - if not file_bytes: - raise UploadError( - "Upload payload is empty", - category=ErrorCategory.VALIDATION, - suggestion="Select a non-empty file and try again.", - ) - - safe_name = Path(filename).name - if not safe_name: - raise UploadError( - "Upload filename is required", - category=ErrorCategory.VALIDATION, - suggestion="Choose a file with a valid filename and retry.", - ) - - suffix = Path(safe_name).suffix.lower() - if suffix not in supported_extensions: - raise UploadError( - f"Unsupported upload extension: {suffix}", - category=ErrorCategory.USER_INPUT, - suggestion="Upload a supported image or document file and retry.", - ) - - def _build_stored_filename(*, filename: str, filename_stem: str | None = None) -> str: safe_name = Path(filename).name suffix = Path(safe_name).suffix.lower() stem = filename_stem or str(uuid4()) return f"{stem}{suffix}" + + +create_upload_job = create_document_job +store_file = store_source_file diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 3bfe990..eb2388e 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -1,634 +1,44 @@ -"""Prompt loading and provider-backed transcription service.""" - -from __future__ import annotations - -from dataclasses import dataclass -import hashlib -import logging -import mimetypes -from collections.abc import Sequence -from contextlib import contextmanager -from datetime import UTC -from datetime import datetime -from pathlib import Path -from uuid import UUID - -from sqlalchemy.ext.asyncio import async_sessionmaker -from sqlalchemy.orm import selectinload -from sqlmodel import select -from sqlmodel.ext.asyncio.session import AsyncSession - -from transcription.config import Settings -from transcription.config import get_settings -from transcription.db.models import Job -from transcription.db.models import JobSource -from transcription.db.models import JobSourceStatus -from transcription.db.models import Source -from transcription.errors import AppError -from transcription.errors import ErrorCategory -from transcription.providers import ProviderAuthError -from transcription.providers import ProviderError -from transcription.providers import ProviderResponseError -from transcription.providers import TranscriptionProvider -from transcription.providers import TranscriptionResult -from transcription.providers import get_transcription_provider - -from .base import ServiceBase - -logger = logging.getLogger(__name__) - -DEFAULT_PROMPT_FILE = "transcribe_document.md" -SUPPORTED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"} - - -@dataclass(frozen=True) -class PromptExecution: - """Resolved prompt inputs captured for one page execution.""" - - prompt_name: str - prompt_hash: str - system_prompt: str | None - user_prompt: str - temperature: float | None - top_p: float | None - - -class PromptLoadError(AppError): - """Raised when prompt artifacts cannot be loaded safely.""" - - -class TranscriptionError(AppError): - """Raised when transcription execution fails.""" - - -class TranscriptionNotFoundError(TranscriptionError): - """Raised when a transcription-related resource is not found.""" - - -class SourceDeleteBlockedError(TranscriptionError): - """Raised when source deletion is blocked by dependency policy.""" - - -class TranscriptionService(ServiceBase): - """Service class for job transcription output and page-level source revisions.""" - - provider: TranscriptionProvider - - def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None): - super().__init__(session_factory=session_factory) - self.provider = get_transcription_provider(settings=self.settings) - - async def create_source(self, source: Source, *, session: AsyncSession | None = None) -> Source: - """Create a new source page record in the database.""" - async with self._session_scope(session) as _session: - _session.add(source) - await self._finalize(session=_session, caller_session=session, refresh=(source,)) - return source - - async def read_source(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source: - """Read an existing source page record.""" - async with self._session_scope(session) as _session: - source = await _session.get(Source, source_id) - if source is None: - raise TranscriptionNotFoundError( - f"Source with id {source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the source id and retry.", - ) - return source - - async def read_source_detail(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source: - """Read a source page record with job-source context for UI detail rendering.""" - async with self._session_scope(session) as _session: - query = ( - select(Source) - .options( - selectinload(Source.job_sources).selectinload(JobSource.job), # pyright: ignore[reportArgumentType] - ) - .where(Source.id == source_id) - .execution_options(populate_existing=True) - ) - source = (await _session.exec(query)).first() - - if source is None: - raise TranscriptionNotFoundError( - f"Source with id {source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the source id and retry.", - ) - return source - - async def update_source(self, source: Source, *, session: AsyncSession | None = None) -> Source: - """Update an existing source page record.""" - async with self._session_scope(session) as _session: - merged = await _session.merge(source) - await self._finalize(session=_session, caller_session=session, refresh=(merged,)) - return merged - - async def delete_source(self, source: Source, *, session: AsyncSession | None = None) -> None: - """Delete a source page record.""" - source_file_path = source.file_path - async with self._session_scope(session) as _session: - await _session.delete(source) - await self._finalize(session=_session, caller_session=session) - - self._delete_source_file(source_file_path=source_file_path) - - async def delete_unlinked_source(self, *, source_id: UUID, session: AsyncSession | None = None) -> None: - """Delete a source only when no JobSource links exist.""" - async with self._session_scope(session) as _session: - source = await _session.get( - Source, - source_id, - options=( - selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] - ), - ) - if source is None: - raise TranscriptionNotFoundError( - f"Source with id {source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the source id and retry.", - ) - - if source.job_sources: - raise SourceDeleteBlockedError( - "Source delete blocked because it is linked to one or more jobs", - category=ErrorCategory.VALIDATION, - suggestion="Remove JobSource links first, then retry deletion.", - ) - - source_file_path = source.file_path - await _session.delete(source) - await self._finalize(session=_session, caller_session=session) - - self._delete_source_file(source_file_path=source_file_path) - - async def list_sources( - self, - *, - document_id: UUID | None = None, - session: AsyncSession | None = None, - ) -> Sequence[Source]: - """List source pages, optionally filtered by document.""" - async with self._session_scope(session) as _session: - query = select(Source) - if document_id is not None: - query = query.where(Source.document_id == document_id) - result = await _session.exec(query) - return result.all() - - async def query_sources( - self, - *, - document_id: UUID | None = None, - page_number: int | None = None, - session: AsyncSession | None = None, - ) -> Sequence[Source]: - """Query source pages using the provided filters.""" - async with self._session_scope(session) as _session: - query = select(Source) - if document_id is not None: - query = query.where(Source.document_id == document_id) - if page_number is not None: - query = query.where(Source.page_number == page_number) - result = await _session.exec(query) - return result.all() - - async def list_sources_detail( - self, - *, - document_id: UUID | None = None, - job_id: UUID | None = None, - session: AsyncSession | None = None, - ) -> Sequence[Source]: - """List source pages with document/job link context for UI rendering.""" - async with self._session_scope(session) as _session: - query = select(Source).options( - selectinload(Source.document), - selectinload(Source.job_sources), - ) - if document_id is not None: - query = query.where(Source.document_id == document_id) - - result = await _session.exec(query) - sources = list(result.all()) - - if job_id is not None: - sources = [ - source for source in sources if any(job_source.job_id == job_id for job_source in source.job_sources) - ] - - return sources - - async def create_job_source( - self, - job_source: JobSource, - *, - session: AsyncSession | None = None, - ) -> JobSource: - """Create a new job_source execution record in the database.""" - async with self._session_scope(session) as _session: - _session.add(job_source) - await self._finalize(session=_session, caller_session=session, refresh=(job_source,)) - return job_source - - async def read_job_source(self, job_source_id: UUID, *, session: AsyncSession | None = None) -> JobSource: - """Read an existing job_source record.""" - async with self._session_scope(session) as _session: - job_source = await _session.get( - JobSource, - job_source_id, - options=(selectinload(JobSource.source),), # pyright: ignore[reportArgumentType] - ) - if job_source is None: - raise TranscriptionNotFoundError( - f"JobSource with id {job_source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the job source id and retry.", - ) - return job_source - - async def update_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> JobSource: - """Update an existing job_source record.""" - async with self._session_scope(session) as _session: - merged = await _session.merge(job_source) - await self._finalize(session=_session, caller_session=session, refresh=(merged,)) - return merged - - async def delete_job_source(self, job_source: JobSource, *, session: AsyncSession | None = None) -> None: - """Delete a job_source record.""" - async with self._session_scope(session) as _session: - await _session.delete(job_source) - await self._finalize(session=_session, caller_session=session) - - async def delete_source_from_job_context( - self, - *, - job_id: UUID, - source_id: UUID, - session: AsyncSession | None = None, - ) -> None: - """Delete a source from an active job context with dependency guardrails. - - Policy: - - Allowed when exactly one JobSource link exists and it points at ``job_id``. - - Blocked when additional JobSource links exist (history/shared dependencies). - """ - async with self._session_scope(session) as _session: - source = await _session.get( - Source, - source_id, - options=( - selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] - ), - ) - if source is None: - raise TranscriptionNotFoundError( - f"Source with id {source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the source id and retry.", - ) - - linked_job_sources = list(source.job_sources) - matching_links = [job_source for job_source in linked_job_sources if job_source.job_id == job_id] - if not matching_links: - raise TranscriptionNotFoundError( - f"Source {source_id} is not linked to job {job_id}", - category=ErrorCategory.NOT_FOUND, - suggestion="Open the source from its linked job context and retry.", - ) - - if len(linked_job_sources) > len(matching_links): - raise SourceDeleteBlockedError( - "Source delete blocked by related job history", - category=ErrorCategory.VALIDATION, - suggestion="Remove additional JobSource links first, then retry deletion.", - ) - - for job_source in matching_links: - await _session.delete(job_source) - - source_file_path = source.file_path - await _session.delete(source) - await self._finalize(session=_session, caller_session=session) - - self._delete_source_file(source_file_path=source_file_path) - - def _delete_source_file(self, *, source_file_path: str) -> None: - """Best-effort cleanup for source media files.""" - candidate_path = Path(source_file_path) - resolved_path = candidate_path if candidate_path.is_absolute() else self.settings.upload_dir / candidate_path - - if not resolved_path.exists(): - return - - try: - resolved_path.unlink() - logger.info("Deleted source file: %s", resolved_path) - except OSError: - logger.warning("Failed to delete source file: %s", resolved_path) - - async def list_job_sources( - self, - *, - job_id: UUID | None = None, - session: AsyncSession | None = None, - ) -> Sequence[JobSource]: - """List job-source records, optionally filtered by job.""" - async with self._session_scope(session) as _session: - query = select(JobSource).options( - selectinload(JobSource.job), # pyright: ignore[reportArgumentType] - selectinload(JobSource.source), # pyright: ignore[reportArgumentType] - ) - if job_id is not None: - query = query.where(JobSource.job_id == job_id) - result = await _session.exec(query) - return result.all() - - async def update_job_source_transcription( - self, - *, - job_id: UUID, - source_id: UUID, - text: str | None, - error_detail: str | None = None, - ai_metadata: dict[str, object] | None = None, - raw_api_response: dict[str, object] | None = None, - provider: str | None = None, - model: str | None = None, - session: AsyncSession | None = None, - ) -> JobSource: - """Persist transcription fields for one source within a specific job.""" - async with self._session_scope(session) as _session: - job = await _session.get(Job, job_id) - if job is None: - raise TranscriptionNotFoundError( - f"Job with id {job_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the job id and retry.", - ) - - source = await _session.get(Source, source_id) - if source is None: - raise TranscriptionNotFoundError( - f"Source with id {source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the source id and retry.", - ) - - if source.document_id != job.document_id: - raise TranscriptionError( - f"Source {source_id} does not belong to job {job_id}", - category=ErrorCategory.VALIDATION, - suggestion="Link the source to the same document as the job and retry.", - ) - - job.provider = provider or job.provider or self.settings.provider.value - job.model = model or job.model or _resolve_transcript_model(provider=self.provider, settings=self.settings) - job.date_updated = datetime.now(UTC) - - if text is not None: - source.raw_transcription = text - - existing_job_source = await _session.exec( - select(JobSource).where(JobSource.job_id == job_id).where(JobSource.source_id == source_id) - ) - job_source = existing_job_source.first() - if job_source is None: - job_source = JobSource( - job_id=job_id, - source_id=source_id, - status=JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED, - raw_transcription=text, - ai_metadata=ai_metadata, - raw_api_response=raw_api_response, - error_detail=error_detail, - ) - _session.add(job_source) - else: - job_source.raw_transcription = text - job_source.ai_metadata = ai_metadata - job_source.raw_api_response = raw_api_response - job_source.error_detail = error_detail - job_source.status = JobSourceStatus.TRANSCRIBED if text is not None else JobSourceStatus.FAILED - job_source.executed_at = datetime.now(UTC) - - await self._finalize(session=_session, caller_session=session, refresh=(job, source, job_source)) - return job_source - - async def upsert_revision_for_source( - self, - *, - source_id: UUID, - text: str, - session: AsyncSession | None = None, - ) -> Source: - """Persist a human revision on a source page.""" - async with self._session_scope(session) as _session: - source = await _session.get(Source, source_id) - if source is None: - raise TranscriptionNotFoundError( - f"Source with id {source_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the source id and retry.", - ) - - source.revised_text = text - source.date_revised = datetime.now(UTC) - await self._finalize(session=_session, caller_session=session, refresh=(source,)) - return source - - async def read_revision_by_source( - self, - source_id: UUID, - *, - session: AsyncSession | None = None, - ) -> Source | None: - """Read the source record for a given page, including any revision text.""" - async with self._session_scope(session) as _session: - return await _session.get(Source, source_id) - - async def list_revisions_by_job( - self, - job_id: UUID, - *, - session: AsyncSession | None = None, - ) -> Sequence[Source]: - """List source pages for a job that carry revision text.""" - async with self._session_scope(session) as _session: - query = ( - select(Source) - .join(JobSource, JobSource.source_id == Source.id) - .where(JobSource.job_id == job_id) - .where(Source.revised_text.is_not(None)) - .order_by(Source.date_revised) # pyright: ignore[reportArgumentType] - ) - result = await _session.exec(query) - return result.all() - - -def _resolve_transcript_model(*, provider: TranscriptionProvider, settings: Settings) -> str: - provider_model = getattr(provider, "model", None) - if isinstance(provider_model, str) and provider_model.strip(): - return provider_model - - if settings.provider_model and settings.provider_model.strip(): - return settings.provider_model - - return "unknown" - - -async def transcribe_document_image( - image_path: str | Path, - *, - prompt_name: str | None = None, - prompt_text: str | None = None, - temperature: float | None = None, - top_p: float | None = None, - settings: Settings | None = None, - provider: TranscriptionProvider | None = None, -) -> TranscriptionResult: - """Transcribe a local image using the configured prompt and provider.""" - runtime_settings = settings or get_settings() - if prompt_text is None: - prompt_execution = build_prompt_execution(prompt_name=prompt_name, settings=runtime_settings) - else: - effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip() - prompt_execution = PromptExecution( - prompt_name=effective_prompt_name, - prompt_hash=hashlib.sha256(prompt_text.encode("utf-8")).hexdigest(), - system_prompt=None, - user_prompt=prompt_text, - temperature=temperature if temperature is not None else runtime_settings.transcription_temperature, - top_p=top_p if top_p is not None else runtime_settings.transcription_top_p, - ) - image_bytes, mime_type = load_image_payload(image_path) - - adapter = provider or get_transcription_provider(settings=runtime_settings) - logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type) - - with handle_transcription_errors(): - result = await adapter.transcribe( - prompt_text=prompt_execution.user_prompt, - image_bytes=image_bytes, - mime_type=mime_type, - temperature=prompt_execution.temperature, - top_p=prompt_execution.top_p, - ) - logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider) - return TranscriptionResult( - text=result.text, - provider=result.provider, - prompt_name=prompt_execution.prompt_name, - prompt_hash=prompt_execution.prompt_hash, - system_prompt=prompt_execution.system_prompt, - user_prompt=result.user_prompt or prompt_execution.user_prompt, - temperature=result.temperature if result.temperature is not None else prompt_execution.temperature, - top_p=result.top_p if result.top_p is not None else prompt_execution.top_p, - model=result.model, - finish_reason=result.finish_reason, - usage_input_tokens=result.usage_input_tokens, - usage_output_tokens=result.usage_output_tokens, - usage_total_tokens=result.usage_total_tokens, - ai_metadata=result.ai_metadata, - raw_api_response=result.raw_api_response, - ) - - -def build_prompt_execution(*, prompt_name: str | None = None, settings: Settings | None = None) -> PromptExecution: - """Resolve the exact prompt payload and provenance for one execution.""" - runtime_settings = settings or get_settings() - effective_prompt_name = (prompt_name or runtime_settings.default_prompt_name or DEFAULT_PROMPT_FILE).strip() - user_prompt = load_prompt_text(prompt_name=effective_prompt_name, settings=runtime_settings) - return PromptExecution( - prompt_name=effective_prompt_name, - prompt_hash=hashlib.sha256(user_prompt.encode("utf-8")).hexdigest(), - system_prompt=None, - user_prompt=user_prompt, - temperature=runtime_settings.transcription_temperature, - top_p=runtime_settings.transcription_top_p, - ) - - -def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str: - """Load and validate prompt text from PROMPT_DIR.""" - runtime_settings = settings or get_settings() - prompt_path = runtime_settings.prompt_dir / prompt_name - - if not prompt_path.exists() or not prompt_path.is_file(): - raise PromptLoadError( - f"Prompt file not found: {prompt_path}", - category=ErrorCategory.INFRA_PERSISTENT, - suggestion="Verify PROMPT_DIR and prompt file configuration, then retry.", - ) - - prompt_text = prompt_path.read_text(encoding="utf-8").strip() - if not prompt_text: - raise PromptLoadError( - f"Prompt file is empty: {prompt_path}", - category=ErrorCategory.VALIDATION, - suggestion="Populate the prompt file with valid instructions and retry.", - ) - - logger.info("Loaded prompt artifact: %s", prompt_path) - return prompt_text - - -def load_image_payload(image_path: str | Path) -> tuple[bytes, str]: - """Read image bytes and detect mime type for supported uploads.""" - path = Path(image_path) - - if not path.exists() or not path.is_file(): - raise TranscriptionError( - f"Image file not found: {path}", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the uploaded file exists and retry from the jobs page.", - ) - - suffix = path.suffix.lower() - if suffix not in SUPPORTED_EXTENSIONS: - raise TranscriptionError( - f"Unsupported file type: {suffix}", - category=ErrorCategory.USER_INPUT, - suggestion="Use JPG, JPEG, PNG, TIFF, or PDF files.", - ) - - mime_type, _ = mimetypes.guess_type(path.name) - if suffix in {".tif", ".tiff"}: - mime_type = "image/tiff" - if not mime_type: - raise TranscriptionError( - f"Unable to determine MIME type for: {path}", - category=ErrorCategory.VALIDATION, - suggestion="Re-save the file in a supported format and retry.", - ) - - return path.read_bytes(), mime_type - - -@contextmanager -def handle_transcription_errors(): - """Context manager to handle transcription errors.""" - try: - yield - except ProviderAuthError as exc: - raise TranscriptionError( - "Provider authentication failed", - category=ErrorCategory.INFRA_PERSISTENT, - suggestion="Verify provider API credentials and retry.", - ) from exc - except ProviderResponseError as exc: - raise TranscriptionError( - "Provider returned an invalid response", - category=ErrorCategory.EXTERNAL_PROVIDER, - suggestion="Retry once. If it persists, try a different provider model or inspect provider status.", - retriable=True, - ) from exc - except ProviderError as exc: - raise TranscriptionError( - "Provider transcription failed", - category=ErrorCategory.EXTERNAL_PROVIDER, - suggestion="Retry the transcription from jobs. If repeated, check provider availability.", - retriable=True, - ) from exc +"""Compatibility exports for the Source-owned transcription implementation.""" + +from .sources import DEFAULT_PROMPT_FILE +from .sources import SOURCE_EXTENSIONS +from .sources import SOURCE_MIME_TYPES +from .sources import PromptExecution +from .sources import PromptLoadError +from .sources import SourceDeleteBlockedError +from .sources import SourceService +from .sources import TranscriptionError +from .sources import TranscriptionNotFoundError +from .sources import build_prompt_execution +from .sources import handle_transcription_errors +from .sources import load_prompt_text +from .sources import load_source_payload +from .sources import source_mime_type +from .sources import transcribe_document_image +from .sources import validate_source_content + +TranscriptionService = SourceService +SUPPORTED_EXTENSIONS = SOURCE_EXTENSIONS +load_image_payload = load_source_payload + +__all__ = [ + "DEFAULT_PROMPT_FILE", + "SOURCE_EXTENSIONS", + "SOURCE_MIME_TYPES", + "SUPPORTED_EXTENSIONS", + "PromptExecution", + "PromptLoadError", + "SourceDeleteBlockedError", + "SourceService", + "TranscriptionError", + "TranscriptionNotFoundError", + "TranscriptionService", + "build_prompt_execution", + "handle_transcription_errors", + "load_image_payload", + "load_prompt_text", + "load_source_payload", + "source_mime_type", + "transcribe_document_image", + "validate_source_content", +] diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 5a73626..ecd2de8 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -15,9 +15,9 @@ from ..errors import classify_unexpected_error from ..errors import format_error_detail from ..providers import TranscriptionResult from . import ServiceBundle -from .transcription import build_prompt_execution -from .transcription import PromptExecution -from .transcription import transcribe_document_image +from .sources import PromptExecution +from .sources import build_prompt_execution +from .sources import transcribe_document_image logger = logging.getLogger(__name__) @@ -79,8 +79,8 @@ async def process_queued_job( source_job = await services.jobs.read_job(job_id=job.id, session=session) sources = _resolve_job_sources(source_job) if not sources and not source_job.job_sources: - candidate_sources = await services.transcriptions.list_sources(document_id=job.document_id, session=session) - sources = list(sorted(candidate_sources, key=lambda item: (item.page_number, item.upload_name.casefold()))) + candidate_sources = await services.sources.list_sources(document_id=job.document_id, session=session) + sources = sorted(candidate_sources, key=lambda item: (item.page_number, item.upload_name.casefold())) if not sources: return await services.jobs.mark_job_status(job.id, JobStatus.TRANSCRIBED, session=session) @@ -106,7 +106,7 @@ async def process_queued_job( temperature=prompt_execution.temperature, top_p=prompt_execution.top_p, settings=runtime_settings, - provider=services.transcriptions.provider, + provider=services.sources.provider, ), timeout=runtime_settings.worker_provider_timeout_seconds, ) @@ -223,7 +223,7 @@ def _resolve_job_sources(job: Job) -> list[Source]: for job_source in job.job_sources if job_source.source is not None and job_source.status != JobSourceStatus.TRANSCRIBED ] - return list(sorted(sources, key=lambda item: (item.page_number, item.upload_name.casefold()))) + return sorted(sources, key=lambda item: (item.page_number, item.upload_name.casefold())) def _resolve_job_prompt_execution(*, source_job: Job, settings: Settings) -> PromptExecution: @@ -264,7 +264,7 @@ async def _finalize_batch_outcome( if session is None: async with services.jobs._session_scope() as local_session: for source, result in successful_pages: - await services.transcriptions.update_job_source_transcription( + await services.sources.update_job_source_transcription( job_id=job.id, source_id=source.id, text=result.text, @@ -277,7 +277,7 @@ async def _finalize_batch_outcome( ) for source, error in failed_pages: - await services.transcriptions.update_job_source_transcription( + await services.sources.update_job_source_transcription( job_id=job.id, source_id=source.id, text=None, @@ -290,7 +290,7 @@ async def _finalize_batch_outcome( return updated_job for source, result in successful_pages: - await services.transcriptions.update_job_source_transcription( + await services.sources.update_job_source_transcription( job_id=job.id, source_id=source.id, text=result.text, @@ -303,7 +303,7 @@ async def _finalize_batch_outcome( ) for source, error in failed_pages: - await services.transcriptions.update_job_source_transcription( + await services.sources.update_job_source_transcription( job_id=job.id, source_id=source.id, text=None, diff --git a/src/transcription/ui/pages/documents_page.py b/src/transcription/ui/pages/documents_page.py index fad8f97..b850eaf 100644 --- a/src/transcription/ui/pages/documents_page.py +++ b/src/transcription/ui/pages/documents_page.py @@ -17,6 +17,7 @@ from transcription.services.documents import ( DocumentError, DocumentService, ) +from transcription.services.people import PeopleService from transcription.ui.components.app_shell import render_navigation_header from transcription.ui.components.cards import archival_card from transcription.ui.components.data_display import archival_badge, metadata_row @@ -32,19 +33,21 @@ from transcription.ui.theme import page_header from ...db.session import SessionFactoryDep + def register_page() -> None: """Register documents list and detail routes.""" @ui.page("/documents/new") async def document_create_page(request: Request, session_factory: SessionFactoryDep) -> None: document_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/documents") with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"): page_header("Create Document", subtitle="Document name and type are required.") - people = sorted(await document_service.list_people(), key=lambda item: item.full_name.casefold()) - role_catalog = await document_service.list_person_roles() + people = sorted(await people_service.list_people(), key=lambda item: item.full_name.casefold()) + role_catalog = await people_service.list_person_roles() type_catalog = await document_service.list_document_types() form = _render_document_form_fields( people=people, @@ -87,10 +90,12 @@ def register_page() -> None: show_error(exc, title="Create failed", operation="documents.create") return - desired_links = _collect_role_link_candidates(form["role_people"], role_codes=[role.code for role in role_catalog]) + desired_links = _collect_role_link_candidates( + form["role_people"], role_codes=[role.code for role in role_catalog] + ) try: for role_code, person_id in sorted(desired_links): - await document_service.add_document_person_link( + await people_service.add_document_person_link( document_id=created.id, person_id=person_id, role_code=role_code, @@ -137,7 +142,9 @@ def register_page() -> None: DocumentTableRow( id=doc.id, name=doc.name, - document_type=(doc.document_type_ref.label if doc.document_type_ref is not None else (doc.document_type or "")), + document_type=( + doc.document_type_ref.label if doc.document_type_ref is not None else (doc.document_type or "") + ), archive_identifier=doc.archive_identifier or "", created_at=doc.created_at.strftime("%b %d, %Y"), ) @@ -165,7 +172,11 @@ def register_page() -> None: return with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"): - type_display = document.document_type_ref.label if document.document_type_ref is not None else (document.document_type or "Unspecified") + type_display = ( + document.document_type_ref.label + if document.document_type_ref is not None + else (document.document_type or "Unspecified") + ) with section_header_row(): page_header(document.name, subtitle=f"Type: {type_display} | ID: {document.id}") @@ -210,8 +221,16 @@ def register_page() -> None: with section_header_row(): page_header(f"Jobs for {document.name}") with ui.row().classes("gap-2"): - ui.button("Back to Document", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props("flat") - ui.button("Create Job", on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), icon="add").classes("ui-btn-primary") + ui.button( + "Back to Document", + on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), + icon="arrow_back", + ).props("flat") + ui.button( + "Create Job", + on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), + icon="add", + ).classes("ui-btn-primary") if not document.jobs: with archival_card(extra_classes="p-6 text-center"): @@ -224,7 +243,11 @@ def register_page() -> None: with ui.row().classes("items-center gap-2"): archival_badge(job.status.value) ui.label(f"Job ID: {job.id}").classes("text-xs font-mono ui-text-primary") - ui.button("Open Job", on_click=lambda _=None, jid=job.id: ui.navigate.to(f"/jobs/{jid}"), icon="open_in_new").props("flat dense").classes("text-xs ui-link-primary") + ui.button( + "Open Job", + on_click=lambda _=None, jid=job.id: ui.navigate.to(f"/jobs/{jid}"), + icon="open_in_new", + ).props("flat dense").classes("text-xs ui-link-primary") @ui.page("/documents/{document_id}/sources") async def document_sources_page(document_id: str, session_factory: SessionFactoryDep) -> RedirectResponse: @@ -234,6 +257,7 @@ def register_page() -> None: @ui.page("/documents/{document_id}/edit") async def document_edit_page(document_id: str, session_factory: SessionFactoryDep) -> None: document_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/documents") parsed_doc_id = _parse_uuid(document_id) @@ -253,8 +277,8 @@ def register_page() -> None: with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"): page_header("Edit Document Record", subtitle="Document name and document type are required.") - people = sorted(await document_service.list_people(), key=lambda item: item.full_name.casefold()) - role_catalog = await document_service.list_person_roles() + people = sorted(await people_service.list_people(), key=lambda item: item.full_name.casefold()) + role_catalog = await people_service.list_person_roles() type_catalog = await document_service.list_document_types(active_only=False) existing_by_role = _existing_people_by_role(document) form = _render_document_form_fields( @@ -305,10 +329,12 @@ def register_page() -> None: for link in document.document_people if _resolve_link_role_code(link) is not None } - desired_links = _collect_role_link_candidates(form["role_people"], role_codes=[role.code for role in role_catalog]) + desired_links = _collect_role_link_candidates( + form["role_people"], role_codes=[role.code for role in role_catalog] + ) try: for role_code, person_id in sorted(desired_links - set(existing_links.keys())): - await document_service.add_document_person_link( + await people_service.add_document_person_link( document_id=document.id, person_id=person_id, role_code=role_code, @@ -316,7 +342,7 @@ def register_page() -> None: for stale_key in sorted(set(existing_links.keys()) - desired_links): stale_link = existing_links[stale_key] - await document_service.remove_document_person_link(document_person_id=stale_link.id) + await people_service.remove_document_person_link(document_person_id=stale_link.id) except Exception as exc: # noqa: BLE001 show_error(exc, title="Relationship update failed", operation="documents.edit.link_people") return @@ -326,7 +352,9 @@ def register_page() -> None: with ui.row().classes("w-full items-center gap-2 mt-2"): ui.button("Save changes", on_click=submit_edit, icon="save").classes("ui-btn-primary") - ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props("flat") + ui.button( + "Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back" + ).props("flat") @ui.page("/documents/{document_id}/delete") async def document_delete_page(document_id: str, session_factory: SessionFactoryDep) -> None: @@ -354,14 +382,28 @@ def register_page() -> None: ui.label(f"Document: {document.name}").classes("text-sm font-semibold ui-text-primary") if document.sources or document.jobs: - ui.label("Delete is blocked because related records exist.").classes("text-xs ui-text-danger font-bold mt-2") - deps = [cat for cat, present in [("Sources", bool(document.sources)), ("Jobs", bool(document.jobs))] if present] + ui.label("Delete is blocked because related records exist.").classes( + "text-xs ui-text-danger font-bold mt-2" + ) + deps = [ + cat + for cat, present in [("Sources", bool(document.sources)), ("Jobs", bool(document.jobs))] + if present + ] ui.label(f"Dependencies present: {', '.join(deps)}").classes("text-xs ui-text-muted") - ui.label("Remove related records first, then retry deletion.").classes("text-xs ui-text-muted italic") + ui.label("Remove related records first, then retry deletion.").classes( + "text-xs ui-text-muted italic" + ) with ui.row().classes("w-full items-center gap-2 mt-4"): - ui.button("Back to Document", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").classes("ui-btn-primary text-xs") - ui.button("Go to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="work_history").props("flat text-xs") + ui.button( + "Back to Document", + on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), + icon="arrow_back", + ).classes("ui-btn-primary text-xs") + ui.button("Go to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="work_history").props( + "flat text-xs" + ) return ui.label("This action permanently deletes the document.").classes("text-xs ui-text-danger font-medium") @@ -388,8 +430,12 @@ def register_page() -> None: ui.navigate.to("/documents") with ui.row().classes("w-full items-center gap-2 mt-2"): - destructive_button("Delete document permanently", on_click=submit_delete, icon="delete_forever", variant="solid") - ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back").props("flat") + destructive_button( + "Delete document permanently", on_click=submit_delete, icon="delete_forever", variant="solid" + ) + ui.button( + "Cancel", on_click=lambda: ui.navigate.to(f"/documents/{document.id}"), icon="arrow_back" + ).props("flat") # --- Helper Sub-Components --- @@ -405,39 +451,84 @@ def _render_document_form_fields( selected_people_by_role: dict[str, list[UUID]] | None = None, ) -> dict[str, Any]: with archival_card(extra_classes="gap-3"): - name_input = ui.input(label="Document name", value=document.name if document else "").props("outlined").classes("w-full ui-form-surface") + name_input = ( + ui.input(label="Document name", value=document.name if document else "") + .props("outlined") + .classes("w-full ui-form-surface") + ) ordered_types = sorted(type_options.items(), key=lambda item: item[1].casefold()) type_display_to_code = {f"{label} ({code})": code for code, label in ordered_types} type_display_options = list(type_display_to_code.keys()) - selected_type = f"{type_options[document.document_type]} ({document.document_type})" if document and document.document_type in type_options else (type_display_options[0] if type_display_options else "") - type_input = ui.select(type_display_options, label="Document type").props("outlined").classes("w-full ui-form-surface") + selected_type = ( + f"{type_options[document.document_type]} ({document.document_type})" + if document and document.document_type in type_options + else (type_display_options[0] if type_display_options else "") + ) + type_input = ( + ui.select(type_display_options, label="Document type").props("outlined").classes("w-full ui-form-surface") + ) if selected_type: type_input.value = selected_type with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"): - date_input = ui.input( - label="Exact date (YYYY-MM-DD)", - value=document.document_date.isoformat() if document and document.document_date else "", - ).props('outlined type="date"').classes("ui-form-surface") - date_raw_input = ui.input(label="Approximate date", value=document.document_date_raw if document and document.document_date_raw else "").props("outlined").classes("ui-form-surface") + date_input = ( + ui.input( + label="Exact date (YYYY-MM-DD)", + value=document.document_date.isoformat() if document and document.document_date else "", + ) + .props('outlined type="date"') + .classes("ui-form-surface") + ) + date_raw_input = ( + ui.input( + label="Approximate date", + value=document.document_date_raw if document and document.document_date_raw else "", + ) + .props("outlined") + .classes("ui-form-surface") + ) - location_input = ui.input(label="Document location", value=document.location_created if document and document.location_created else "").props("outlined").classes("w-full ui-form-surface") - archive_input = ui.input(label="Archive identifier", value=document.archive_identifier if document and document.archive_identifier else "").props("outlined").classes("w-full ui-form-surface") - notes_input = ui.textarea(label="Notes", value=document.notes if document and document.notes else "").props("outlined autogrow").classes("w-full ui-form-surface") + location_input = ( + ui.input( + label="Document location", + value=document.location_created if document and document.location_created else "", + ) + .props("outlined") + .classes("w-full ui-form-surface") + ) + archive_input = ( + ui.input( + label="Archive identifier", + value=document.archive_identifier if document and document.archive_identifier else "", + ) + .props("outlined") + .classes("w-full ui-form-surface") + ) + notes_input = ( + ui.textarea(label="Notes", value=document.notes if document and document.notes else "") + .props("outlined autogrow") + .classes("w-full ui-form-surface") + ) ui.label("Linked People by Role").classes("text-sm font-semibold ui-text-primary mt-2") - ui.button("Create new person", on_click=lambda: ui.navigate.to("/people/new"), icon="person_add").props("flat dense").classes("self-start") + ui.button("Create new person", on_click=lambda: ui.navigate.to("/people/new"), icon="person_add").props( + "flat dense" + ).classes("self-start") people_options = {str(p.id): p.full_name for p in people} existing = selected_people_by_role or {} role_people_inputs: dict[str, Any] = {} for role_code in role_codes: label = role_labels.get(role_code, role_code.replace("_", " ").title()) current_people = [str(person_id) for person_id in existing.get(role_code, [])] - role_people_inputs[role_code] = ui.select( - people_options, - label=f"{label} people", - multiple=True, - ).props("outlined use-chips").classes("w-full ui-form-surface") + role_people_inputs[role_code] = ( + ui.select( + people_options, + label=f"{label} people", + multiple=True, + ) + .props("outlined use-chips") + .classes("w-full ui-form-surface") + ) if current_people: role_people_inputs[role_code].value = current_people @@ -459,8 +550,14 @@ def _render_bento_viewer_zone(document: Document) -> None: source_path = document.sources[0].file_path if document.sources else None dark_room_viewer(source_path, count_label=f"{len(document.sources)} Source(s) Linked") with ui.row().classes("w-full justify-between items-center mt-2"): - ui.button("View All Sources", on_click=lambda: ui.navigate.to(f"/sources?document_id={document.id}"), icon="description").props("flat dense text-xs").classes("ui-link-primary") - ui.button("+ Add Source", on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), icon="add").classes("ui-btn-primary text-xs") + ui.button( + "View All Sources", + on_click=lambda: ui.navigate.to(f"/sources?document_id={document.id}"), + icon="description", + ).props("flat dense text-xs").classes("ui-link-primary") + ui.button( + "+ Add Source", on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), icon="add" + ).classes("ui-btn-primary text-xs") def _render_bento_metadata_zone(document: Document) -> None: @@ -503,8 +600,12 @@ def _render_bento_relations_zone(document: Document) -> None: ui.label(f"{len(document.jobs)} Active Jobs").classes("text-xs ui-link-primary font-bold") with ui.row().classes("w-full gap-2 mt-2"): - ui.button("View Jobs", on_click=lambda: ui.navigate.to(f"/documents/{document.id}/jobs"), icon="work_history").props("flat dense text-xs").classes("ui-link-primary") - ui.button("+ Add Job", on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), icon="add").classes("ui-btn-primary text-xs") + ui.button( + "View Jobs", on_click=lambda: ui.navigate.to(f"/documents/{document.id}/jobs"), icon="work_history" + ).props("flat dense text-xs").classes("ui-link-primary") + ui.button( + "+ Add Job", on_click=lambda: ui.navigate.to(f"/jobs/new?document_id={document.id}"), icon="add" + ).classes("ui-btn-primary text-xs") def _parse_uuid(value: str | None) -> UUID | None: @@ -527,7 +628,7 @@ def _parse_iso_date(value: str | None) -> date | None: def _resolve_selected_document_type_code(selected_value: Any, type_options: dict[str, str]) -> str | None: - candidate = (str(selected_value).strip() if selected_value is not None else "") + candidate = str(selected_value).strip() if selected_value is not None else "" if not candidate: return None return type_options.get(candidate) @@ -553,7 +654,9 @@ def _existing_people_by_role(document: Document) -> dict[str, list[UUID]]: return people_by_role -def _collect_role_link_candidates(role_people_inputs: dict[str, Any], *, role_codes: list[str]) -> set[tuple[str, UUID]]: +def _collect_role_link_candidates( + role_people_inputs: dict[str, Any], *, role_codes: list[str] +) -> set[tuple[str, UUID]]: desired: set[tuple[str, UUID]] = set() for role_code in role_codes: selected = role_people_inputs[role_code].value or [] @@ -577,4 +680,4 @@ def _group_people_labels_by_role(document: Document) -> dict[str, list[str]]: continue person_label = link.person.full_name if link.person is not None else "Unknown person" grouped.setdefault(role_code, []).append(person_label) - return grouped \ No newline at end of file + return grouped diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py index 163d5f3..c8c4148 100644 --- a/src/transcription/ui/pages/jobs_page.py +++ b/src/transcription/ui/pages/jobs_page.py @@ -74,7 +74,9 @@ def register_page() -> None: # noqa: PLR0915 render_navigation_header(current_path="/jobs") with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"): - page_header("Create Processing Job", subtitle="Queue source files for AI transcription and entity processing.") + page_header( + "Create Processing Job", subtitle="Queue source files for AI transcription and entity processing." + ) documents = await documents_service.list_documents() if not documents: @@ -85,8 +87,12 @@ def register_page() -> None: # noqa: PLR0915 with archival_card(extra_classes="gap-3"): document_options = {str(doc.id): doc.name for doc in documents} - document_select = ui.select(document_options, label="Target Document").props("outlined").classes("w-full ui-form-surface") - + document_select = ( + ui.select(document_options, label="Target Document") + .props("outlined") + .classes("w-full ui-form-surface") + ) + requested_document_id = request.query_params.get("document_id") if requested_document_id in document_options: document_select.value = requested_document_id @@ -115,7 +121,7 @@ def register_page() -> None: # noqa: PLR0915 async with session_scope(session_factory=session_factory) as session: result = await create_job_for_document( document_id=document_id, - uploads=uploaded_files, + source_files=uploaded_files, provider=(provider_input.value or None), model=(model_input.value or None), session=session, @@ -129,7 +135,9 @@ def register_page() -> None: # noqa: PLR0915 ui.navigate.to(f"/jobs/{result.job_id}") with ui.row().classes("w-full items-center gap-2 mt-2"): - ui.button("Submit for transcription", on_click=submit_create, icon="play_arrow").classes("ui-btn-primary") + ui.button("Submit for transcription", on_click=submit_create, icon="play_arrow").classes( + "ui-btn-primary" + ) ui.button("Back to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="arrow_back").props("flat") @ui.page("/jobs/{job_id}") @@ -201,7 +209,9 @@ def register_page() -> None: # noqa: PLR0915 with ui.row().classes("w-full items-center gap-2 mt-2"): destructive_button("Cancel job", on_click=submit_cancel, icon="stop_circle", variant="solid") - ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props("flat") + ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props( + "flat" + ) @ui.page("/jobs/{job_id}/resubmit") async def job_resubmit_page(job_id: str, request: Request, session_factory: SessionFactoryDep) -> None: @@ -252,7 +262,9 @@ def register_page() -> None: # noqa: PLR0915 with ui.row().classes("w-full items-center gap-2 mt-2"): ui.button("Resubmit now", on_click=submit_resubmit, icon="replay").classes("ui-btn-primary") - ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props("flat") + ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props( + "flat" + ) @ui.page("/jobs/{job_id}/delete") async def job_delete_page(job_id: str, session_factory: SessionFactoryDep) -> None: @@ -277,12 +289,16 @@ def register_page() -> None: # noqa: PLR0915 ui.label(f"Job ID: {job.id}").classes("text-sm font-semibold font-mono ui-text-primary") if job.status == JobStatus.PROCESSING: - ui.label("Delete is blocked while the job is processing.").classes("text-xs ui-text-danger font-bold mt-2") - ui.label("Wait for processing to complete, then retry delete.").classes("text-xs ui-text-muted italic") + ui.label("Delete is blocked while the job is processing.").classes( + "text-xs ui-text-danger font-bold mt-2" + ) + ui.label("Wait for processing to complete, then retry delete.").classes( + "text-xs ui-text-muted italic" + ) with ui.row().classes("w-full items-center gap-2 mt-4"): - ui.button("Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").classes( - "ui-btn-primary text-xs" - ) + ui.button( + "Back to Job", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back" + ).classes("ui-btn-primary text-xs") ui.button("Back to Jobs", on_click=lambda: ui.navigate.to("/jobs"), icon="work_history").props( "flat text-xs" ) @@ -290,7 +306,9 @@ def register_page() -> None: # noqa: PLR0915 ui.label("This action permanently deletes the job.").classes("text-xs ui-text-danger font-medium") if job.job_sources: - ui.label("Related JobSource links will be removed as part of delete.").classes("text-xs ui-text-muted") + ui.label("Related JobSource links will be removed as part of delete.").classes( + "text-xs ui-text-muted" + ) async def submit_delete() -> None: try: @@ -310,7 +328,9 @@ def register_page() -> None: # noqa: PLR0915 ui.navigate.to("/jobs") with ui.row().classes("w-full items-center gap-2 mt-2"): - destructive_button("Delete job permanently", on_click=submit_delete, icon="delete_forever", variant="solid") + destructive_button( + "Delete job permanently", on_click=submit_delete, icon="delete_forever", variant="solid" + ) ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/jobs/{job.id}"), icon="arrow_back").props("flat") @@ -450,4 +470,4 @@ def _parse_uuid(value: str | None) -> UUID | None: def _latest_prompt_name(job: Job) -> str | None: - return job.prompt_name \ No newline at end of file + return job.prompt_name diff --git a/src/transcription/ui/pages/people_page.py b/src/transcription/ui/pages/people_page.py index b5c7af2..d12f53c 100644 --- a/src/transcription/ui/pages/people_page.py +++ b/src/transcription/ui/pages/people_page.py @@ -14,8 +14,8 @@ from nicegui import ui from transcription.config import Settings, get_settings from transcription.db.models import Person from transcription.errors import ErrorCategory -from transcription.services.documents import DocumentError, DocumentService -from transcription.services.store import UploadError, store_person_portrait +from transcription.services.people import PeopleError, PeopleService +from transcription.services.people import PersonMediaError, store_person_portrait from transcription.ui.components.app_shell import render_navigation_header from transcription.ui.components.cards import archival_card from transcription.ui.components.data_display import metadata_row @@ -37,7 +37,7 @@ def register_page() -> None: # noqa: PLR0915 @ui.page("/people") async def people_page(session_factory: SessionFactoryDep) -> None: - people_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/people") with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"): @@ -73,7 +73,7 @@ def register_page() -> None: # noqa: PLR0915 @ui.page("/people/new") async def person_create_page(request: Request, session_factory: SessionFactoryDep) -> None: - people_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/people") draft_person_id = uuid4() @@ -124,7 +124,7 @@ def register_page() -> None: # noqa: PLR0915 @ui.page("/people/{person_id}") async def person_detail_page(person_id: str, request: Request, session_factory: SessionFactoryDep) -> None: - people_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/people") parsed_person_id = _parse_uuid(person_id) @@ -134,7 +134,7 @@ def register_page() -> None: # noqa: PLR0915 try: person = await people_service.read_person_detail(parsed_person_id) - except DocumentError: + except PeopleError: ui.label("Person not found").classes("text-h6 ui-text-danger p-4") return except Exception as exc: # noqa: BLE001 @@ -169,7 +169,7 @@ def register_page() -> None: # noqa: PLR0915 @ui.page("/people/{person_id}/edit") async def person_edit_page(person_id: str, request: Request, session_factory: SessionFactoryDep) -> None: - people_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/people") parsed_person_id = _parse_uuid(person_id) @@ -179,7 +179,7 @@ def register_page() -> None: # noqa: PLR0915 try: person = await people_service.read_person_detail(parsed_person_id) - except DocumentError: + except PeopleError: ui.label("Person not found").classes("text-h6 ui-text-danger p-4") return except Exception as exc: # noqa: BLE001 @@ -233,11 +233,13 @@ def register_page() -> None: # noqa: PLR0915 with ui.row().classes("w-full items-center gap-2 mt-2"): ui.button("Save changes", on_click=submit_edit, icon="save").classes("ui-btn-primary") - ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props("flat") + ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props( + "flat" + ) @ui.page("/people/{person_id}/delete") async def person_delete_page(person_id: str, session_factory: SessionFactoryDep) -> None: - people_service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) render_navigation_header(current_path="/people") parsed_person_id = _parse_uuid(person_id) @@ -247,7 +249,7 @@ def register_page() -> None: # noqa: PLR0915 try: person = await people_service.read_person_detail(parsed_person_id) - except DocumentError: + except PeopleError: ui.label("Person not found").classes("text-h6 ui-text-danger p-4") return except Exception as exc: # noqa: BLE001 @@ -265,12 +267,14 @@ def register_page() -> None: # noqa: PLR0915 f"This will also remove {len(person.document_people)} linked document relationship(s)." ).classes("text-xs ui-text-danger font-bold mt-2") - ui.label("This action permanently deletes the person record.").classes("text-xs ui-text-danger font-medium") + ui.label("This action permanently deletes the person record.").classes( + "text-xs ui-text-danger font-medium" + ) async def submit_delete() -> None: try: await people_service.delete_person(person) - except DocumentError as exc: + except PeopleError as exc: if exc.category == ErrorCategory.NOT_FOUND: ui.notify("Person not found.", type="warning") ui.navigate.to("/people") @@ -285,8 +289,12 @@ def register_page() -> None: # noqa: PLR0915 ui.navigate.to("/people") with ui.row().classes("w-full items-center gap-2 mt-2"): - destructive_button("Delete person permanently", on_click=submit_delete, icon="delete_forever", variant="solid") - ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props("flat") + destructive_button( + "Delete person permanently", on_click=submit_delete, icon="delete_forever", variant="solid" + ) + ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/people/{person.id}"), icon="arrow_back").props( + "flat" + ) # --- Helper Sub-Components & Form Builders --- @@ -300,29 +308,79 @@ def _render_person_form_fields( ) -> dict[str, Any]: with archival_card(extra_classes="gap-3"): with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-3"): - full_name_input = ui.input(label="Full name", value=person.full_name if person else "").props("outlined").classes("ui-form-surface") - display_name_input = ui.input(label="Display name", value=person.display_name if person and person.display_name else "").props("outlined").classes("ui-form-surface") - maiden_name_input = ui.input(label="Maiden name", value=person.maiden_name if person and person.maiden_name else "").props("outlined").classes("ui-form-surface") + full_name_input = ( + ui.input(label="Full name", value=person.full_name if person else "") + .props("outlined") + .classes("ui-form-surface") + ) + display_name_input = ( + ui.input(label="Display name", value=person.display_name if person and person.display_name else "") + .props("outlined") + .classes("ui-form-surface") + ) + maiden_name_input = ( + ui.input(label="Maiden name", value=person.maiden_name if person and person.maiden_name else "") + .props("outlined") + .classes("ui-form-surface") + ) with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-3"): - birth_date_input = ui.input( - label="Birth date (YYYY-MM-DD)", - value=person.birth_date.isoformat() if person and person.birth_date else "", - ).props('outlined type="date"').classes("ui-form-surface") - birth_date_raw_input = ui.input(label="Birth date (approximate)", value=person.birth_date_raw if person and person.birth_date_raw else "").props("outlined").classes("ui-form-surface") - birth_place_input = ui.input(label="Birth place", value=person.birth_place if person and person.birth_place else "").props("outlined").classes("ui-form-surface") + birth_date_input = ( + ui.input( + label="Birth date (YYYY-MM-DD)", + value=person.birth_date.isoformat() if person and person.birth_date else "", + ) + .props('outlined type="date"') + .classes("ui-form-surface") + ) + birth_date_raw_input = ( + ui.input( + label="Birth date (approximate)", + value=person.birth_date_raw if person and person.birth_date_raw else "", + ) + .props("outlined") + .classes("ui-form-surface") + ) + birth_place_input = ( + ui.input(label="Birth place", value=person.birth_place if person and person.birth_place else "") + .props("outlined") + .classes("ui-form-surface") + ) with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-3"): - death_date_input = ui.input( - label="Death date (YYYY-MM-DD)", - value=person.death_date.isoformat() if person and person.death_date else "", - ).props('outlined type="date"').classes("ui-form-surface") - death_date_raw_input = ui.input(label="Death date (approximate)", value=person.death_date_raw if person and person.death_date_raw else "").props("outlined").classes("ui-form-surface") - death_place_input = ui.input(label="Death place", value=person.death_place if person and person.death_place else "").props("outlined").classes("ui-form-surface") + death_date_input = ( + ui.input( + label="Death date (YYYY-MM-DD)", + value=person.death_date.isoformat() if person and person.death_date else "", + ) + .props('outlined type="date"') + .classes("ui-form-surface") + ) + death_date_raw_input = ( + ui.input( + label="Death date (approximate)", + value=person.death_date_raw if person and person.death_date_raw else "", + ) + .props("outlined") + .classes("ui-form-surface") + ) + death_place_input = ( + ui.input(label="Death place", value=person.death_place if person and person.death_place else "") + .props("outlined") + .classes("ui-form-surface") + ) + + biography_input = ( + ui.textarea(label="Biography", value=person.biography if person and person.biography else "") + .props("outlined autogrow") + .classes("w-full ui-form-surface") + ) + portrait_path_input = ( + ui.input(label="Portrait path", value=person.portrait_path if person and person.portrait_path else "") + .props("outlined") + .classes("w-full ui-form-surface") + ) - biography_input = ui.textarea(label="Biography", value=person.biography if person and person.biography else "").props("outlined autogrow").classes("w-full ui-form-surface") - portrait_path_input = ui.input(label="Portrait path", value=person.portrait_path if person and person.portrait_path else "").props("outlined").classes("w-full ui-form-surface") - _bind_portrait_file_picker( portrait_path_input, settings=_resolve_runtime_settings(request), @@ -408,7 +466,7 @@ def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Setti file_bytes=payload, settings=settings, ) - except UploadError as exc: + except PersonMediaError as exc: ui.notify(str(exc), type="negative") return except Exception: # noqa: BLE001 @@ -512,4 +570,4 @@ def _parse_iso_date(value: str | None) -> date | None: try: return date.fromisoformat(candidate) except ValueError: - return None \ No newline at end of file + return None diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index 673c376..d84312d 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/src/transcription/ui/pages/sources_page.py @@ -11,10 +11,10 @@ from nicegui import ui from transcription.config import Settings, get_settings from transcription.db.models import JobSource, Source -from transcription.services.transcription import ( +from transcription.services.sources import ( SourceDeleteBlockedError, + SourceService, TranscriptionNotFoundError, - TranscriptionService, ) from transcription.ui.components.app_shell import render_navigation_header from transcription.ui.components.cards import archival_card @@ -37,7 +37,7 @@ def register_page() -> None: document_id: str | None = None, job_id: str | None = None, ) -> None: - sources_service = TranscriptionService(session_factory=session_factory) + sources_service = SourceService(session_factory=session_factory) parsed_doc_id = _parse_uuid(document_id) parsed_job_id = _parse_uuid(job_id) @@ -93,11 +93,13 @@ def register_page() -> None: render_sources_table(rows) if parsed_doc_id is None and parsed_job_id is None: - ui.label("Open a source row to inspect AI output and add human revisions.").classes("text-xs ui-text-muted") + ui.label("Open a source row to inspect AI output and add human revisions.").classes( + "text-xs ui-text-muted" + ) @ui.page("/sources/{source_id}") async def source_detail_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None: - sources_service = TranscriptionService(session_factory=session_factory) + sources_service = SourceService(session_factory=session_factory) parsed_source_id = _parse_uuid(source_id) render_navigation_header(current_path="/sources") @@ -153,7 +155,7 @@ def register_page() -> None: @ui.page("/sources/{source_id}/delete") async def source_delete_page(source_id: str, session_factory: SessionFactoryDep) -> None: - sources_service = TranscriptionService(session_factory=session_factory) + sources_service = SourceService(session_factory=session_factory) parsed_source_id = _parse_uuid(source_id) render_navigation_header(current_path="/sources") @@ -178,8 +180,12 @@ def register_page() -> None: ui.label(f"Source: {source.upload_name}").classes("text-sm font-semibold ui-text-primary") if source.job_sources: - ui.label("Delete is only available for unlinked sources.").classes("text-xs ui-text-danger font-bold mt-2") - ui.label("Open the related job record and remove job links first.").classes("text-xs ui-text-muted italic") + ui.label("Delete is only available for unlinked sources.").classes( + "text-xs ui-text-danger font-bold mt-2" + ) + ui.label("Open the related job record and remove job links first.").classes( + "text-xs ui-text-muted italic" + ) with ui.row().classes("w-full items-center gap-2 mt-4"): ui.button( "Back to Source", @@ -193,7 +199,9 @@ def register_page() -> None: ).props("flat text-xs") return - ui.label("This action permanently deletes the source record.").classes("text-xs ui-text-danger font-medium") + ui.label("This action permanently deletes the source record.").classes( + "text-xs ui-text-danger font-medium" + ) async def submit_delete() -> None: try: @@ -213,8 +221,12 @@ def register_page() -> None: ui.navigate.to("/sources") with ui.row().classes("w-full items-center gap-2 mt-2"): - destructive_button("Delete source permanently", on_click=submit_delete, icon="delete_forever", variant="solid") - ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/sources/{source.id}"), icon="arrow_back").props("flat") + destructive_button( + "Delete source permanently", on_click=submit_delete, icon="delete_forever", variant="solid" + ) + ui.button("Cancel", on_click=lambda: ui.navigate.to(f"/sources/{source.id}"), icon="arrow_back").props( + "flat" + ) def _render_source_viewer_zone(source: Source, *, settings: Settings, request: Request) -> None: @@ -230,7 +242,7 @@ def _render_source_transcription_column( source: Source, original_transcription: str | None, latest_job_source: JobSource | None, - sources_service: TranscriptionService, + sources_service: SourceService, ) -> None: with ui.column().classes("col-span-12 lg:col-span-4 gap-4"): _render_source_transcription_zone( @@ -281,7 +293,9 @@ def _render_source_job_metadata_zone(latest_job_source: JobSource | None) -> Non ) metadata_row( "Prompt:", - latest_job_source.job.prompt_name if latest_job_source.job and latest_job_source.job.prompt_name else "unknown", + latest_job_source.job.prompt_name + if latest_job_source.job and latest_job_source.job.prompt_name + else "unknown", ) if latest_job_source.error_detail: @@ -305,7 +319,7 @@ def _render_source_transcription_zone( source: Source, original_transcription: str | None, latest_job_source: JobSource | None, - sources_service: TranscriptionService, + sources_service: SourceService, ) -> None: with archival_card(title="Transcription Text"): if original_transcription: @@ -315,10 +329,14 @@ def _render_source_transcription_zone( with archival_card(title="Editable Revision"): seed_revision = source.revised_text if source.revised_text is not None else (original_transcription or "") - revision_input = ui.textarea( - label="Revised transcription", - value=seed_revision, - ).props("outlined autogrow").classes("w-full ui-form-surface") + revision_input = ( + ui.textarea( + label="Revised transcription", + value=seed_revision, + ) + .props("outlined autogrow") + .classes("w-full ui-form-surface") + ) save_state = ui.label( f"Last saved: {source.date_revised.isoformat()}" @@ -348,13 +366,19 @@ def _render_source_transcription_zone( source.revised_text = updated.revised_text source.date_revised = updated.date_revised save_state.text = ( - f"Last saved: {updated.date_revised.isoformat()}" if updated.date_revised is not None else "Revision saved." + f"Last saved: {updated.date_revised.isoformat()}" + if updated.date_revised is not None + else "Revision saved." ) ui.notify("Revision saved", type="positive") with ui.row().classes("w-full items-center gap-2 mt-2"): ui.button("Save revision", on_click=submit_revision, icon="save").classes("ui-btn-primary") - ui.button("Reset", on_click=lambda: _reset_revision_text(revision_input, source, original_transcription), icon="refresh").props("flat") + ui.button( + "Reset", + on_click=lambda: _reset_revision_text(revision_input, source, original_transcription), + icon="refresh", + ).props("flat") if latest_job_source is not None and latest_job_source.status.value == "failed": ui.label("Source has a failed job execution. Save a human revision to preserve corrected text.").classes( diff --git a/src/transcription/worker.py b/src/transcription/worker.py index a28ac5d..c6f8586 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -21,7 +21,8 @@ from transcription.errors import classify_unexpected_error from .services import ServiceBundle from .services.documents import DocumentService from .services.jobs import JobService -from .services.transcription import TranscriptionService +from .services.people import PeopleService +from .services.sources import SourceService from .services.workflows import advance_job from .services.workflows import process_next_queued_job as process_next_queued_job_workflow @@ -98,7 +99,7 @@ async def worker_consumer_lifespan( async def queue_consumer_loop(queue: asyncio.Queue[UUID], stop_event: asyncio.Event): """Main worker loop that consumes jobs from the queue and processes them. - The queue is for Job UUIDs, and the corresponding documents should already have been uploaded. + The queue contains Job UUIDs whose Document and Source records already exist. """ service = JobService() while not stop_event.is_set(): @@ -180,8 +181,9 @@ async def process_next_queued_job( else: services = ServiceBundle( documents=DocumentService(session_factory=session_factory), + sources=SourceService(session_factory=session_factory), jobs=JobService(session_factory=session_factory), - transcriptions=TranscriptionService(session_factory=session_factory), + people=PeopleService(session_factory=session_factory), ) if session is None: diff --git a/tests/api/test_v4_documents.py b/tests/api/test_v4_documents.py index 3e53073..e9c68e4 100644 --- a/tests/api/test_v4_documents.py +++ b/tests/api/test_v4_documents.py @@ -13,20 +13,24 @@ from fastapi.testclient import TestClient from transcription.api.errors import register_error_handlers from transcription.api.v4_documents import get_document_service +from transcription.api.v4_documents import get_people_service from transcription.api.v4_documents import router from transcription.config import Settings from transcription.config import SqliteSettings from transcription.db import create_all from transcription.db.engine import get_database_url from transcription.db.engine import get_engine -from transcription.db.session import dispose_session_factory -from transcription.db.session import session_scope from transcription.db.models import Document from transcription.db.models import Person +from transcription.db.session import dispose_session_factory +from transcription.db.session import session_scope from transcription.services.documents import DocumentService +from transcription.services.people import PeopleService -def _seed_document_and_person(*, db_url: str, document_name: str = "API Doc", person_name: str = "API Person") -> tuple[UUID, UUID]: +def _seed_document_and_person( + *, db_url: str, document_name: str = "API Doc", person_name: str = "API Person" +) -> tuple[UUID, UUID]: async def _seed() -> tuple[UUID, UUID]: async with session_scope(database_url=db_url) as session: document = Document(name=document_name) @@ -42,7 +46,7 @@ def _seed_document_and_person(*, db_url: str, document_name: str = "API Doc", pe @contextmanager -def _v4_api_client(tmp_path: Path, *, db_filename: str) -> Generator[tuple[TestClient, str], None, None]: +def _v4_api_client(tmp_path: Path, *, db_filename: str) -> Generator[tuple[TestClient, str]]: settings = Settings( openrouter_api_key="test-key", database=SqliteSettings(path=str(tmp_path / db_filename)), @@ -64,10 +68,12 @@ def _v4_api_client(tmp_path: Path, *, db_filename: str) -> Generator[tuple[TestC asyncio.run(_bootstrap()) service = DocumentService(session_factory=session_factory) + people_service = PeopleService(session_factory=session_factory) app = FastAPI() register_error_handlers(app) app.include_router(router) app.dependency_overrides[get_document_service] = lambda: service + app.dependency_overrides[get_people_service] = lambda: people_service try: with TestClient(app) as client: diff --git a/tests/integration/test_pipeline_flow.py b/tests/integration/test_pipeline_flow.py index 44d30f3..1e611b2 100644 --- a/tests/integration/test_pipeline_flow.py +++ b/tests/integration/test_pipeline_flow.py @@ -7,13 +7,12 @@ import pytest from transcription.config import Settings from transcription.db.models import Document -from transcription.db.models import Job from transcription.db.models import JobSourceStatus from transcription.db.models import JobStatus from transcription.providers.base import TranscriptionResult from transcription.services import ServiceBundle +from transcription.services.store import create_document_job from transcription.services.store import create_job_for_document -from transcription.services.store import create_upload_job from transcription.services.workflows import advance_job @@ -31,8 +30,8 @@ def _build_services(default_session_factory) -> ServiceBundle: ) object.__setattr__( services, - "transcriptions", - services.transcriptions.__class__(session_factory=default_session_factory), + "sources", + services.sources.__class__(session_factory=default_session_factory), ) return services @@ -53,7 +52,7 @@ class TestPipelineSuccessFlow: transcription_temperature=0.2, transcription_top_p=0.85, ) - upload_result = await create_upload_job( + upload_result = await create_document_job( filename="pipeline.jpg", file_bytes=b"pipeline-bytes", session=async_session, @@ -109,9 +108,13 @@ class TestPipelineSuccessFlow: assert job.user_prompt is not None assert job.temperature == 0.2 assert job.top_p == 0.85 - assert any(job_source.ai_metadata == {"finish_reason": "stop", "usage": {"total_tokens": 42}} for job_source in job.job_sources) assert any( - job_source.raw_api_response == {"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]} + job_source.ai_metadata == {"finish_reason": "stop", "usage": {"total_tokens": 42}} + for job_source in job.job_sources + ) + assert any( + job_source.raw_api_response + == {"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]} for job_source in job.job_sources ) assert all(job_source.error_detail is None for job_source in job.job_sources) @@ -132,7 +135,7 @@ class TestPipelineSuccessFlow: create_result = await create_job_for_document( document_id=document.id, - uploads=[ + source_files=[ ("page-01.jpg", b"one"), ("page-02.jpg", b"two"), ("page-03.jpg", b"three"), @@ -176,7 +179,9 @@ class TestPipelineSuccessFlow: assert len(job.job_sources) == 3 assert all(job_source.status == JobSourceStatus.TRANSCRIBED for job_source in job.job_sources) assert all(job_source.raw_transcription for job_source in job.job_sources) - assert all(job_source.source is not None and job_source.source.raw_transcription for job_source in job.job_sources) + assert all( + job_source.source is not None and job_source.source.raw_transcription for job_source in job.job_sources + ) assert job.prompt_name == "transcribe_document.md" @pytest.mark.asyncio @@ -195,7 +200,7 @@ class TestPipelineSuccessFlow: create_result = await create_job_for_document( document_id=document.id, - uploads=[ + source_files=[ ("page-01.jpg", b"one"), ("page-02.jpg", b"two"), ], @@ -261,7 +266,7 @@ class TestPipelineSuccessFlow: create_result = await create_job_for_document( document_id=document.id, - uploads=[ + source_files=[ ("page-01.jpg", b"one"), ("page-02.jpg", b"two"), ], @@ -278,8 +283,8 @@ class TestPipelineSuccessFlow: page_one.raw_transcription = "existing transcript" page_two.status = JobSourceStatus.PENDING page_two.raw_transcription = None - await services.transcriptions.update_job_source(job_source=page_one, session=async_session) - await services.transcriptions.update_job_source(job_source=page_two, session=async_session) + await services.sources.update_job_source(job_source=page_one, session=async_session) + await services.sources.update_job_source(job_source=page_two, session=async_session) await services.jobs.update_job_state(job_id=job.id, status=JobStatus.QUEUED, session=async_session) await async_session.commit() @@ -334,7 +339,7 @@ class TestPipelineFailureFlow: ): """Upload followed by worker processing persists error detail and failed status on the job.""" settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) - upload_result = await create_upload_job( + upload_result = await create_document_job( filename="pipeline.jpg", file_bytes=b"pipeline-bytes", session=async_session, @@ -371,7 +376,9 @@ class TestPipelineFailureFlow: assert job.status == JobStatus.FAILED assert all(job_source.raw_transcription is None for job_source in job.job_sources) assert any(job_source.error_detail is not None for job_source in job.job_sources) - error_detail = next(job_source.error_detail for job_source in job.job_sources if job_source.error_detail is not None) + error_detail = next( + job_source.error_detail for job_source in job.job_sources if job_source.error_detail is not None + ) assert "pipeline provider failure" in error_detail assert "[internal_unexpected_error]" in error_detail assert "error_id=" in error_detail diff --git a/tests/services/test_document_service.py b/tests/services/test_document_service.py index 3e3c83e..c70391f 100644 --- a/tests/services/test_document_service.py +++ b/tests/services/test_document_service.py @@ -2,23 +2,24 @@ from __future__ import annotations from datetime import UTC from datetime import datetime -from pathlib import Path from uuid import uuid4 import pytest from sqlmodel import select from transcription.db.models import Document -from transcription.db.models import Job from transcription.db.models import DocumentPerson from transcription.db.models import DocumentPersonRole from transcription.db.models import DocumentType +from transcription.db.models import Job from transcription.db.models import Person from transcription.db.models import PersonRole from transcription.db.models import Source from transcription.services.documents import DocumentDeleteBlockedError from transcription.services.documents import DocumentError from transcription.services.documents import DocumentService +from transcription.services.people import PeopleError +from transcription.services.people import PeopleService @pytest.mark.asyncio @@ -123,6 +124,7 @@ async def test_delete_document_succeeds_when_unlinked(default_session_factory, t @pytest.mark.asyncio async def test_delete_document_removes_person_links(default_session_factory, tmp_path): service = DocumentService(session_factory=default_session_factory) + people_service = PeopleService(session_factory=default_session_factory) service.settings.upload_dir = tmp_path document = await service.create_document( @@ -132,8 +134,8 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp document_type="memo", ) ) - person = await service.create_person(Person(full_name="Linked Person")) - await service.create_document_person( + person = await people_service.create_person(Person(full_name="Linked Person")) + await people_service.create_document_person( DocumentPerson( document_id=document.id, person_id=person.id, @@ -141,7 +143,7 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp ) ) - links_before_delete = await service.list_document_people(document_id=document.id) + links_before_delete = await people_service.list_document_people(document_id=document.id) assert len(links_before_delete) == 1 assert links_before_delete[0].role_id is not None assert links_before_delete[0].role_ref is not None @@ -153,7 +155,7 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp await service.delete_document(document) assert not document_dir.exists() - assert await service.list_document_people(document_id=document.id) == [] + assert await people_service.list_document_people(document_id=document.id) == [] with pytest.raises(DocumentError): await service.read_document_detail(document.id) @@ -189,6 +191,7 @@ async def test_delete_document_removes_populated_storage_tree(default_session_fa @pytest.mark.asyncio async def test_read_person_detail_loads_document_links(default_session_factory): service = DocumentService(session_factory=default_session_factory) + people_service = PeopleService(session_factory=default_session_factory) document = await service.create_document( Document( @@ -197,8 +200,8 @@ async def test_read_person_detail_loads_document_links(default_session_factory): document_type="letter", ) ) - person = await service.create_person(Person(full_name="Linked Person")) - await service.create_document_person( + person = await people_service.create_person(Person(full_name="Linked Person")) + await people_service.create_document_person( DocumentPerson( document_id=document.id, person_id=person.id, @@ -206,7 +209,7 @@ async def test_read_person_detail_loads_document_links(default_session_factory): ) ) - detail = await service.read_person_detail(person.id) + detail = await people_service.read_person_detail(person.id) assert detail.id == person.id assert len(detail.document_people) == 1 @@ -216,7 +219,7 @@ async def test_read_person_detail_loads_document_links(default_session_factory): @pytest.mark.asyncio async def test_update_person_refreshes_updated_timestamp(default_session_factory): - service = DocumentService(session_factory=default_session_factory) + service = PeopleService(session_factory=default_session_factory) created = await service.create_person( Person( @@ -235,9 +238,10 @@ async def test_update_person_refreshes_updated_timestamp(default_session_factory @pytest.mark.asyncio async def test_delete_person_removes_links_when_linked_documents_exist(default_session_factory): - service = DocumentService(session_factory=default_session_factory) + documents_service = DocumentService(session_factory=default_session_factory) + service = PeopleService(session_factory=default_session_factory) - document = await service.create_document( + document = await documents_service.create_document( Document( id=uuid4(), name="block-person-delete-doc", @@ -258,19 +262,19 @@ async def test_delete_person_removes_links_when_linked_documents_exist(default_s links = await service.list_document_people(person_id=person.id) assert links == [] - with pytest.raises(DocumentError): + with pytest.raises(PeopleError): await service.read_person_detail(person.id) @pytest.mark.asyncio async def test_delete_person_succeeds_when_unlinked(default_session_factory): - service = DocumentService(session_factory=default_session_factory) + service = PeopleService(session_factory=default_session_factory) person = await service.create_person(Person(full_name="Free Person")) await service.delete_person(person) - with pytest.raises(DocumentError): + with pytest.raises(PeopleError): await service.read_person_detail(person.id) @@ -292,9 +296,12 @@ async def test_create_document_reuses_existing_document_type_registry(default_se @pytest.mark.asyncio async def test_update_document_person_sets_role_id_from_legacy_role(default_session_factory): - service = DocumentService(session_factory=default_session_factory) + documents_service = DocumentService(session_factory=default_session_factory) + service = PeopleService(session_factory=default_session_factory) - document = await service.create_document(Document(id=uuid4(), name="role-sync-doc", document_type="letter")) + document = await documents_service.create_document( + Document(id=uuid4(), name="role-sync-doc", document_type="letter") + ) person = await service.create_person(Person(full_name="Role Sync Person")) link = await service.create_document_person( DocumentPerson( diff --git a/tests/services/test_store.py b/tests/services/test_store.py index c05c597..45721ae 100644 --- a/tests/services/test_store.py +++ b/tests/services/test_store.py @@ -9,31 +9,33 @@ from transcription.db.models import Document from transcription.db.models import Job from transcription.db.models import JobSource from transcription.db.models import Source -from transcription.services.store import UploadError -from transcription.services.store import create_upload_job +from transcription.services.people import store_person_portrait +from transcription.services.sources import source_mime_type +from transcription.services.store import SourceStorageError +from transcription.services.store import create_document_job from transcription.services.store import create_job_for_document -from transcription.services.store import store_person_portrait +from transcription.services.store import store_source_file @pytest.mark.asyncio -async def test_create_job_for_document_requires_at_least_one_upload(async_session, tmp_path): +async def test_create_job_for_document_requires_at_least_one_source(async_session, tmp_path): document = Document(id=uuid4(), name="needs-upload") async_session.add(document) await async_session.commit() settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) - with pytest.raises(UploadError): + with pytest.raises(SourceStorageError): await create_job_for_document( document_id=document.id, - uploads=[], + source_files=[], session=async_session, settings=settings, ) @pytest.mark.asyncio -async def test_create_job_for_document_sorts_uploads_and_creates_links(async_session, tmp_path): +async def test_create_job_for_document_sorts_sources_and_creates_links(async_session, tmp_path): document = Document(id=uuid4(), name="ordered-upload-doc") async_session.add(document) await async_session.commit() @@ -42,7 +44,7 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses result = await create_job_for_document( document_id=document.id, - uploads=[ + source_files=[ ("folder/b_page.pdf", b"b"), ("folder/A_page.pdf", b"a"), ], @@ -61,9 +63,7 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses sources = ( await async_session.exec( - select(Source) - .where(Source.document_id == document.id) - .order_by(Source.page_number) # pyright: ignore[reportArgumentType] + select(Source).where(Source.document_id == document.id).order_by(Source.page_number) # pyright: ignore[reportArgumentType] ) ).all() assert [source.upload_name for source in sources] == ["A_page.pdf", "b_page.pdf"] @@ -83,10 +83,10 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses @pytest.mark.asyncio -async def test_create_upload_job_stores_source_under_document_id_directory(async_session, tmp_path): +async def test_create_document_job_stores_source_under_document_id_directory(async_session, tmp_path): settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) - result = await create_upload_job( + result = await create_document_job( filename="single-page.jpg", file_bytes=b"image-bytes", session=async_session, @@ -99,9 +99,7 @@ async def test_create_upload_job_stores_source_under_document_id_directory(async source = ( await async_session.exec( - select(Source) - .where(Source.document_id == result.document_id) - .order_by(Source.page_number) # pyright: ignore[reportArgumentType] + select(Source).where(Source.document_id == result.document_id).order_by(Source.page_number) # pyright: ignore[reportArgumentType] ) ).first() assert source is not None @@ -129,3 +127,25 @@ def test_store_person_portrait_stores_file_under_person_id_directory(tmp_path): assert stored_path.parent == (tmp_path / "persons" / str(person_id)) assert stored_path.exists() + + +@pytest.mark.parametrize( + ("filename", "expected_mime_type"), + [ + ("page.jpg", "image/jpeg"), + ("page.JPEG", "image/jpeg"), + ("page.png", "image/png"), + ("page.tif", "image/tiff"), + ("page.TIFF", "image/tiff"), + ("page.pdf", "application/pdf"), + ], +) +def test_source_mime_type_uses_canonical_source_policy(filename, expected_mime_type): + assert source_mime_type(filename) == expected_mime_type + + +def test_source_storage_rejects_unsupported_format(tmp_path): + settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) + + with pytest.raises(SourceStorageError): + store_source_file(filename="page.txt", file_bytes=b"text", settings=settings) diff --git a/tests/services/test_transcription_service.py b/tests/services/test_transcription_service.py index 51c96af..a10c5a8 100644 --- a/tests/services/test_transcription_service.py +++ b/tests/services/test_transcription_service.py @@ -1,4 +1,4 @@ -"""Tests for source revision behavior in TranscriptionService.""" +"""Tests for SourceService revision behavior.""" from uuid import uuid4 @@ -12,20 +12,20 @@ from transcription.db.models import JobStatus from transcription.db.models import Source from transcription.services.documents import DocumentService from transcription.services.jobs import JobService -from transcription.services.transcription import SourceDeleteBlockedError -from transcription.services.transcription import TranscriptionNotFoundError -from transcription.services.transcription import TranscriptionService +from transcription.services.sources import SourceDeleteBlockedError +from transcription.services.sources import SourceService +from transcription.services.sources import TranscriptionNotFoundError @pytest.mark.integration -class TestTranscriptionServiceRevisionUpsert: +class TestSourceServiceRevisionUpsert: """Verify page-level source revision semantics.""" @pytest.mark.asyncio async def test_upsert_revision_creates_new_revision(self, default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = Document(id=uuid4(), name="revision-create") await documents.create_document(document=document) @@ -62,7 +62,7 @@ class TestTranscriptionServiceRevisionUpsert: async def test_upsert_revision_updates_existing_single_revision(self, default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = Document(id=uuid4(), name="revision-update") await documents.create_document(document=document) @@ -97,10 +97,12 @@ class TestTranscriptionServiceRevisionUpsert: assert revisions[0].revised_text == "Revision v2" @pytest.mark.asyncio - async def test_delete_source_from_job_context_removes_source_and_single_link(self, default_session_factory, tmp_path): + async def test_delete_source_from_job_context_removes_source_and_single_link( + self, default_session_factory, tmp_path + ): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) transcriptions.settings.upload_dir = tmp_path document = Document(id=uuid4(), name="delete-source-success") @@ -139,7 +141,7 @@ class TestTranscriptionServiceRevisionUpsert: async def test_delete_source_from_job_context_blocks_when_other_job_links_exist(self, default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = Document(id=uuid4(), name="delete-source-blocked") await documents.create_document(document=document) @@ -172,7 +174,7 @@ class TestTranscriptionServiceRevisionUpsert: @pytest.mark.asyncio async def test_delete_unlinked_source_succeeds(self, default_session_factory, tmp_path): documents = DocumentService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) transcriptions.settings.upload_dir = tmp_path document = Document(id=uuid4(), name="delete-unlinked-source") @@ -203,7 +205,7 @@ class TestTranscriptionServiceRevisionUpsert: async def test_delete_unlinked_source_blocks_when_linked(self, default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = Document(id=uuid4(), name="delete-unlinked-blocked") await documents.create_document(document=document) diff --git a/tests/services/test_v2_crud.py b/tests/services/test_v2_crud.py index 478a964..8b774f2 100644 --- a/tests/services/test_v2_crud.py +++ b/tests/services/test_v2_crud.py @@ -13,48 +13,50 @@ from transcription.db.models import Source from transcription.services.documents import DocumentDeleteBlockedError from transcription.services.documents import DocumentService from transcription.services.jobs import JobService -from transcription.services.transcription import SourceDeleteBlockedError -from transcription.services.transcription import TranscriptionService +from transcription.services.people import PeopleService +from transcription.services.sources import SourceDeleteBlockedError +from transcription.services.sources import SourceService @pytest.mark.asyncio -async def test_document_service_handles_person_and_document_person_crud(default_session_factory): +async def test_people_service_handles_person_and_document_person_crud(default_session_factory): documents = DocumentService(session_factory=default_session_factory) + people_service = PeopleService(session_factory=default_session_factory) document = await documents.create_document(Document(id=uuid4(), name="person-doc")) - person = await documents.create_person(Person(full_name="Ada Lovelace")) + person = await people_service.create_person(Person(full_name="Ada Lovelace")) assert document.document_type_id is None - link = await documents.create_document_person( + link = await people_service.create_document_person( DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR) ) - fetched = await documents.read_document_person(link.id) + fetched = await people_service.read_document_person(link.id) assert fetched.id == link.id assert fetched.role == DocumentPersonRole.AUTHOR assert fetched.role_id is not None - updated_link = await documents.update_document_person( + updated_link = await people_service.update_document_person( DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role=DocumentPersonRole.RECIPIENT) ) assert updated_link.role == DocumentPersonRole.RECIPIENT assert updated_link.role_id is not None - listed = await documents.list_document_people(document_id=document.id) + listed = await people_service.list_document_people(document_id=document.id) assert len(listed) == 1 - people = await documents.list_people() + people = await people_service.list_people() assert len(people) == 1 - await documents.delete_document_person(updated_link) - assert len(await documents.list_document_people(document_id=document.id)) == 0 + await people_service.delete_document_person(updated_link) + assert len(await people_service.list_document_people(document_id=document.id)) == 0 @pytest.mark.asyncio async def test_transcription_service_manages_source_crud(default_session_factory): documents = DocumentService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = await documents.create_document(Document(id=uuid4(), name="source-doc")) source = await transcriptions.create_source( @@ -88,9 +90,7 @@ async def test_transcription_service_manages_source_crud(default_session_factory @pytest.mark.asyncio async def test_transcription_service_job_source_crud_uses_caller_session(default_session_factory): - documents = DocumentService(session_factory=default_session_factory) - jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) async with transcriptions._session_scope() as session: document = Document(id=uuid4(), name="job-source-doc") @@ -138,10 +138,11 @@ async def test_transcription_service_job_source_crud_uses_caller_session(default @pytest.mark.asyncio async def test_document_detail_loads_linked_person_relationship(default_session_factory): documents = DocumentService(session_factory=default_session_factory) + people_service = PeopleService(session_factory=default_session_factory) document = await documents.create_document(Document(id=uuid4(), name="detail-person-doc")) - person = await documents.create_person(Person(full_name="Grace Hopper")) - await documents.create_document_person( + person = await people_service.create_person(Person(full_name="Grace Hopper")) + await people_service.create_document_person( DocumentPerson( document_id=document.id, person_id=person.id, @@ -162,7 +163,7 @@ async def test_document_detail_loads_linked_person_relationship(default_session_ async def test_document_delete_is_blocked_with_source_and_job_dependencies(default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = await documents.create_document(Document(id=uuid4(), name="blocked-by-deps")) job = await jobs.create_job(Job(document_id=document.id)) @@ -197,7 +198,7 @@ async def test_document_delete_is_blocked_with_source_and_job_dependencies(defau async def test_source_delete_blocks_when_linked_to_multiple_jobs(default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = await documents.create_document(Document(id=uuid4(), name="multi-job-source-doc")) job_one = await jobs.create_job(Job(document_id=document.id)) @@ -229,7 +230,7 @@ async def test_source_delete_blocks_when_linked_to_multiple_jobs(default_session async def test_update_job_source_transcription_persists_provider_json_payloads(default_session_factory): documents = DocumentService(session_factory=default_session_factory) jobs = JobService(session_factory=default_session_factory) - transcriptions = TranscriptionService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) document = await documents.create_document(Document(id=uuid4(), name="provider-payloads-doc")) job = await jobs.create_job(Job(document_id=document.id)) diff --git a/tests/services/test_workflows_reliability.py b/tests/services/test_workflows_reliability.py index bcfcfa2..20ddd61 100644 --- a/tests/services/test_workflows_reliability.py +++ b/tests/services/test_workflows_reliability.py @@ -36,8 +36,8 @@ class TestWorkflowReliability: ) object.__setattr__( services, - "transcriptions", - services.transcriptions.__class__(session_factory=default_session_factory), + "sources", + services.sources.__class__(session_factory=default_session_factory), ) async with services.jobs._session_scope() as session: