Add Jellyfin handoff progress job
This commit is contained in:
Dymas
+321
@@ -25,6 +25,7 @@ from app_support import (
|
||||
job_output_dir,
|
||||
job_staging_dir,
|
||||
load_json,
|
||||
normalize_config,
|
||||
now_iso,
|
||||
printable_command,
|
||||
send_discord_webhook_event,
|
||||
@@ -518,6 +519,326 @@ class WatchlistAutoRefreshScheduler:
|
||||
self.wakeup.clear()
|
||||
|
||||
|
||||
class JellyfinSyncJobs:
|
||||
CONFIG_KEYS = (
|
||||
"download_dir",
|
||||
"jellyfin_sync_enabled",
|
||||
"jellyfin_tv_dir",
|
||||
"jellyfin_movie_dir",
|
||||
"discord_webhook_url",
|
||||
"discord_webhook_events",
|
||||
)
|
||||
|
||||
def __init__(self, store, sync_fn, config_getter=None, start_worker=True):
|
||||
self.store = store
|
||||
self.sync_fn = sync_fn
|
||||
self.config_getter = config_getter or (lambda: {})
|
||||
self.lock = threading.RLock()
|
||||
self.wakeup = threading.Event()
|
||||
self.stop_event = threading.Event()
|
||||
self.pending = []
|
||||
self.jobs = []
|
||||
self.current_job_id = None
|
||||
self._init_db()
|
||||
self._restore_interrupted_jobs()
|
||||
self._load_jobs()
|
||||
self.worker = None
|
||||
if start_worker:
|
||||
self.ensure_worker()
|
||||
|
||||
def _connect(self):
|
||||
conn = sqlite3.connect(STATE_DB_PATH, timeout=30)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
def ensure_worker(self):
|
||||
with self.lock:
|
||||
if self.worker is not None and self.worker.is_alive():
|
||||
return
|
||||
self.stop_event.clear()
|
||||
self.worker = threading.Thread(target=self._run, daemon=True)
|
||||
self.worker.start()
|
||||
|
||||
def shutdown(self, wait=False):
|
||||
self.stop_event.set()
|
||||
self.wakeup.set()
|
||||
worker = self.worker
|
||||
if wait and worker is not None and worker.is_alive():
|
||||
worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT_SECONDS)
|
||||
return worker is None or not worker.is_alive()
|
||||
|
||||
def _init_db(self):
|
||||
STATE_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS jellyfin_sync_jobs (
|
||||
id TEXT PRIMARY KEY,
|
||||
status TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
payload TEXT NOT NULL
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_jellyfin_sync_created_at ON jellyfin_sync_jobs(created_at)")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_jellyfin_sync_status ON jellyfin_sync_jobs(status)")
|
||||
|
||||
def _job_from_row(self, row):
|
||||
payload = row["payload"] if isinstance(row, sqlite3.Row) else row[0]
|
||||
return json.loads(payload)
|
||||
|
||||
def _upsert_job_conn(self, conn, job):
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO jellyfin_sync_jobs (id, status, created_at, updated_at, payload)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
status = excluded.status,
|
||||
updated_at = excluded.updated_at,
|
||||
payload = excluded.payload
|
||||
""",
|
||||
(
|
||||
job["id"],
|
||||
job["status"],
|
||||
job["created_at"],
|
||||
job["updated_at"],
|
||||
json.dumps(job, sort_keys=True),
|
||||
),
|
||||
)
|
||||
|
||||
def _trim_history_conn(self, conn):
|
||||
rows = conn.execute(
|
||||
"SELECT id FROM jellyfin_sync_jobs ORDER BY created_at DESC LIMIT -1 OFFSET 10"
|
||||
).fetchall()
|
||||
if rows:
|
||||
conn.executemany("DELETE FROM jellyfin_sync_jobs WHERE id = ?", [(row["id"],) for row in rows])
|
||||
|
||||
def _save_job_locked(self, job):
|
||||
with self._connect() as conn:
|
||||
self._upsert_job_conn(conn, job)
|
||||
self._trim_history_conn(conn)
|
||||
|
||||
def _load_jobs(self):
|
||||
with self.lock, self._connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT payload FROM jellyfin_sync_jobs ORDER BY created_at DESC LIMIT 10"
|
||||
).fetchall()
|
||||
self.jobs = [self._job_from_row(row) for row in rows]
|
||||
|
||||
def _restore_interrupted_jobs(self):
|
||||
with self.lock, self._connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT payload FROM jellyfin_sync_jobs WHERE status IN ('pending', 'running')"
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
job = self._job_from_row(row)
|
||||
job["status"] = "interrupted"
|
||||
job["finished_at"] = now_iso()
|
||||
job["updated_at"] = job["finished_at"]
|
||||
job["current_title"] = None
|
||||
job["current_file"] = None
|
||||
job["message"] = "Interrupted by web app restart."
|
||||
self._upsert_job_conn(conn, job)
|
||||
self._trim_history_conn(conn)
|
||||
|
||||
def _copy_job(self, job):
|
||||
return json.loads(json.dumps(job))
|
||||
|
||||
def _find_locked(self, job_id):
|
||||
for job in self.jobs:
|
||||
if job["id"] == job_id:
|
||||
return job
|
||||
return None
|
||||
|
||||
def _job_config(self, config):
|
||||
normalized = normalize_config(config or self.config_getter() or {})
|
||||
return {key: normalized.get(key) for key in self.CONFIG_KEYS}
|
||||
|
||||
def status(self):
|
||||
with self.lock:
|
||||
job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None)
|
||||
if job is None:
|
||||
job = self.jobs[0] if self.jobs else None
|
||||
active = bool(job and job["status"] in {"pending", "running"})
|
||||
return {"job": self._copy_job(job) if job else None, "running": active}
|
||||
|
||||
def start(self, config=None, source="manual"):
|
||||
source_name = str(source or "manual").strip().lower() or "manual"
|
||||
with self.lock:
|
||||
job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None)
|
||||
if job is not None:
|
||||
return {
|
||||
"message": "Jellyfin handoff already queued." if job["status"] == "pending" else "Jellyfin handoff already running.",
|
||||
"job": self._copy_job(job),
|
||||
"created": False,
|
||||
}
|
||||
now = now_iso()
|
||||
job = {
|
||||
"id": os.urandom(16).hex(),
|
||||
"status": "pending",
|
||||
"source": source_name,
|
||||
"config": self._job_config(config),
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
"started_at": None,
|
||||
"finished_at": None,
|
||||
"total": 0,
|
||||
"completed": 0,
|
||||
"moved": 0,
|
||||
"errors": 0,
|
||||
"moved_files": 0,
|
||||
"current_title": None,
|
||||
"current_file": None,
|
||||
"message": "Queued Jellyfin handoff.",
|
||||
"items": [],
|
||||
}
|
||||
self.jobs.insert(0, job)
|
||||
self.jobs = self.jobs[:10]
|
||||
self._save_job_locked(job)
|
||||
self.pending.append(job["id"])
|
||||
self.wakeup.set()
|
||||
return {"message": "Jellyfin handoff started.", "job": self._copy_job(job), "created": True}
|
||||
|
||||
def _next_pending(self):
|
||||
with self.lock:
|
||||
while self.pending:
|
||||
job_id = self.pending.pop(0)
|
||||
job = self._find_locked(job_id)
|
||||
if job is not None:
|
||||
self.current_job_id = job_id
|
||||
return job
|
||||
return None
|
||||
|
||||
def _run(self):
|
||||
while not self.stop_event.is_set():
|
||||
job = self._next_pending()
|
||||
if not job:
|
||||
self.wakeup.wait(2)
|
||||
self.wakeup.clear()
|
||||
continue
|
||||
self._run_job(job)
|
||||
|
||||
def _run_job(self, job):
|
||||
interrupted = False
|
||||
result_items = []
|
||||
try:
|
||||
active_config = self._job_config(job.get("config") or self.config_getter() or {})
|
||||
show_ids = self.store.all_show_ids()
|
||||
started_at = now_iso()
|
||||
with self.lock:
|
||||
job["status"] = "running"
|
||||
job["started_at"] = started_at
|
||||
job["updated_at"] = started_at
|
||||
job["total"] = len(show_ids)
|
||||
job["completed"] = 0
|
||||
job["moved"] = 0
|
||||
job["errors"] = 0
|
||||
job["moved_files"] = 0
|
||||
job["current_title"] = None
|
||||
job["current_file"] = None
|
||||
job["items"] = []
|
||||
job["message"] = f"Moving 0/{len(show_ids)} watchlist entries to Jellyfin."
|
||||
self._save_job_locked(job)
|
||||
|
||||
for index, show_id in enumerate(show_ids, start=1):
|
||||
if self.stop_event.is_set():
|
||||
interrupted = True
|
||||
break
|
||||
try:
|
||||
item = self.store.get(show_id)
|
||||
except KeyError:
|
||||
item = None
|
||||
if item is None:
|
||||
with self.lock:
|
||||
job["completed"] = index
|
||||
job["updated_at"] = now_iso()
|
||||
job["message"] = f"Moving {index}/{len(show_ids)}: removed entry"
|
||||
self._save_job_locked(job)
|
||||
continue
|
||||
|
||||
title = str(item.get("title") or show_id)
|
||||
|
||||
def progress(update):
|
||||
with self.lock:
|
||||
job["current_title"] = title
|
||||
job["current_file"] = update.get("current_file")
|
||||
job["updated_at"] = now_iso()
|
||||
file_total = int(update.get("file_total") or 0)
|
||||
file_index = int(update.get("file_index") or 0)
|
||||
if file_total > 0:
|
||||
job["message"] = (
|
||||
f"Moving {index}/{len(show_ids)}: {title} "
|
||||
f"({file_index}/{file_total} files)"
|
||||
)
|
||||
else:
|
||||
job["message"] = f"Moving {index}/{len(show_ids)}: {title}"
|
||||
self._save_job_locked(job)
|
||||
|
||||
result = self.sync_fn(item, automatic=False, config=active_config, progress_fn=progress)
|
||||
result_items.append(
|
||||
{
|
||||
"show_id": item.get("show_id"),
|
||||
"title": item.get("title"),
|
||||
"media_type": item.get("media_type"),
|
||||
"moved": bool(result.get("moved")),
|
||||
"reason": result.get("reason"),
|
||||
"target": result.get("target"),
|
||||
}
|
||||
)
|
||||
moved_files = len(result.get("moved_files") or []) if isinstance(result.get("moved_files"), list) else 0
|
||||
errored = str(result.get("reason") or "").strip() in {"error", "target_missing", "source_missing", "target_incomplete"}
|
||||
with self.lock:
|
||||
job["completed"] = index
|
||||
job["moved"] += 1 if result.get("moved") else 0
|
||||
job["moved_files"] += moved_files
|
||||
job["errors"] += 1 if errored else 0
|
||||
job["current_title"] = title
|
||||
job["current_file"] = None
|
||||
job["updated_at"] = now_iso()
|
||||
job["items"] = result_items[-20:]
|
||||
if errored and result.get("error"):
|
||||
job["last_error"] = str(result.get("error"))
|
||||
outcome = "moved" if result.get("moved") else str(result.get("reason") or "skipped")
|
||||
job["message"] = f"Moving {index}/{len(show_ids)}: {title} ({outcome})"
|
||||
self._save_job_locked(job)
|
||||
|
||||
finished_at = now_iso()
|
||||
with self.lock:
|
||||
job["status"] = "interrupted" if interrupted else "done"
|
||||
job["finished_at"] = finished_at
|
||||
job["updated_at"] = finished_at
|
||||
job["current_title"] = None
|
||||
job["current_file"] = None
|
||||
job["items"] = result_items[-20:]
|
||||
if interrupted:
|
||||
job["message"] = f"Jellyfin handoff interrupted after {job['completed']} entries."
|
||||
elif job["errors"]:
|
||||
job["message"] = (
|
||||
f"Moved {job['moved']} of {job['completed']} processed watchlist entries to Jellyfin "
|
||||
f"with {job['errors']} issues."
|
||||
)
|
||||
else:
|
||||
job["message"] = (
|
||||
f"Moved {job['moved']} watchlist entr{'y' if job['moved'] == 1 else 'ies'} to Jellyfin libraries."
|
||||
)
|
||||
self._save_job_locked(job)
|
||||
except Exception as exc:
|
||||
finished_at = now_iso()
|
||||
with self.lock:
|
||||
job["status"] = "failed"
|
||||
job["finished_at"] = finished_at
|
||||
job["updated_at"] = finished_at
|
||||
job["current_title"] = None
|
||||
job["current_file"] = None
|
||||
job["message"] = f"Jellyfin handoff failed: {exc}"
|
||||
job["last_error"] = str(exc)
|
||||
self._save_job_locked(job)
|
||||
finally:
|
||||
with self.lock:
|
||||
self.current_job_id = None
|
||||
|
||||
|
||||
class DownloadQueue:
|
||||
def __init__(self, config_getter, watchlist_sync_fn=None, job_prepare_fn=None, start_worker=True):
|
||||
self.config_getter = config_getter
|
||||
|
||||
Reference in New Issue
Block a user