diff --git a/CHANGELOG.md b/CHANGELOG.md index d8ce129..c7815ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## 0.2.0 - 2026-05-11 + +- Replaced JSON queue storage with SQLite-backed queue persistence. +- Added automatic migration from the previous `queue.json` queue file. +- Limited the queue view to 10 jobs per page with previous and next controls. +- Added bulk queue actions to remove finished jobs and retry all failed jobs. + ## 0.1.2 - 2026-05-11 - Updated finalized download layout to save media under `ANI_CLI_DOWNLOAD_DIR/anime_name/Season /`. diff --git a/README.md b/README.md index ade2d24..454a132 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A small local web UI for a system-wide `ani-cli` install. -Current version: `0.1.2` +Current version: `0.2.0` ## Project Layout @@ -29,6 +29,8 @@ http://127.0.0.1:8421/ The app searches anime, queues downloads, supports original or dubbed mode, quality, episode ranges, editable library names, season numbers, and a default download folder. +The queue is stored in SQLite, shown 10 jobs at a time, and includes bulk actions for removing finished jobs and retrying failed jobs. + Downloads are staged in the web app state directory first. After `ani-cli` finishes, the wrapper moves them into a Plex/Jellyfin-friendly layout: ```text @@ -44,6 +46,6 @@ ANI_CLI_DOWNLOAD_DIR/anime_name/Season /anime_name - S01E01.mp4 - `ANI_CLI_WEB_HOST`: server host, default `127.0.0.1`. - `ANI_CLI_WEB_PORT`: server port, default `8421`. -Settings and the queue are stored under XDG config/state directories when writable, with a local `.ani-cli-web/` fallback inside this project. +Settings, queue data, and staging files are stored under XDG config/state directories when writable, with a local `.ani-cli-web/` fallback inside this project. Existing `queue.json` data is migrated into SQLite automatically. Runtime folders such as `.ani-cli-web/`, `downloads/`, and Python bytecode cache directories at any depth are ignored by git. diff --git a/VERSION b/VERSION index d917d3e..0ea3a94 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.1.2 +0.2.0 diff --git a/app.py b/app.py index 0156ca3..3f95bab 100644 --- a/app.py +++ b/app.py @@ -4,6 +4,7 @@ import os import re import shutil import signal +import sqlite3 import subprocess import threading import time @@ -20,7 +21,7 @@ from uuid import uuid4 PROJECT_ROOT = Path(__file__).resolve().parent ANI_CLI = os.environ.get("ANI_CLI_BIN") or shutil.which("ani-cli") or "ani-cli" APP_NAME = "ani-cli-web" -VERSION = "0.1.2" +VERSION = "0.2.0" ALLANIME_BASE = "allanime.day" ALLANIME_API = f"https://api.{ALLANIME_BASE}" ALLANIME_REFERER = "https://allmanga.to" @@ -71,6 +72,7 @@ def app_dir(kind): CONFIG_PATH = app_dir("config") / "config.json" QUEUE_PATH = app_dir("state") / "queue.json" +QUEUE_DB_PATH = app_dir("state") / "queue.sqlite3" STAGING_ROOT = app_dir("state") / "staging" @@ -202,41 +204,114 @@ class DownloadQueue: def __init__(self): self.lock = threading.RLock() self.wakeup = threading.Event() - self.jobs = load_json(QUEUE_PATH, []) self.current_process = None self.current_job_id = None + self._init_db() + self._migrate_json_queue() self._restore_interrupted_jobs() self.worker = threading.Thread(target=self._run, daemon=True) self.worker.start() + def _connect(self): + conn = sqlite3.connect(QUEUE_DB_PATH, timeout=30) + conn.row_factory = sqlite3.Row + return conn + + def _init_db(self): + QUEUE_DB_PATH.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS 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_jobs_status ON jobs(status)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_created_at ON jobs(created_at)") + + def _migrate_json_queue(self): + if not QUEUE_PATH.exists(): + return + jobs = load_json(QUEUE_PATH, []) + if not jobs: + QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated")) + return + with self.lock, self._connect() as conn: + existing = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"] + if existing: + return + for job in jobs: + if isinstance(job, dict) and job.get("id"): + self._upsert_job_conn(conn, job) + QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated")) + def _restore_interrupted_jobs(self): - changed = False - for job in self.jobs: - if job.get("status") == "running": + with self.lock, self._connect() as conn: + rows = conn.execute("SELECT payload FROM jobs WHERE status = 'running'").fetchall() + for row in rows: + job = self._job_from_row(row) job["status"] = "pending" job["pid"] = None job.setdefault("log", []).append("Restarted after web app restart.") - changed = True - if changed: - self._save_locked() + self._upsert_job_conn(conn, job) - def _save_locked(self): - clean_jobs = [] - for job in self.jobs: - clean = dict(job) - clean.pop("process", None) - clean_jobs.append(clean) - write_json(QUEUE_PATH, clean_jobs) + def _job_from_row(self, row): + payload = row["payload"] if isinstance(row, sqlite3.Row) else row[0] + return json.loads(payload) - def list(self): - with self.lock: - return [dict(job) for job in self.jobs] + def _upsert_job_conn(self, conn, job): + clean = dict(job) + clean.pop("process", None) + conn.execute( + """ + INSERT INTO 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 + """, + ( + clean["id"], + clean.get("status", "pending"), + clean.get("created_at") or now_iso(), + clean.get("updated_at") or now_iso(), + json.dumps(clean, sort_keys=True), + ), + ) + + def _save_job_locked(self, job): + with self._connect() as conn: + self._upsert_job_conn(conn, job) + + def list(self, page=1, per_page=10): + page = max(1, int(page or 1)) + per_page = min(50, max(1, int(per_page or 10))) + offset = (page - 1) * per_page + with self.lock, self._connect() as conn: + total = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"] + rows = conn.execute( + "SELECT payload FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?", + (per_page, offset), + ).fetchall() + pages = max(1, (total + per_page - 1) // per_page) + return { + "jobs": [self._job_from_row(row) for row in rows], + "page": page, + "per_page": per_page, + "total": total, + "pages": pages, + } def add(self, payload): job = build_job(payload) with self.lock: - self.jobs.append(job) - self._save_locked() + self._save_job_locked(job) self.wakeup.set() return job @@ -252,17 +327,36 @@ class DownloadQueue: job["finished_at"] = None job["updated_at"] = now_iso() job["log"] = [] - self._save_locked() + self._save_job_locked(job) self.wakeup.set() return job + def retry_all_failed(self): + count = 0 + with self.lock, self._connect() as conn: + rows = conn.execute("SELECT payload FROM jobs WHERE status = 'failed'").fetchall() + for row in rows: + job = self._job_from_row(row) + job["status"] = "pending" + job["exit_code"] = None + job["pid"] = None + job["started_at"] = None + job["finished_at"] = None + job["updated_at"] = now_iso() + job["log"] = [] + self._upsert_job_conn(conn, job) + count += 1 + if count: + self.wakeup.set() + return {"ok": True, "count": count} + def cancel(self, job_id): with self.lock: job = self._find(job_id) if job["status"] == "pending": job["status"] = "canceled" job["updated_at"] = now_iso() - self._save_locked() + self._save_job_locked(job) return job if job["status"] != "running": return job @@ -273,7 +367,7 @@ class DownloadQueue: self.current_process.terminate() job["cancel_requested"] = True job["updated_at"] = now_iso() - self._save_locked() + self._save_job_locked(job) return job raise ValueError("Could not find the running process") @@ -282,27 +376,30 @@ class DownloadQueue: job = self._find(job_id) if job["status"] == "running": raise ValueError("Running jobs must be canceled before removing") - self.jobs = [item for item in self.jobs if item["id"] != job_id] - self._save_locked() + with self._connect() as conn: + conn.execute("DELETE FROM jobs WHERE id = ?", (job_id,)) return {"ok": True} def clear_finished(self): - with self.lock: - self.jobs = [job for job in self.jobs if job.get("status") in {"pending", "running"}] - self._save_locked() - return {"ok": True} + with self.lock, self._connect() as conn: + cursor = conn.execute("DELETE FROM jobs WHERE status IN ('done', 'canceled')") + return {"ok": True, "count": cursor.rowcount} def _find(self, job_id): - for job in self.jobs: - if job.get("id") == job_id: - return job + with self._connect() as conn: + row = conn.execute("SELECT payload FROM jobs WHERE id = ?", (job_id,)).fetchone() + if row: + return self._job_from_row(row) raise KeyError("Job not found") def _next_pending(self): with self.lock: - for job in self.jobs: - if job.get("status") == "pending": - return job + with self._connect() as conn: + row = conn.execute( + "SELECT payload FROM jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1" + ).fetchone() + if row: + return self._job_from_row(row) return None def _append_log(self, job, line): @@ -313,7 +410,7 @@ class DownloadQueue: job.setdefault("log", []).append(text) job["log"] = job["log"][-MAX_LOG_LINES:] job["updated_at"] = now_iso() - self._save_locked() + self._save_job_locked(job) def _run(self): while True: @@ -353,7 +450,7 @@ class DownloadQueue: f"Final folder: {job['target_dir']}", ] job["cancel_requested"] = False - self._save_locked() + self._save_job_locked(job) process = subprocess.Popen( command, @@ -369,7 +466,7 @@ class DownloadQueue: self.current_process = process self.current_job_id = job["id"] job["pid"] = process.pid - self._save_locked() + self._save_job_locked(job) assert process.stdout is not None for line in process.stdout: @@ -405,7 +502,7 @@ class DownloadQueue: job.setdefault("log", []).append(f"Download failed with exit code {exit_code}.") self.current_process = None self.current_job_id = None - self._save_locked() + self._save_job_locked(job) def strip_control(value): @@ -622,7 +719,10 @@ class Handler(BaseHTTPRequestHandler): raise ValueError("Mode must be sub or dub") self.json({"episodes": episode_list(show_id, mode)}) elif parsed.path == "/api/queue": - self.json({"jobs": DOWNLOAD_QUEUE.list()}) + params = parse_qs(parsed.query) + page = (params.get("page") or ["1"])[0] + per_page = (params.get("per_page") or ["10"])[0] + self.json(DOWNLOAD_QUEUE.list(page=page, per_page=per_page)) else: self.error(HTTPStatus.NOT_FOUND, "Not found") except Exception as exc: @@ -637,6 +737,8 @@ class Handler(BaseHTTPRequestHandler): self.json(DOWNLOAD_QUEUE.add(self.body_json()), HTTPStatus.CREATED) elif parsed.path == "/api/queue/clear-finished": self.json(DOWNLOAD_QUEUE.clear_finished()) + elif parsed.path == "/api/queue/retry-failed": + self.json(DOWNLOAD_QUEUE.retry_all_failed()) elif parsed.path.startswith("/api/queue/"): parts = parsed.path.strip("/").split("/") if len(parts) != 4: @@ -1032,9 +1134,13 @@ INDEX_HTML = r"""

