importarr/importarr/worker.py

#!/usr/bin/env python3
"""Import manual SABnzbd downloads into Jellyfin movie/TV library roots."""

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(os.getenv("IMPORTARR_DOWNLOAD_ROOT", "/data/downloads/manual"))
LEGACY_DANISH_DOWNLOADS = Path(os.getenv("IMPORTARR_LEGACY_DOWNLOAD_ROOT", "/data/downloads/legacy"))
DOWNLOAD_ROOTS = [MANUAL_DOWNLOADS, LEGACY_DANISH_DOWNLOADS]
SAB_TRANSIENT_PREFIXES = ("_UNPACK_", "__UNPACK__", "_FAILED_", "_ADMIN_")

HOST_MEDIA_PREFIX = os.getenv("IMPORTARR_HOST_MEDIA_PREFIX", "/data")
MOVIES_ROOT = Path(os.getenv("IMPORTARR_MOVIES_ROOT", "/data/movies"))
TV_ROOT = Path(os.getenv("IMPORTARR_TV_ROOT", "/data/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 = os.getenv("IMPORTARR_HOME_ASSISTANT_WEBHOOK", "")

RADARR_CONFIG = Path(os.getenv("IMPORTARR_RADARR_CONFIG", "/config/radarr/config.xml"))
RADARR_URL = os.getenv("IMPORTARR_RADARR_URL", "http://radarr:7878")
RADARR_DB = Path(os.getenv("IMPORTARR_RADARR_DB", "/config/radarr/radarr.db"))
SONARR_CONFIG = Path(os.getenv("IMPORTARR_SONARR_CONFIG", "/config/sonarr/config.xml"))
SONARR_URL = os.getenv("IMPORTARR_SONARR_URL", "http://sonarr:8989")
SONARR_DB = Path(os.getenv("IMPORTARR_SONARR_DB", "/config/sonarr/sonarr.db"))
SABNZBD_CONFIG = Path(os.getenv("IMPORTARR_SABNZBD_CONFIG", "/config/sabnzbd/sabnzbd.ini"))
SABNZBD_URL = os.getenv("IMPORTARR_SABNZBD_URL", "http://sabnzbd: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("/data/imdb"),
    Path("/data/imdb-datasets"),
    Path("/data/.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:
    if not HOME_ASSISTANT_WEBHOOK:
        return
    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 "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())