diff --git a/dockerfile b/dockerfile index 6f85541..a3c4f0d 100644 --- a/dockerfile +++ b/dockerfile @@ -24,4 +24,4 @@ COPY entrypoint.sh /entrypoint.sh RUN chmod +x /entrypoint.sh EXPOSE 5000 -ENTRYPOINT ["/entrypoint.sh"] \ No newline at end of file +ENTRYPOINT ["/entrypoint.sh \ No newline at end of file diff --git a/image_import_worker.py b/image_import_worker.py index 61ded1c..030f7f4 100644 --- a/image_import_worker.py +++ b/image_import_worker.py @@ -1,287 +1,354 @@ # image_import_worker.py import os import time +import logging +from dataclasses import dataclass +from typing import Callable, Optional + from sqlalchemy.exc import OperationalError from app import create_app from app.utils.image_importer import import_images_task +# ----------------------------- +# Config +# ----------------------------- 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) +CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "10")) # seconds (polling for flags) + +# Run a periodic import even without trigger.flag +IMPORT_EVERY_SECONDS = int(os.getenv("IMPORT_EVERY_SECONDS", str(8 * 60 * 60))) # default 8h + +# Verbosity controls (thumb regen) 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 +THUMBS_LOG_EVERY = int(os.getenv("THUMBS_LOG_EVERY", "50")) +THUMBS_BATCH_SIZE = int(os.getenv("THUMBS_BATCH_SIZE", "500")) + +MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3")) + +# ----------------------------- +# Logging +# ----------------------------- +logging.basicConfig( + level=os.getenv("LOG_LEVEL", "INFO").upper(), + format="%(asctime)s %(levelname)s %(message)s", +) +log = logging.getLogger("image-import-worker") app = create_app() -def monitor_and_import(): +@dataclass(frozen=True) +class FlagJob: + name: str + flag_path: str + + @property + def run_path(self) -> str: + return self.flag_path + ".run" + + @property + def err_path(self) -> str: + return self.flag_path + ".err" + + +def claim_job(job: FlagJob) -> bool: + """ + Atomically claim a job by renaming -> .run. + Returns True if claimed, False otherwise. + """ + try: + os.replace(job.flag_path, job.run_path) + log.info("[%s] Claimed job -> %s", job.name, job.run_path) + return True + except FileNotFoundError: + return False + except Exception as e: + log.warning("[%s] Could not claim job: %s", job.name, e) + return False + + +def complete_job(job: FlagJob) -> None: + """Remove the .run file on success.""" + try: + os.remove(job.run_path) + log.info("[%s] Cleared run flag.", job.name) + except FileNotFoundError: + # If something else cleaned it up, that’s fine. + log.info("[%s] Run flag already cleared.", job.name) + except Exception as e: + log.warning("[%s] Could not remove run flag: %s", job.name, e) + + +def fail_job(job: FlagJob) -> None: + """Move the .run file to .err on failure.""" + try: + os.replace(job.run_path, job.err_path) + log.error("[%s] Marked failure -> %s", job.name, job.err_path) + except FileNotFoundError: + log.error("[%s] Could not mark failure: run flag missing.", job.name) + except Exception as e: + log.error("[%s] Could not mark failure: %s", job.name, e) + + +def db_ping() -> None: + """Validate DB session connection quickly.""" + from sqlalchemy import text + from app import db + db.session.execute(text("SELECT 1")) + + +def recover_db_session(dispose_engine: bool = False) -> None: + """Rollback session and optionally dispose engine pool to force fresh connections.""" + try: + from app import db + db.session.rollback() + if dispose_engine: + db.engine.dispose() + except Exception: + pass + + +def run_with_retries( + job: FlagJob, + fn: Callable[[], None], + *, + dispose_engine_on_operational_error: bool = False, + max_retries: int = MAX_RETRIES, +) -> bool: + attempt = 0 + while attempt < max_retries: + try: + # Catch stale pool conns early + db_ping() + + fn() + return True + + except OperationalError as e: + attempt += 1 + log.warning("[%s] DB error (attempt %d/%d): %s", job.name, attempt, max_retries, e) + recover_db_session(dispose_engine=dispose_engine_on_operational_error) + + if attempt >= max_retries: + return False + + sleep_time = 2 ** attempt + log.info("[%s] Retrying in %ss...", job.name, sleep_time) + time.sleep(sleep_time) + + except Exception as e: + log.exception("[%s] Unexpected error: %s", job.name, e) + recover_db_session(dispose_engine=True) + return False + + return False + + +def do_import() -> None: + result = import_images_task(SOURCE_DIR, DEST_DIR) + log.info("[Import] %s", result) + + +def do_thumbnail_regen() -> None: + from pathlib import Path + from app import db + from app.models import ImageRecord + from app.utils.image_importer import generate_thumbnail, generate_video_thumbnail + + total = db.session.query(ImageRecord).count() + log.info("[Thumbs] Starting regen; ImageRecord count = %d", total) + + images_root = Path(DEST_DIR).resolve() + batch_size = THUMBS_BATCH_SIZE + + processed = 0 + writes = 0 + path_updates = 0 + 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")) + + 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: + log.info("[Thumbs] SKIP (not under %s): id=%s %s", images_root, img.id, img.filepath) + 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) + + if new_path: + writes += 1 + batch_writes += 1 + + if new_path and new_path != img.thumb_path: + img.thumb_path = new_path + path_updates += 1 + batch_updates += 1 + + if THUMBS_VERBOSE and (processed <= 10 or processed % THUMBS_LOG_EVERY == 0): + kind = "video" if is_video else "image" + log.info("[Thumbs] #%d (%s) id=%s name='%s' -> %s", processed, kind, img.id, img.filename, new_path or "None") + + except Exception as e: + failures += 1 + batch_failures += 1 + log.warning("[Thumbs] Failed for id=%s name='%s': %s", img.id, img.filename, e) + + db.session.commit() + last_id = rows[-1].id + db.session.expunge_all() + + bt = time.time() - bt0 + rate = batch_processed / bt if bt > 0 else 0 + log.info( + "[Thumbs] Batch %s-%s: processed=%d writes=%d db_updates=%d failures=%d (%.1f items/s) " + "— totals: processed=%d writes=%d db_updates=%d failures=%d (last_id=%d)", + batch_start_id, batch_end_id, + batch_processed, batch_writes, batch_updates, batch_failures, rate, + processed, writes, path_updates, failures, last_id, + ) + + elapsed = time.time() - t_start + overall_rate = processed / elapsed if elapsed > 0 else 0 + log.info( + "[Thumbs] COMPLETE in %.1fs — processed=%d writes=%d db_updates=%d failures=%d (%.1f items/s).", + elapsed, processed, writes, path_updates, failures, overall_rate, + ) + + +def handle_flag_job(job: FlagJob, fn: Callable[[], None], *, dispose_engine_on_operational_error: bool = False) -> bool: + """ + If exists, claim it and run fn with retries. Returns True if it did anything. + """ + if not os.path.exists(job.flag_path): + return False + + log.info("[%s] Flag detected: %s", job.name, job.flag_path) + + if not claim_job(job): + return True # it existed, but someone else may have claimed it; still counts as "did work" + + ok = run_with_retries( + job, + fn, + dispose_engine_on_operational_error=dispose_engine_on_operational_error, + ) + if ok: + complete_job(job) + else: + fail_job(job) + + return True + + +def monitor_and_import() -> None: + thumbnail_job = FlagJob(name="Thumbs", flag_path=THUMBNAIL_FLAG) + import_job = FlagJob(name="Import", flag_path=TRIGGER_FLAG) + + # next time we should run periodic import (monotonic avoids wall-clock jumps) + next_periodic_import = time.monotonic() + IMPORT_EVERY_SECONDS + with app.app_context(): - print("[Image Importer] Monitoring for trigger flags... (CHECK_INTERVAL=%ss)" % CHECK_INTERVAL, flush=True) + log.info( + "[Worker] Started. Poll=%ss, periodic import every=%ss (%.2fh).", + CHECK_INTERVAL, + IMPORT_EVERY_SECONDS, + IMPORT_EVERY_SECONDS / 3600.0, + ) 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) + # 1) Thumbnail regen is prioritized + did_work |= handle_flag_job(thumbnail_job, do_thumbnail_regen) - RUN_FLAG = THUMBNAIL_FLAG + ".run" - ERR_FLAG = THUMBNAIL_FLAG + ".err" + # 2) Flag-triggered import + did_work |= handle_flag_job( + import_job, + do_import, + dispose_engine_on_operational_error=True, # matches your original logic for import + ) - # 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 + # 3) Periodic import (every 8h by default) — uses the same .run lock convention + now = time.monotonic() + if now >= next_periodic_import: + next_periodic_import = now + IMPORT_EVERY_SECONDS - max_retries = 3 - attempt = 0 - - while attempt < max_retries: + # Don’t run if an import is already in progress/failed state + if os.path.exists(import_job.run_path): + log.info("[Import] Periodic run skipped (import already running).") + elif os.path.exists(import_job.err_path): + log.info("[Import] Periodic run skipped (import in error state: %s).", import_job.err_path) + else: + # Create/claim an import run lock without needing trigger.flag + # (We create the .run file atomically by replacing a temp file.) + tmp = import_job.run_path + ".tmp" 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 + with open(tmp, "w") as f: + f.write("periodic\n") + os.replace(tmp, import_job.run_path) + log.info("[Import] Periodic run claimed -> %s", import_job.run_path) - # 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 + ok = run_with_retries( + import_job, + do_import, + dispose_engine_on_operational_error=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) + if ok: + complete_job(import_job) else: - sleep_time = 2 ** attempt - print(f"[Image Importer] Retrying in {sleep_time} seconds...", flush=True) - time.sleep(sleep_time) + fail_job(import_job) + did_work = True except Exception as e: - print(f"[Image Importer] Unexpected error during import: {e}", flush=True) - # Make sure the session is usable next time + log.warning("[Import] Could not start periodic run: %s", e) try: - from app import db - db.session.rollback() - db.engine.dispose() + if os.path.exists(tmp): + os.remove(tmp) 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)