From af2808189d45fe5345d32b829aa84f50c20f7626 Mon Sep 17 00:00:00 2001 From: Daniel Gradman-Svendsen Date: Wed, 29 Jul 2026 13:19:55 +0200 Subject: [PATCH] Solve Importarr queue controls issues #1 #2 #3 --- importarr/main.py | 84 +++++++++++++++++++++++++--------- importarr/readiness.py | 11 +++-- importarr/state.py | 75 ++++++++++++++++++++++++++++++ importarr/static/importarr.css | 2 +- importarr/templates/index.html | 12 ++++- tests/test_manual_batches.py | 25 ++++++++++ tests/test_sab_readiness.py | 11 +++++ 7 files changed, 191 insertions(+), 29 deletions(-) diff --git a/importarr/main.py b/importarr/main.py index 8b36a2e..c2fe286 100644 --- a/importarr/main.py +++ b/importarr/main.py @@ -29,6 +29,10 @@ class ManualBatchCreate(BaseModel): path: str +class RunNowRequest(BaseModel): + force: bool = False + + def require_write_auth(authorization: Annotated[str | None, Header()] = None) -> None: if not settings.auth_token: return @@ -68,10 +72,11 @@ def status() -> dict[str, object]: @app.get("/api/manual-batches") def manual_batches() -> list[dict[str, object]]: + sync_manual_queue() rows = [] for batch in state.list_manual_batches(): - videos = scan_videos(Path(batch["path"])) if batch["status"] == "active" else [] - rows.append({**batch, "videos": [{"file": v.path.name, "relative_path": str(v.relative_path), "size": v.size} for v in videos]}) + videos = [item for item in state.list_queue_items() if item["batch_id"] == batch["id"]] if batch["status"] == "active" else [] + rows.append({**batch, "videos": [{"file": item["name"], "relative_path": item["relative_path"], "size": item["size"]} for item in videos]}) return rows @@ -102,45 +107,76 @@ async def jobs() -> dict[str, object]: @app.get("/api/preview") async def preview() -> dict[str, object]: + await sync_queue() + jobs = queue_jobs() + return {"sab_status": "ok", "jobs": jobs, "would_import": sum(1 for row in jobs if row["state"] in {"ready", "manual_batch"})} + + +async def sync_queue() -> None: + sync_manual_queue() client = SabnzbdClient(settings.sab_url, settings.sab_api_key) try: active = await client.active_nzo_ids() data = await client.history() except Exception as exc: # do not leak keys in URL/params - jobs = manual_batch_jobs() - return {"sab_status": "error", "error": exc.__class__.__name__, "jobs": jobs, "would_import": len(jobs)} + state.upsert_queue_item(source_type="system", source_id="sab-sync", name="SABnzbd", state="failed", reason=exc.__class__.__name__) + return slots = data.get("history", {}).get("slots", []) - rows = [] for item in slots: readiness = classify_history_item(item, active, settings.sab_category, settings.download_root) + job_id = str(item.get("nzo_id") or item.get("nzoid") or item.get("name") or "") + if not job_id: + continue if readiness.ready and readiness.storage: for video in scan_videos(readiness.storage): - rows.append({"name": video.path.name, "state": "ready", "relative_path": str(video.relative_path), "storage": str(readiness.storage), "size": video.size}) + state.upsert_queue_item(source_type="sab", source_id=str(video.path), source_path=video.path, name=video.path.name, state="ready", reason=readiness.reason, relative_path=str(video.relative_path), size=video.size, job_id=job_id) else: - rows.append({"name": item.get("name"), "state": readiness.state, "reason": readiness.reason, "storage": str(readiness.storage) if readiness.storage else None}) - jobs = rows + manual_batch_jobs() - return {"sab_status": "ok", "jobs": jobs, "would_import": sum(1 for row in jobs if row["state"] in {"ready", "manual_batch"})} + state.upsert_queue_item(source_type="sab", source_id=job_id, source_path=readiness.storage, name=str(item.get("name") or job_id), state=readiness.state, reason=readiness.reason, job_id=job_id) + + +def queue_jobs() -> list[dict[str, object]]: + return [ + { + "name": item["name"], + "state": item["state"], + "reason": item["reason"], + "relative_path": item["relative_path"], + "storage": item["source_path"], + "size": item["size"], + "source_type": item["source_type"], + } + for item in state.list_queue_items() + if item["source_type"] != "system" + ] def manual_batch_jobs() -> list[dict[str, object]]: - rows: list[dict[str, object]] = [] + sync_manual_queue() + return queue_jobs() + + +def sync_manual_queue() -> None: root = settings.download_root.resolve() for batch in state.list_manual_batches(active_only=True): + seen: set[str] = set() for video in scan_videos(Path(batch["path"])): - rows.append({"name": video.path.name, "state": "manual_batch", "relative_path": str(video.path.relative_to(root)), "size": video.size}) - return rows + source_id = str(video.path) + seen.add(source_id) + state.upsert_queue_item(source_type="manual", source_id=source_id, source_path=video.path, name=video.path.name, state="manual_batch", relative_path=str(video.path.relative_to(root)), size=video.size, batch_id=batch["id"]) + state.remove_missing_manual_items(batch["id"], seen) @app.post("/api/import/run-now") -async def run_now(_: None = Depends(require_write_auth)) -> dict[str, object]: +async def run_now(payload: RunNowRequest | None = None, _: None = Depends(require_write_auth)) -> dict[str, object]: importer = Importer(settings.movies_root, settings.tv_root) imported = 0 - imported += await _import_ready_sab_jobs(importer) + force = bool(payload.force) if payload else False + imported += await _import_ready_sab_jobs(importer, force=force) imported += _import_manual_batches(importer) return {"imported": imported} -async def _import_ready_sab_jobs(importer: Importer) -> int: +async def _import_ready_sab_jobs(importer: Importer, force: bool = False) -> int: client = SabnzbdClient(settings.sab_url, settings.sab_api_key) try: active = await client.active_nzo_ids() @@ -149,31 +185,37 @@ async def _import_ready_sab_jobs(importer: Importer) -> int: return 0 imported = 0 for item in data.get("history", {}).get("slots", []): - readiness = classify_history_item(item, active, settings.sab_category, settings.download_root) - if not readiness.ready or readiness.storage is None: + readiness = classify_history_item(item, active, settings.sab_category, settings.download_root, force_status=force) + if readiness.storage is None or (not readiness.ready and not force): continue for video in scan_videos(readiness.storage): try: result = importer.import_file(video.path) state.add_history(result.source, result.target, "imported", result.bytes) + state.mark_queue_item("sab", str(video.path), "imported") imported += 1 except Exception as exc: state.add_history(video.path, video.path, "failed", 0, exc.__class__.__name__) + state.mark_queue_item("sab", str(video.path), "failed", exc.__class__.__name__) return imported def _import_manual_batches(importer: Importer) -> int: + sync_manual_queue() imported = 0 for batch in state.list_manual_batches(active_only=True): path = Path(batch["path"]) - videos = scan_videos(path) - for video in videos: + items = [item for item in state.list_queue_items() if item["source_type"] == "manual" and item["batch_id"] == batch["id"]] + for item in items: + source = Path(item["source_path"]) try: - result = importer.import_file(video.path) + result = importer.import_file(source) state.add_history(result.source, result.target, "imported", result.bytes) + state.mark_queue_item("manual", item["source_id"], "imported") imported += 1 except Exception as exc: - state.add_history(video.path, video.path, "failed", 0, exc.__class__.__name__) + state.add_history(source, source, "failed", 0, exc.__class__.__name__) + state.mark_queue_item("manual", item["source_id"], "failed", exc.__class__.__name__) if not scan_videos(path): state.complete_manual_batch(batch["id"]) return imported diff --git a/importarr/readiness.py b/importarr/readiness.py index df2b39a..2848a41 100644 --- a/importarr/readiness.py +++ b/importarr/readiness.py @@ -26,16 +26,16 @@ def has_transient_part(path: Path) -> bool: return any(part in TRANSIENT_PARTS or any(token in part for token in TRANSIENT_PARTS) for part in path.parts) -def classify_history_item(item: dict[str, Any], active_nzo_ids: set[str], category: str, download_root: Path) -> Readiness: +def classify_history_item(item: dict[str, Any], active_nzo_ids: set[str], category: str, download_root: Path, force_status: bool = False) -> Readiness: nzo_id = str(item.get("nzo_id") or item.get("nzoid") or "") - if nzo_id and nzo_id in active_nzo_ids: + if not force_status and nzo_id and nzo_id in active_nzo_ids: return Readiness("processing", "SAB job is still present in queue") if str(item.get("category") or "") != category: return Readiness("ignored", "SAB category is not owned by Importarr") status = str(item.get("status") or "") - if status == "Failed": + if not force_status and status == "Failed": return Readiness("failed", "SAB history reports failure") - if status in NOT_READY_STATUSES or status != "Completed": + if not force_status and (status in NOT_READY_STATUSES or status != "Completed"): return Readiness("processing", f"SAB status is {status or 'unknown'}") storage_value = str(item.get("storage") or "") if not storage_value: @@ -46,4 +46,5 @@ def classify_history_item(item: dict[str, Any], active_nzo_ids: set[str], catego return Readiness("ignored", "SAB storage is outside configured download root", storage) if has_transient_part(storage): return Readiness("processing", "SAB storage path contains transient unpack/admin marker", storage) - return Readiness("ready", "SAB completed in owned category with final storage", storage) + reason = "forced despite SAB status" if force_status and status != "Completed" else "SAB completed in owned category with final storage" + return Readiness("ready", reason, storage) diff --git a/importarr/state.py b/importarr/state.py index a5d4e94..febac87 100644 --- a/importarr/state.py +++ b/importarr/state.py @@ -34,6 +34,23 @@ class State: error text ); create table if not exists app_state (key text primary key, value text not null); + create table if not exists import_queue_items ( + id integer primary key autoincrement, + source_type text not null, + source_id text not null, + source_path text, + name text not null, + state text not null, + reason text, + relative_path text, + size integer not null default 0, + batch_id integer, + job_id text, + first_seen_at text not null default current_timestamp, + updated_at text not null default current_timestamp, + completed_at text, + unique(source_type, source_id) + ); """ ) self.conn.commit() @@ -56,6 +73,7 @@ class State: def delete_manual_batch(self, batch_id: int) -> None: self.conn.execute("delete from manual_batches where id = ?", (batch_id,)) + self.conn.execute("delete from import_queue_items where batch_id = ? and source_type = 'manual'", (batch_id,)) self.conn.commit() def complete_manual_batch(self, batch_id: int) -> None: @@ -69,5 +87,62 @@ class State: ) self.conn.commit() + def upsert_queue_item( + self, + *, + source_type: str, + source_id: str, + name: str, + state: str, + source_path: Path | None = None, + reason: str | None = None, + relative_path: str | None = None, + size: int = 0, + batch_id: int | None = None, + job_id: str | None = None, + ) -> dict[str, Any]: + self.conn.execute( + """ + insert into import_queue_items(source_type, source_id, source_path, name, state, reason, relative_path, size, batch_id, job_id) + values (?,?,?,?,?,?,?,?,?,?) + on conflict(source_type, source_id) do update set + source_path=excluded.source_path, + name=excluded.name, + state=excluded.state, + reason=excluded.reason, + relative_path=excluded.relative_path, + size=excluded.size, + batch_id=excluded.batch_id, + job_id=excluded.job_id, + updated_at=current_timestamp, + completed_at=case when excluded.state in ('imported','failed','skipped') then current_timestamp else null end + """, + (source_type, source_id, str(source_path) if source_path else None, name, state, reason, relative_path, size, batch_id, job_id), + ) + self.conn.commit() + row = self.conn.execute("select * from import_queue_items where source_type = ? and source_id = ?", (source_type, source_id)).fetchone() + return dict(row) + + def mark_queue_item(self, source_type: str, source_id: str, state: str, reason: str | None = None) -> None: + self.conn.execute( + "update import_queue_items set state=?, reason=?, updated_at=current_timestamp, completed_at=case when ? in ('imported','failed','skipped') then current_timestamp else completed_at end where source_type=? and source_id=?", + (state, reason, state, source_type, source_id), + ) + self.conn.commit() + + def remove_missing_manual_items(self, batch_id: int, source_ids: set[str]) -> None: + rows = self.conn.execute("select source_id from import_queue_items where source_type='manual' and batch_id=?", (batch_id,)).fetchall() + for row in rows: + if row["source_id"] not in source_ids: + self.conn.execute("delete from import_queue_items where source_type='manual' and source_id=?", (row["source_id"],)) + self.conn.commit() + + def list_queue_items(self, active_only: bool = True) -> list[dict[str, Any]]: + sql = "select * from import_queue_items" + if active_only: + sql += " where state not in ('imported','failed','skipped')" + sql += " order by updated_at desc, id desc" + return [dict(row) for row in self.conn.execute(sql)] + def list_history(self, limit: int = 100) -> list[dict[str, Any]]: return [dict(row) for row in self.conn.execute("select * from import_history order by id desc limit ?", (limit,))] diff --git a/importarr/static/importarr.css b/importarr/static/importarr.css index 035d3d4..d5970d4 100644 --- a/importarr/static/importarr.css +++ b/importarr/static/importarr.css @@ -1 +1 @@ -body{font-family:system-ui,sans-serif;margin:0;background:#111827;color:#e5e7eb}header,main{max-width:1100px;margin:auto;padding:1rem}.topbar{display:flex;justify-content:space-between;gap:1rem;align-items:center;background:#0f172a}.build{text-align:right}.build strong{font-size:1.2rem}.cards{display:grid;grid-template-columns:repeat(auto-fit,minmax(12rem,1fr));gap:1rem}.cards article,.panel{background:#1f2937;border-radius:.75rem;padding:1rem;margin-top:1rem}strong{display:block;font-size:2rem}span,small,dd{color:#9ca3af}table{width:100%;border-collapse:collapse;background:#1f2937;margin-top:1rem}th,td{padding:.6rem;border-bottom:1px solid #374151;text-align:left}input,button{padding:.6rem;border-radius:.4rem;border:1px solid #374151}button{background:#38bdf8;color:#082f49;font-weight:700}.info{display:grid;grid-template-columns:10rem 1fr;gap:.4rem 1rem}.info dt{font-weight:700}.info dd{margin:0;overflow-wrap:anywhere}.state{background:#0f172a;border:1px solid #374151;border-radius:999px;padding:.15rem .5rem} +body{font-family:system-ui,sans-serif;margin:0;background:#111827;color:#e5e7eb}header,main{max-width:1100px;margin:auto;padding:1rem}.topbar{display:flex;justify-content:space-between;gap:1rem;align-items:center;background:#0f172a}.build{text-align:right}.build strong{font-size:1.2rem}.cards{display:grid;grid-template-columns:repeat(auto-fit,minmax(12rem,1fr));gap:1rem}.cards article,.panel{background:#1f2937;border-radius:.75rem;padding:1rem;margin-top:1rem}strong{display:block;font-size:2rem}span,small,dd{color:#9ca3af}table{width:100%;border-collapse:collapse;background:#1f2937;margin-top:1rem}th,td{padding:.6rem;border-bottom:1px solid #374151;text-align:left}input,button{padding:.6rem;border-radius:.4rem;border:1px solid #374151}button{background:#38bdf8;color:#082f49;font-weight:700}.inline-form{display:flex;gap:.5rem;flex-wrap:wrap}.inline-form input[name=path]{min-width:min(100%,28rem);flex:1}.info{display:grid;grid-template-columns:10rem 1fr;gap:.4rem 1rem}.info dt{font-weight:700}.info dd{margin:0;overflow-wrap:anywhere}.state{background:#0f172a;border:1px solid #374151;border-radius:999px;padding:.15rem .5rem} diff --git a/importarr/templates/index.html b/importarr/templates/index.html index abf157a..3005867 100644 --- a/importarr/templates/index.html +++ b/importarr/templates/index.html @@ -36,18 +36,26 @@

