diff --git a/app/database.py b/app/database.py index 03c6add..6b19bf1 100644 --- a/app/database.py +++ b/app/database.py @@ -2,8 +2,10 @@ Configuration SQLAlchemy — session async """ import logging +from datetime import datetime, timezone, timedelta from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import DeclarativeBase +from sqlalchemy import select, text from app.config import settings logger = logging.getLogger(__name__) @@ -44,9 +46,9 @@ async def get_db() -> AsyncSession: async def init_db(): """Crée toutes les tables, applique les migrations, et initialise un client par défaut si nécessaire.""" import secrets - from sqlalchemy import select, text from app.dependencies.auth import hash_api_key from app.models.client import APIClient, ClientPlan + from app.models.image import Image, ProcessingStatus async with engine.begin() as conn: from app.models import image # noqa: F401 @@ -59,7 +61,7 @@ async def init_db(): "ALTER TABLE api_clients ADD COLUMN IF NOT EXISTS ai_provider VARCHAR(64)" )) except Exception: - pass # SQLite ne supporte pas ADD COLUMN IF NOT EXISTS + pass try: await conn.execute(text( @@ -68,8 +70,27 @@ async def init_db(): except Exception: pass - # Vérifier s'il y a déjà des clients + # Vérifier s'il y a déjà des clients + recovery des images bloquées async with AsyncSessionLocal() as session: + # Recovery : marquer les images bloquées en "processing" depuis > 10 min comme ERROR + cutoff = datetime.now(timezone.utc) - timedelta(minutes=10) + stuck_query = select(Image).where( + Image.processing_status == ProcessingStatus.PROCESSING, + Image.processing_started_at < cutoff, + ) + stuck_result = await session.execute(stuck_query) + stuck_images = stuck_result.scalars().all() + for img in stuck_images: + img.processing_status = ProcessingStatus.ERROR + img.processing_error = "[Recovery] Pipeline interrompu (timeout ou crash worker)" + img.processing_done_at = datetime.now(timezone.utc) + if stuck_images: + await session.commit() + logger.warning( + "init_db.recovery", + extra={"stuck_images_recovered": len(stuck_images)}, + ) + result = await session.execute(select(APIClient).limit(1)) if result.scalar_one_or_none() is None: # Table vide -> Création du client bootstrap diff --git a/app/services/pipeline.py b/app/services/pipeline.py index bce1016..c5c364d 100644 --- a/app/services/pipeline.py +++ b/app/services/pipeline.py @@ -21,7 +21,6 @@ from app.metrics import ( hub_pipeline_duration, hub_pipeline_step_duration, hub_pipeline_errors, hub_ai_tokens_consumed ) -import asyncio logger = logging.getLogger(__name__) @@ -44,20 +43,17 @@ async def _publish_event( payload["data"] = data raw = json.dumps(payload) - # Publier sur le channel dédié à l'image await redis.publish(f"pipeline:{image_id}", raw) - # Buffer de reconnexion (10 derniers événements, TTL 60s) buffer_key = f"pipeline:buffer:{image_id}" await redis.rpush(buffer_key, raw) await redis.ltrim(buffer_key, -10, -1) await redis.expire(buffer_key, 60) - # Publier sur le channel admin pour le monitoring global if event in ("pipeline.started", "pipeline.done", "pipeline.error"): await redis.publish("pipeline:admin", raw) except Exception: - pass # Pub/Sub non critique — ne doit pas bloquer le pipeline + pass async def process_image_pipeline( @@ -70,10 +66,9 @@ async def process_image_pipeline( 3. Vision AI — description + tags (async) 4. Sauvegarde finale en BDD - Le statut est mis à jour à chaque étape pour permettre le polling. - Publie des événements Redis sur le channel pipeline:{image_id}. + Try/finally garantit que le statut est TOUJOURS mis à jour, + même en cas de crash inattendu (timeout, OOM, etc.). """ - # ── Chargement de l'image ───────────────────────────────── result = await db.execute(select(Image).where(Image.id == image_id)) image = result.scalar_one_or_none() @@ -81,7 +76,6 @@ async def process_image_pipeline( logger.warning("pipeline.image_not_found", extra={"image_id": image_id}) return - # ── Démarrage ───────────────────────────────────────────── image.processing_status = ProcessingStatus.PROCESSING image.processing_started_at = datetime.now(timezone.utc) t_start = time.time() @@ -93,172 +87,176 @@ async def process_image_pipeline( errors: list[str] = [] file_path = image.file_path - # ════════════════════════════════════════════════════════════ - # ÉTAPE 1 — Extraction EXIF - # ════════════════════════════════════════════════════════════ try: - logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "exif", "step_num": "1/3"}) - t0 = time.time() - - # Maintenant async et utilise le backend - exif = await extract_exif(file_path) + # ════════════════════════════════════════════════════════ + # ÉTAPE 1 — Extraction EXIF + # ════════════════════════════════════════════════════════ + try: + logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "exif", "step_num": "1/3"}) + t0 = time.time() - image.exif_raw = exif.get("raw") - image.exif_make = exif.get("make") - image.exif_model = exif.get("model") - image.exif_lens = exif.get("lens") - image.exif_taken_at = exif.get("taken_at") - image.exif_gps_lat = exif.get("gps_lat") - image.exif_gps_lon = exif.get("gps_lon") - image.exif_altitude = exif.get("altitude") - image.exif_iso = exif.get("iso") - image.exif_aperture = exif.get("aperture") - image.exif_shutter = exif.get("shutter") - image.exif_focal = exif.get("focal") - image.exif_flash = exif.get("flash") - image.exif_orientation = exif.get("orientation") - image.exif_software = exif.get("software") + exif = await extract_exif(file_path) - await db.commit() - elapsed = int((time.time() - t0) * 1000) - hub_pipeline_step_duration.labels(step="exif").observe(time.time() - t0) - - logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "exif", "duration_ms": elapsed, "camera": image.exif_make}) - await _publish_event(redis, image_id, "step.completed", { - "step": "exif", "duration_ms": elapsed, "camera": image.exif_make, - }) + image.exif_raw = exif.get("raw") + image.exif_make = exif.get("make") + image.exif_model = exif.get("model") + image.exif_lens = exif.get("lens") + image.exif_taken_at = exif.get("taken_at") + image.exif_gps_lat = exif.get("gps_lat") + image.exif_gps_lon = exif.get("gps_lon") + image.exif_altitude = exif.get("altitude") + image.exif_iso = exif.get("iso") + image.exif_aperture = exif.get("aperture") + image.exif_shutter = exif.get("shutter") + image.exif_focal = exif.get("focal") + image.exif_flash = exif.get("flash") + image.exif_orientation = exif.get("orientation") + image.exif_software = exif.get("software") - except Exception as e: - msg = f"EXIF : {str(e)}" - errors.append(msg) - hub_pipeline_errors.labels(step="exif").inc() - logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "exif", "error": str(e)}) + await db.commit() + elapsed = int((time.time() - t0) * 1000) + hub_pipeline_step_duration.labels(step="exif").observe(time.time() - t0) - # ════════════════════════════════════════════════════════════ - # ÉTAPE 2 — OCR - # ════════════════════════════════════════════════════════════ - try: - logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ocr", "step_num": "2/3"}) - t0 = time.time() - - # Maintenant async et utilise le backend - ocr = await extract_text(file_path) + logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "exif", "duration_ms": elapsed, "camera": image.exif_make}) + await _publish_event(redis, image_id, "step.completed", { + "step": "exif", "duration_ms": elapsed, "camera": image.exif_make, + }) - # Fallback AI si OCR classique échoue ou ne trouve rien - if not ocr.get("has_text", False): - logger.info("pipeline.ocr.fallback", extra={"image_id": image_id, "reason": "tesseract_empty"}) - ai_ocr = await extract_text_with_ai(file_path) + except Exception as e: + msg = f"EXIF : {str(e)}" + errors.append(msg) + hub_pipeline_errors.labels(step="exif").inc() + logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "exif", "error": str(e)}) - if ai_ocr.get("has_text"): - ocr = ai_ocr - logger.info("pipeline.ocr.fallback_success", extra={"image_id": image_id, "chars": len(ocr.get("text", ""))}) - else: - logger.info("pipeline.ocr.fallback_empty", extra={"image_id": image_id}) + # ════════════════════════════════════════════════════════ + # ÉTAPE 2 — OCR + # ════════════════════════════════════════════════════════ + try: + logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ocr", "step_num": "2/3"}) + t0 = time.time() - image.ocr_text = ocr.get("text") - image.ocr_language = ocr.get("language") - image.ocr_confidence = ocr.get("confidence") - image.ocr_has_text = ocr.get("has_text", False) + ocr = await extract_text(file_path) - await db.commit() - elapsed = int((time.time() - t0) * 1000) - hub_pipeline_step_duration.labels(step="ocr").observe(time.time() - t0) + if not ocr.get("has_text", False): + logger.info("pipeline.ocr.fallback", extra={"image_id": image_id, "reason": "tesseract_empty"}) + ai_ocr = await extract_text_with_ai(file_path) - logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text}) - await _publish_event(redis, image_id, "step.completed", { - "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text, - }) + if ai_ocr.get("has_text"): + ocr = ai_ocr + logger.info("pipeline.ocr.fallback_success", extra={"image_id": image_id, "chars": len(ocr.get("text", ""))}) + else: + logger.info("pipeline.ocr.fallback_empty", extra={"image_id": image_id}) - except Exception as e: - msg = f"OCR : {str(e)}" - errors.append(msg) - hub_pipeline_errors.labels(step="ocr").inc() - logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ocr", "error": str(e)}) + image.ocr_text = ocr.get("text") + image.ocr_language = ocr.get("language") + image.ocr_confidence = ocr.get("confidence") + image.ocr_has_text = ocr.get("has_text", False) - # ════════════════════════════════════════════════════════════ - # ÉTAPE 3 — Vision AI (description + tags) - # ════════════════════════════════════════════════════════════ - try: - logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ai", "step_num": "3/3"}) - t0 = time.time() + await db.commit() + elapsed = int((time.time() - t0) * 1000) + hub_pipeline_step_duration.labels(step="ocr").observe(time.time() - t0) - # Charger les préférences AI du client - provider_override = None - model_override = None - if image.client_id: - from app.models.client import APIClient - client_result = await db.execute( - select(APIClient).where(APIClient.id == image.client_id) + logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text}) + await _publish_event(redis, image_id, "step.completed", { + "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text, + }) + + except Exception as e: + msg = f"OCR : {str(e)}" + errors.append(msg) + hub_pipeline_errors.labels(step="ocr").inc() + logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ocr", "error": str(e)}) + + # ════════════════════════════════════════════════════════ + # ÉTAPE 3 — Vision AI (description + tags) + # ════════════════════════════════════════════════════════ + try: + logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ai", "step_num": "3/3"}) + t0 = time.time() + + provider_override = None + model_override = None + if image.client_id: + from app.models.client import APIClient + client_result = await db.execute( + select(APIClient).where(APIClient.id == image.client_id) + ) + owner = client_result.scalar_one_or_none() + if owner: + provider_override = owner.ai_provider + model_override = owner.ai_model + + ai = await analyze_image( + file_path=file_path, + ocr_hint=image.ocr_text, + provider_override=provider_override, + model_override=model_override, ) - owner = client_result.scalar_one_or_none() - if owner: - provider_override = owner.ai_provider - model_override = owner.ai_model - ai = await analyze_image( - file_path=file_path, - ocr_hint=image.ocr_text, - provider_override=provider_override, - model_override=model_override, - ) + image.ai_description = ai.get("description") + image.ai_tags = ai.get("tags", []) + image.ai_confidence = ai.get("confidence") + image.ai_model_used = ai.get("model") + image.ai_processed_at = datetime.now(timezone.utc) + image.ai_prompt_tokens = ai.get("prompt_tokens") + image.ai_output_tokens = ai.get("output_tokens") - image.ai_description = ai.get("description") - image.ai_tags = ai.get("tags", []) - image.ai_confidence = ai.get("confidence") - image.ai_model_used = ai.get("model") - image.ai_processed_at = datetime.now(timezone.utc) - image.ai_prompt_tokens = ai.get("prompt_tokens") - image.ai_output_tokens = ai.get("output_tokens") + await db.commit() + elapsed = int((time.time() - t0) * 1000) + hub_pipeline_step_duration.labels(step="ai").observe(time.time() - t0) - await db.commit() - elapsed = int((time.time() - t0) * 1000) - hub_pipeline_step_duration.labels(step="ai").observe(time.time() - t0) - - # Token metrics - if image.ai_prompt_tokens or image.ai_output_tokens: - total_tokens = (image.ai_prompt_tokens or 0) + (image.ai_output_tokens or 0) - hub_ai_tokens_consumed.labels( - client_id=image.client_id, - model=image.ai_model_used or "unknown" - ).inc(total_tokens) + if image.ai_prompt_tokens or image.ai_output_tokens: + total_tokens = (image.ai_prompt_tokens or 0) + (image.ai_output_tokens or 0) + hub_ai_tokens_consumed.labels( + client_id=image.client_id, + model=image.ai_model_used or "unknown" + ).inc(total_tokens) - logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or [])}) - await _publish_event(redis, image_id, "step.completed", { - "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or []), - }) + logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or [])}) + await _publish_event(redis, image_id, "step.completed", { + "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or []), + }) - except Exception as e: - msg = f"AI Vision : {str(e)}" - errors.append(msg) - hub_pipeline_errors.labels(step="ai").inc() - logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ai", "error": str(e)}) + except Exception as e: + msg = f"AI Vision : {str(e)}" + errors.append(msg) + hub_pipeline_errors.labels(step="ai").inc() + logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ai", "error": str(e)}) - # ════════════════════════════════════════════════════════════ - # FINALISATION - # ════════════════════════════════════════════════════════════ - image.processing_done_at = datetime.now(timezone.utc) + finally: + # ════════════════════════════════════════════════════════ + # FINALISATION — toujours exécuté, même après un crash + # ════════════════════════════════════════════════════════ + try: + image.processing_done_at = datetime.now(timezone.utc) - if errors: - if image.ai_description: - image.processing_status = ProcessingStatus.DONE - image.processing_error = f"Avertissements : {'; '.join(errors)}" - else: - image.processing_status = ProcessingStatus.ERROR - image.processing_error = "; ".join(errors) - else: - image.processing_status = ProcessingStatus.DONE - image.processing_error = None + if errors: + if image.ai_description: + image.processing_status = ProcessingStatus.DONE + image.processing_error = f"Avertissements : {'; '.join(errors)}" + else: + image.processing_status = ProcessingStatus.ERROR + image.processing_error = "; ".join(errors) + else: + image.processing_status = ProcessingStatus.DONE + image.processing_error = None - await db.commit() - hub_pipeline_duration.observe(time.time() - t_start) - - logger.info("pipeline.completed", extra={ - "image_id": image_id, - "status": image.processing_status.value, - "errors": len(errors), - }) + await db.commit() + hub_pipeline_duration.observe(time.time() - t_start) - if errors: - await _publish_event(redis, image_id, "pipeline.error", {"errors": errors}) - else: - await _publish_event(redis, image_id, "pipeline.done") + logger.info("pipeline.completed", extra={ + "image_id": image_id, + "status": image.processing_status.value, + "errors": len(errors), + }) + + if errors: + await _publish_event(redis, image_id, "pipeline.error", {"errors": errors}) + else: + await _publish_event(redis, image_id, "pipeline.done") + except Exception as final_error: + # Si même la finalisation échoue, log seulement — ne pas laisser l'image bloquée + logger.critical("pipeline.finalization_failed", extra={ + "image_id": image_id, + "error": str(final_error), + })