""" Exécution réelle du pipeline sur un job page (Sprint 4 — Session C). Point d'entrée principal : execute_page_job(job_id) Séquence stricte (CLAUDE.md §8 pipeline) : 1. job → RUNNING 2. Charger page / manuscrit / corpus depuis BDD 3. Charger CorpusProfile depuis profiles/{profile_id}.json 4. Charger ModelConfig depuis BDD (erreur explicite si absent) 5. fetch_and_normalize() → ImageDerivativeInfo 6. run_primary_analysis() → PageMaster (+ double stockage R05) 7. generate_alto() + write_alto() 8. page.processing_status → ANALYZED 9. job → DONE Sur toute exception : job → FAILED + error_message, page → ERROR. Aucun échec silencieux (CLAUDE.md §7). """ # 1. stdlib import asyncio import json import logging from datetime import datetime, timezone from pathlib import Path # 2. third-party from sqlalchemy.ext.asyncio import AsyncSession # 3. local from app import config as _config_module from app.models.corpus import CorpusModel, ManuscriptModel, PageModel from app.models.database import async_session_factory from app.models.job import JobModel from app.models.model_config_db import ModelConfigDB from app.schemas.corpus_profile import CorpusProfile from app.schemas.image import ImageSourceInfo from app.schemas.model_config import ModelConfig, ProviderType from app.services.image.normalizer import ( create_derivatives, fetch_ai_derivative_bytes, fetch_and_normalize, ) logger = logging.getLogger(__name__) # Racine du projet — résolue via .resolve() pour garantir un chemin absolu # même si __file__ est relatif au CWD (comportement variable selon l'environnement # d'exécution : Docker, HuggingFace Spaces, tests...). # job_runner.py est à backend/app/services/job_runner.py → 4 parents → racine. _PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent # ── Point d'entrée public ────────────────────────────────────────────────── async def execute_page_job(job_id: str, db: AsyncSession | None = None) -> None: """BackgroundTask : exécute le pipeline complet sur une page. Args: job_id: identifiant du JobModel en BDD. db: session optionnelle (fournie dans les tests ; None = nouvelle session). """ if db is None: async with async_session_factory() as session: await _run_job_impl(job_id, session) else: await _run_job_impl(job_id, db) # ── Implémentation interne (testable directement) ────────────────────────── async def _run_job_impl(job_id: str, db: AsyncSession) -> None: """Exécution du pipeline sur un job, avec la session fournie. Exposé (préfixe _ conservé) pour les tests unitaires. """ # ── 1. Charger le job, passer status → RUNNING ────────────────────────── job = await db.get(JobModel, job_id) if job is None: logger.error("Job introuvable — exécution abandonnée", extra={"job_id": job_id}) return job.status = "running" job.started_at = datetime.now(timezone.utc) await db.commit() page: PageModel | None = None try: # ── 2. Charger page / manuscrit / corpus ───────────────────────────── if job.page_id is None: raise ValueError("Ce job n'a pas de page_id — impossible d'exécuter le pipeline") page = await db.get(PageModel, job.page_id) if page is None: raise ValueError(f"Page introuvable en BDD : {job.page_id}") manuscript = await db.get(ManuscriptModel, page.manuscript_id) if manuscript is None: raise ValueError(f"Manuscrit introuvable en BDD : {page.manuscript_id}") corpus = await db.get(CorpusModel, manuscript.corpus_id) if corpus is None: raise ValueError(f"Corpus introuvable en BDD : {manuscript.corpus_id}") # ── 3. Charger le CorpusProfile ────────────────────────────────────── # settings.profiles_dir est la source canonique du chemin (config.py). # Résolu depuis PROFILES_DIR en Docker, ou _REPO_ROOT/profiles en local. profile_path = _config_module.settings.profiles_dir / f"{corpus.profile_id}.json" if not profile_path.exists(): raise FileNotFoundError( f"Fichier de profil introuvable : {profile_path}. " f"Profil attendu : «{corpus.profile_id}»" ) profile_data = json.loads(profile_path.read_text(encoding="utf-8")) corpus_profile = CorpusProfile.model_validate(profile_data) # ── 4. Charger le ModelConfig (erreur explicite si absent) ─────────── model_db = await db.get(ModelConfigDB, corpus.id) if model_db is None: raise ValueError( f"Aucun modèle IA configuré pour le corpus «{corpus.id}». " "Sélectionnez un modèle via PUT /api/v1/corpora/{id}/model avant " "de lancer le pipeline." ) model_config = ModelConfig( corpus_id=corpus.id, selected_model_id=model_db.selected_model_id, selected_model_display_name=model_db.selected_model_display_name, provider=ProviderType(model_db.provider_type), supports_vision=model_db.supports_vision, last_fetched_at=model_db.updated_at, ) # ── 5. Obtenir l'image pour l'IA ───────────────────────────────────── data_dir = _config_module.settings.data_dir image_source = page.image_master_path or "" from app.services.ai.analyzer import run_primary_analysis if page.iiif_service_url: # ── Mode IIIF natif : fetch en mémoire, zéro stockage ──────────── deriv_bytes, deriv_w, deriv_h = await asyncio.to_thread( fetch_ai_derivative_bytes, iiif_service_url=page.iiif_service_url, fallback_url=None, ) image_source_info = ImageSourceInfo( original_url=image_source or page.iiif_service_url, iiif_service_url=page.iiif_service_url, manifest_url=page.manifest_url, is_iiif=True, original_width=page.canvas_width or deriv_w, original_height=page.canvas_height or deriv_h, ) # ── 6. Analyse primaire IA (R05 : double stockage) ─────────────── page_master = await asyncio.to_thread( run_primary_analysis, derivative_image_bytes=deriv_bytes, derivative_width=deriv_w, derivative_height=deriv_h, corpus_profile=corpus_profile, model_config=model_config, page_id=page.id, manuscript_id=manuscript.id, corpus_slug=corpus.slug, folio_label=page.folio_label, sequence=page.sequence, image_info=image_source_info, base_data_dir=data_dir, project_root=_PROJECT_ROOT, ) elif image_source.startswith(("http://", "https://")): # ── Mode fallback URL : télécharge + stocke sur disque (legacy) ── image_info = await asyncio.to_thread( fetch_and_normalize, image_source, corpus.slug, page.folio_label, data_dir ) page_master = await asyncio.to_thread( run_primary_analysis, derivative_image_path=Path(image_info.derivative_path), corpus_profile=corpus_profile, model_config=model_config, page_id=page.id, manuscript_id=manuscript.id, corpus_slug=corpus.slug, folio_label=page.folio_label, sequence=page.sequence, image_info=image_info, base_data_dir=data_dir, project_root=_PROJECT_ROOT, ) elif image_source: # ── Mode fichier local (upload) ────────────────────────────────── source_path = Path(image_source).resolve() data_dir_resolved = data_dir.resolve() if not source_path.is_relative_to(data_dir_resolved): raise ValueError( f"Chemin image hors du répertoire de données interdit : " f"{image_source!r} (résolu : {source_path})" ) source_bytes = source_path.read_bytes() image_info = await asyncio.to_thread( create_derivatives, source_bytes, image_source, corpus.slug, page.folio_label, data_dir ) page_master = await asyncio.to_thread( run_primary_analysis, derivative_image_path=Path(image_info.derivative_path), corpus_profile=corpus_profile, model_config=model_config, page_id=page.id, manuscript_id=manuscript.id, corpus_slug=corpus.slug, folio_label=page.folio_label, sequence=page.sequence, image_info=image_info, base_data_dir=data_dir, project_root=_PROJECT_ROOT, ) else: raise ValueError( f"La page {page.id} n'a pas d'image source " "(ni iiif_service_url, ni image_master_path)" ) # ── 6b. Index pour la recherche ───────────────────────────────────── from app.services.search.indexer import index_page await index_page(db, page_master) # ── 7. Générer et écrire l'ALTO XML ────────────────────────────────── from app.services.export.alto import generate_alto, write_alto alto_xml = generate_alto(page_master) alto_path = ( data_dir / "corpora" / corpus.slug / "pages" / page.folio_label / "alto.xml" ) write_alto(alto_xml, alto_path) # ── 8. Page → ANALYZED ─────────────────────────────────────────────── page.processing_status = "ANALYZED" if page_master.ocr is not None: page.confidence_summary = page_master.ocr.confidence # ── 9. Job → DONE ──────────────────────────────────────────────────── job.status = "done" job.finished_at = datetime.now(timezone.utc) await db.commit() logger.info( "Job terminé avec succès", extra={ "job_id": job_id, "page_id": page.id, "folio": page.folio_label, "corpus": corpus.slug, }, ) except Exception as exc: logger.error( "Échec du job", extra={"job_id": job_id, "error": str(exc)}, exc_info=True, ) job.status = "failed" job.error_message = str(exc) job.finished_at = datetime.now(timezone.utc) if page is not None: page.processing_status = "ERROR" try: await db.commit() except Exception as commit_exc: logger.error( "Impossible de persister l'état d'échec du job", extra={"job_id": job_id, "commit_error": str(commit_exc)}, )