diff --git a/AGENTS.md b/AGENTS.md index dab4622..30d2d13 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,6 +5,7 @@ Importarr is owned as a Linux-ops-managed service repository. Treat this checkou ## Repository Source Of Truth - Work from `/srv/opencode-workspace/importarr` for Importarr code and deploy changes. +- The live service is installed on `dgsserver1` from a repo checkout at `/opt/importarr/repo`; gmk1 installs are not the public Importarr service. - Do not edit host-local legacy scripts as the normal workflow: - `/usr/local/sbin/importarr-status.py` - `/usr/local/sbin/manual-media-import.py` @@ -26,10 +27,10 @@ When asked to implement an Importarr feature, fix, UI change, deployment change, ``` 4. Inspect `git diff` and ensure no secrets, raw `.env`, tokens, databases, or private material are included. 5. Commit and push completed Importarr changes by default unless the user explicitly asks not to publish or verification is blocked. -6. Install/restart from the repository so the running local service matches the repo: - ```sh - make upgrade-local - ``` +6. Install/restart on `dgsserver1` from the repository so the public service matches the repo: + ```sh + ssh -p 2222 opencode@dgsserver1 'sudo -n sh /opt/importarr/repo-upgrade.sh' + ``` 7. Verify the live service: ```sh make verify-live @@ -58,7 +59,8 @@ make verify-live ## Runtime Defaults On This Host -- Local URL: `http://127.0.0.1:8095/` +- Public URL: `https://importarr.delphas.dk/` +- dgsserver1 local URL: `http://127.0.0.1:8095/` - Health: `http://127.0.0.1:8095/health` - Status: `http://127.0.0.1:8095/api/status` - Systemd service: `importarr.service` diff --git a/deploy/importarr.service b/deploy/importarr.service index dd381e2..194394a 100644 --- a/deploy/importarr.service +++ b/deploy/importarr.service @@ -1,16 +1,15 @@ [Unit] -Description=Importarr manual media importer +Description=Importarr manual media importer status UI After=network-online.target Wants=network-online.target [Service] -EnvironmentFile=/etc/importarr/importarr.env -ExecStart=/opt/importarr/venv/bin/importarr +EnvironmentFile=-/etc/importarr/importarr.env +EnvironmentFile=-/opt/importarr/build.env +ExecStart=/opt/importarr/venv/bin/importarr-status Restart=on-failure RestartSec=5s -User=importarr -Group=importarr -StateDirectory=importarr +User=root [Install] WantedBy=multi-user.target diff --git a/deploy/manual-media-import-failure.service b/deploy/manual-media-import-failure.service new file mode 100644 index 0000000..5fc77fd --- /dev/null +++ b/deploy/manual-media-import-failure.service @@ -0,0 +1,6 @@ +[Unit] +Description=Critical alert when manual media importer fails + +[Service] +Type=oneshot +ExecStart=/usr/bin/python3 /usr/local/sbin/ha-critical-notify.py service-recovery "manual-media-import.service failed" --host dgsserver1 --resource manual-media-import --details "Check journalctl -u manual-media-import.service and /var/log/manual-media-import.log" diff --git a/deploy/manual-media-import.service b/deploy/manual-media-import.service new file mode 100644 index 0000000..59c5efd --- /dev/null +++ b/deploy/manual-media-import.service @@ -0,0 +1,12 @@ +[Unit] +Description=Import manual SABnzbd media into Jellyfin library roots +Wants=network-online.target +After=network-online.target +OnFailure=manual-media-import-failure.service + +[Service] +Type=oneshot +EnvironmentFile=-/etc/importarr/importarr.env +EnvironmentFile=-/opt/importarr/build.env +ExecStart=/opt/importarr/venv/bin/manual-media-import +TimeoutStartSec=30min diff --git a/deploy/manual-media-import.timer b/deploy/manual-media-import.timer new file mode 100644 index 0000000..e4f647a --- /dev/null +++ b/deploy/manual-media-import.timer @@ -0,0 +1,12 @@ +[Unit] +Description=Run manual media importer periodically + +[Timer] +OnBootSec=5min +OnUnitActiveSec=15min +AccuracySec=1min +Persistent=true +Unit=manual-media-import.service + +[Install] +WantedBy=timers.target diff --git a/deploy/repo-upgrade.sh b/deploy/repo-upgrade.sh index 7330c5c..6555d10 100644 --- a/deploy/repo-upgrade.sh +++ b/deploy/repo-upgrade.sh @@ -31,5 +31,12 @@ fi git fetch --prune origin git pull --ff-only "$VENV/bin/pip" install --upgrade "$REPO_DIR" +GIT_SHA="$(git rev-parse --short=12 HEAD 2>/dev/null || printf development)" +BUILD_DATE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" +cat > /opt/importarr/build.env </dev/null 2>&1; then - useradd --system --home /var/lib/importarr --shell /usr/sbin/nologin importarr -fi -chown importarr:importarr /var/lib/importarr python3 -m venv /opt/importarr/venv /opt/importarr/venv/bin/pip install --upgrade pip /opt/importarr/venv/bin/pip install --upgrade "$REPO_DIR" @@ -24,7 +20,19 @@ if ! grep -q '^IMPORTARR_REPO_DIR=' /etc/importarr/importarr.env; then printf '\nIMPORTARR_REPO_DIR=%s\n' "$REPO_DIR" >> /etc/importarr/importarr.env fi install -m 0644 "$REPO_DIR/deploy/importarr.service" /etc/systemd/system/importarr.service +install -m 0644 "$REPO_DIR/deploy/manual-media-import.service" /etc/systemd/system/manual-media-import.service +install -m 0644 "$REPO_DIR/deploy/manual-media-import.timer" /etc/systemd/system/manual-media-import.timer +install -m 0644 "$REPO_DIR/deploy/manual-media-import-failure.service" /etc/systemd/system/manual-media-import-failure.service install -m 0755 "$REPO_DIR/deploy/repo-upgrade.sh" /opt/importarr/repo-upgrade.sh +GIT_SHA="$(git -C "$REPO_DIR" rev-parse --short=12 HEAD 2>/dev/null || printf development)" +BUILD_DATE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" +cat > /opt/importarr/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/importarr/worker.py b/importarr/worker.py new file mode 100644 index 0000000..a671604 --- /dev/null +++ b/importarr/worker.py @@ -0,0 +1,1096 @@ +#!/usr/bin/env python3 +"""Import manual SABnzbd downloads into Jellyfin movie/TV library roots. + +Run on dgsserver1. Intended live install path: +`/usr/local/sbin/manual-media-import.py`, managed by +`manual-media-import.timer`. +""" + +from __future__ import annotations + +import argparse +import fcntl +import gzip +import json +import os +import re +import shutil +import sqlite3 +import sys +import time +import urllib.parse +import urllib.request +import xml.etree.ElementTree as ET +from collections import Counter +from datetime import datetime +from pathlib import Path + +MANUAL_DOWNLOADS = Path("/srv/scrypted/sabnzbd-data/downloads/manual") +LEGACY_DANISH_DOWNLOADS = Path("/srv/scrypted/sabnzbd-data/downloads/danish") +DOWNLOAD_ROOTS = [MANUAL_DOWNLOADS, LEGACY_DANISH_DOWNLOADS] +SAB_TRANSIENT_PREFIXES = ("_UNPACK_", "__UNPACK__", "_FAILED_", "_ADMIN_") + +HOST_MEDIA_PREFIX = "/srv/media" +MOVIES_ROOT = Path("/srv/media/movies") +TV_ROOT = Path("/srv/media/tv") + +LOG = Path("/var/log/manual-media-import.log") +LOCK = Path("/run/manual-media-import.lock") +STATUS = Path("/run/manual-media-import/status.json") +MANUAL_BATCHES = Path("/var/lib/importarr/manual-batches.json") +HOME_ASSISTANT_WEBHOOK = "https://hass.delphas.dk/api/webhook/jellyfin_event" + +RADARR_CONFIG = Path("/opt/stacks/media-transform/radarr/config/config.xml") +RADARR_URL = "http://127.0.0.1:7878" +RADARR_DB = Path("/opt/stacks/media-transform/radarr/config/radarr.db") +SONARR_CONFIG = Path("/opt/stacks/media-transform/sonarr/config/config.xml") +SONARR_URL = "http://127.0.0.1:8989" +SONARR_DB = Path("/opt/stacks/media-transform/sonarr/config/sonarr.db") +SABNZBD_CONFIG = Path("/opt/stacks/media-transform/sabnzbd/config/sabnzbd.ini") +SABNZBD_URL = "http://127.0.0.1:8080/api" +ARR_API_TIMEOUT = 10 + +VIDEO_EXT = {".mkv", ".mp4", ".m4v", ".avi", ".mov", ".wmv", ".mpg", ".mpeg", ".ts", ".m2ts", ".webm"} +SIDECAR_EXT = {".srt", ".ass", ".sub", ".idx", ".nfo"} +MIN_SIZE = 100 * 1024 * 1024 +IMDB_DATASET_CANDIDATES = [ + Path("/srv/media/imdb"), + Path("/srv/media/imdb-datasets"), + Path("/srv/media/.cache/imdb"), + Path("/var/lib/imdb"), + Path("/opt/imdb"), +] + +STOP_TOKENS = { + "x264", + "x265", + "h264", + "h265", + "hevc", + "aac", + "ac3", + "dts", + "ddp5", + "atmos", + "hdr", + "dv", + "repack", + "proper", + "remux", + "internal", + "sample", + "mkv", + "mp4", + "avi", +} + + +class Skip(Exception): + pass + + +def log(level: str, msg: str, **kw: object) -> None: + LOG.parent.mkdir(parents=True, exist_ok=True) + rec = {"ts": datetime.now().isoformat(timespec="seconds"), "level": level, "msg": msg, **kw} + with LOG.open("a", encoding="utf-8") as f: + f.write(json.dumps(rec, ensure_ascii=False) + "\n") + print(f"[{level}] {msg} {kw}") + + +def write_status(phase: str, **kw: object) -> None: + STATUS.parent.mkdir(parents=True, exist_ok=True) + payload = {"ts": datetime.now().isoformat(timespec="seconds"), "phase": phase, **kw} + tmp = STATUS.with_suffix(".json.tmp") + tmp.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + tmp.replace(STATUS) + + +def norm(text: str) -> str: + return re.sub(r"[^a-z0-9]+", " ", text.lower()).strip() + + +def toks(text: str) -> list[str]: + return [t for t in norm(text).split() if t not in {"the", "and", "in", "of", "a", "an"}] + + +def title_match_score(query: str, candidate: str) -> int: + qn = norm(query) + cn = norm(candidate) + if not qn or not cn: + return 0 + if qn == cn: + return 100 + qt = set(toks(query)) + ct = set(toks(candidate)) + if not qt or not ct: + return 0 + overlap = len(qt & ct) + if overlap == 0: + return 0 + return overlap * 10 - abs(len(qt) - overlap) + + +def sanitize_title(title: str) -> str: + title = re.sub(r"[\\/:*?\"<>|]", " ", title) + return re.sub(r"\s+", " ", title).strip().strip(".") + + +def open_tsv_dataset(base_dir: Path, stem: str): + plain = base_dir / f"{stem}.tsv" + gz = base_dir / f"{stem}.tsv.gz" + if plain.is_file(): + return plain.open("rt", encoding="utf-8", errors="replace") + if gz.is_file(): + return gzip.open(gz, "rt", encoding="utf-8", errors="replace") + return None + + +def discover_imdb_dataset_dir() -> Path | None: + env_dir = os.environ.get("IMDB_DATASET_DIR") + candidates = [Path(env_dir)] if env_dir else [] + candidates.extend(IMDB_DATASET_CANDIDATES) + for base in candidates: + if (base / "title.basics.tsv").is_file() or (base / "title.basics.tsv.gz").is_file(): + return base + return None + + +def resolve_with_imdb_datasets(source_title: str, year: int, imdb_dir: Path) -> tuple[str, int] | None: + basics = open_tsv_dataset(imdb_dir, "title.basics") + if basics is None: + return None + wanted = range(max(1888, year - 1), year + 2) + candidates: list[dict[str, object]] = [] + try: + next(basics, None) + for line in basics: + cols = line.rstrip("\n").split("\t") + if len(cols) < 9: + continue + tconst, title_type, primary, original, _, _, start_year, _, _ = cols[:9] + if title_type not in {"movie", "tvMovie"} or start_year == "\\N": + continue + try: + movie_year = int(start_year) + except ValueError: + continue + if movie_year not in wanted: + continue + score = max(title_match_score(source_title, primary), title_match_score(source_title, original)) + if score > 0: + candidates.append({"tconst": tconst, "title": primary, "year": movie_year, "score": score}) + finally: + basics.close() + + if not candidates: + return None + + akas = open_tsv_dataset(imdb_dir, "title.akas") + if akas is not None: + boosts = {str(c["tconst"]): 0 for c in candidates} + try: + next(akas, None) + wanted_ids = set(boosts) + for line in akas: + cols = line.rstrip("\n").split("\t") + if len(cols) < 3 or cols[0] not in wanted_ids: + continue + boosts[cols[0]] = max(boosts[cols[0]], title_match_score(source_title, cols[2])) + finally: + akas.close() + for c in candidates: + c["score"] = int(c["score"]) + boosts.get(str(c["tconst"]), 0) + + candidates.sort(key=lambda c: (-int(c["score"]), abs(int(c["year"]) - year), str(c["title"]))) + top = candidates[0] + if len(candidates) > 1 and candidates[1]["score"] == top["score"]: + return None + return sanitize_title(str(top["title"])), int(top["year"]) + + +def quality_source_tokens(name: str) -> list[str]: + text = Path(name).stem.replace(".", " ").replace("_", " ") + words = text.split() + lower_words = [w.lower() for w in words] + parts: list[str] = [] + + quality = next((w for w in words if re.fullmatch(r"(?i)(2160p|1080p|720p|480p)", w)), None) + if quality: + parts.append(quality) + + if "bluray" in lower_words or ("blu" in lower_words and "ray" in lower_words): + parts.append("BluRay") + elif any(w in lower_words for w in {"web", "web-dl", "webdl", "webrip"}): + parts.append("WEB-DL") + elif "hdtv" in lower_words: + parts.append("HDTV") + + for w in reversed(words): + clean = re.sub(r"[^A-Za-z0-9-]", "", w) + if not clean: + continue + if clean.lower() in STOP_TOKENS: + continue + if re.fullmatch(r"\d{4}|\d+", clean): + continue + parts.append(clean) + break + + # preserve order and dedupe + out: list[str] = [] + seen: set[str] = set() + for p in parts: + k = p.lower() + if k in seen: + continue + seen.add(k) + out.append(p) + return out + + +def safe_label(filename: str, source_tag: str, for_existing: bool = False) -> str: + tokens = quality_source_tokens(filename) + if for_existing: + prefix = "Existing" + elif source_tag == "manual": + prefix = "Manual" + elif source_tag == "danish-legacy": + prefix = "Nordic" + else: + prefix = "Imported" + if tokens: + return f"{prefix} {' '.join(tokens)}" + return prefix + + +def stable(path: Path) -> bool: + s1 = path.stat().st_size + time.sleep(5) + return s1 == path.stat().st_size + + +def unique(folder: Path, base: str, ext: str) -> Path: + p = folder / (base + ext) + i = 2 + while p.exists(): + p = folder / (f"{base} {i}" + ext) + i += 1 + return p + + +def api_key_from_config(config_path: Path) -> str | None: + if not config_path.is_file(): + return None + key = ET.parse(config_path).findtext("ApiKey") + return key or None + + +def sabnzbd_api_key() -> str | None: + try: + m = re.search(r"^api_key\s*=\s*(\S+)", SABNZBD_CONFIG.read_text(errors="replace"), re.M) + except OSError: + return None + return m.group(1) if m else None + + +def sabnzbd_history() -> list[dict[str, object]]: + key = sabnzbd_api_key() + if not key: + return [] + url = SABNZBD_URL + "?" + urllib.parse.urlencode({"mode": "history", "output": "json", "limit": 200, "apikey": key}) + try: + data = json.load(urllib.request.urlopen(url, timeout=ARR_API_TIMEOUT)) + except Exception as exc: + log("WARN", "SABnzbd history unavailable; using filesystem-only readiness", error=str(exc)) + return [] + return data.get("history", {}).get("slots", []) + + +def sabnzbd_readiness_by_name() -> dict[str, bool]: + out: dict[str, bool] = {} + for item in sabnzbd_history(): + name = str(item.get("name") or "") + if not name: + continue + status = str(item.get("status") or "") + storage = str(item.get("storage") or "") + category = str(item.get("category") or item.get("cat") or "") + out[name] = category == "manual" and status == "Completed" and bool(storage) and "_UNPACK_" not in storage + return out + + +def manual_batch_roots() -> list[Path]: + try: + raw = json.loads(MANUAL_BATCHES.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return [] + out = [] + for item in raw if isinstance(raw, list) else []: + try: + p = Path(str(item)).resolve() + if any(p == root.resolve() or p.is_relative_to(root.resolve()) for root in DOWNLOAD_ROOTS): + out.append(p) + except OSError: + continue + return out + + +def under_manual_batch(path: Path, batches: list[Path]) -> bool: + try: + resolved = path.resolve() + except OSError: + return False + return any(resolved == batch or resolved.is_relative_to(batch) for batch in batches) + + +def prune_manual_batches() -> None: + batches = manual_batch_roots() + keep = [] + for batch in batches: + if any(p.is_file() and p.suffix.lower() in VIDEO_EXT for p in batch.rglob("*")): + keep.append(str(batch)) + MANUAL_BATCHES.parent.mkdir(parents=True, exist_ok=True) + MANUAL_BATCHES.write_text(json.dumps(keep, indent=2), encoding="utf-8") + + +def radarr_movies() -> list[dict[str, object]]: + key = api_key_from_config(RADARR_CONFIG) + if not key: + return [] + req = urllib.request.Request(RADARR_URL + "/api/v3/movie", headers={"X-Api-Key": key}) + return json.load(urllib.request.urlopen(req, timeout=ARR_API_TIMEOUT)) + + +def sonarr_series() -> list[dict[str, object]]: + key = api_key_from_config(SONARR_CONFIG) + if not key: + return [] + req = urllib.request.Request(SONARR_URL + "/api/v3/series", headers={"X-Api-Key": key}) + return json.load(urllib.request.urlopen(req, timeout=ARR_API_TIMEOUT)) + + +def arr_get_json(base_url: str, api_key: str, api_path: str, params: dict[str, object] | None = None) -> object: + url = base_url + api_path + if params: + url += "?" + urllib.parse.urlencode(params) + req = urllib.request.Request(url, headers={"X-Api-Key": api_key}) + return json.load(urllib.request.urlopen(req, timeout=ARR_API_TIMEOUT)) + + +def arr_post_json(base_url: str, api_key: str, api_path: str, payload: dict[str, object]) -> dict[str, object]: + req = urllib.request.Request( + base_url + api_path, + data=json.dumps(payload).encode("utf-8"), + headers={"X-Api-Key": api_key, "Content-Type": "application/json"}, + method="POST", + ) + return json.load(urllib.request.urlopen(req, timeout=ARR_API_TIMEOUT)) + + +def most_common_nonempty(values: list[object]) -> object | None: + usable = [v for v in values if v not in (None, "")] + if not usable: + return None + return Counter(usable).most_common(1)[0][0] + + +def arr_path_from_host_path(host_path: Path) -> str | None: + try: + rel = host_path.resolve().relative_to(Path(HOST_MEDIA_PREFIX)) + except ValueError: + return None + return "/data/" + str(rel).replace("\\", "/") + + +def ensure_radarr_unmonitored_movie(meta: dict[str, object], radarr_catalog: list[dict[str, object]]) -> object | None: + if meta.get("id"): + return meta.get("id") + + api_key = api_key_from_config(RADARR_CONFIG) + if not api_key: + return None + + title = str(meta.get("title") or "").strip() + year = meta.get("year") + if not title: + return None + + root_folder = arr_path_from_host_path(MOVIES_ROOT) or "/data/movies" + default_quality = most_common_nonempty([m.get("qualityProfileId") for m in radarr_catalog]) + default_min_avail = most_common_nonempty([m.get("minimumAvailability") for m in radarr_catalog]) + + term = title if year is None else f"{title} {year}" + lookup = arr_get_json(RADARR_URL, api_key, "/api/v3/movie/lookup", {"term": term}) + if not isinstance(lookup, list): + return None + + best: dict[str, object] | None = None + best_score = 0 + for item in lookup: + if not isinstance(item, dict): + continue + item_title = str(item.get("title") or "") + score = title_match_score(title, item_title) + item_year = item.get("year") + if year is not None and item_year == year: + score += 20 + if score > best_score: + best_score = score + best = item + if not best or best_score <= 0: + return None + + payload = dict(best) + payload["rootFolderPath"] = root_folder + payload["monitored"] = False + payload["addOptions"] = {"searchForMovie": False} + if default_quality is not None: + payload["qualityProfileId"] = default_quality + if default_min_avail is not None: + payload["minimumAvailability"] = default_min_avail + + created = arr_post_json(RADARR_URL, api_key, "/api/v3/movie", payload) + movie_id = created.get("id") + log("INFO", "added missing movie to Radarr as unmonitored", title=title, year=year, id=movie_id) + return movie_id + + +def ensure_sonarr_unmonitored_series(meta: dict[str, object], sonarr_catalog: list[dict[str, object]]) -> object | None: + if meta.get("id"): + return meta.get("id") + + api_key = api_key_from_config(SONARR_CONFIG) + if not api_key: + return None + + show_title = str(meta.get("series_name") or meta.get("title") or "").strip() + if not show_title: + return None + + root_folder = arr_path_from_host_path(TV_ROOT) or "/data/tv" + default_quality = most_common_nonempty([s.get("qualityProfileId") for s in sonarr_catalog]) + default_language = most_common_nonempty([s.get("languageProfileId") for s in sonarr_catalog]) + + lookup = arr_get_json(SONARR_URL, api_key, "/api/v3/series/lookup", {"term": show_title}) + if not isinstance(lookup, list): + return None + + best: dict[str, object] | None = None + best_score = 0 + for item in lookup: + if not isinstance(item, dict): + continue + item_title = str(item.get("title") or "") + score = title_match_score(show_title, item_title) + if score > best_score: + best_score = score + best = item + if not best or best_score <= 0: + return None + + payload = dict(best) + payload["rootFolderPath"] = root_folder + payload["monitored"] = False + payload["addOptions"] = {"searchForMissingEpisodes": False} + if default_quality is not None: + payload["qualityProfileId"] = default_quality + if default_language is not None: + payload["languageProfileId"] = default_language + + seasons = payload.get("seasons") + if isinstance(seasons, list): + for season in seasons: + if isinstance(season, dict): + season["monitored"] = False + + created = arr_post_json(SONARR_URL, api_key, "/api/v3/series", payload) + series_id = created.get("id") + log("INFO", "added missing show to Sonarr as unmonitored", title=show_title, id=series_id) + return series_id + + +def parse_bool(value: object) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes", "on"} + return False + + +def jellyfin_settings_from_arr_db(db_path: Path, source: str) -> dict[str, object] | None: + if not db_path.is_file(): + return None + try: + with sqlite3.connect(db_path) as conn: + row = conn.execute( + "SELECT Settings FROM Notifications WHERE Implementation = ? AND Name = ? ORDER BY Id DESC LIMIT 1", + ("MediaBrowser", "Jellyfin"), + ).fetchone() + except sqlite3.Error as exc: + log("WARN", "failed reading Arr notification settings", source=source, db=str(db_path), error=str(exc)) + return None + + if not row or not row[0]: + return None + + try: + settings = json.loads(str(row[0])) + except json.JSONDecodeError as exc: + log("WARN", "invalid Arr Jellyfin settings JSON", source=source, db=str(db_path), error=str(exc)) + return None + + host = str(settings.get("host") or "").strip() + api_key = str(settings.get("apiKey") or "").strip() + if not host or not api_key: + return None + + host = re.sub(r"^https?://", "", host).rstrip("/") + if "/" in host: + host = host.split("/", 1)[0] + if not host: + return None + + raw_port = settings.get("port") + try: + port = int(raw_port) + except (TypeError, ValueError): + port = 8096 + + return { + "source": source, + "host": host, + "port": port, + "use_ssl": parse_bool(settings.get("useSsl")), + "api_key": api_key, + } + + +def jellyfin_settings_from_arr_notifications() -> dict[str, object] | None: + for source, db_path in (("radarr", RADARR_DB), ("sonarr", SONARR_DB)): + settings = jellyfin_settings_from_arr_db(db_path, source) + if settings: + return settings + return None + + +def refresh_jellyfin_library() -> None: + settings = jellyfin_settings_from_arr_notifications() + if not settings: + log("WARN", "Jellyfin refresh skipped: no Arr Jellyfin connector settings found") + return + + scheme = "https" if bool(settings["use_ssl"]) else "http" + host = str(settings["host"]) + port = int(settings["port"]) + url = f"{scheme}://{host}:{port}/Library/Refresh" + payload = b"{}" + + req = urllib.request.Request( + url, + data=payload, + headers={"Content-Type": "application/json", "X-Emby-Token": str(settings["api_key"])}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=ARR_API_TIMEOUT) as resp: + log("INFO", "requested Jellyfin library refresh", status=resp.status, source=settings["source"], host=host, port=port) + except Exception as exc: + log( + "WARN", + "Jellyfin library refresh failed; continuing", + source=settings["source"], + host=host, + port=port, + error=str(exc), + ) + + +def host_path_from_arr_path(arr_path: str | None) -> Path | None: + if not arr_path or not arr_path.startswith("/data"): + return None + return Path(HOST_MEDIA_PREFIX + arr_path[5:]) + + +def parse_movie_title_year(name: str) -> tuple[str, int] | None: + stem = Path(name).stem.replace(".", " ").replace("_", " ") + ym = re.search(r"\b(19\d{2}|20\d{2})\b", stem) + if not ym: + return None + year = int(ym.group(1)) + title = re.sub(r"\s+", " ", stem[: ym.start()]).strip(" .-_") + if not title: + return None + return title, year + + +def parse_episode_info(name: str) -> dict[str, object] | None: + stem = Path(name).stem.replace("_", " ").replace(".", " ") + + m = re.search(r"\b[Ss](\d{1,2})[ ._-]*[Ee](\d{2})(?:[ ._-]*[Ee](\d{2}))?\b", stem) + if m: + season = int(m.group(1)) + e1 = int(m.group(2)) + e2 = int(m.group(3)) if m.group(3) else None + title = re.sub(r"\s+", " ", stem[: m.start()]).strip(" .-_") + episode_title = re.sub(r"\s+", " ", stem[m.end() :]).strip(" .-_") or None + if title: + return { + "show_title": title, + "season": season, + "episode": e1, + "episode_end": e2, + "air_date": None, + "episode_title": episode_title, + } + + m = re.search(r"\b(\d{1,2})x(\d{2})(?:x(\d{2}))?\b", stem) + if m: + season = int(m.group(1)) + e1 = int(m.group(2)) + e2 = int(m.group(3)) if m.group(3) else None + title = re.sub(r"\s+", " ", stem[: m.start()]).strip(" .-_") + episode_title = re.sub(r"\s+", " ", stem[m.end() :]).strip(" .-_") or None + if title: + return { + "show_title": title, + "season": season, + "episode": e1, + "episode_end": e2, + "air_date": None, + "episode_title": episode_title, + } + + m = re.search(r"\b(20\d{2})[ ._-](\d{2})[ ._-](\d{2})\b", stem) + if m: + air_date = f"{m.group(1)}-{m.group(2)}-{m.group(3)}" + season = int(m.group(1)) + title = re.sub(r"\s+", " ", stem[: m.start()]).strip(" .-_") + if title: + return { + "show_title": title, + "season": season, + "episode": None, + "episode_end": None, + "air_date": air_date, + "episode_title": None, + } + + return None + + +def useful_release_name(src: Path) -> str: + ignored = {"manual", "danish", "downloads"} + for parent in [src.parent, *src.parents]: + name = parent.name.strip() + lower = name.lower() + if not name or lower in ignored: + continue + if lower.startswith(("leftover-cleanup-", "_unpack_", "__unpack__", "_failed_", "_admin_")): + continue + if parent in DOWNLOAD_ROOTS: + break + return name + return src.stem + + +def movie_label_source_name(src: Path) -> str: + return src.name if parse_movie_title_year(src.name) else useful_release_name(src) + + +def infer_media_type(src: Path) -> str: + info = parse_episode_info(src.name) + if info: + return "tv" + rel = str(src).lower() + if any(token in rel for token in ["/tv/", "season", "episode", "s0", " s1", " e0"]): + return "tv" + return "movie" + + +def resolve_movie_folder(src: Path, radarr_catalog: list[dict[str, object]]) -> tuple[Path, dict[str, object]]: + release_name = useful_release_name(src) + parsed = parse_movie_title_year(src.name) or parse_movie_title_year(release_name) + matches: list[dict[str, object]] = [] + if parsed and radarr_catalog: + source_title, year = parsed + for movie in radarr_catalog: + if movie.get("year") != year: + continue + title = str(movie.get("title") or "") + if title_match_score(source_title, title) > 0: + matches.append(movie) + matches.sort(key=lambda m: -title_match_score(parsed[0], str(m.get("title") or ""))) + + if len(matches) == 1: + m = matches[0] + host = host_path_from_arr_path(str(m.get("path") or "")) + if host: + host.mkdir(parents=True, exist_ok=True) + return host, {"title": m.get("title"), "year": m.get("year"), "id": m.get("id")} + + if parsed: + source_title, year = parsed + imdb_dir = discover_imdb_dataset_dir() + if imdb_dir: + resolved = resolve_with_imdb_datasets(source_title, year, imdb_dir) + if resolved: + title, resolved_year = resolved + target = MOVIES_ROOT / f"{title} ({resolved_year})" + target.mkdir(parents=True, exist_ok=True) + log("INFO", "resolved movie via local IMDb datasets", src=str(src), dataset=str(imdb_dir), title=title, year=resolved_year) + return target, {"title": title, "year": resolved_year, "id": None} + log("WARN", "IMDb datasets present but no unique movie match", src=str(src), dataset=str(imdb_dir), source_title=source_title, year=year) + + target = MOVIES_ROOT / f"{sanitize_title(source_title)} ({year})" + target.mkdir(parents=True, exist_ok=True) + return target, {"title": sanitize_title(source_title), "year": year, "id": None} + + fallback_title = sanitize_title(release_name) + target = MOVIES_ROOT / fallback_title + target.mkdir(parents=True, exist_ok=True) + return target, {"title": fallback_title, "year": None, "id": None} + + +def find_best_series_folder(show_title: str, sonarr_catalog: list[dict[str, object]]) -> tuple[str, Path, object | None]: + sonarr_matches: list[tuple[int, dict[str, object]]] = [] + for series in sonarr_catalog: + title = str(series.get("title") or "") + score = title_match_score(show_title, title) + if score > 0: + sonarr_matches.append((score, series)) + sonarr_matches.sort(key=lambda item: -item[0]) + + if sonarr_matches: + best_score = sonarr_matches[0][0] + top = [item for item in sonarr_matches if item[0] == best_score] + if len(top) == 1: + series = top[0][1] + host = host_path_from_arr_path(str(series.get("path") or "")) + if host: + host.mkdir(parents=True, exist_ok=True) + return str(series.get("title") or show_title), host, series.get("id") + + if TV_ROOT.is_dir(): + library_matches: list[tuple[int, Path]] = [] + for entry in TV_ROOT.iterdir(): + if not entry.is_dir(): + continue + score = title_match_score(show_title, entry.name) + if score > 0: + library_matches.append((score, entry)) + library_matches.sort(key=lambda item: (-item[0], item[1].name.lower())) + if library_matches: + top_score = library_matches[0][0] + top = [item for item in library_matches if item[0] == top_score] + if len(top) == 1: + return top[0][1].name, top[0][1], None + + folder = TV_ROOT / sanitize_title(show_title) + folder.mkdir(parents=True, exist_ok=True) + return sanitize_title(show_title), folder, None + + +def resolve_tv_destination(src: Path, sonarr_catalog: list[dict[str, object]]) -> tuple[Path, dict[str, object], str]: + info = parse_episode_info(src.name) + if not info: + raise Skip("TV candidate missing season/episode number") + + show_name, show_folder, series_id = find_best_series_folder(str(info["show_title"]), sonarr_catalog) + season_num = int(info["season"]) + season_folder = show_folder / f"Season {season_num}" + season_folder.mkdir(parents=True, exist_ok=True) + + if info["air_date"]: + ep = f"{show_name} - {info['air_date']}" + episode_num = None + else: + e1 = int(info["episode"]) + e2 = info["episode_end"] + episode_num = e1 + if e2 is not None: + ep = f"{show_name} - S{season_num:02d}E{e1:02d}-E{int(e2):02d}" + else: + ep = f"{show_name} - S{season_num:02d}E{e1:02d}" + + meta = { + "title": show_name, + "year": None, + "id": series_id, + "item_type": "Episode", + "series_name": show_name, + "season_number": season_num, + "episode_number": episode_num, + "name": info.get("episode_title"), + } + return season_folder, meta, ep + + +def normalize_existing_movie_versions(folder: Path) -> None: + prefix = folder.name + " - " + for p in list(folder.iterdir()): + if not p.is_file() or p.suffix.lower() not in VIDEO_EXT: + continue + if p.name.startswith(prefix): + continue + label = safe_label(p.name, source_tag="existing", for_existing=True) + dest = unique(folder, f"{folder.name} - {label}", p.suffix) + log("MOVE", "normalizing existing movie filename", src=str(p), dest=str(dest)) + p.rename(dest) + + +def copy_then_remove(src: Path, dest: Path, *, media_type: str | None = None, source_tag: str | None = None, kind: str = "media") -> None: + partial = dest.with_name(dest.name + ".partial") + if partial.exists(): + log("WARN", "removing stale partial", partial=str(partial)) + partial.unlink() + total = src.stat().st_size + copied = 0 + last_status = 0.0 + write_status("copying", src=str(src), dest=str(dest), partial=str(partial), bytes_copied=0, bytes_total=total, percent=0, media_type=media_type, source_tag=source_tag, kind=kind) + with src.open("rb") as source, partial.open("wb") as target: + shutil.copystat(str(src), str(partial), follow_symlinks=True) + while True: + chunk = source.read(16 * 1024 * 1024) + if not chunk: + break + target.write(chunk) + copied += len(chunk) + now = time.monotonic() + if now - last_status >= 1 or copied == total: + last_status = now + write_status("copying", src=str(src), dest=str(dest), partial=str(partial), bytes_copied=copied, bytes_total=total, percent=round((copied / total * 100), 2) if total else 100, media_type=media_type, source_tag=source_tag, kind=kind) + if partial.stat().st_size != total: + raise RuntimeError(f"partial size mismatch {partial.stat().st_size} != {total}") + write_status("finalizing", src=str(src), dest=str(dest), partial=str(partial), bytes_copied=total, bytes_total=total, percent=100, media_type=media_type, source_tag=source_tag, kind=kind) + partial.rename(dest) + src.unlink() + + +def notify_home_assistant(item: dict[str, object], dest: Path, source_tag: str) -> None: + item_type = str(item.get("item_type") or "Movie") + name = item.get("title") or dest.parent.name + payload_body: dict[str, object] = { + "NotificationType": "ItemAdded", + "Name": name, + "ItemType": item_type, + "Year": item.get("year"), + "ItemId": str(item.get("id") or ""), + "Source": f"manual-media-import:{source_tag}", + "Path": str(dest), + } + if item_type == "Episode": + if not item.get("series_name") or item.get("season_number") is None or item.get("episode_number") is None: + log("WARN", "not sending episode notification without series/season/episode", item=item, dest=str(dest)) + return + payload_body["Name"] = item.get("name") or "" + payload_body["SeriesName"] = item.get("series_name") or name + payload_body["SeasonNumber"] = item.get("season_number") + payload_body["EpisodeNumber"] = item.get("episode_number") + + payload = json.dumps( + payload_body + ).encode() + req = urllib.request.Request( + HOME_ASSISTANT_WEBHOOK, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=15) as resp: + log("INFO", "sent Home Assistant media notification", item=item.get("title"), status=resp.status) + + +def source_tag_for(path: Path) -> str: + for root in DOWNLOAD_ROOTS: + try: + path.relative_to(root) + return "manual" if root == MANUAL_DOWNLOADS else "danish-legacy" + except ValueError: + continue + return "manual" + + +def candidates() -> list[Path]: + out: list[Path] = [] + sab_ready = sabnzbd_readiness_by_name() + manual_batches = manual_batch_roots() + for root in DOWNLOAD_ROOTS: + if not root.exists(): + continue + for dirpath, _, files in os.walk(root): + rel_parts = Path(dirpath).relative_to(root).parts if Path(dirpath) != root else () + if any(part.startswith(SAB_TRANSIENT_PREFIXES) for part in rel_parts): + continue + if rel_parts and rel_parts[0] in sab_ready and not sab_ready[rel_parts[0]]: + continue + if rel_parts and rel_parts[0] not in sab_ready and not under_manual_batch(Path(dirpath), manual_batches): + continue + if "sample" in dirpath.lower(): + continue + for f in files: + p = Path(dirpath) / f + if p.suffix.lower() in VIDEO_EXT and "sample" not in f.lower(): + out.append(p) + return out + + +def ignored_source_files() -> list[Path]: + out: list[Path] = [] + for root in DOWNLOAD_ROOTS: + if not root.exists(): + continue + for dirpath, _, files in os.walk(root): + rel_parts = Path(dirpath).relative_to(root).parts if Path(dirpath) != root else () + if any(part.startswith(SAB_TRANSIENT_PREFIXES) for part in rel_parts): + continue + sample_root = "sample" in dirpath.lower() + for f in files: + p = Path(dirpath) / f + if sample_root or "sample" in f.lower() or p.suffix.lower() not in VIDEO_EXT: + out.append(p) + return out + + +def cleanup_source_directories() -> tuple[int, int]: + removed_files = removed_dirs = 0 + for p in ignored_source_files(): + try: + log("DELETE", "removing ignored source file", src=str(p)) + p.unlink() + removed_files += 1 + except FileNotFoundError: + continue + for root in DOWNLOAD_ROOTS: + if not root.exists(): + continue + for dirpath, dirs, _ in os.walk(root, topdown=False): + for d in dirs: + p = Path(dirpath) / d + try: + p.rmdir() + log("DELETE", "removing empty source directory", src=str(p)) + removed_dirs += 1 + except OSError: + pass + return removed_files, removed_dirs + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + LOCK.parent.mkdir(parents=True, exist_ok=True) + lock_f = LOCK.open("w", encoding="utf-8") + try: + fcntl.flock(lock_f, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + log("WARN", "another importer run is active") + return 0 + + try: + write_status("starting") + radarr_catalog = radarr_movies() + except Exception as exc: # optional hint only + log("WARN", "Radarr lookup unavailable; continuing without it", error=str(exc)) + radarr_catalog = [] + + try: + sonarr_catalog = sonarr_series() + except Exception as exc: # optional hint only + log("WARN", "Sonarr lookup unavailable; continuing without it", error=str(exc)) + sonarr_catalog = [] + + moved = skipped = errors = 0 + radarr_added: dict[tuple[str, object], object | None] = {} + sonarr_added: dict[str, object | None] = {} + for src in candidates(): + source_tag = source_tag_for(src) + try: + if src.stat().st_size < MIN_SIZE: + raise Skip("too small") + if not stable(src): + raise Skip("file still changing") + + media_type = infer_media_type(src) + if media_type == "tv": + dest_folder, meta, episode_stub = resolve_tv_destination(src, sonarr_catalog) + sonarr_key = norm(str(meta.get("series_name") or meta.get("title") or "")) + if sonarr_key and meta.get("id") is None: + if sonarr_key in sonarr_added: + meta["id"] = sonarr_added[sonarr_key] + else: + try: + meta["id"] = ensure_sonarr_unmonitored_series(meta, sonarr_catalog) + except Exception as exc: + log("WARN", "failed adding missing show to Sonarr; continuing", title=meta.get("series_name") or meta.get("title"), error=str(exc)) + sonarr_added[sonarr_key] = meta.get("id") + label = safe_label(src.name, source_tag=source_tag) + dest = unique(dest_folder, f"{episode_stub} - {label}", src.suffix) + else: + dest_folder, meta = resolve_movie_folder(src, radarr_catalog) + radarr_key = (norm(str(meta.get("title") or "")), meta.get("year")) + if radarr_key[0] and meta.get("id") is None: + if radarr_key in radarr_added: + meta["id"] = radarr_added[radarr_key] + else: + try: + meta["id"] = ensure_radarr_unmonitored_movie(meta, radarr_catalog) + except Exception as exc: + log("WARN", "failed adding missing movie to Radarr; continuing", title=meta.get("title"), year=meta.get("year"), error=str(exc)) + radarr_added[radarr_key] = meta.get("id") + normalize_existing_movie_versions(dest_folder) + label = safe_label(movie_label_source_name(src), source_tag=source_tag) + dest = unique(dest_folder, f"{dest_folder.name} - {label}", src.suffix) + + if args.dry_run: + log("DRYRUN", "would move", src=str(src), dest=str(dest), media_type=media_type) + continue + + log("MOVE", "moving", src=str(src), dest=str(dest), media_type=media_type, source_tag=source_tag) + copy_then_remove(src, dest, media_type=media_type, source_tag=source_tag) + try: + notify_home_assistant(meta, dest, source_tag=source_tag) + except Exception as exc: + log("WARN", "failed Home Assistant media notification", item=meta.get("title"), error=str(exc)) + + moved += 1 + + for side in list(src.parent.iterdir()): + if side.is_file() and side.stem == src.stem and side.suffix.lower() in SIDECAR_EXT: + side_dest = unique(dest.parent, dest.stem, side.suffix) + log("MOVE", "moving sidecar", src=str(side), dest=str(side_dest)) + copy_then_remove(side, side_dest, media_type=media_type, source_tag=source_tag, kind="sidecar") + except Skip as exc: + skipped += 1 + log("SKIP", str(exc), src=str(src)) + except Exception as exc: + errors += 1 + log("ERROR", "failed candidate", src=str(src), error=str(exc)) + + cleaned_files = cleaned_dirs = 0 + if moved > 0 and not args.dry_run: + refresh_jellyfin_library() + + if errors == 0 and not args.dry_run: + cleaned_files, cleaned_dirs = cleanup_source_directories() + prune_manual_batches() + + log( + "INFO", + "summary", + moved=moved, + skipped=skipped, + errors=errors, + dry_run=args.dry_run, + cleaned_files=cleaned_files, + cleaned_dirs=cleaned_dirs, + ) + write_status("done", moved=moved, skipped=skipped, errors=errors, dry_run=args.dry_run, cleaned_files=cleaned_files, cleaned_dirs=cleaned_dirs) + return 2 if errors else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index 513bd63..cc6a006 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,8 @@ 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] testpaths = ["tests"]