diff --git a/importarr/importer.py b/importarr/importer.py index fefc2fc..2730941 100644 --- a/importarr/importer.py +++ b/importarr/importer.py @@ -4,6 +4,7 @@ import os import shutil from dataclasses import dataclass from pathlib import Path +from typing import Callable @dataclass @@ -13,6 +14,10 @@ class ImportResult: bytes: int +class ImportCancelled(Exception): + """Raised when an import is cancelled at a safe copy boundary.""" + + class Importer: def __init__(self, movies_root: Path, tv_root: Path): self.movies_root = movies_root @@ -22,14 +27,24 @@ class Importer: target_root = self.tv_root if _looks_like_tv(source) else self.movies_root return _unique_path(target_root / source.name) - def import_file(self, source: Path) -> ImportResult: + def import_file(self, source: Path, should_cancel: Callable[[], bool] | None = None) -> ImportResult: target = self.target_for(source) target.parent.mkdir(parents=True, exist_ok=True) partial = target.with_name(target.name + ".partial") - with source.open("rb") as src, partial.open("wb") as dst: - shutil.copyfileobj(src, dst, length=1024 * 1024) - dst.flush() - os.fsync(dst.fileno()) + try: + with source.open("rb") as src, partial.open("wb") as dst: + while True: + if should_cancel and should_cancel(): + raise ImportCancelled("import cancelled") + chunk = src.read(1024 * 1024) + if not chunk: + break + dst.write(chunk) + dst.flush() + os.fsync(dst.fileno()) + except ImportCancelled: + partial.unlink(missing_ok=True) + raise if partial.stat().st_size != source.stat().st_size: raise IOError("partial copy size mismatch") partial.rename(target) diff --git a/importarr/main.py b/importarr/main.py index c2fe286..be89765 100644 --- a/importarr/main.py +++ b/importarr/main.py @@ -12,7 +12,7 @@ from pydantic import BaseModel from .build_info import build_info from .config import Settings -from .importer import Importer +from .importer import ImportCancelled, Importer from .sabnzbd import SabnzbdClient from .readiness import classify_history_item from .scanner import scan_videos @@ -33,6 +33,10 @@ class RunNowRequest(BaseModel): force: bool = False +class QueueControlRequest(BaseModel): + mode: str + + def require_write_auth(authorization: Annotated[str | None, Header()] = None) -> None: if not settings.auth_token: return @@ -53,6 +57,7 @@ def index(request: Request) -> HTMLResponse: @app.get("/api/status") def status() -> dict[str, object]: history = state.list_history() + control = control_status() return { "app": "Importarr", "build": build_info(), @@ -66,13 +71,81 @@ def status() -> dict[str, object]: "manual_batches": len(state.list_manual_batches(active_only=True)), "imported_total": sum(1 for row in history if row["status"] == "imported"), "failed_total": sum(1 for row in history if row["status"] == "failed"), - "current": None, + "current": control["current"], + "control": control, } +def control_status() -> dict[str, object]: + mode = state.get_app_state("queue_mode", "running") or "running" + current = state.get_app_state("current_job") + cancel_requested = (state.get_app_state("cancel_requested", "false") or "false") == "true" + return { + "queue_mode": mode, + "queue_accepting_new_jobs": mode == "running", + "cancel_requested": cancel_requested, + "current": current, + } + + +def queue_accepting_new_jobs() -> bool: + return (state.get_app_state("queue_mode", "running") or "running") == "running" + + +def cancel_requested() -> bool: + return (state.get_app_state("cancel_requested", "false") or "false") == "true" + + +def consume_cancel_request() -> bool: + if not cancel_requested(): + return False + state.set_app_state("cancel_requested", "false") + return True + + +def set_current_job(name: str | None) -> None: + state.set_app_state("current_job", name or "") + + +@app.post("/api/control/queue") +def set_queue_control(payload: QueueControlRequest, _: None = Depends(require_write_auth)) -> dict[str, object]: + if payload.mode not in {"running", "paused", "stopped"}: + raise HTTPException(status_code=400, detail="mode must be running, paused, or stopped") + state.set_app_state("queue_mode", payload.mode) + if payload.mode == "running": + state.set_app_state("cancel_requested", "false") + return control_status() + + +@app.post("/api/control/start") +def start_queue(_: None = Depends(require_write_auth)) -> dict[str, object]: + state.set_app_state("queue_mode", "running") + state.set_app_state("cancel_requested", "false") + return control_status() + + +@app.post("/api/control/pause") +def pause_queue(_: None = Depends(require_write_auth)) -> dict[str, object]: + state.set_app_state("queue_mode", "paused") + return control_status() + + +@app.post("/api/control/stop") +def stop_queue(_: None = Depends(require_write_auth)) -> dict[str, object]: + state.set_app_state("queue_mode", "stopped") + return control_status() + + +@app.post("/api/control/cancel-current") +def cancel_current(_: None = Depends(require_write_auth)) -> dict[str, object]: + state.set_app_state("cancel_requested", "true") + return control_status() + + @app.get("/api/manual-batches") def manual_batches() -> list[dict[str, object]]: - sync_manual_queue() + if queue_accepting_new_jobs(): + sync_manual_queue() rows = [] for batch in state.list_manual_batches(): videos = [item for item in state.list_queue_items() if item["batch_id"] == batch["id"]] if batch["status"] == "active" else [] @@ -107,9 +180,10 @@ async def jobs() -> dict[str, object]: @app.get("/api/preview") async def preview() -> dict[str, object]: - await sync_queue() + if queue_accepting_new_jobs(): + 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"})} + return {"sab_status": "ok", "jobs": jobs, "would_import": sum(1 for row in jobs if row["state"] in {"ready", "manual_batch"}), "control": control_status()} async def sync_queue() -> None: @@ -156,6 +230,8 @@ def manual_batch_jobs() -> list[dict[str, object]]: def sync_manual_queue() -> None: + if not queue_accepting_new_jobs(): + return root = settings.download_root.resolve() for batch in state.list_manual_batches(active_only=True): seen: set[str] = set() @@ -185,37 +261,58 @@ async def _import_ready_sab_jobs(importer: Importer, force: bool = False) -> int return 0 imported = 0 for item in data.get("history", {}).get("slots", []): + if consume_cancel_request(): + break 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): + if consume_cancel_request(): + return imported + set_current_job(str(video.path)) try: - result = importer.import_file(video.path) + result = importer.import_file(video.path, should_cancel=consume_cancel_request) state.add_history(result.source, result.target, "imported", result.bytes) state.mark_queue_item("sab", str(video.path), "imported") imported += 1 + except ImportCancelled: + state.add_history(video.path, video.path, "cancelled", 0, "cancelled") + state.mark_queue_item("sab", str(video.path), "skipped", "cancelled") + return imported 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__) + finally: + set_current_job(None) return imported def _import_manual_batches(importer: Importer) -> int: - sync_manual_queue() + if queue_accepting_new_jobs(): + sync_manual_queue() imported = 0 for batch in state.list_manual_batches(active_only=True): path = Path(batch["path"]) items = [item for item in state.list_queue_items() if item["source_type"] == "manual" and item["batch_id"] == batch["id"]] for item in items: + if consume_cancel_request(): + return imported source = Path(item["source_path"]) + set_current_job(str(source)) try: - result = importer.import_file(source) + result = importer.import_file(source, should_cancel=consume_cancel_request) state.add_history(result.source, result.target, "imported", result.bytes) state.mark_queue_item("manual", item["source_id"], "imported") imported += 1 + except ImportCancelled: + state.add_history(source, source, "cancelled", 0, "cancelled") + state.mark_queue_item("manual", item["source_id"], "skipped", "cancelled") + return imported except Exception as exc: state.add_history(source, source, "failed", 0, exc.__class__.__name__) state.mark_queue_item("manual", item["source_id"], "failed", exc.__class__.__name__) + finally: + set_current_job(None) if not scan_videos(path): state.complete_manual_batch(batch["id"]) return imported diff --git a/importarr/state.py b/importarr/state.py index febac87..d5c6270 100644 --- a/importarr/state.py +++ b/importarr/state.py @@ -55,6 +55,17 @@ class State: ) self.conn.commit() + def get_app_state(self, key: str, default: str | None = None) -> str | None: + row = self.conn.execute("select value from app_state where key = ?", (key,)).fetchone() + return row["value"] if row else default + + def set_app_state(self, key: str, value: str) -> None: + self.conn.execute( + "insert into app_state(key, value) values (?, ?) on conflict(key) do update set value=excluded.value", + (key, value), + ) + self.conn.commit() + def add_manual_batch(self, path: Path) -> dict[str, Any]: self.conn.execute("insert or ignore into manual_batches(path) values (?)", (str(path),)) self.conn.commit() diff --git a/importarr/static/importarr.css b/importarr/static/importarr.css index d5970d4..f42c6a8 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}.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} +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}.danger{background:#f87171;color:#450a0a}.controls{display:flex;gap:.5rem;flex-wrap:wrap}.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 3005867..a3899fd 100644 --- a/importarr/templates/index.html +++ b/importarr/templates/index.html @@ -32,8 +32,20 @@
Movies root
{{ status.movies_root }}
TV root
{{ status.tv_root }}
Write auth
{{ 'enabled' if status.auth_enabled else 'disabled' }}
+
Queue mode
{{ status.control.queue_mode }}
+
Current job
{{ status.current or 'idle' }}
+
+

Queue controls

+

Pause and stop prevent new jobs from being added to the queue. They do not interrupt an import already in progress; use cancel current job for that.

+
+ + + + +
+

Manual batches

@@ -51,7 +63,8 @@