updated import script to have 8h standing import job

dockerfile

@@ -24,4 +24,4 @@ COPY entrypoint.sh /entrypoint.sh
 RUN chmod +x /entrypoint.sh
 
 EXPOSE 5000
-ENTRYPOINT ["/entrypoint.sh"]
+ENTRYPOINT ["/entrypoint.sh

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_LOG_EVERY   = int(os.getenv("THUMBS_LOG_EVERY", "50"))
-THUMBS_BATCH_SIZE  = int(os.getenv("THUMBS_BATCH_SIZE", "500")) # DB chunk size per commit
+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 <flag> -> <flag>.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 <flag> 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
 
-            # ------------------------------------------------------------------
+            # 1) Thumbnail regen is prioritized
-            # THUMBNAIL REGEN (PRIORITIZED) — chunked loop, detailed progress
+            did_work |= handle_flag_job(thumbnail_job, do_thumbnail_regen)
-            # ------------------------------------------------------------------
-            if os.path.exists(THUMBNAIL_FLAG):
-                print("[Image Importer] Thumbnail regeneration flag detected.", flush=True)
 
-                RUN_FLAG = THUMBNAIL_FLAG + ".run"
+            # 2) Flag-triggered import
-                ERR_FLAG = THUMBNAIL_FLAG + ".err"
+            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
+            # 3) Periodic import (every 8h by default) — uses the same .run lock convention
-                try:
+            now = time.monotonic()
-                    os.replace(THUMBNAIL_FLAG, RUN_FLAG)
+            if now >= next_periodic_import:
-                    print(f"[Image Importer] Claimed thumbnail job -> {RUN_FLAG}", flush=True)
+                next_periodic_import = now + IMPORT_EVERY_SECONDS
-                except Exception as e:
-                    print(f"[Image Importer] Could not claim thumbnail job: {e}", flush=True)
-                    time.sleep(CHECK_INTERVAL)
-                    continue
 
-                max_retries = 3
+                # Don’t run if an import is already in progress/failed state
-                attempt = 0
+                if os.path.exists(import_job.run_path):
-
+                    log.info("[Import] Periodic run skipped (import already running).")
-                while attempt < max_retries:
+                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
+                        with open(tmp, "w") as f:
-                        from sqlalchemy import text
+                            f.write("periodic\n")
-                        from app import db
+                        os.replace(tmp, import_job.run_path)
-                        from app.models import ImageRecord
+                        log.info("[Import] Periodic run claimed -> %s", import_job.run_path)
-                        from app.utils.image_importer import generate_thumbnail, generate_video_thumbnail
 
-                        # Early diagnostics
+                        ok = run_with_retries(
-                        db.session.execute(text("SELECT 1"))
+                            import_job,
-                        total = db.session.query(ImageRecord).count()
+                            do_import,
-                        print(f"[Thumbs] Starting regen; ImageRecord count = {total}", flush=True)
+                            dispose_engine_on_operational_error=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
                         )
-
+                        if ok:
-                        # Success — remove the .run flag and exit
+                            complete_job(import_job)
-                        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
+                            fail_job(import_job)
-                            print(f"[Image Importer] Retrying in {sleep_time} seconds...", flush=True)
-                            time.sleep(sleep_time)
 
+                        did_work = True
                     except Exception as e:
-                        print(f"[Image Importer] Unexpected error during import: {e}", flush=True)
+                        log.warning("[Import] Could not start periodic run: %s", e)
-                        # Make sure the session is usable next time
                         try:
-                            from app import db
+                            if os.path.exists(tmp):
-                            db.session.rollback()
+                                os.remove(tmp)
-                            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)