# image_import_worker.py import os import time from sqlalchemy.exc import OperationalError from app import create_app from app.utils.image_importer import import_images_task THUMBNAIL_FLAG = "/import/thumbnail.flag" TRIGGER_FLAG = "/import/trigger.flag" SOURCE_DIR = "/import" DEST_DIR = "/images" CHECK_INTERVAL = 10 # seconds # Verbosity controls (set via environment) THUMBS_VERBOSE = os.getenv("THUMBS_VERBOSE", "0").lower() in ("1", "true", "yes") THUMBS_LOG_EVERY = int(os.getenv("THUMBS_LOG_EVERY", "50")) # log every N images if verbose THUMBS_BATCH_SIZE = int(os.getenv("THUMBS_BATCH_SIZE", "500")) # DB chunk size per commit app = create_app() def monitor_and_import(): with app.app_context(): print("[Image Importer] Monitoring for trigger flags... (CHECK_INTERVAL=%ss)" % CHECK_INTERVAL, flush=True) while True: did_work = False # ------------------------------------------------------------------ # THUMBNAIL REGEN (PRIORITIZED) — chunked loop, detailed progress # ------------------------------------------------------------------ if os.path.exists(THUMBNAIL_FLAG): print("[Image Importer] Thumbnail regeneration flag detected.", flush=True) RUN_FLAG = THUMBNAIL_FLAG + ".run" ERR_FLAG = THUMBNAIL_FLAG + ".err" # Claim atomically so only one worker runs try: os.replace(THUMBNAIL_FLAG, RUN_FLAG) print(f"[Image Importer] Claimed thumbnail job -> {RUN_FLAG}", flush=True) except Exception as e: print(f"[Image Importer] Could not claim thumbnail job: {e}", flush=True) time.sleep(CHECK_INTERVAL) continue max_retries = 3 attempt = 0 while attempt < max_retries: try: from pathlib import Path from sqlalchemy import text from app import db from app.models import ImageRecord from app.utils.image_importer import generate_thumbnail, generate_video_thumbnail # Early diagnostics db.session.execute(text("SELECT 1")) total = db.session.query(ImageRecord).count() print(f"[Thumbs] Starting regen; ImageRecord count = {total}", flush=True) images_root = Path("/images").resolve() batch_size = THUMBS_BATCH_SIZE processed = 0 # total processed rows writes = 0 # files written/overwritten path_updates = 0 # DB thumb_path changes failures = 0 last_id = 0 t_start = time.time() while True: rows = ( ImageRecord.query .filter(ImageRecord.id > last_id) .order_by(ImageRecord.id.asc()) .limit(batch_size) .all() ) if not rows: break batch_start_id = rows[0].id batch_end_id = rows[-1].id batch_processed = 0 batch_writes = 0 batch_updates = 0 batch_failures = 0 bt0 = time.time() for img in rows: try: processed += 1 batch_processed += 1 ext = (img.format or "").lower() is_video = ext in ("mp4", "mov") or img.filepath.lower().endswith((".mp4", ".mov")) # Determine output path for videos if needed if is_video: out_path = img.thumb_path if not out_path: try: rel = Path(img.filepath).resolve().relative_to(images_root) except Exception: batch_failures += 1 failures += 1 if THUMBS_VERBOSE: print(f"[Thumbs] SKIP (not under /images): id={img.id} {img.filepath}", flush=True) continue out_path = str((images_root / "thumbs" / rel).with_suffix(".jpg")) new_path = generate_video_thumbnail(img.filepath, out_path) else: new_path = generate_thumbnail(img.filepath, overwrite=True) # Count file writes: our helpers return the path on success if new_path: writes += 1 batch_writes += 1 # Update DB if path changed if new_path and new_path != img.thumb_path: img.thumb_path = new_path path_updates += 1 batch_updates += 1 # Optional per-file log if THUMBS_VERBOSE and (processed <= 10 or processed % THUMBS_LOG_EVERY == 0): kind = "video" if is_video else "image" print(f"[Thumbs] #{processed} ({kind}) id={img.id} name='{img.filename}' -> {new_path or 'None'}", flush=True) except Exception as e: failures += 1 batch_failures += 1 print(f"[Thumbs] Failed for id={img.id} name='{img.filename}': {e}", flush=True) # Commit AFTER each chunk; safe for Postgres (no server-side cursor) db.session.commit() last_id = rows[-1].id db.session.expunge_all() # keep memory stable # Batch progress bt = time.time() - bt0 rate = batch_processed / bt if bt > 0 else 0 print( f"[Thumbs] Batch {batch_start_id}-{batch_end_id}: " f"processed={batch_processed}, writes={batch_writes}, " f"db_updates={batch_updates}, failures={batch_failures} " f"({rate:.1f} items/s) — totals: processed={processed}, writes={writes}, " f"db_updates={path_updates}, failures={failures} (last_id={last_id})", flush=True ) # Final summary elapsed = time.time() - t_start overall_rate = processed / elapsed if elapsed > 0 else 0 print( f"[Image Importer] Thumbnail regen COMPLETE in {elapsed:.1f}s — " f"processed={processed}, writes={writes}, db_updates={path_updates}, failures={failures} " f"({overall_rate:.1f} items/s).", flush=True ) # Success — remove the .run flag and exit try: os.remove(RUN_FLAG) print("[Image Importer] Thumbnail .run flag cleared.", flush=True) except Exception as e: print(f"[Image Importer] Could not remove .run flag: {e}", flush=True) break # success except OperationalError as e: attempt += 1 print(f"[Image Importer] DB error during thumbnail regen (attempt {attempt}): {e}", flush=True) # Recover the session for the next attempt try: from app import db db.session.rollback() except Exception: pass time.sleep(2 ** attempt) if attempt >= max_retries: try: os.replace(RUN_FLAG, ERR_FLAG) print(f"[Image Importer] Moved thumbnail flag to {ERR_FLAG}", flush=True) except Exception as re: print(f"[Image Importer] Could not mark error flag: {re}", flush=True) except Exception as e: print(f"[Image Importer] Unexpected error during thumbnail regen: {e}", flush=True) try: os.replace(RUN_FLAG, ERR_FLAG) print(f"[Image Importer] Moved thumbnail flag to {ERR_FLAG}", flush=True) except Exception as re: print(f"[Image Importer] Could not mark error flag: {re}", flush=True) break did_work = True # handled a job (success or fail) # ------------------------------------------------------------------ # IMPORT TRIGGER (robust: claim -> run -> clear/err; rollback on fail) # ------------------------------------------------------------------ if os.path.exists(TRIGGER_FLAG): print("[Image Importer] Import flag found. Starting import...", flush=True) RUN_FLAG = TRIGGER_FLAG + ".run" ERR_FLAG = TRIGGER_FLAG + ".err" # Claim atomically so only one worker takes it try: os.replace(TRIGGER_FLAG, RUN_FLAG) print(f"[Image Importer] Claimed import job -> {RUN_FLAG}", flush=True) except Exception as e: print(f"[Image Importer] Could not claim import job: {e}", flush=True) time.sleep(CHECK_INTERVAL) continue max_retries = 3 attempt = 0 while attempt < max_retries: try: from sqlalchemy import text from app import db # Fast DB ping to catch stale pool conns db.session.execute(text("SELECT 1")) # Run the import result = import_images_task(SOURCE_DIR, DEST_DIR) print(f"[Image Importer] {result}", flush=True) # Success — remove the .run flag and exit try: os.remove(RUN_FLAG) print("[Image Importer] Import .run flag cleared.", flush=True) except Exception as e: print(f"[Image Importer] Could not remove import .run flag: {e}", flush=True) break # success except OperationalError as e: attempt += 1 print(f"[Image Importer] DB error during import (attempt {attempt}): {e}", flush=True) # IMPORTANT: rollback failed transaction & drop pool to force clean conns try: from app import db db.session.rollback() db.engine.dispose() except Exception: pass if attempt >= max_retries: try: os.replace(RUN_FLAG, ERR_FLAG) print(f"[Image Importer] Moved import flag to {ERR_FLAG}", flush=True) except Exception as re: print(f"[Image Importer] Could not mark import error flag: {re}", flush=True) else: sleep_time = 2 ** attempt print(f"[Image Importer] Retrying in {sleep_time} seconds...", flush=True) time.sleep(sleep_time) except Exception as e: print(f"[Image Importer] Unexpected error during import: {e}", flush=True) # Make sure the session is usable next time try: from app import db db.session.rollback() db.engine.dispose() except Exception: pass try: os.replace(RUN_FLAG, ERR_FLAG) print(f"[Image Importer] Moved import flag to {ERR_FLAG}", flush=True) except Exception as re: print(f"[Image Importer] Could not mark import error flag: {re}", flush=True) break did_work = True # ------------------------------------------------------------------ # IDLE SLEEP # ------------------------------------------------------------------ if not did_work: time.sleep(CHECK_INTERVAL) if __name__ == "__main__": monitor_and_import()