Download queue

- +
+ + +
+
@@ -1044,6 +1150,9 @@ INDEX_HTML = r""" config: { mode: "sub", quality: "best", download_dir: "" }, selected: null, episodes: [], + queuePage: 1, + queuePages: 1, + queueTotal: 0, queueTimer: null }; @@ -1167,14 +1276,19 @@ INDEX_HTML = r""" } } - function renderQueue(jobs) { + function renderQueue(data) { + const jobs = data.jobs || []; + state.queuePage = data.page || 1; + state.queuePages = data.pages || 1; + state.queueTotal = data.total || 0; const el = $("queue"); el.innerHTML = ""; if (!jobs.length) { el.innerHTML = '
Queue is empty.
'; + renderQueuePager(); return; } - for (const job of jobs.slice().reverse()) { + for (const job of jobs) { const item = document.createElement("article"); item.className = "job"; const log = (job.log || []).slice(-28).join("\n"); @@ -1215,6 +1329,29 @@ INDEX_HTML = r""" } el.appendChild(item); } + renderQueuePager(); + } + + function renderQueuePager() { + const el = $("queuePager"); + const start = state.queueTotal ? ((state.queuePage - 1) * 10) + 1 : 0; + const end = Math.min(state.queuePage * 10, state.queueTotal); + el.innerHTML = ` + +
+ `; + el.querySelector(".muted").textContent = + `${start}-${end} of ${state.queueTotal} ยท page ${state.queuePage} of ${state.queuePages}`; + const controls = el.querySelector(".row"); + const prev = actionButton("Previous", () => { + if (state.queuePage > 1) loadQueue(state.queuePage - 1); + }); + const next = actionButton("Next", () => { + if (state.queuePage < state.queuePages) loadQueue(state.queuePage + 1); + }); + prev.disabled = state.queuePage <= 1; + next.disabled = state.queuePage >= state.queuePages; + controls.append(prev, next); } function actionButton(text, handler, extra = "") { @@ -1229,15 +1366,26 @@ INDEX_HTML = r""" async function queueAction(id, action) { try { await api(`/api/queue/${id}/${action}`, { method: "POST", body: "{}" }); - await loadQueue(); + await loadQueue(state.queuePage); } catch (error) { setNotice(error.message); } } - async function loadQueue() { - const data = await api("/api/queue"); - renderQueue(data.jobs); + async function queueBulkAction(path, message) { + try { + const data = await api(path, { method: "POST", body: "{}" }); + setNotice(`${message}: ${data.count || 0}`); + await loadQueue(state.queuePage); + } catch (error) { + setNotice(error.message); + } + } + + async function loadQueue(page = state.queuePage || 1) { + const data = await api(`/api/queue?page=${page}&per_page=10`); + if (data.page > data.pages && data.pages > 0) return loadQueue(data.pages); + renderQueue(data); } async function loadConfig() { @@ -1289,10 +1437,8 @@ INDEX_HTML = r""" }); $("addBtn").addEventListener("click", addSelected); $("saveConfigBtn").addEventListener("click", saveConfig); - $("clearBtn").addEventListener("click", async () => { - await api("/api/queue/clear-finished", { method: "POST", body: "{}" }); - await loadQueue(); - }); + $("removeFinishedBtn").addEventListener("click", () => queueBulkAction("/api/queue/clear-finished", "Removed finished jobs")); + $("retryFailedBtn").addEventListener("click", () => queueBulkAction("/api/queue/retry-failed", "Retried failed jobs")); $("quality").addEventListener("change", () => { state.config.quality = $("quality").value; }); $("firstEpBtn").addEventListener("click", () => { $("episodesInput").value = state.episodes[0] || ""; }); $("latestEpBtn").addEventListener("click", () => { @@ -1340,7 +1486,7 @@ def main(): install_signal_handlers(httpd) print(f"ani-cli web is running at http://{host}:{port}/") print(f"Config: {CONFIG_PATH}") - print(f"Queue: {QUEUE_PATH}") + print(f"Queue: {QUEUE_DB_PATH}") httpd.serve_forever()