Dymas/ani-cli-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Pace bulk watchlist refresh requests

Browse Source
This commit is contained in:
Dymas
2026-05-17 21:15:53 +02:00
parent a8d7d3de00
commit ae090c954b
8 changed files with 100 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
queue_jobs.py
+28 -31
View File
@@ -2,7 +2,6 @@
"""Queue and background job services for ani-cli-web."""
import concurrent.futures
import json
import os
import shutil
@@ -33,7 +32,6 @@ from app_support import (
LOG_PERSIST_INTERVAL_SECONDS = 0.75
LOG_PERSIST_LINE_BATCH = 8
WATCHLIST_REFRESH_MAX_WORKERS = 4
JOB_STRUCTURED_COLUMNS = (
"show_id",
"title",
@@ -59,8 +57,9 @@ JOB_EXTRA_PAYLOAD_COLUMN = "payload"
class WatchlistRefreshJobs:
def __init__(self, store, start_worker=True):
def __init__(self, store, config_getter=None, start_worker=True):
self.store = store
self.config_getter = config_getter or (lambda: {})
self.lock = threading.RLock()
self.wakeup = threading.Event()
self.stop_event = threading.Event()
@@ -153,6 +152,14 @@ class WatchlistRefreshJobs:
self._upsert_job_conn(conn, job)
self._trim_history_conn(conn)
def _refresh_delay_seconds(self):
config = self.config_getter() or {}
try:
delay = int(str(config.get("watchlist_refresh_delay_seconds", 5)).strip() or "5")
except (TypeError, ValueError):
delay = 5
return max(0, min(300, delay))
def _load_jobs(self):
with self.lock, self._connect() as conn:
rows = conn.execute(
@@ -257,6 +264,7 @@ class WatchlistRefreshJobs:
source_name = str(job.get("source") or "manual").strip().lower() or "manual"
try:
show_ids = self.store.all_show_ids()
delay_seconds = self._refresh_delay_seconds()
started_at = now_iso()
with self.lock:
job["status"] = "running"
@@ -271,13 +279,14 @@ class WatchlistRefreshJobs:
if source_name == "scheduled":
self._stdout_log(f"Scheduled refresh started (job={job['id']}, total={len(show_ids)}).")
def refresh_one(show_id):
for index, show_id in enumerate(show_ids, start=1):
if self.stop_event.is_set():
interrupted = True
break
title = show_id
errored = False
skipped = False
error_message = ""
if self.stop_event.is_set():
return {"title": title, "errored": False, "skipped": True, "error": ""}
try:
item = self.store.refresh(show_id, include_thumbnail=False)
if item is None:
@@ -290,34 +299,22 @@ class WatchlistRefreshJobs:
errored = True
error_message = str(exc)
debug_log("watchlist.refresh_all.item_error", show_id=show_id, error=exc)
return {"title": title, "errored": errored, "skipped": skipped, "error": error_message}
max_workers = max(1, min(WATCHLIST_REFRESH_MAX_WORKERS, len(show_ids) or 1))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
with self.lock:
self.current_executor = executor
futures = [executor.submit(refresh_one, show_id) for show_id in show_ids]
for index, future in enumerate(concurrent.futures.as_completed(futures), start=1):
if self.stop_event.is_set():
job["completed"] = index
job["errors"] += 1 if errored else 0
job["current_title"] = title
job["updated_at"] = now_iso()
if errored and error_message:
job["last_error"] = error_message
if skipped:
job["message"] = f"Refreshing {index}/{len(show_ids)}: removed entry"
else:
job["message"] = f"Refreshing {index}/{len(show_ids)}: {title}"
self._save_job_locked(job)
if index < len(show_ids) and delay_seconds > 0:
if self.stop_event.wait(delay_seconds):
interrupted = True
executor.shutdown(wait=False, cancel_futures=True)
break
result = future.result()
title = result["title"]
with self.lock:
job["completed"] = index
job["errors"] += 1 if result["errored"] else 0
job["current_title"] = title
job["updated_at"] = now_iso()
if result["errored"] and result["error"]:
job["last_error"] = result["error"]
if result["skipped"]:
job["message"] = f"Refreshing {index}/{len(show_ids)}: removed entry"
else:
job["message"] = f"Refreshing {index}/{len(show_ids)}: {title}"
self._save_job_locked(job)
with self.lock:
self.current_executor = None
finished_at = now_iso()
with self.lock: