diff --git a/backend/app/routers/library.py b/backend/app/routers/library.py index 9a2b6d0..86ff42d 100644 --- a/backend/app/routers/library.py +++ b/backend/app/routers/library.py @@ -137,6 +137,35 @@ async def trigger_scan(current_user: User = Depends(get_current_user)): return {"status": "success", "message": "Library scan started"} +@router.post("/maintenance/recover-stuck") +async def recover_stuck_photos( + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Reset photos stuck in 'processing' for more than 30 minutes back to + 'pending' so the pipeline can retry them. Returns the count of recovered + photos.""" + from datetime import datetime, timedelta, timezone + + cutoff = datetime.now(timezone.utc) - timedelta(minutes=30) + result = await db.execute( + update(Photo) + .where( + Photo.processing_status == 'processing', + Photo.updated_at < cutoff, + ) + .values( + processing_status='pending', + processing_error='Auto-recovered from stuck processing state', + ) + ) + await db.commit() + count = result.rowcount + if count: + logger.info("Recovered %d stuck photos back to pending", count) + return {"status": "success", "recovered": count} + + @router.post("/backfill-gps") async def trigger_backfill_gps(current_user: User = Depends(get_current_user)): """Re-run EXIF metadata extraction on every photo that's still missing diff --git a/backend/app/services/metadata.py b/backend/app/services/metadata.py index b54cbfa..4f251f1 100644 --- a/backend/app/services/metadata.py +++ b/backend/app/services/metadata.py @@ -209,21 +209,24 @@ async def _extract_metadata_async(photo_id: str): cmd, capture_output=True, text=True, - timeout=30 + timeout=30, + stdin=subprocess.DEVNULL, ) - + if result.returncode != 0: logger.error(f"ExifTool error: {result.stderr}") + photo.processing_error = f"ExifTool: {result.stderr[:500]}" + await session.commit() return {'status': 'error', 'message': result.stderr} - + # Parse JSON output metadata = json.loads(result.stdout) if metadata and len(metadata) > 0: exif_data = metadata[0] - + # Store full metadata as JSON photo.exif_json = json.dumps(exif_data) - + # Extract taken_at date date_fields = [ 'EXIF:DateTimeOriginal', @@ -231,7 +234,7 @@ async def _extract_metadata_async(photo_id: str): 'QuickTime:MediaCreateDate', 'EXIF:ModifyDate' ] - + for field in date_fields: if field in exif_data: taken_at = parse_exif_datetime(exif_data[field]) @@ -247,7 +250,7 @@ async def _extract_metadata_async(photo_id: str): photo.has_date_warning = has_date_warning( photo.filepath, photo.taken_at ) - + # Extract dimensions if not already set if not photo.width: photo.width = exif_data.get('EXIF:ImageWidth') or exif_data.get('File:ImageWidth') @@ -262,26 +265,27 @@ async def _extract_metadata_async(photo_id: str): # Extract and store key metadata for search key_metadata = extract_key_metadata(exif_data) - - # Update FTS table (would be done via trigger in production) - # For now, we'll store it in a comment - + await session.commit() - + logger.info(f"Metadata extracted for photo {photo_id}") return { 'status': 'success', 'photo_id': photo_id, 'taken_at': photo.taken_at.isoformat() if photo.taken_at else None } - + except subprocess.TimeoutExpired: logger.error(f"ExifTool timeout for {photo.filepath}") + photo.processing_error = 'ExifTool timeout' + await session.commit() return {'status': 'error', 'message': 'ExifTool timeout'} except json.JSONDecodeError as e: logger.error(f"Failed to parse ExifTool output: {e}") + photo.processing_error = f"Invalid ExifTool output: {e}" + await session.commit() return {'status': 'error', 'message': 'Invalid ExifTool output'} - + except Exception as e: logger.error(f"Error extracting metadata for {photo_id}: {e}") return {'status': 'error', 'message': str(e)} \ No newline at end of file diff --git a/backend/app/services/vision/bootstrap_models.py b/backend/app/services/vision/bootstrap_models.py index 61f9e85..57f38e5 100644 --- a/backend/app/services/vision/bootstrap_models.py +++ b/backend/app/services/vision/bootstrap_models.py @@ -103,6 +103,16 @@ def bootstrap(models_dir: str | None = None): else: logger.info("All model files present in %s", base) + # Signal readiness via Redis so the scan pipeline knows the vision + # worker can accept tasks. + try: + import redis as _redis + r = _redis.from_url(settings.redis_url) + r.set("mulita:vision:ready", "1") + logger.info("Set mulita:vision:ready in Redis") + except Exception as e: + logger.warning("Could not set vision readiness flag in Redis: %s", e) + if __name__ == "__main__": logging.basicConfig(level=logging.INFO) diff --git a/backend/app/tasks/celery.py b/backend/app/tasks/celery.py index cc218e6..514c5aa 100644 --- a/backend/app/tasks/celery.py +++ b/backend/app/tasks/celery.py @@ -1,15 +1,26 @@ """ Celery configuration and app initialization """ +import logging +import os + from celery import Celery +from celery.signals import worker_process_init from app.config import settings +logger = logging.getLogger(__name__) + # Create Celery app celery_app = Celery( 'mulita', broker=settings.celery_broker_url, backend=settings.celery_result_backend, - include=['app.tasks.scan', 'app.tasks.thumbs', 'app.tasks.vision'] + include=[ + 'app.tasks.scan', + 'app.tasks.thumbs', + 'app.tasks.vision', + 'app.services.metadata', # extract_metadata lives here + ] ) # Configure Celery @@ -19,16 +30,34 @@ celery_app.conf.update( result_serializer='json', timezone='UTC', enable_utc=True, + # Robust acknowledgment: keep message in broker until task succeeds. + task_acks_late=True, + task_reject_on_worker_lost=True, + # Global time limits — individual tasks can override via decorator. + task_soft_time_limit=300, # 5 min — raises SoftTimeLimitExceeded + task_time_limit=600, # 10 min — SIGKILL + # Explicit routes for every task name. Wildcard patterns don't match + # short names produced by @shared_task(name='...'). task_routes={ - 'app.tasks.thumbs.*': {'queue': 'high'}, - 'app.tasks.scan.*': {'queue': 'low'}, - 'app.tasks.vision.*': {'queue': 'vision'}, + # Vision queue — GPU/CPU-bound inference 'embed_photo': {'queue': 'vision'}, 'ocr_photo': {'queue': 'vision'}, 'detect_objects': {'queue': 'vision'}, 'extract_faces': {'queue': 'vision'}, 'classify_content': {'queue': 'vision'}, 'vision_fanout': {'queue': 'vision'}, + 'recluster_faces': {'queue': 'vision'}, + # High-priority queue — thumbnails & duplicates + 'generate_thumbnails': {'queue': 'high'}, + 'regenerate_all_thumbnails': {'queue': 'high'}, + 'backfill_phashes': {'queue': 'high'}, + 'regroup_duplicates': {'queue': 'high'}, + 'incremental_regroup_duplicates': {'queue': 'high'}, + # Low-priority queue — scans + 'scan_folder': {'queue': 'low'}, + 'scan_all_source_roots': {'queue': 'low'}, + 'backfill_gps': {'queue': 'low'}, + # Dedicated watcher queue 'watch_folders': {'queue': 'watcher'}, }, task_default_queue='default', @@ -36,4 +65,22 @@ celery_app.conf.update( task_default_exchange_type='direct', task_default_routing_key='default', broker_connection_retry_on_startup=True, -) \ No newline at end of file +) + + +@worker_process_init.connect +def _warmup_vision_models(**kwargs): + """Pre-load vision models in the worker process so the first task + doesn't pay cold-start latency. Only runs on the vision queue.""" + # The worker name contains the queue — only warm up vision workers. + worker_queues = os.environ.get("CELERY_QUEUES", "") + if "vision" not in worker_queues: + # Heuristic: check the celery command line for -Q vision + import sys + if "vision" not in " ".join(sys.argv): + return + try: + from app.services.vision.registry import registry + registry.warmup() + except Exception: + logger.exception("Vision model warmup failed") \ No newline at end of file diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 8fcbb55..13e027f 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -95,11 +95,13 @@ async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], ta r = _get_redis() + PROGRESS_TTL = 3600 # 1 hour — auto-expire if scan crashes + def progress_set(key: str, value) -> None: if r is None: return try: - r.set(key, str(value)) + r.set(key, str(value), ex=PROGRESS_TTL) except Exception as e: logger.debug(f"scan progress set failed: {e}") @@ -467,12 +469,20 @@ async def _scan_all_source_roots_async(): except Exception as e: logger.warning(f"Could not queue post-scan face recluster: {e}") + # Re-extract metadata for photos missing GPS coordinates. + # Runs on every startup so photos scanned before the GPS fix + # eventually get their coordinates populated. + try: + backfill_gps.apply_async(countdown=30) + except Exception as e: + logger.warning(f"Could not queue post-scan GPS backfill: {e}") + WATCHER_LOCK_KEY = "mulita:watch_folders:lock" -WATCHER_LOCK_TTL = 300 # 5 min — renewed every 60s +WATCHER_LOCK_TTL = 60 # 1 min — renewed every event batch via wall-clock check -@shared_task(name='watch_folders', bind=True) +@shared_task(name='watch_folders', bind=True, soft_time_limit=None, time_limit=None) def watch_folders(self): """ Watch folders for changes using watchfiles. Long-running task that @@ -527,16 +537,20 @@ def watch_folders(self): return root_id return None - renew_counter = 0 - for changes in watch(*paths): - # Renew the Redis lock periodically so it doesn't expire - # while the watcher is idle between events. - renew_counter += 1 - if renew_counter % 10 == 0: + import time + last_renew = time.monotonic() + for changes in watch(*paths, rust_timeout=30_000): + # Renew the Redis lock on a wall-clock schedule (every 30s) + # instead of every N events, so quiet directories don't let + # the lock expire. watchfiles' rust_timeout ensures we wake + # at least every 30s even with no FS events. + now = time.monotonic() + if now - last_renew >= 30: try: lock.extend(WATCHER_LOCK_TTL) + last_renew = now except Exception: - pass + logger.warning("watch_folders: failed to renew Redis lock") for change_type, filepath in changes: filepath = str(filepath) @@ -557,7 +571,7 @@ def watch_folders(self): try: lock.release() except Exception: - pass + logger.warning("watch_folders: could not release Redis lock (may have expired)") async def handle_file_deletion(filepath: str): """Handle deletion of a file from the filesystem""" diff --git a/backend/app/tasks/thumbs.py b/backend/app/tasks/thumbs.py index b5067e3..40f45a9 100644 --- a/backend/app/tasks/thumbs.py +++ b/backend/app/tasks/thumbs.py @@ -218,23 +218,28 @@ def auto_rotate_image(image: Image.Image) -> Image.Image: if orientation in rotation_map: image = image.rotate(rotation_map[orientation], expand=True) - except: + except (AttributeError, KeyError, TypeError): pass # No orientation data available return image def generate_thumbnail(image: Image.Image, size: int, output_path: str): - """Generate a thumbnail of the specified size""" - # Maintain aspect ratio - image.thumbnail((size, size), Image.Resampling.LANCZOS) - - # Save as WebP with specified quality - image.save( + """Generate a thumbnail of the specified size. + + Works on a copy so the caller's image is never mutated — this is + critical because the thumbnail loop iterates multiple sizes and + in-place shrinking would degrade later (larger) sizes. + """ + img = image.copy() + img.thumbnail((size, size), Image.Resampling.LANCZOS) + + img.save( output_path, 'WEBP', quality=settings.thumbnails.quality, method=4 # Balance between speed and compression ) + img.close() @shared_task(bind=True, name='generate_thumbnails') def generate_thumbnails(self, photo_id: str): @@ -325,11 +330,11 @@ async def _generate_thumbnails_async(photo_id: str, task): photo.processing_status = 'completed' photo.processing_error = None await session.commit() - + logger.info(f"Thumbnails generated for photo {photo_id}") - # Dispatch vision pipeline (embedding, OCR, detection, faces) - # after thumbs are ready so vision tasks have images to read. + # Dispatch vision pipeline only after thumbnails succeeded — + # vision tasks need the generated thumbnails to run inference. try: from app.tasks.vision import vision_fanout vision_fanout.delay(photo_id) diff --git a/backend/app/tasks/vision.py b/backend/app/tasks/vision.py index f4005bf..b5edb83 100644 --- a/backend/app/tasks/vision.py +++ b/backend/app/tasks/vision.py @@ -23,12 +23,33 @@ from app.config import settings logger = logging.getLogger(__name__) +VISION_READY_KEY = "mulita:vision:ready" + + +def _vision_worker_ready() -> bool: + """Check whether the vision worker has finished model bootstrap.""" + try: + import redis as _redis + return bool(_redis.from_url(settings.redis_url).exists(VISION_READY_KEY)) + except Exception: + return False + + +_sync_engine = None + + +def _get_sync_engine(): + """Return a module-level singleton engine (one per worker process).""" + global _sync_engine + if _sync_engine is None: + sync_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace("+aiosqlite", "") + _sync_engine = create_engine(sync_url, pool_pre_ping=True, pool_size=3, max_overflow=5) + return _sync_engine + def _get_sync_session() -> Session: - """Create a sync DB session for use in Celery workers.""" - sync_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace("+aiosqlite", "") - engine = create_engine(sync_url, pool_pre_ping=True) - return sessionmaker(bind=engine)() + """Create a sync DB session backed by the shared engine.""" + return sessionmaker(bind=_get_sync_engine())() def _load_thumb(photo_id: str, size: str = "medium") -> np.ndarray | None: @@ -37,12 +58,19 @@ def _load_thumb(photo_id: str, size: str = "medium") -> np.ndarray | None: if not thumb_path.exists(): logger.warning("Thumbnail not found: %s", thumb_path) return None - img = Image.open(thumb_path).convert("RGB") - return np.array(img) + try: + img = Image.open(thumb_path).convert("RGB") + img.load() # force decode to catch corruption early + arr = np.array(img) + img.close() + return arr + except Exception as e: + logger.warning("Corrupt or unreadable thumbnail for %s: %s", photo_id, e) + return None -@shared_task(name='embed_photo', queue='vision') -def embed_photo(photo_id: str): +@shared_task(name='embed_photo', queue='vision', bind=True, max_retries=3) +def embed_photo(self, photo_id: str): """Generate CLIP embedding for a photo and store in pgvector.""" if not settings.vision.enabled: return {'status': 'skipped', 'reason': 'vision disabled'} @@ -51,9 +79,13 @@ def embed_photo(photo_id: str): if image is None: return {'status': 'error', 'message': 'thumbnail not found'} - from app.services.vision.registry import registry - embedder = registry.get_embedder() - vector = embedder.embed_image(image) + try: + from app.services.vision.registry import registry + embedder = registry.get_embedder() + vector = embedder.embed_image(image) + except Exception as exc: + logger.exception("embed_photo failed for %s", photo_id) + raise self.retry(exc=exc, countdown=60) model_name = settings.vision.embedder.name @@ -72,10 +104,13 @@ def embed_photo(photo_id: str): ) session.add(emb) session.commit() + except Exception: + session.rollback() + raise finally: session.close() - logger.info("Embedded photo %s with %s", photo_id, model_name) + logger.info("[%s] Embedded photo %s with %s", self.request.id, photo_id, model_name) return {'status': 'success', 'photo_id': photo_id} @@ -99,8 +134,8 @@ def vision_fanout(photo_id: str): return {'status': 'dispatched', 'photo_id': photo_id} -@shared_task(name='ocr_photo', queue='vision') -def ocr_photo(photo_id: str): +@shared_task(name='ocr_photo', queue='vision', bind=True, max_retries=3) +def ocr_photo(self, photo_id: str): """Run OCR on a photo and store text regions.""" if not settings.vision.enabled or not settings.vision.ocr.enabled: return {'status': 'skipped', 'reason': 'OCR disabled'} @@ -109,9 +144,13 @@ def ocr_photo(photo_id: str): if image is None: return {'status': 'error', 'message': 'thumbnail not found'} - from app.services.vision.registry import registry - ocr_engine = registry.get_ocr() - results = ocr_engine.run(image) + try: + from app.services.vision.registry import registry + ocr_engine = registry.get_ocr() + results = ocr_engine.run(image) + except Exception as exc: + logger.exception("ocr_photo failed for %s", photo_id) + raise self.retry(exc=exc, countdown=60) if not results: logger.info("No OCR text found for photo %s", photo_id) @@ -131,15 +170,18 @@ def ocr_photo(photo_id: str): bbox=r.bbox, )) session.commit() + except Exception: + session.rollback() + raise finally: session.close() - logger.info("OCR: %d text regions for photo %s", len(results), photo_id) + logger.info("[%s] OCR: %d text regions for photo %s", self.request.id, len(results), photo_id) return {'status': 'success', 'photo_id': photo_id, 'regions': len(results)} -@shared_task(name='detect_objects', queue='vision') -def detect_objects(photo_id: str): +@shared_task(name='detect_objects', queue='vision', bind=True, max_retries=3) +def detect_objects(self, photo_id: str): """Detect objects in a photo, create Tag(kind=object) rows, and link via photo_tags with confidence/bbox/source.""" if not settings.vision.enabled or not settings.vision.detector.enabled: @@ -149,14 +191,19 @@ def detect_objects(photo_id: str): if image is None: return {'status': 'error', 'message': 'thumbnail not found'} - from app.services.vision.registry import registry - detector = registry.get_detector() - detections = detector.detect(image) + try: + from app.services.vision.registry import registry + detector = registry.get_detector() + detections = detector.detect(image) + except Exception as exc: + logger.exception("detect_objects failed for %s", photo_id) + raise self.retry(exc=exc, countdown=60) if not detections: logger.info("No objects detected for photo %s", photo_id) return {'status': 'success', 'photo_id': photo_id, 'objects': 0} + from app.models import Photo from app.models.tags import Tag, photo_tags source_name = "vision:yolov8n" @@ -206,16 +253,19 @@ def detect_objects(photo_id: str): ) session.commit() + except Exception: + session.rollback() + raise finally: session.close() labels = [d.label for d in detections] - logger.info("Detected %d objects in photo %s: %s", len(detections), photo_id, labels) + logger.info("[%s] Detected %d objects in photo %s: %s", self.request.id, len(detections), photo_id, labels) return {'status': 'success', 'photo_id': photo_id, 'objects': len(detections)} -@shared_task(name='classify_content', queue='vision') -def classify_content(photo_id: str): +@shared_task(name='classify_content', queue='vision', bind=True, max_retries=3) +def classify_content(self, photo_id: str): """Classify image content type (screenshot, document, artwork, etc.) using CLIP zero-shot classification. Writes Tag(kind=content_type).""" if not settings.vision.enabled or not settings.vision.classifier.enabled: @@ -225,14 +275,19 @@ def classify_content(photo_id: str): if image is None: return {'status': 'error', 'message': 'thumbnail not found'} - from app.services.vision.registry import registry - classifier = registry.get_classifier() - results = classifier.classify(image) + try: + from app.services.vision.registry import registry + classifier = registry.get_classifier() + results = classifier.classify(image) + except Exception as exc: + logger.exception("classify_content failed for %s", photo_id) + raise self.retry(exc=exc, countdown=60) if not results: logger.info("No confident classification for photo %s", photo_id) return {'status': 'success', 'photo_id': photo_id, 'content_type': None} + from app.models import Photo from app.models.tags import Tag, photo_tags source_name = "vision:clip_classifier" @@ -273,10 +328,13 @@ def classify_content(photo_id: str): ) ) session.commit() + except Exception: + session.rollback() + raise finally: session.close() - logger.info("Classified photo %s as '%s' (%.2f)", photo_id, best.label, best.confidence) + logger.info("[%s] Classified photo %s as '%s' (%.2f)", self.request.id, photo_id, best.label, best.confidence) return {'status': 'success', 'photo_id': photo_id, 'content_type': best.label} @@ -309,15 +367,19 @@ def _load_original(photo_id: str) -> np.ndarray | None: w, h = img.size if max(w, h) > max_dim: scale = max_dim / max(w, h) - img = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC) - return np.array(img) + resized = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC) + img.close() + img = resized + arr = np.array(img) + img.close() + return arr except Exception as e: logger.warning("Failed to load original %s: %s", filepath, e) return None -@shared_task(name='extract_faces', queue='vision') -def extract_faces(photo_id: str): +@shared_task(name='extract_faces', queue='vision', bind=True, max_retries=3) +def extract_faces(self, photo_id: str): """Detect faces and store recognition embeddings using InsightFace (RetinaFace + ArcFace). No YOLO workaround needed — RetinaFace has strong human-vs-non-human precision on its own.""" @@ -330,9 +392,13 @@ def extract_faces(photo_id: str): if image is None: return {'status': 'error', 'message': 'no image available'} - from app.services.vision.registry import registry - face_proc = registry.get_face_processor() - faces = face_proc.process(image) + try: + from app.services.vision.registry import registry + face_proc = registry.get_face_processor() + faces = face_proc.process(image) + except Exception as exc: + logger.exception("extract_faces failed for %s", photo_id) + raise self.retry(exc=exc, countdown=60) if not faces: logger.info("No faces detected for photo %s", photo_id) @@ -355,6 +421,9 @@ def _save_faces(photo_id: str, faces) -> dict: cluster_id=None, )) session.commit() + except Exception: + session.rollback() + raise finally: session.close() @@ -383,8 +452,8 @@ def _schedule_recluster_debounced(): logger.debug("recluster debounce check failed: %s", e) -@shared_task(name='recluster_faces', queue='vision') -def recluster_faces(): +@shared_task(name='recluster_faces', queue='vision', bind=True, max_retries=10) +def recluster_faces(self): """Run DBSCAN clustering over all face embeddings and assign/create Tag(kind=face_cluster) entries.""" # Clear debounce key so new face extractions can schedule another round. @@ -394,9 +463,14 @@ def recluster_faces(): except Exception: pass + if not _vision_worker_ready(): + logger.info("Vision worker not ready yet — retrying in 30s") + raise self.retry(countdown=30) + if not settings.vision.enabled or not settings.vision.faces.enabled: return {'status': 'skipped', 'reason': 'faces disabled'} + from app.models import Photo from app.models.face_embedding import FaceEmbedding from app.models.tags import Tag, photo_tags from app.services.vision.clustering import cluster_faces @@ -477,6 +551,9 @@ def recluster_faces(): ) session.commit() + except Exception: + session.rollback() + raise finally: session.close() @@ -485,45 +562,98 @@ def recluster_faces(): return {'status': 'success', 'clusters': n_clusters, 'faces': len(face_rows)} -@shared_task(name='backfill_vision') -def backfill_vision(task: str | None = None, limit: int | None = None): +@shared_task(name='backfill_vision', bind=True, max_retries=10) +def backfill_vision(self, task: str | None = None, limit: int | None = None): """Queue vision tasks for photos that haven't been processed yet. Uses a sync DB connection to avoid asyncpg conflicts in Celery.""" + if not _vision_worker_ready(): + logger.info("Vision worker not ready yet — retrying in 30s") + raise self.retry(countdown=30) + model_name = settings.vision.embedder.name - # Newest-first ordering — matches regenerate_all_thumbnails so the - # whole ingestion pipeline sweeps the library top-down and the user - # sees recent photos fully-indexed long before the backlog drains. - # `taken_at` is the canonical capture timestamp (from EXIF, falls - # back to filesystem mtime in scan); `added_at` is the tie-breaker - # when taken_at is null. - sql = """ - SELECT p.id FROM photos p - LEFT JOIN embeddings e ON e.photo_id = p.id AND e.model = :model - WHERE e.photo_id IS NULL - AND p.processing_status = 'completed' - ORDER BY p.taken_at DESC NULLS LAST, p.added_at DESC NULLS LAST - """ + ordering = "ORDER BY p.taken_at DESC NULLS LAST, p.added_at DESC NULLS LAST" + limit_clause = " LIMIT :lim" if limit else "" + params: dict = {"model": model_name} if limit: - sql += f" LIMIT {limit}" + params["lim"] = int(limit) session = _get_sync_session() try: - result = session.execute(sa_text(sql), {"model": model_name}) - photo_ids = [row[0] for row in result.fetchall()] + # Each query finds photos missing a specific pipeline output so + # enabling a new processor after import still back-fills. + embed_ids = [] + if task in ('embed', None): + sql = f""" + SELECT p.id FROM photos p + LEFT JOIN embeddings e ON e.photo_id = p.id AND e.model = :model + WHERE e.photo_id IS NULL AND p.processing_status = 'completed' + {ordering}{limit_clause} + """ + embed_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()] + + ocr_ids = [] + if task in ('ocr', None) and settings.vision.ocr.enabled: + sql = f""" + SELECT p.id FROM photos p + LEFT JOIN ocr_text o ON o.photo_id = p.id + WHERE o.photo_id IS NULL AND p.processing_status = 'completed' + {ordering}{limit_clause} + """ + ocr_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()] + + detect_ids = [] + if task in ('detect', None) and settings.vision.detector.enabled: + sql = f""" + SELECT p.id FROM photos p + WHERE p.processing_status = 'completed' + AND NOT EXISTS ( + SELECT 1 FROM photo_tags pt WHERE pt.photo_id = p.id + AND pt.source = 'vision:yolov8n' + ) + {ordering}{limit_clause} + """ + detect_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()] + + face_ids = [] + if task in ('faces', None) and settings.vision.faces.enabled: + sql = f""" + SELECT p.id FROM photos p + LEFT JOIN face_embeddings fe ON fe.photo_id = p.id + WHERE fe.photo_id IS NULL AND p.processing_status = 'completed' + {ordering}{limit_clause} + """ + face_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()] + + classify_ids = [] + if task in ('classify', None) and settings.vision.classifier.enabled: + sql = f""" + SELECT p.id FROM photos p + WHERE p.processing_status = 'completed' + AND NOT EXISTS ( + SELECT 1 FROM photo_tags pt WHERE pt.photo_id = p.id + AND pt.source = 'vision:clip_classifier' + ) + {ordering}{limit_clause} + """ + classify_ids = [r[0] for r in session.execute(sa_text(sql), params).fetchall()] + except Exception: + session.rollback() + raise finally: session.close() - count = 0 - for pid in photo_ids: - if task == 'embed' or task is None: - embed_photo.delay(pid) - if task == 'ocr' or task is None: - ocr_photo.delay(pid) - if task == 'detect' or task is None: - detect_objects.delay(pid) - if task == 'faces' or task is None: - extract_faces.delay(pid) - count += 1 + # Dispatch — deduplicate across query results. + all_ids = set(embed_ids) | set(ocr_ids) | set(detect_ids) | set(face_ids) | set(classify_ids) + for pid in embed_ids: + embed_photo.delay(pid) + for pid in ocr_ids: + ocr_photo.delay(pid) + for pid in detect_ids: + detect_objects.delay(pid) + for pid in face_ids: + extract_faces.delay(pid) + for pid in classify_ids: + classify_content.delay(pid) - logger.info("Backfill queued %d photos for vision processing", count) - return {'status': 'queued', 'count': count} + logger.info("Backfill queued %d photos for vision processing", len(all_ids)) + return {'status': 'queued', 'count': len(all_ids)} diff --git a/docker-compose.yml b/docker-compose.yml index 816c580..7b0f86b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -116,6 +116,12 @@ services: condition: service_started db: condition: service_healthy + healthcheck: + test: ["CMD-SHELL", "celery -A app.tasks.celery inspect ping -d light@$$HOSTNAME 2>/dev/null | grep -q OK"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 120s networks: - mulita-network restart: unless-stopped @@ -197,6 +203,12 @@ services: # - driver: nvidia # count: all # capabilities: [gpu] + healthcheck: + test: ["CMD-SHELL", "celery -A app.tasks.celery inspect ping -d vision@$$HOSTNAME 2>/dev/null | grep -q OK"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 300s depends_on: redis: condition: service_started @@ -239,6 +251,11 @@ services: - mulita-network restart: unless-stopped command: redis-server --appendonly yes + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 5s + retries: 5 networks: mulita-network: