correcting accidental isses

2025-12-26 09:41:17 -05:00
parent 81aa4d5117
commit 5f568f43bc
1 changed files with 144 additions and 0 deletions
@@ -305,3 +305,147 @@ def do_thumbnail_regen() -> None:
        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,
+    )
+
+
+# -------------------------------------------------------------------
+# Orchestration helpers
+# -------------------------------------------------------------------
+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 (including another worker claiming it).
+    """
+    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):
+        # It existed, but someone else may have claimed it.
+        return True
+
+    ok, err_text = run_with_retries(
+        job,
+        fn,
+        dispose_engine_on_operational_error=dispose_engine_on_operational_error,
+    )
+    if ok:
+        complete_job(job)
+    else:
+        fail_job(job, error_text=err_text)
+
+    return True
+
+
+# -------------------------------------------------------------------
+# Main loop
+# -------------------------------------------------------------------
+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():
+        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 (priority)
+            did_work |= handle_flag_job(thumbnail_job, do_thumbnail_regen)
+
+            # 2) Flag-triggered import
+            did_work |= handle_flag_job(
+                import_job,
+                do_import,
+                dispose_engine_on_operational_error=True,
+            )
+
+            # 3) Periodic import (every IMPORT_EVERY_SECONDS)
+            now = time.monotonic()
+            if IMPORT_EVERY_SECONDS > 0 and now >= next_periodic_import:
+                next_periodic_import = now + IMPORT_EVERY_SECONDS
+
+                # Skip if an import is already in progress or in error
+                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 a .run lock for periodic run (no trigger.flag required)
+                    tmp = import_job.run_path + ".tmp"
+                    try:
+                        with open(tmp, "w", encoding="utf-8") as f:
+                            f.write("periodic\n")
+                        os.replace(tmp, import_job.run_path)
+                        log.info("[Import] Periodic run claimed -> %s", import_job.run_path)
+
+                        ok, err_text = run_with_retries(
+                            import_job,
+                            do_import,
+                            dispose_engine_on_operational_error=True,
+                        )
+                        if ok:
+                            complete_job(import_job)
+                        else:
+                            fail_job(import_job, error_text=err_text)
+
+                        did_work = True
+                    except Exception as e:
+                        log.warning("[Import] Could not start periodic run: %s", e)
+                        try:
+                            if os.path.exists(tmp):
+                                os.remove(tmp)
+                        except Exception:
+                            pass
+
+            if not did_work:
+                time.sleep(CHECK_INTERVAL)
+
+
+if __name__ == "__main__":
+    monitor_and_import()