diff --git a/README.md b/README.md index 6bb5c08..beaf567 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,11 @@ Release-worthy changes should be committed, tagged with SemVer (`v0.1.1`, `v0.2. - `GET /api/manual-batches` - `POST /api/manual-batches` with `{ "path": "relative/or/absolute/path" }` - `DELETE /api/manual-batches/{id}` +- `POST /api/control/start` +- `POST /api/control/pause` +- `POST /api/control/stop` +- `POST /api/control/cancel-current` +- `POST /api/queue-items/{id}/action` with `{ "action": "retry|ignore|remove" }` - `POST /api/import/run-now` Set `IMPORTARR_AUTH_TOKEN_FILE` or `IMPORTARR_AUTH_TOKEN` to require `Authorization: Bearer ` for write endpoints. diff --git a/deploy/importarr.service b/deploy/importarr.service index 194394a..0282b13 100644 --- a/deploy/importarr.service +++ b/deploy/importarr.service @@ -1,12 +1,12 @@ [Unit] -Description=Importarr manual media importer status UI +Description=Importarr manual media importer web UI After=network-online.target Wants=network-online.target [Service] EnvironmentFile=-/etc/importarr/importarr.env EnvironmentFile=-/opt/importarr/build.env -ExecStart=/opt/importarr/venv/bin/importarr-status +ExecStart=/opt/importarr/venv/bin/importarr Restart=on-failure RestartSec=5s User=root diff --git a/deploy/repo-upgrade.sh b/deploy/repo-upgrade.sh index b318daf..691771d 100644 --- a/deploy/repo-upgrade.sh +++ b/deploy/repo-upgrade.sh @@ -32,6 +32,8 @@ fi git fetch --prune origin git pull --ff-only "$VENV/bin/pip" install --upgrade "$REPO_DIR" +install -m 0644 "$REPO_DIR/deploy/importarr.service" /etc/systemd/system/importarr.service +systemctl daemon-reload GIT_SHA="$(git rev-parse --short=12 HEAD 2>/dev/null || printf development)" BUILD_DATE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" cat > "$PREFIX/build.env" < str: - return subprocess.run(args, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL).stdout - - -def systemctl_show(unit: str) -> dict[str, str]: - data = {} - for line in run(["systemctl", "show", unit, "--no-pager"]).splitlines(): - if "=" in line: - key, value = line.split("=", 1) - data[key] = value - return data - - -def int_value(value: str | None) -> int | None: - try: - return int(value or "") - except ValueError: - return None - - -def timestamp_to_iso(usec: str | None) -> str | None: - value = int_value(usec) - if not value or value <= 0: - return None - return datetime.fromtimestamp(value / 1_000_000, tz=timezone.utc).isoformat() - - -def monotonic_runtime_seconds(service: dict[str, str]) -> int | None: - started = int_value(service.get("ActiveEnterTimestampMonotonic")) - if not started: - main_pid = int_value(service.get("MainPID")) - if not main_pid: - return None - etimes = run(["ps", "-o", "etimes=", "-p", str(main_pid)]).strip() - return int_value(etimes) - boot_ns = time.clock_gettime_ns(time.CLOCK_BOOTTIME) - runtime = int((boot_ns / 1000 - started) / 1_000_000) - return max(runtime, 0) - - -def scan_queue(root: Path) -> dict[str, object]: - files = dirs = bytes_total = 0 - processing_files = processing_dirs = processing_bytes = 0 - oldest = newest = None - top_level: list[dict[str, object]] = [] - if not root.exists(): - return {"path": str(root), "exists": False, "files": 0, "dirs": 0, "bytes": 0, "oldest": None, "newest": None, "topLevel": [], "items": [], "processingFiles": 0, "processingDirs": 0, "processingBytes": 0, "processing": [], "processingItems": []} - top_level_map: dict[Path, dict[str, object]] = {} - processing_map: dict[Path, dict[str, object]] = {} - ready_items: list[dict[str, object]] = [] - processing_items: list[dict[str, object]] = [] - for dirpath, dirnames, filenames in os.walk(root): - for filename in filenames: - path = Path(dirpath) / filename - if path.suffix.lower() not in VIDEO_EXT or "sample" in filename.lower() or "sample" in str(path.parent).lower(): - continue - try: - st = path.stat() - except FileNotFoundError: - continue - oldest = st.st_mtime if oldest is None else min(oldest, st.st_mtime) - newest = st.st_mtime if newest is None else max(newest, st.st_mtime) - try: - rel = path.relative_to(root) - except ValueError: - rel = path - top = root / rel.parts[0] if rel.parts else path - sab = sab_state_for(top.name) - transient = top.name.startswith(('_UNPACK_', '__UNPACK__', '_FAILED_', '_ADMIN_')) - is_processing = transient or (sab and sab.get("ready") is False and sab.get("status") != "manual") - relative_dir = str(Path(*rel.parts[:-1])) if len(rel.parts) > 1 else "" - item_label = path.name if not relative_dir else f"{relative_dir} / {path.name}" - state = str(sab.get("state") if sab else ("unpacking" if transient else "ready")) - file_item = {"name": path.name, "label": item_label, "release": top.name, "relativeDir": relative_dir, "path": str(path), "bytes": st.st_size, "mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(), "state": state} - if is_processing: - processing_files += 1 - processing_bytes += st.st_size - processing_items.append(file_item) - item = processing_map.setdefault(top, {"name": top.name, "type": "dir" if top.is_dir() else "file", "files": 0, "bytes": 0, "mtime": None}) - item["files"] = int(item["files"]) + 1 - item["bytes"] = int(item["bytes"]) + st.st_size - item["mtime"] = datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat() - continue - - files += 1 - bytes_total += st.st_size - ready_items.append(file_item) - item = top_level_map.setdefault(top, {"name": top.name, "type": "dir" if top.is_dir() else "file", "files": 0, "bytes": 0, "mtime": None}) - item["files"] = int(item["files"]) + 1 - item["bytes"] = int(item["bytes"]) + st.st_size - item["mtime"] = datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat() - dirs = sum(1 for item in top_level_map.values() if item["type"] == "dir") - processing_dirs = sum(1 for item in processing_map.values() if item["type"] == "dir") - top_level = sorted(top_level_map.values(), key=lambda item: str(item["name"]).lower()) - processing = sorted(processing_map.values(), key=lambda item: str(item["name"]).lower()) - return { - "path": str(root), - "exists": True, - "files": files, - "dirs": dirs, - "bytes": bytes_total, - "oldest": datetime.fromtimestamp(oldest, tz=timezone.utc).isoformat() if oldest else None, - "newest": datetime.fromtimestamp(newest, tz=timezone.utc).isoformat() if newest else None, - "topLevel": top_level[:100], - "items": ready_items[:500], - "processingFiles": processing_files, - "processingDirs": processing_dirs, - "processingBytes": processing_bytes, - "processing": processing[:100], - "processingItems": processing_items[:500], - } - - -def read_logs(limit: int = 100) -> list[dict[str, object]]: - if not LOG.exists(): - return [] - lines = LOG.read_text(errors="replace").splitlines()[-max(1, min(limit, 1000)):] - records = [] - for line in lines: - try: - records.append(json.loads(line)) - except json.JSONDecodeError: - records.append({"level": "RAW", "msg": line}) - return records - - -def read_summaries() -> list[dict[str, object]]: - if not LOG.exists(): - return [] - summaries = [] - for line in LOG.read_text(errors="replace").splitlines(): - try: - record = json.loads(line) - except json.JSONDecodeError: - continue - if record.get("msg") == "summary": - summaries.append(record) - return summaries - - -def read_importer_status() -> dict[str, object] | None: - if not IMPORTER_STATUS.exists(): - return None - - -def infer_current_from_logs(logs: list[dict[str, object]]) -> dict[str, object] | None: - for record in reversed(logs): - if record.get("level") != "MOVE" or record.get("msg") not in {"moving", "moving sidecar"}: - continue - src = record.get("src") - dest = record.get("dest") - if not src or not dest: - continue - src_path = Path(str(src)) - partial = Path(str(dest) + ".partial") - total = None - copied = None - try: - total = src_path.stat().st_size - except FileNotFoundError: - pass - try: - copied = partial.stat().st_size - except FileNotFoundError: - copied = None - percent = round((copied / total * 100), 2) if copied is not None and total else None - return {"phase": "copying", "src": str(src), "dest": str(dest), "partial": str(partial), "bytes_copied": copied, "bytes_total": total, "percent": percent, "media_type": record.get("media_type"), "source_tag": record.get("source_tag"), "kind": "inferred"} - return None - try: - return json.loads(IMPORTER_STATUS.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return None - - -def queue_status() -> dict[str, object]: - global _SAB_HISTORY_CACHE - _SAB_HISTORY_CACHE = None - roots = {name: scan_queue(path) for name, path in QUEUE_ROOTS.items()} - return { - "roots": roots, - "files": sum(int(r["files"]) for r in roots.values()), - "dirs": sum(int(r["dirs"]) for r in roots.values()), - "bytes": sum(int(r["bytes"]) for r in roots.values()), - "processingFiles": sum(int(r["processingFiles"]) for r in roots.values()), - "processingDirs": sum(int(r["processingDirs"]) for r in roots.values()), - "processingBytes": sum(int(r["processingBytes"]) for r in roots.values()), - } - - -def sab_api_key() -> str | None: - try: - import re - m = re.search(r"^api_key\s*=\s*(\S+)", SAB_CONFIG.read_text(errors="replace"), re.M) - return m.group(1) if m else None - except OSError: - return None - - -def sab_history() -> list[dict[str, object]]: - key = sab_api_key() - if not key: - return [] - try: - q = urllib.parse.urlencode({"mode": "history", "output": "json", "limit": 200, "apikey": key}) - data = json.load(urllib.request.urlopen(f"{SAB_API}?{q}", timeout=10)) - return data.get("history", {}).get("slots", []) - except Exception: - return [] - - -_SAB_HISTORY_CACHE: list[dict[str, object]] | None = None - - -def sab_state_for(folder_name: str) -> dict[str, object] | None: - global _SAB_HISTORY_CACHE - if _SAB_HISTORY_CACHE is None: - _SAB_HISTORY_CACHE = sab_history() - normalized = folder_name.removeprefix("_UNPACK_").removeprefix("__UNPACK__") - for item in _SAB_HISTORY_CACHE: - name = str(item.get("name") or "") - if name != normalized and name != folder_name: - continue - status = str(item.get("status") or "") - storage = str(item.get("storage") or "") - action = str(item.get("action_line") or "") - category = str(item.get("category") or item.get("cat") or "") - owned = category == "manual" - ready = owned and status == "Completed" and bool(storage) and "_UNPACK_" not in storage - state = "ready" if ready else ("ignored category " + category if not owned else (status.lower() if status else "sab pending")) - if action: - state = action - return {"ready": ready, "owned": owned, "category": category, "status": status, "storage": storage, "state": state} - return None - - -def read_manual_batches() -> list[str]: - try: - raw = json.loads(MANUAL_BATCHES.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return [] - return [str(x) for x in raw] if isinstance(raw, list) else [] - - -def write_manual_batches(items: list[str]) -> None: - MANUAL_BATCHES.parent.mkdir(parents=True, exist_ok=True) - MANUAL_BATCHES.write_text(json.dumps(sorted(set(items)), indent=2), encoding="utf-8") - - -def add_manual_batch(value: str) -> tuple[bool, str]: - value = value.strip().strip("/") - if not value: - return False, "missing folder" - base = QUEUE_ROOTS["manual"].resolve() - path = (base / value).resolve() if not value.startswith("/srv/") else Path(value).resolve() - if not (path == base or path.is_relative_to(base)): - return False, "folder must be under manual downloads" - if not path.is_dir(): - return False, "folder does not exist" - items = read_manual_batches() - items.append(str(path)) - write_manual_batches(items) - return True, str(path) - - -def status() -> dict[str, object]: - service = systemctl_show(SERVICE) - timer = systemctl_show(TIMER) - logs = read_logs(300) - all_logs = read_logs(1000) - summaries = read_summaries() - last_summary = summaries[-1] if summaries else None - cutoff_1h = datetime.now() - timedelta(hours=1) - cutoff_24h = datetime.now() - timedelta(hours=24) - processed_1h = 0 - processed_24h = 0 - processed_total = 0 - runs_1h = 0 - runs_24h = 0 - runs_total = 0 - for summary in summaries: - moved = int(summary.get("moved") or 0) - processed_total += moved - runs_total += 1 - try: - ts = datetime.fromisoformat(str(summary.get("ts"))) - except ValueError: - ts = None - if ts and ts >= cutoff_1h: - processed_1h += moved - runs_1h += 1 - if ts and ts >= cutoff_24h: - processed_24h += moved - runs_24h += 1 - queue = queue_status() - running = service.get("ActiveState") == "activating" or service.get("SubState") in {"start", "running"} - current = read_importer_status() - if running and (not current or current.get("phase") == "done"): - current = infer_current_from_logs(all_logs) - runtime = monotonic_runtime_seconds(service) if running else None - memory_current = int_value(service.get("MemoryCurrent")) - memory_peak = int_value(service.get("MemoryPeak")) - warnings = [] - if runtime and runtime > LONG_RUNTIME_SECONDS: - warnings.append("manual-media-import.service has been running longer than 25 minutes") - if memory_peak and memory_peak > HIGH_MEMORY_BYTES: - warnings.append("manual-media-import.service peak memory is over 8 GiB") - if last_summary and int(last_summary.get("errors") or 0) > 0: - warnings.append("last importer summary reported errors") - now = time.time() - for name, root in queue["roots"].items(): - oldest = root.get("oldest") - if oldest: - try: - age = now - datetime.fromisoformat(str(oldest)).timestamp() - if age > STALE_QUEUE_SECONDS: - warnings.append(f"{name} queue contains files older than 6 hours") - except ValueError: - pass - health_state = "warning" if warnings else ("running" if running else service.get("Result", "unknown")) - return { - "name": "Importarr", - "service": { - "unit": SERVICE, - "activeState": service.get("ActiveState"), - "subState": service.get("SubState"), - "result": service.get("Result"), - "running": running, - "mainPid": int_value(service.get("MainPID")), - "startedAt": timestamp_to_iso(service.get("ActiveEnterTimestampUSec")), - "runtimeSeconds": runtime, - "memoryCurrentBytes": memory_current, - "memoryPeakBytes": memory_peak, - }, - "timer": { - "unit": TIMER, - "activeState": timer.get("ActiveState"), - "subState": timer.get("SubState"), - "lastTrigger": timestamp_to_iso(timer.get("LastTriggerUSec")), - "nextElapse": timestamp_to_iso(timer.get("NextElapseUSecRealtime")), - }, - "queue": queue, - "lastSummary": last_summary, - "processed": {"last1h": processed_1h, "last24h": processed_24h, "total": processed_total, "runsLast1h": runs_1h, "runsLast24h": runs_24h, "runsTotal": runs_total}, - "current": current, - "manualBatches": read_manual_batches(), - "health": {"state": health_state, "warnings": warnings}, - } - - -def fast_health() -> dict[str, object]: - service = systemctl_show(SERVICE) - timer = systemctl_show(TIMER) - running = service.get("ActiveState") == "activating" or service.get("SubState") in {"start", "running"} - runtime = monotonic_runtime_seconds(service) if running else None - warnings = [] - if runtime and runtime > LONG_RUNTIME_SECONDS: - warnings.append("manual-media-import.service has been running longer than 25 minutes") - return {"state": "warning" if warnings else ("running" if running else service.get("Result", "unknown")), "warnings": warnings, "service": service.get("ActiveState"), "timer": timer.get("ActiveState")} - - -def page() -> bytes: - s = status() - logs = read_logs(80) - def esc(value: object) -> str: - return html.escape("—" if value is None else str(value)) - def gib(value: object) -> str: - return "—" if value is None else f"{int(value) / 1024**3:.2f} GiB" - def secs(value: object) -> str: - if value is None: - return "—" - seconds = int(value) - return f"{seconds // 60}m {seconds % 60}s" - def job_name(path: object) -> str: - if not path: - return "" - name = Path(str(path)).name - return name[:37] + "..." if len(name) > 40 else name - title_icon = "📥" - title_runtime = secs(s["service"]["runtimeSeconds"]) - current = s.get("current") or {} - progress = current.get("percent") - full_current_name = Path(str(current.get("src", ""))).name if current.get("src") else "" - current_src_path = Path(str(current.get("src", ""))) if current.get("src") else None - current_top_name = "" - current_relative_dir = "" - if current_src_path: - for root in QUEUE_ROOTS.values(): - try: - rel = current_src_path.relative_to(root) - current_top_name = rel.parts[0] if rel.parts else current_src_path.name - current_relative_dir = str(Path(*rel.parts[:-1])) if len(rel.parts) > 1 else "" - break - except ValueError: - continue - current_name = job_name(current.get("src")) - title = f"{title_icon} {progress}% {title_runtime} - {current_name}" if progress is not None and current_name else f"{title_icon} - idle" - warnings = s["health"]["warnings"] - warning_html = "".join(f"

{esc(w)}

" for w in warnings) or "

None

" - log_text = "\n".join(esc(f"[{r.get('ts','')}] {r.get('level','')} {r.get('msg','')} {json.dumps(r, ensure_ascii=False)}") for r in logs) - source_target_html = f"

Source: {esc(current.get('src'))}

Target: {esc(current.get('dest'))}

" if current else "" - progress_html = f"

{esc(full_current_name)}
{esc(current_relative_dir)}

{esc(progress)}% · runtime {title_runtime}

{esc(current.get('phase'))} · {gib(current.get('bytes_copied'))} / {gib(current.get('bytes_total'))}

{source_target_html}" if current else "

—

" - current_row = f"▶{esc(full_current_name)}
{esc(current_relative_dir)}{esc(progress)}%{title_runtime}" if current else "" - current_path = str(current_src_path) if current_src_path else "" - processing_items = [item for item in s["queue"]["roots"]["manual"]["processingItems"] if item["path"] != current_path] - ready_items = [item for item in s["queue"]["roots"]["manual"]["items"] if item["path"] != current_path] - processing_rows = "".join(f"⏳{esc(item['name'])}
{esc(item.get('relativeDir') or item['release'])}unpacking{gib(item.get('bytes'))}waiting" for item in processing_items[:30]) - ready_rows = "".join(f"◷{esc(item['name'])}
{esc(item.get('relativeDir') or item['release'])}ready{gib(item.get('bytes'))}queued" for item in ready_items[:30]) - queue_rows = processing_rows + ready_rows or "✓No video files waiting" - history = [r for r in reversed(logs) if r.get("level") == "MOVE" and r.get("msg") == "moving"][:12] - history_rows = "".join(f"✓{esc(Path(str(r.get('dest',''))).name)}{esc(r.get('media_type',''))}{esc(r.get('ts',''))}" for r in history) or "—No recent imports" - body = f"""{esc(title)}

📥 Importarr

{esc(s['queue']['files'])} videos ready{esc(s['queue']['processingFiles'])} unpackingready {gib(s['queue']['bytes'])}1h {esc(s['processed']['last1h'])} · 24h {esc(s['processed']['last24h'])} · total {esc(s['processed']['total'])}

Status

{esc(s['health']['state'])}

Service: {esc(s['service']['activeState'])}/{esc(s['service']['subState'])}

Runtime: {secs(s['service']['runtimeSeconds'])}

Memory current: {gib(s['service']['memoryCurrentBytes'])}

Memory peak: {gib(s['service']['memoryPeakBytes'])}

Scheduled automatically every 15 minutes. `_UNPACK_` folders are shown as unpacking, not ready.

Current file

{progress_html}

Processed

Last 1h: {esc(s['processed']['last1h'])} imported

Last 24h: {esc(s['processed']['last24h'])} imported

Total: {esc(s['processed']['total'])} imported

Warnings

{warning_html}

Add manual batch

Jobs

{current_row}{queue_rows}
NameStateProgress / SizeRuntime

History

{history_rows}
NameTypeTime

Recent log

{log_text}

/api/status · /api/queue · /api/logs

""" - return body.encode() - - -class Handler(BaseHTTPRequestHandler): - def send(self, code: int, content_type: str, data: bytes) -> None: - self.send_response(code) - self.send_header("Content-Type", content_type) - self.send_header("Cache-Control", "no-store") - self.end_headers() - self.wfile.write(data) - - def do_GET(self) -> None: - parsed = urllib.parse.urlparse(self.path) - if parsed.path == "/": - self.send(200, "text/html; charset=utf-8", page()) - elif parsed.path == "/api/status": - self.send(200, "application/json", json.dumps(status()).encode()) - elif parsed.path == "/api/queue": - self.send(200, "application/json", json.dumps(queue_status()).encode()) - elif parsed.path == "/api/logs": - params = urllib.parse.parse_qs(parsed.query) - limit = int(params.get("limit", ["100"])[0]) - self.send(200, "application/json", json.dumps(read_logs(limit)).encode()) - elif parsed.path == "/health": - self.send(200, "application/json", json.dumps(fast_health()).encode()) - else: - self.send(404, "text/plain", b"not found") - - def do_POST(self) -> None: - parsed = urllib.parse.urlparse(self.path) - if parsed.path != "/api/manual-batches": - self.send(404, "text/plain", b"not found") - return - length = int(self.headers.get("content-length") or 0) - try: - payload = json.loads(self.rfile.read(length) or b"{}") - except json.JSONDecodeError: - self.send(400, "application/json", json.dumps({"ok": False, "error": "invalid json"}).encode()) - return - ok, message = add_manual_batch(str(payload.get("path") or "")) - self.send(200 if ok else 400, "application/json", json.dumps({"ok": ok, "result": message}).encode()) - - def log_message(self, fmt: str, *args: object) -> None: - return - - -def main() -> None: - host = os.getenv("IMPORTARR_BIND_HOST", "0.0.0.0") - port = int(os.getenv("IMPORTARR_BIND_PORT", "8095")) - ThreadingHTTPServer((host, port), Handler).serve_forever() - - -if __name__ == "__main__": - main() diff --git a/pyproject.toml b/pyproject.toml index cc6a006..087c84c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,6 @@ test = ["pytest>=8.2", "pytest-asyncio>=0.23"] [project.scripts] importarr = "importarr.main:run" -importarr-status = "importarr.status_ui:main" manual-media-import = "importarr.worker:main" [tool.pytest.ini_options]