Manual batches

-
+
+ + + + +
{% for batch in batches %}{% endfor %}
IDStatusPath
{{ batch.id }}{{ batch.status }}{{ batch.path }}
-

Jobs

Loading…
+

Jobs

Loading…
diff --git a/tests/test_manual_batches.py b/tests/test_manual_batches.py index 9a3a6eb..4e37851 100644 --- a/tests/test_manual_batches.py +++ b/tests/test_manual_batches.py @@ -28,3 +28,28 @@ def test_manual_batch_completes_when_empty(tmp_path, monkeypatch): rows = main.state.list_manual_batches() assert rows[0]["status"] == "completed" assert (movies / "Movie.mkv").exists() + + +def test_manual_queue_items_are_persisted_and_imported(tmp_path, monkeypatch): + monkeypatch.setenv("IMPORTARR_STATE_PATH", str(tmp_path / "initial.db")) + import importarr.main as main + + download = tmp_path / "downloads" + movies = tmp_path / "movies" + tv = tmp_path / "tv" + batch = download / "Release" + batch.mkdir(parents=True) + (batch / "Movie.mkv").write_bytes(b"movie") + monkeypatch.setattr(main, "settings", Settings(download_root=download, movies_root=movies, tv_root=tv, state_path=tmp_path / "state.db")) + monkeypatch.setattr(main, "state", State(tmp_path / "state.db")) + main.state.add_manual_batch(batch) + + main.sync_manual_queue() + queued = main.state.list_queue_items() + assert len(queued) == 1 + assert queued[0]["source_type"] == "manual" + assert queued[0]["state"] == "manual_batch" + + assert main._import_manual_batches(main.Importer(movies, tv)) == 1 + assert main.state.list_queue_items() == [] + assert main.state.list_queue_items(active_only=False)[0]["state"] == "imported" diff --git a/tests/test_sab_readiness.py b/tests/test_sab_readiness.py index e793a2f..2c7463f 100644 --- a/tests/test_sab_readiness.py +++ b/tests/test_sab_readiness.py @@ -27,6 +27,12 @@ def test_queue_item_not_ready(): assert result.state == "processing" +def test_force_status_overrides_queue_and_processing_status(): + result = classify_history_item(item(status="Extracting"), {"1"}, "manual", ROOT, force_status=True) + assert result.ready + assert result.storage == ROOT / "Movie" + + def test_post_processing_not_ready(): for status in ["Queued", "Repairing", "Extracting", "Moving"]: assert classify_history_item(item(status=status), set(), "manual", ROOT).state == "processing" @@ -39,3 +45,8 @@ def test_empty_storage_unknown(): def test_unpack_path_not_ready(): result = classify_history_item(item(storage=str(ROOT / "_UNPACK_Movie")), set(), "manual", ROOT) assert result.state == "processing" + + +def test_force_status_keeps_transient_path_safety(): + result = classify_history_item(item(storage=str(ROOT / "_UNPACK_Movie"), status="Extracting"), {"1"}, "manual", ROOT, force_status=True) + assert result.state == "processing"