diff --git a/api/video/__init__.py b/api/video/__init__.py index 4b042dd4..bce3348f 100644 --- a/api/video/__init__.py +++ b/api/video/__init__.py @@ -51,6 +51,7 @@ def create_video_blueprint() -> Blueprint: from .wishlist import register_routes as reg_wishlist from .youtube import register_routes as reg_youtube from .downloads import register_routes as reg_downloads + from .manual_import import register_routes as reg_manual_import from .automations import register_routes as reg_automations reg_dashboard(bp) reg_scan(bp) @@ -66,6 +67,7 @@ def create_video_blueprint() -> Blueprint: reg_wishlist(bp) reg_youtube(bp) reg_downloads(bp) + reg_manual_import(bp) reg_automations(bp) return bp diff --git a/api/video/downloads.py b/api/video/downloads.py index 54a6244b..4255a955 100644 --- a/api/video/downloads.py +++ b/api/video/downloads.py @@ -135,6 +135,20 @@ def register_routes(bp): body = request.get_json(silent=True) or {} return jsonify(save(get_video_db(), body)) + @bp.route("/organization", methods=["GET"]) + def video_organization(): + """The library-organisation settings: naming templates + post-process toggles.""" + from . import get_video_db + from core.video.organization import load + return jsonify(load(get_video_db())) + + @bp.route("/organization", methods=["POST"]) + def video_organization_save(): + from . import get_video_db + from core.video.organization import save + body = request.get_json(silent=True) or {} + return jsonify(save(get_video_db(), body)) + @bp.route("/downloads/evaluate", methods=["POST"]) def video_quality_evaluate(): """Judge a video file the user already owns against their quality profile — diff --git a/api/video/manual_import.py b/api/video/manual_import.py new file mode 100644 index 00000000..9c0df803 --- /dev/null +++ b/api/video/manual_import.py @@ -0,0 +1,160 @@ +"""Manual / failed-import resolution API (isolated). + +When a finished download can't be auto-placed it's parked as ``import_failed`` with +the file left on disk (``dest_path`` points at it). The Import page surfaces these and +lets the user place them by hand: + + GET /api/video/import/failed → the queue of unplaced downloads + POST /api/video/import//place → force-import to the user's chosen identity + POST /api/video/import//dismiss → drop the row (optionally delete the file) + +The identity picker on the page reuses the existing /api/video/search (TMDB, with a +``library_id`` annotation for owned titles) — no new search endpoint needed here. +Reads only the video engine + video.db; nothing from the music side. +""" + +from __future__ import annotations + +import json +import os + +from flask import jsonify, request + +from utils.logging_config import get_logger + +logger = get_logger("video_api.manual_import") + +_KIND_FOR_SCOPE = {"movie": "movie", "episode": "show"} + + +def _ctx(row): + try: + c = json.loads(row.get("search_ctx") or "{}") + return c if isinstance(c, dict) else {} + except (ValueError, TypeError): + return {} + + +def _failed_view(row): + """The render-ready shape for one unplaced download.""" + c = _ctx(row) + return { + "id": row.get("id"), + "title": row.get("title"), + "kind": row.get("kind"), + "year": row.get("year"), + "reason": row.get("error"), + "file": row.get("dest_path"), # where the file is sitting, unplaced + "release_title": row.get("release_title"), + "poster_url": row.get("poster_url"), + "scope": c.get("scope"), + "season": c.get("season"), + "episode": c.get("episode"), + } + + +def register_routes(bp): + @bp.route("/import/failed", methods=["GET"]) + def video_import_failed(): + from . import get_video_db + rows = get_video_db().get_import_failed_video_downloads() + return jsonify({"success": True, "items": [_failed_view(r) for r in rows]}) + + @bp.route("/import//place", methods=["POST"]) + def video_import_place(dl_id): + """Force-import an unplaced file to the user's chosen identity. Body: + {scope, title, year, season, episode, episode_title, media_id}.""" + from . import get_video_db + from core.video import organization + from core.video.download_pipeline import target_dir_for + from core.video.importer import real_fs, run_import + from core.video.mediainfo import probe + + db = get_video_db() + row = db.get_video_download(dl_id) + if not row or row.get("status") != "import_failed": + return jsonify({"success": False, "error": "Not an unplaced import."}), 404 + src = row.get("dest_path") + if not src or not os.path.exists(src): + return jsonify({"success": False, "error": "The file is no longer on disk."}), 410 + + body = request.get_json(silent=True) or {} + scope = str(body.get("scope") or "").lower() + if scope not in ("movie", "episode"): + return jsonify({"success": False, "error": "Choose a movie or an episode."}), 400 + + paths = {k: db.get_setting(k) or "" for k in ("movies_path", "tv_path", "youtube_path")} + if not paths["movies_path"]: + paths["movies_path"] = db.get_setting("transfer_path") or "" + override = { + "scope": scope, + "title": body.get("title"), + "year": body.get("year"), + "season": body.get("season"), + "episode": body.get("episode"), + "episode_title": body.get("episode_title"), + "media_id": body.get("media_id"), + "target_dir": target_dir_for(_KIND_FOR_SCOPE[scope], paths), + } + settings = organization.load(db) + prober = probe if settings.get("verify_with_ffprobe", True) else None + patch = run_import(row, src, fs=real_fs(), prober=prober, settings=settings, + force=True, override=override) + try: + db.update_video_download(dl_id, **patch) + except Exception: + logger.exception("manual place: failed to persist import %s", dl_id) + return jsonify({"success": False, "error": "Couldn't save the result."}), 500 + ok = patch.get("status") == "completed" + if ok: + # Write NFO + artwork sidecars for the chosen identity (best-effort), then + # refresh the server + DB the same way the auto-download path does + # (batch-complete → scan chain), so the manually-placed title shows up + # without waiting for a scheduled scan. + sidecar_dl = {"kind": _KIND_FOR_SCOPE[scope], "media_source": "tmdb", + "media_id": override.get("media_id"), + "poster_url": row.get("poster_url"), + "search_ctx": json.dumps({"scope": scope, "season": override.get("season"), + "episode": override.get("episode")})} + if settings.get("save_artwork") or settings.get("write_nfo"): + try: + from core.video.download_monitor import write_sidecars + from core.video.importer import real_fs + write_sidecars(db, sidecar_dl, patch["dest_path"], settings, real_fs()) + except Exception: + logger.exception("manual place: sidecar write failed for %s", dl_id) + if settings.get("download_subtitles"): + try: + from core.video.download_monitor import write_subtitles_for + from core.video.importer import real_fs + write_subtitles_for(db, sidecar_dl, patch["dest_path"], settings, real_fs()) + except Exception: + logger.exception("manual place: subtitle fetch failed for %s", dl_id) + try: + from core.video.download_events import notify_batch_complete + notify_batch_complete({"completed": 1, "manual": True}) + except Exception: + logger.exception("manual place: batch-complete notify failed for %s", dl_id) + return jsonify({"success": ok, "status": patch.get("status"), + "dest_path": patch.get("dest_path"), "error": patch.get("error")}) + + @bp.route("/import//dismiss", methods=["POST"]) + def video_import_dismiss(dl_id): + """Drop a failed-import row. Body {delete_file: bool} optionally removes the + unplaced file from disk too.""" + from . import get_video_db + db = get_video_db() + row = db.get_video_download(dl_id) + if not row or row.get("status") != "import_failed": + return jsonify({"success": False, "error": "Not an unplaced import."}), 404 + body = request.get_json(silent=True) or {} + if body.get("delete_file"): + src = row.get("dest_path") + if src and os.path.exists(src): + try: + os.remove(src) + except OSError: + logger.warning("dismiss: could not delete %s", src) + db.update_video_download(dl_id, status="cancelled", + error="Dismissed from manual import") + return jsonify({"success": True}) diff --git a/core/video/download_monitor.py b/core/video/download_monitor.py index ce0abf99..4fdc8962 100644 --- a/core/video/download_monitor.py +++ b/core/video/download_monitor.py @@ -37,12 +37,17 @@ _started = False _lock = threading.Lock() -def _complete_via_file(dl, download_dir, lister, mover): - """Locate the finished file in the download dir and move it to the library. Returns - a completed/failed patch, or {'progress':100} if the file isn't there yet.""" +def _complete_via_file(dl, download_dir, lister, mover, organizer): + """Locate the finished file in the download dir and post-process it into the + library. Returns a completed/import_failed patch, or {'progress':100} if the file + isn't on disk yet. ``organizer(dl, src)`` (when supplied) runs the full Radarr-style + import (parse → templated rename → copy/replace → carry subs); otherwise we fall + back to the legacy flat move by basename.""" src = find_completed_file(download_dir, dl.get("filename"), lister) if not src: return {"progress": 100.0} + if organizer is not None: + return organizer(dl, src) dest = dest_path_for(dl.get("target_dir"), src) try: mover(src, dest) @@ -51,7 +56,7 @@ def _complete_via_file(dl, download_dir, lister, mover): return {"status": "completed", "progress": 100.0, "dest_path": dest} -def process_download(dl: dict, transfers: list, download_dir: str, *, lister, mover) -> dict | None: +def process_download(dl: dict, transfers: list, download_dir: str, *, lister, mover, organizer=None) -> dict | None: """Decide the next state for one active download given the current slskd transfers. Returns a patch dict for the DB row, or {'_missing': True} when slskd no longer knows the transfer (the caller decides when to give up). Robust to slskd clearing @@ -60,7 +65,7 @@ def process_download(dl: dict, transfers: list, download_dir: str, *, lister, mo t = find_transfer(transfers, dl.get("username"), dl.get("filename")) if not t: # slskd forgot it — could be done+cleared. If the file's there, finish it. - done = _complete_via_file(dl, download_dir, lister, mover) + done = _complete_via_file(dl, download_dir, lister, mover, organizer) if done.get("status"): return done return {"_missing": True} @@ -71,7 +76,7 @@ def process_download(dl: dict, transfers: list, download_dir: str, *, lister, mo return {"status": "cancelled", "error": "Cancelled on Soulseek"} if state == "failed": return {"status": "failed", "error": "Soulseek transfer " + str(t.get("state") or "failed")} - return _complete_via_file(dl, download_dir, lister, mover) # completed + return _complete_via_file(dl, download_dir, lister, mover, organizer) # completed def _move(src: str, dest: str) -> None: @@ -79,6 +84,117 @@ def _move(src: str, dest: str) -> None: shutil.move(src, dest) +def _make_organizer(db): + """A per-tick organizer closure: post-process a finished download into the library + via the importer (Radarr-style parse → ffprobe-verify → templated rename → + copy/replace → carry subs) against the real filesystem. The upgrade decision reads + the destination folder (filesystem-as-truth), so no DB/profile lookup is needed. + ffprobe is best-effort — when it isn't installed, ``probe`` returns None and the + importer falls back to scene-name parsing.""" + from core.video import organization + from core.video.importer import real_fs, run_import + from core.video.mediainfo import probe + fs = real_fs() + try: + settings = organization.load(db) + except Exception: # noqa: BLE001 - a settings-load hiccup must never wedge the monitor + settings = organization.default_settings() + prober = probe if settings.get("verify_with_ffprobe", True) else None + + def organize(dl, src): + patch = run_import(dl, src, fs=fs, prober=prober, settings=settings) + if patch.get("status") == "completed" and patch.get("dest_path"): + if settings.get("save_artwork") or settings.get("write_nfo"): + write_sidecars(db, dl, patch["dest_path"], settings, fs) + if settings.get("download_subtitles"): + write_subtitles_for(db, dl, patch["dest_path"], settings, fs) + return patch + + return organize + + +def _media_ids(db, dl): + """(tmdb_id, imdb_id) for a download's title — taken directly when it was grabbed + from TMDB, or looked up from the LIBRARY row for an owned re-grab (whose ``media_id`` + is the library id, not a TMDB id). (None, None) when it can't be resolved.""" + dl = dl or {} + mid = dl.get("media_id") + if not mid: + return (None, None) + src = str(dl.get("media_source") or "").lower() + if src == "tmdb": + try: + return (int(mid), None) + except (TypeError, ValueError): + return (None, None) + if src == "library" and db is not None: + try: + kind = "movie" if str(dl.get("kind") or "").lower() == "movie" else "show" + return db.media_tmdb_id(kind, mid) + except Exception: # noqa: BLE001 + return (None, None) + return (None, None) + + +def write_sidecars(db, dl, dest_path, settings, fs): + """Best-effort: fetch full TMDB metadata for the imported title (resolving a library + re-grab's id when needed) and write NFO + the artwork set next to it — movie folder, + or the show root for an episode. Uses the detail's ABSOLUTE image URLs (the download + row's poster is an internal/relative path, not fetchable). Never raises. Shared by + the monitor and the manual-import endpoint.""" + try: + from core.video import sidecars + scope = "movie" if str(dl.get("kind") or "").lower() == "movie" else "episode" + tmdb_id, _imdb = _media_ids(db, dl) + detail = None + if tmdb_id is not None: + try: + from core.video.enrichment.engine import get_video_enrichment_engine + # full_detail (not tmdb_detail) so OWNED titles don't redirect — sidecars + # need the raw metadata + absolute image URLs regardless of ownership. + d = get_video_enrichment_engine().tmdb_full_detail( + "movie" if scope == "movie" else "show", tmdb_id) + if isinstance(d, dict): + detail = d + except Exception: # noqa: BLE001 - a metadata fetch hiccup → skip, don't fail + detail = None + sidecars.write_for(dest_path, scope, (detail or {}).get("poster_url"), detail, settings, fs) + except Exception: # noqa: BLE001 - sidecars are a nice-to-have, never fatal + logger.exception("sidecar write failed for download %s", (dl or {}).get("id")) + + +def write_subtitles_for(db, dl, dest_path, settings, fs): + """Best-effort: download external .srt files (OpenSubtitles) next to the imported + video for the user's preferred languages. The .srt sits NEXT TO the file (so an + episode's subs land in its Season folder, not the show root). Never raises.""" + try: + import json as _json + from core.video import subtitles + api_key = db.get_setting("opensubtitles_api_key") if db else None + fetch = subtitles.opensubtitles_fetcher(api_key) + if not fetch: + return + tmdb_id, imdb_id = _media_ids(db, dl) + identity = {} + if tmdb_id is not None: + identity["tmdb_id"] = tmdb_id + if imdb_id: + identity["imdb_id"] = imdb_id + try: + ctx = _json.loads(dl.get("search_ctx") or "{}") + except (ValueError, TypeError): + ctx = {} + if isinstance(ctx, dict) and ctx.get("season") is not None: + identity["season"] = ctx.get("season") + identity["episode"] = ctx.get("episode") + if not (identity.get("tmdb_id") or identity.get("imdb_id")): + return + langs = subtitles.parse_langs(settings.get("subtitle_langs")) + subtitles.write_subtitles(dest_path, langs, identity, fetch, fs) + except Exception: # noqa: BLE001 - subtitle fetch is best-effort, never fatal + logger.exception("subtitle fetch failed for download %s", (dl or {}).get("id")) + + def _walk(root: str): for dirpath, _dirs, files in os.walk(str(root or ".")): for f in files: @@ -233,11 +349,13 @@ def _tick(db) -> None: from config.settings import config_manager download_dir = str(config_manager.get("soulseek.download_path", "") or "") transfers = list_downloads() + organizer = _make_organizer(db) live_ids = set() completed_now = 0 for dl in active: live_ids.add(dl["id"]) - upd = process_download(dl, transfers, download_dir, lister=_walk, mover=_move) + upd = process_download(dl, transfers, download_dir, lister=_walk, mover=_move, + organizer=organizer) if not upd: continue if upd.get("status") == "completed": @@ -253,7 +371,10 @@ def _tick(db) -> None: if upd.get("status") == "failed": _fail_or_retry(db, dl, upd.get("error")) # auto-retry before truly failing continue - if upd.get("status") in ("completed", "cancelled"): + # import_failed = the file downloaded fine but couldn't be placed (sample, wrong + # episode, not an upgrade, …). Terminal + needs manual import — NOT a download + # failure, so don't burn the retry budget re-downloading the same good file. + if upd.get("status") in ("completed", "cancelled", "import_failed"): upd.setdefault("completed_at", _now()) try: db.update_video_download(dl["id"], **upd) diff --git a/core/video/enrichment/engine.py b/core/video/enrichment/engine.py index 6c3b878b..3e60abce 100644 --- a/core/video/enrichment/engine.py +++ b/core/video/enrichment/engine.py @@ -435,6 +435,19 @@ class VideoEnrichmentEngine: return None return cached or None + def tmdb_full_detail(self, kind, tmdb_id) -> dict | None: + """Raw TMDB full detail (absolute image URLs + metadata) WITHOUT the + owned→library redirect — for sidecar / NFO writing, which needs the data even + for titles already in the library.""" + w = self.workers.get("tmdb") + if not w or not w.enabled: + return None + try: + return w.client.full_detail(kind, tmdb_id, region=self._region()) + except Exception: + logger.exception("tmdb_full_detail failed for %s %s", kind, tmdb_id) + return None + def tmdb_detail(self, kind, tmdb_id) -> dict | None: """Full detail for a TMDB title not in the library — same shape as the library detail (source='tmdb', direct image URLs, nothing owned). If it IS diff --git a/core/video/importer.py b/core/video/importer.py new file mode 100644 index 00000000..8e3e0f5c --- /dev/null +++ b/core/video/importer.py @@ -0,0 +1,386 @@ +"""Post-process a finished video download into the library — the Radarr/Sonarr step. + +On completion the monitor hands us the located file. We: + 1. parse the release + SANITY-GATE it (reject non-video / samples / wrong episode / + multi-file packs we can't safely place), + 2. build the canonical library path (``library_paths``), + 3. decide IMPORT vs UPGRADE-replace vs not-an-upgrade by looking at what is already + in the destination folder — the filesystem is the source of truth (like Radarr), + which avoids leaning on DB columns the schema doesn't have, + 4. COPY it in (renamed), carry sibling subtitles, on an upgrade delete the worse + existing file, and remove the source unless it's a torrent (preserve seeding). + +``plan_import`` is pure (directory reads injected via ``list_dir``); ``run_import`` +executes the plan through an injected ``fs`` facade so orchestration is unit-tested +without touching disk. Isolated — sibling video modules + stdlib only; no music imports. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +from typing import Any, Callable + +from core.video import organization +from core.video.download_pipeline import basename_of +from core.video.library_paths import quality_full +from core.video.quality_eval import resolution_rank +from core.video.release_parse import parse_release + +VIDEO_EXTS = frozenset({ + ".mkv", ".mp4", ".avi", ".m4v", ".mov", ".ts", ".wmv", + ".mpg", ".mpeg", ".webm", ".flv", ".m2ts", +}) +SUB_EXTS = frozenset({".srt", ".sub", ".ass", ".ssa", ".idx", ".vtt", ".smi"}) + +_SAMPLE_MAX_BYTES = 150 * 1024 * 1024 # a "sample"-named file under this is a sample +_SAMPLE = re.compile(r"(^|[.\-_ ])sample([.\-_ ]|$)", re.I) +_SXXEXX = re.compile(r"\bS(\d{1,2})[ .]?E(\d{1,3})\b", re.I) + +# A probed runtime under this (seconds) is a sample/clip, not the real thing. Movies +# get a generous floor; episodes vary wildly (shorts/cartoons) so only the absurdly +# short get caught. +_RUNTIME_FLOOR = {"movie": 15 * 60, "episode": 90} + +# Source ranking for the upgrade comparison (mirrors the quality ladder order). +_SRC_RANK = {"remux": 6, "bluray": 5, "web-dl": 4, "webrip": 3, "hdtv": 2, "dvd": 1} + + +def ext_of(path: Any) -> str: + """Lower-cased extension (with dot) of a path's basename, '' if none.""" + return os.path.splitext(basename_of(path))[1].lower() + + +def is_video(path: Any) -> bool: + return ext_of(path) in VIDEO_EXTS + + +def is_sample(name: Any, size_bytes: Any) -> bool: + """A 'sample'-tagged file that's also small (or of unknown size) is a sample.""" + if not _SAMPLE.search(basename_of(name)): + return False + try: + sz = int(size_bytes or 0) + except (TypeError, ValueError): + sz = 0 + return sz == 0 or sz < _SAMPLE_MAX_BYTES + + +def quality_score(parsed: Any) -> int: + """A self-contained quality score (resolution dominates, source breaks ties) used + only for the local upgrade comparison. Higher = better.""" + parsed = parsed if isinstance(parsed, dict) else {} + return resolution_rank(parsed.get("resolution")) * 10 + _SRC_RANK.get(parsed.get("source"), 0) + + +def _scope_of(dl: dict) -> str: + """The import scope from the search context, falling back to the download kind. + Only 'movie' and 'episode' are placeable; packs/youtube are gated out upstream.""" + ctx = _search_ctx(dl) + sc = str(ctx.get("scope") or "").lower() + if sc in ("movie", "episode", "season", "series"): + return sc + k = str(dl.get("kind") or "").lower() + if k == "movie": + return "movie" + if k in ("show", "tv", "episode"): + return "episode" + return k or "movie" + + +def _search_ctx(dl: dict) -> dict: + try: + ctx = json.loads((dl or {}).get("search_ctx") or "{}") + return ctx if isinstance(ctx, dict) else {} + except (ValueError, TypeError): + return {} + + +def _ctx(dl: dict) -> dict: + sc = _search_ctx(dl) + return { + "title": sc.get("title") or (dl or {}).get("title") or "", + "year": sc.get("year") if sc.get("year") is not None else (dl or {}).get("year"), + "season": sc.get("season"), + "episode": sc.get("episode"), + "episode_title": sc.get("episode_title"), + } + + +def _reject(reason: str) -> dict: + return {"action": "reject", "reason": reason, "manual": True} + + +def _existing_match(scope: str, dest_dir: str, ctx: dict, list_dir: Callable) -> str | None: + """The basename of a file ALREADY in the destination folder that represents this + same item (any video for a movie; a matching SxxExx for an episode), or None. + ``list_dir(dir)`` yields basenames; a missing dir yields nothing.""" + try: + names = [str(n) for n in (list_dir(dest_dir) or [])] + except Exception: # noqa: BLE001 - a missing/denied dir simply means "nothing there" + return None + vids = [n for n in names if ext_of(n) in VIDEO_EXTS and not is_sample(n, None)] + if scope == "movie": + return max(vids, key=len) if vids else None + if scope == "episode": + try: + ws, we = int(ctx.get("season")), int(ctx.get("episode")) + except (TypeError, ValueError): + return None + for n in vids: + m = _SXXEXX.search(n) + if m and int(m.group(1)) == ws and int(m.group(2)) == we: + return n + return None + return None + + +def plan_import(dl: dict, src_path: str, *, list_dir: Callable, probe: dict | None = None, + settings: dict | None = None, force: bool = False, + override: dict | None = None) -> dict: + """Decide what to do with a finished download. Returns one of: + + {"action": "import", "dest": {...}, "quality_label": str} + {"action": "upgrade", "dest": {...}, "replace_path": str, "quality_label": str} + {"action": "reject", "reason": str, "manual": True} + + Pure: all directory reads go through ``list_dir`` (injected). ``probe`` is the + ffprobe ``mediainfo`` result (or None when ffprobe is unavailable) — when present + we trust the FILE's real resolution over the scene name and reject corrupt / + too-short junk. ``settings`` are the user's organisation settings (naming templates + + replace policy); None = defaults. + + MANUAL placement: ``force=True`` with an ``override`` ({scope, title, year, season, + episode, episode_title, target_dir, media_id}) trusts the user's chosen identity — + it skips the auto sanity-gates (sample / wrong-episode / pack / not-an-upgrade) and + files the file exactly where they said, replacing any worse copy. ffprobe is still + used for the true resolution, but never to reject.""" + dl = dl or {} + settings = organization.normalize(settings) + override = override or {} + scope = str(override.get("scope") or _scope_of(dl)).lower() if force else _scope_of(dl) + name = basename_of(src_path) + ext = ext_of(src_path) + parsed = parse_release(dl.get("release_title") or name) + ctx = _ctx(dl) + if force: # the user told us what it is — let their identity win + for k in ("title", "year", "season", "episode", "episode_title"): + if override.get(k) is not None: + ctx[k] = override.get(k) + + if not is_video(src_path): # can't place a non-video, even on a forced import + return _reject("Not a video file (%s)" % (ext or "no extension")) + if not force: + if is_sample(name, dl.get("size_bytes")): + return _reject("Looks like a sample, not the feature") + if scope not in ("movie", "episode"): + return _reject("Season/complete packs need manual import") + if scope == "episode": + if ctx.get("season") is None or ctx.get("episode") is None: + return _reject("Missing season/episode info") + # If the release name itself names a DIFFERENT episode, don't mis-file it. + if parsed.get("episode") is not None and ( + parsed.get("season") != ctx.get("season") + or parsed.get("episode") != ctx.get("episode")): + return _reject("Release is S%02dE%02d, not the episode requested" + % (parsed.get("season") or 0, parsed.get("episode") or 0)) + else: + if scope not in ("movie", "episode"): + return _reject("Pick a movie or an episode to place this file") + if scope == "episode" and (ctx.get("season") is None or ctx.get("episode") is None): + return _reject("Pick a season and episode to place this file") + + # ffprobe verification — best-effort; on a forced placement we use the real + # resolution but never reject on it (the user has decided). + if probe is not None: + if not force: + if not probe.get("ok"): + return _reject("No readable video stream — corrupt or fake file") + dur = probe.get("duration_sec") or 0 + floor = _RUNTIME_FLOOR.get(scope) + if floor and 0 < dur < floor: + return _reject("Runtime is only %d min — looks like a sample/clip, not the %s" + % (int(dur // 60), scope)) + # Trust the FILE over the (often lying) scene name: real resolution always, + # real codec only when the name didn't carry one. + parsed = dict(parsed) + if probe.get("resolution"): + parsed["resolution"] = probe["resolution"] + if probe.get("video_codec") and not parsed.get("codec"): + parsed["codec"] = probe["video_codec"] + + root = (override.get("target_dir") if force else None) or dl.get("target_dir") or "" + if not root: + return _reject("No library folder configured for this type") + + media_id = override.get("media_id") if force else dl.get("media_id") + quality = quality_full(parsed) + fields = { + "title": ctx.get("title"), "year": ctx.get("year"), + "series": ctx.get("title"), "season": ctx.get("season"), + "episode": ctx.get("episode"), "episode_title": ctx.get("episode_title"), + "quality": quality, "resolution": parsed.get("resolution"), + "source": parsed.get("source"), "codec": parsed.get("codec"), + "tmdbid": media_id if scope == "movie" else None, + "tvdbid": media_id if scope == "episode" else None, + } + dest = organization.render_path(scope, root, fields, settings, ext) + # Where poster.jpg goes: the movie folder, or the SHOW root for an episode + # (parent of the Season folder) — so it isn't dropped per-season. + artwork_dir = dest["dir"] if scope == "movie" else os.path.dirname(dest["dir"]) + + existing = _existing_match(scope, dest["dir"], ctx, list_dir) + if existing: + # A forced placement replaces whatever's there (the user chose to put it here). + if force: + return {"action": "upgrade", "dest": dest, "quality_label": quality, + "replace_path": os.path.join(dest["dir"], existing), "artwork_dir": artwork_dir} + if not settings.get("replace_existing", True): + return _reject("Already in the library (%s) — replace is turned off" % existing) + new_score = quality_score(parsed) + old_score = quality_score(parse_release(existing)) + if new_score > old_score: + return {"action": "upgrade", "dest": dest, "quality_label": quality, + "replace_path": os.path.join(dest["dir"], existing), "artwork_dir": artwork_dir} + return _reject("Not an upgrade over the copy already in the library (%s)" % existing) + + return {"action": "import", "dest": dest, "quality_label": quality, "artwork_dir": artwork_dir} + + +def plan_subs(src_path: str, dest_path: str, list_dir: Callable) -> list: + """Sibling subtitle files to carry alongside the video, renamed to match the + destination stem (preserving any language suffix, e.g. '.en.srt'). Returns a list + of (src_abs, dest_abs) pairs. ``list_dir`` lists the SOURCE directory.""" + src_dir = os.path.dirname(src_path) or "." + v_stem = os.path.splitext(basename_of(src_path))[0] + d_stem = os.path.splitext(basename_of(dest_path))[0] + d_dir = os.path.dirname(dest_path) + out = [] + try: + names = [str(n) for n in (list_dir(src_dir) or [])] + except Exception: # noqa: BLE001 + return out + for n in names: + if ext_of(n) not in SUB_EXTS: + continue + stem, ext = os.path.splitext(n) + if stem == v_stem: + extra = "" # movie.srt → .srt + elif stem.startswith(v_stem + "."): + extra = stem[len(v_stem):] # movie.en.srt → .en.srt + else: + continue + out.append((os.path.join(src_dir, n), os.path.join(d_dir, d_stem + extra + ext))) + return out + + +def run_import(dl: dict, src_path: str, *, fs: Any, prober: Callable | None = None, + settings: dict | None = None, force: bool = False, + override: dict | None = None) -> dict: + """Execute the import and return a DB patch dict for the download row. + + ``fs`` is an injected facade with: ``list_dir(dir)->iterable[name]``, + ``makedirs(dir)``, ``copy(src, dst)``, ``move(src, dst)``, ``remove(path)``. + ``prober(path)->mediainfo`` is an optional ffprobe hook (None = skip verification). + ``settings`` are the user's organisation settings (transfer mode, subtitle carry); + None = defaults. ``force``/``override`` drive a MANUAL placement (see ``plan_import``). + A reject becomes an ``import_failed`` row with ``dest_path`` pointing at the file's + current (unplaced) location so the Import page can resolve it; a success becomes a + ``completed`` row with ``dest_path`` set to its final home.""" + settings = organization.normalize(settings) + probe_info = None + if prober is not None: + try: + probe_info = prober(src_path) + except Exception: # noqa: BLE001 - a probe crash must not block the import + probe_info = None + plan = plan_import(dl, src_path, list_dir=fs.list_dir, probe=probe_info, + settings=settings, force=force, override=override) + if plan["action"] == "reject": + # Leave the file where it is; remember WHERE so manual import can find it. + return {"status": "import_failed", "progress": 100.0, "error": plan["reason"], + "dest_path": src_path} + + dest = plan["dest"] + move_mode = settings.get("transfer_mode") == "move" + try: + fs.makedirs(dest["dir"]) + if move_mode: + fs.move(src_path, dest["path"]) + else: + fs.copy(src_path, dest["path"]) + if settings.get("carry_subtitles", True): + for sub_src, sub_dst in plan_subs(src_path, dest["path"], fs.list_dir): + try: + fs.copy(sub_src, sub_dst) + except Exception: # noqa: BLE001 - a subtitle that won't copy isn't fatal + pass + if plan["action"] == "upgrade" and plan.get("replace_path"): + try: + fs.remove(plan["replace_path"]) + except Exception: # noqa: BLE001 - failing to delete the old file isn't fatal + pass + # Copy mode reclaims the download copy UNLESS it's a torrent (keep seeding); + # move mode already relocated it. + if not move_mode and str(dl.get("source") or "").lower() != "torrent": + try: + fs.remove(src_path) + except Exception: # noqa: BLE001 + pass + except Exception as e: # noqa: BLE001 - any copy/mkdir failure → manual import + return {"status": "import_failed", "progress": 100.0, "error": "Import failed: " + str(e), + "dest_path": src_path} + + return {"status": "completed", "progress": 100.0, "dest_path": dest["path"], + "quality_label": plan.get("quality_label") or dl.get("quality_label")} + + +class _RealFS: + """The production filesystem facade for ``run_import`` (os/shutil).""" + + @staticmethod + def list_dir(path): + try: + return os.listdir(str(path or "")) + except OSError: + return [] + + @staticmethod + def makedirs(path): + os.makedirs(str(path or "."), exist_ok=True) + + @staticmethod + def copy(src, dst): + shutil.copy2(src, dst) + + @staticmethod + def move(src, dst): + shutil.move(src, dst) + + @staticmethod + def save_url(url, dst): + import urllib.request + req = urllib.request.Request(url, headers={"User-Agent": "SoulSync"}) + with urllib.request.urlopen(req, timeout=20) as resp, open(dst, "wb") as f: + shutil.copyfileobj(resp, f) + + @staticmethod + def write_text(path, content): + with open(path, "w", encoding="utf-8") as f: + f.write(content) + + @staticmethod + def remove(path): + os.remove(path) + + +def real_fs() -> _RealFS: + return _RealFS() + + +__all__ = [ + "VIDEO_EXTS", "SUB_EXTS", "ext_of", "is_video", "is_sample", "quality_score", + "plan_import", "plan_subs", "run_import", "real_fs", +] diff --git a/core/video/library_paths.py b/core/video/library_paths.py new file mode 100644 index 00000000..35555920 --- /dev/null +++ b/core/video/library_paths.py @@ -0,0 +1,165 @@ +"""Organize a finished video download into a Radarr/Sonarr-standard library path. + +Given the SCOPE/intent of a grab (movie vs episode, plus title/year/season/episode) +and the PARSED release quality, build the canonical folder + filename the file should +land at: + + /The Matrix (1999)/The Matrix (1999) Bluray-1080p.mkv + /Breaking Bad/Season 01/Breaking Bad - S01E01 - Pilot WEBDL-1080p.mkv + +These are the community-standard templates and also exactly what Plex/Jellyfin want +for reliable identification (we organise on disk; the server just refreshes). + +Pure (no filesystem, no DB) so it's unit-tested in isolation. Isolated — stdlib + +the sibling ``download_pipeline.basename_of`` only; nothing from the music side. +""" + +from __future__ import annotations + +import os +import re +from typing import Any + +from core.video.download_pipeline import basename_of + +# Characters illegal on Windows / awkward on most filesystems, plus control chars. +_ILLEGAL = re.compile(r'[\\/:*?"<>|\x00-\x1f]') +_TRAILING_DOTSPACE = re.compile(r"[ .]+$") + +# Parsed source token → the label used in the Radarr ``{Quality Full}`` tag. +_SRC_LABEL = { + "remux": "Remux", "bluray": "Bluray", "web-dl": "WEBDL", + "webrip": "WEBRip", "hdtv": "HDTV", "dvd": "DVD", +} + + +def sanitize(name: Any) -> str: + """Filesystem-safe path COMPONENT: strip illegal chars, collapse whitespace, and + trim trailing dots/spaces (Windows rejects those). Never contains a separator.""" + s = _ILLEGAL.sub("", str(name or "")) + s = re.sub(r"\s+", " ", s).strip() + s = _TRAILING_DOTSPACE.sub("", s) + return s + + +def _year_suffix(year: Any) -> str: + """' (1999)' for a plausible release year, else '' (keeps titles clean).""" + try: + y = int(year) + except (TypeError, ValueError): + return "" + return " (%d)" % y if 1870 <= y <= 2999 else "" + + +def source_label(source: Any) -> str: + """The display label for a parsed source token ('bluray' → 'Bluray', 'web-dl' → + 'WEBDL'), '' when unknown. Used for the $source template token.""" + return _SRC_LABEL.get(str(source or "").strip().lower(), "") + + +def quality_full(parsed: Any) -> str: + """The Radarr-style ``{Quality Full}`` tag, e.g. 'Bluray-1080p' / 'WEBDL-1080p' / + 'Remux-2160p'. '' when neither source nor resolution can be determined. A + proper/repack is appended ('… Proper') so an upgrade reads at a glance.""" + parsed = parsed if isinstance(parsed, dict) else {} + src = _SRC_LABEL.get(parsed.get("source")) + res = parsed.get("resolution") + if src and res: + tag = src + "-" + res + elif res: + tag = res + elif src: + tag = src + else: + return "" + if parsed.get("proper") or parsed.get("repack"): + tag += " Proper" + return tag + + +def movie_folder(title: Any, year: Any) -> str: + """'The Matrix (1999)' — the per-movie folder name.""" + return (sanitize(title) or "Unknown") + _year_suffix(year) + + +def movie_filename(title: Any, year: Any, quality: Any, ext: Any) -> str: + """'The Matrix (1999) Bluray-1080p.mkv' (quality tag omitted when unknown).""" + stem = (sanitize(title) or "Unknown") + _year_suffix(year) + q = sanitize(quality) + if q: + stem += " " + q + return stem + _ext(ext) + + +def show_folder(series: Any) -> str: + """'Breaking Bad' — the per-series folder name.""" + return sanitize(series) or "Unknown" + + +def season_folder(season: Any) -> str: + """'Season 01' (or 'Specials' for season 0, Sonarr's convention).""" + try: + n = int(season) + except (TypeError, ValueError): + n = None + if n == 0: + return "Specials" + return "Season %02d" % (n if n is not None else 0) + + +def episode_filename(series: Any, season: Any, episode: Any, ep_title: Any, + quality: Any, ext: Any) -> str: + """'Breaking Bad - S01E01 - Pilot WEBDL-1080p.mkv'. The ' - {title}' segment and + the trailing quality tag are each omitted when unknown.""" + try: + s = int(season) + except (TypeError, ValueError): + s = 0 + try: + e = int(episode) + except (TypeError, ValueError): + e = 0 + stem = "%s - S%02dE%02d" % (sanitize(series) or "Unknown", s, e) + t = sanitize(ep_title) + if t: + stem += " - " + t + q = sanitize(quality) + if q: + stem += " " + q + return stem + _ext(ext) + + +def _ext(ext: Any) -> str: + e = str(ext or "").strip().lower() + if not e: + return "" + return e if e.startswith(".") else "." + e + + +def plan_path(scope: Any, root: Any, ctx: dict, quality: Any, ext: Any) -> dict: + """Resolve the canonical destination for a finished download. + + ``ctx`` = {title, year, season, episode, episode_title}. Returns + ``{"dir": , "filename": , "path": }``. For an + unsupported scope it falls back to a flat drop (root + sanitized basename) so a + file is never lost — the caller gates scope before relying on this.""" + ctx = ctx if isinstance(ctx, dict) else {} + root = str(root or "") + sc = str(scope or "").lower() + if sc == "movie": + d = os.path.join(root, movie_folder(ctx.get("title"), ctx.get("year"))) + fn = movie_filename(ctx.get("title"), ctx.get("year"), quality, ext) + elif sc == "episode": + d = os.path.join(root, show_folder(ctx.get("title")), season_folder(ctx.get("season"))) + fn = episode_filename(ctx.get("title"), ctx.get("season"), ctx.get("episode"), + ctx.get("episode_title"), quality, ext) + else: + d = root + fn = sanitize(basename_of(ctx.get("src") or "")) or "download" + return {"dir": d, "filename": fn, "path": os.path.join(d, fn)} + + +__all__ = [ + "sanitize", "source_label", "quality_full", "movie_folder", "movie_filename", + "show_folder", "season_folder", "episode_filename", "plan_path", +] diff --git a/core/video/mediainfo.py b/core/video/mediainfo.py new file mode 100644 index 00000000..481dc95c --- /dev/null +++ b/core/video/mediainfo.py @@ -0,0 +1,137 @@ +"""Probe a finished video file with ffprobe for its TRUE media info. + +We otherwise trust the release NAME for resolution/quality, and names lie: 720p +upscales labelled 1080p, trailers/samples labelled as the feature, broken muxes. +ffprobe reads the real container — duration, dimensions → resolution, codecs — so the +importer can tag the file by its actual quality and reject corrupt / too-short junk. + +The parsing (``parse_ffprobe`` / ``resolution_from_dimensions``) is pure and unit-tested +on canned JSON; the subprocess runner is injected, so nothing here needs ffmpeg to be +tested. ffmpeg is OPTIONAL — when ffprobe isn't installed, or it errors, ``probe`` +returns None and the caller falls back to the scene name. + +Three outcomes, deliberately distinct: + - None → couldn't verify (ffprobe missing / crashed / timed out) → skip + - {"ok": False, …} → ffprobe ran and found NO video stream → corrupt / fake + - {"ok": True, …} → real media info to trust over the name + +Isolated: stdlib only; no music imports. +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +from typing import Any, Callable + +_FFPROBE = "ffprobe" + + +def _int(v: Any) -> int: + try: + return int(v) + except (TypeError, ValueError): + return 0 + + +def _float(v: Any) -> float: + try: + return float(v) + except (TypeError, ValueError): + return 0.0 + + +def resolution_from_dimensions(width: Any, height: Any) -> str | None: + """Bucket real pixel dimensions into a resolution label. Uses the LARGER axis so + a letterboxed 1920x800 movie reads as 1080p (not 720p by its short side).""" + ref = max(_int(width), _int(height)) + if ref <= 0: + return None + if ref >= 3000: + return "2160p" + if ref >= 1700: + return "1080p" + if ref >= 1100: + return "720p" + return "480p" + + +def _norm_codec(name: Any) -> str | None: + s = str(name or "").strip().lower() + if not s: + return None + if s in ("hevc", "h265", "x265"): + return "hevc" + if s in ("h264", "avc", "x264"): + return "x264" + if s == "av1": + return "av1" + return s + + +def parse_ffprobe(data: Any) -> dict: + """Parse ffprobe's ``-show_format -show_streams`` JSON into the fields we use. + ``ok`` is True only when a video stream is present (else: corrupt / not a video).""" + data = data if isinstance(data, dict) else {} + streams = data.get("streams") or [] + fmt = data.get("format") or {} + video = next((s for s in streams if s.get("codec_type") == "video"), None) + audio = next((s for s in streams if s.get("codec_type") == "audio"), None) + duration = _float(fmt.get("duration")) or _float((video or {}).get("duration")) + width = (video or {}).get("width") + height = (video or {}).get("height") + return { + "ok": video is not None, + "duration_sec": duration, + "width": _int(width), + "height": _int(height), + "resolution": resolution_from_dimensions(width, height) if video else None, + "video_codec": _norm_codec((video or {}).get("codec_name")), + "audio_codec": str((audio or {}).get("codec_name") or "") or None, + } + + +def ffprobe_available() -> bool: + return shutil.which(_FFPROBE) is not None + + +def _default_runner(path: str) -> str | None: + """Run ffprobe and return its JSON stdout, or None on any failure (so a transient + ffprobe error degrades to 'unverified', never to a false 'corrupt').""" + try: + proc = subprocess.run( + [_FFPROBE, "-v", "quiet", "-print_format", "json", + "-show_format", "-show_streams", str(path)], + capture_output=True, text=True, timeout=120, check=False, + ) + except Exception: # noqa: BLE001 - missing binary / timeout / OS error → unverified + return None + if proc.returncode != 0 or not (proc.stdout or "").strip(): + return None + return proc.stdout + + +def probe(path: Any, runner: Callable | None = None) -> dict | None: + """Probe ``path`` and return parsed media info, or None when it can't be verified. + ``runner(path)->json_str|None`` is injected (real ffprobe in prod, canned in tests). + When no runner is given and ffprobe isn't installed, returns None (skip verify).""" + use = runner if runner is not None else (_default_runner if ffprobe_available() else None) + if use is None: + return None + try: + raw = use(path) + except Exception: # noqa: BLE001 + return None + if not raw: + return None + try: + data = json.loads(raw) + except (ValueError, TypeError): + return None + return parse_ffprobe(data) + + +__all__ = [ + "resolution_from_dimensions", "parse_ffprobe", "ffprobe_available", "probe", +] diff --git a/core/video/organization.py b/core/video/organization.py new file mode 100644 index 00000000..e344ba86 --- /dev/null +++ b/core/video/organization.py @@ -0,0 +1,213 @@ +"""Video library-organisation settings: naming templates + post-process toggles. + +Mirrors the MUSIC side's file-organisation standard — editable ``$token`` path +templates with per-component sanitisation and dangling-separator cleanup — but for +video's movie/episode shape, plus the optional post-process behaviours the user can +turn on and off. + +Settings (persisted as JSON in video.db ``video_settings['organization']``): + - ``movie_template`` : path template for movies (folders via '/', last = file) + - ``episode_template`` : path template for episodes + - ``verify_with_ffprobe`` : probe the real file (true quality + reject junk) + - ``replace_existing`` : upgrade-replace a worse copy already in the library + - ``transfer_mode`` : 'copy' (reclaim source unless torrent) | 'move' + - ``carry_subtitles`` : bring sibling .srt/.ass alongside the video + +Template tokens + Movies: $title $titlefirst $year $quality $resolution $source $codec $edition + $tmdbid $imdbid + Episodes: $series $season $seasonraw $episode $episodetitle $year $quality + $resolution $source $codec $tvdbid + ($season/$episode are zero-padded to 2; $seasonraw is the bare number.) + +Pure data + a pure renderer (no DB, no FS) so it's unit-tested in isolation. Isolated — +stdlib + sibling video ``library_paths`` only; nothing from the music side. +""" + +from __future__ import annotations + +import json +import os +import re +from typing import Any + +from core.video.library_paths import sanitize, source_label + +DEFAULTS = { + "version": 1, + "movie_template": "$title ($year)/$title ($year) $quality", + "episode_template": "$series/Season $season/$series - S$seasonE$episode - $episodetitle $quality", + "verify_with_ffprobe": True, + "replace_existing": True, + "transfer_mode": "copy", + "carry_subtitles": True, + "save_artwork": False, + "write_nfo": False, + "download_subtitles": False, + "subtitle_langs": "en", +} + +_TRANSFER_MODES = ("copy", "move") + + +def default_settings() -> dict: + return dict(DEFAULTS) + + +def normalize(raw: Any) -> dict: + """Coerce stored/posted settings to a valid shape, filling gaps from the default. + Blank templates fall back to the default; never raises.""" + d = default_settings() + if not isinstance(raw, dict): + return d + for key in ("movie_template", "episode_template"): + v = raw.get(key) + if isinstance(v, str) and v.strip(): + d[key] = v.strip() + for key in ("verify_with_ffprobe", "replace_existing", "carry_subtitles", + "save_artwork", "write_nfo", "download_subtitles"): + if key in raw: + d[key] = bool(raw.get(key)) + tm = str(raw.get("transfer_mode") or "").strip().lower() + if tm in _TRANSFER_MODES: + d["transfer_mode"] = tm + if "subtitle_langs" in raw: + from core.video.subtitles import parse_langs + d["subtitle_langs"] = ",".join(parse_langs(raw.get("subtitle_langs"))) + return d + + +def load(db) -> dict: + raw = db.get_setting("organization") + if raw: + try: + return normalize(json.loads(raw)) + except (ValueError, TypeError): + pass + return default_settings() + + +def save(db, raw: Any) -> dict: + s = normalize(raw) + db.set_setting("organization", json.dumps(s)) + return s + + +# ── the template engine (the music $token standard, video tokens) ───────────── +def _str(v: Any) -> str: + if v is None: + return "" + return str(v) + + +def _pad2(v: Any) -> str: + try: + return "%02d" % int(v) + except (TypeError, ValueError): + return _str(v) + + +def _plausible_year(v: Any) -> bool: + try: + return 1870 <= int(v) <= 2999 + except (TypeError, ValueError): + return False + + +def _ext(ext: Any) -> str: + e = str(ext or "").strip().lower() + if not e: + return "" + return e if e.startswith(".") else "." + e + + +def _movie_values(f: dict) -> dict: + title = f.get("title") or "Unknown" + return { + "title": title, + "titlefirst": (str(title)[:1] or "U").upper(), + "year": _str(f.get("year")) if _plausible_year(f.get("year")) else "", + "quality": _str(f.get("quality")), + "resolution": _str(f.get("resolution")), + "source": source_label(f.get("source")), + "codec": _str(f.get("codec")).upper(), + "edition": _str(f.get("edition")), + "tmdbid": _str(f.get("tmdbid")), + "imdbid": _str(f.get("imdbid")), + } + + +def _episode_values(f: dict) -> dict: + series = f.get("series") or f.get("title") or "Unknown" + return { + "series": series, + "season": _pad2(f.get("season")), + "seasonraw": _str(f.get("season")), + "episode": _pad2(f.get("episode")), + "episodetitle": _str(f.get("episode_title")), + "year": _str(f.get("year")) if _plausible_year(f.get("year")) else "", + "quality": _str(f.get("quality")), + "resolution": _str(f.get("resolution")), + "source": source_label(f.get("source")), + "codec": _str(f.get("codec")).upper(), + "tvdbid": _str(f.get("tvdbid")), + } + + +def render_template(template: Any, values: dict) -> str: + """Substitute ``$token`` / ``${token}`` from ``values`` into ``template``. Each + value is path-sanitised first (so a title with '/' can't spawn a folder), and + tokens are replaced longest-name-first ($episodetitle before $episode).""" + clean = {k: sanitize(v) for k, v in (values or {}).items()} + out = str(template or "") + for tok in sorted(clean, key=len, reverse=True): + out = out.replace("${" + tok + "}", clean[tok]) + for tok in sorted(clean, key=len, reverse=True): + out = out.replace("$" + tok, clean[tok]) + return out + + +def _tidy_component(part: str) -> str: + """Clean one path segment: drop a ' - ' left dangling by an empty token, remove + empty ()/[] left by an empty $year, collapse whitespace, trim stray dashes and + Windows-hostile trailing dots/spaces.""" + p = re.sub(r"\s+-\s+(?=(\s|$))", " ", part) # ' - ' before an empty token + p = re.sub(r"\(\s*\)", "", p) # empty ( ) from a missing $year + p = re.sub(r"\[\s*\]", "", p) + p = re.sub(r"\s+", " ", p).strip() + p = p.strip("-").strip() + return p.rstrip(". ") + + +def render_path(scope: Any, root: Any, fields: dict, settings: Any, ext: Any) -> dict: + """Render the destination for a finished download from the user's templates. + Returns ``{"dir", "filename", "path"}`` (same shape as ``library_paths.plan_path``). + An unsupported scope falls back to a flat drop so a file is never lost.""" + settings = settings if isinstance(settings, dict) else {} + fields = fields if isinstance(fields, dict) else {} + root = str(root or "") + sc = str(scope or "").lower() + + if sc == "movie": + tmpl = settings.get("movie_template") or DEFAULTS["movie_template"] + values = _movie_values(fields) + elif sc == "episode": + tmpl = settings.get("episode_template") or DEFAULTS["episode_template"] + values = _episode_values(fields) + else: + base = (sanitize(fields.get("title")) or "download") + _ext(ext) + return {"dir": root, "filename": base, "path": os.path.join(root, base)} + + rendered = render_template(tmpl, values) + parts = [p for p in (_tidy_component(seg) for seg in rendered.split("/")) if p] + if not parts: + parts = ["download"] + d = os.path.join(root, *parts[:-1]) if len(parts) > 1 else root + filename = parts[-1] + _ext(ext) + return {"dir": d, "filename": filename, "path": os.path.join(d, filename)} + + +__all__ = [ + "DEFAULTS", "default_settings", "normalize", "load", "save", + "render_template", "render_path", +] diff --git a/core/video/sidecars.py b/core/video/sidecars.py new file mode 100644 index 00000000..fabd9448 --- /dev/null +++ b/core/video/sidecars.py @@ -0,0 +1,229 @@ +"""Write Kodi/Jellyfin-style sidecars next to imported video — NFO metadata + the full +artwork set — so the library is self-describing on disk (server-agnostic + portable). + +The metadata comes from the on-demand TMDB detail fetch (``engine.tmdb_detail``), passed +in as ``meta``. This module is pure — NFO XML building + a write plan — with the +filesystem injected, so it's unit-tested without disk or network. Best-effort BY +CONTRACT: the caller treats every failure as non-fatal (a missing poster or a flaky +fetch never breaks an import). + +Layout written: + Movie folder: movie.nfo, poster.jpg, fanart.jpg, clearlogo.png + Show root: tvshow.nfo, poster.jpg, fanart.jpg, clearlogo.png, seasonNN-poster.jpg + +Isolated: stdlib only; no music imports. +""" + +from __future__ import annotations + +from typing import Any +from xml.sax.saxutils import escape as _xesc + +_HEAD = '\n' + + +def _t(tag: str, value: Any) -> str: + """A single XML element, or '' when the value is empty (so absent fields are + simply omitted rather than written blank).""" + if value is None: + return "" + s = str(value).strip() + if not s: + return "" + return " <%s>%s\n" % (tag, _xesc(s), tag) + + +def _year(meta: dict) -> str: + y = meta.get("year") + return str(y) if y else "" + + +def _uniqueids(meta: dict, *, tmdb_default: bool) -> str: + out = "" + tmdb = meta.get("tmdb_id") + if tmdb: + out += ' %s\n' % ( + ' default="true"' if tmdb_default else "", _xesc(str(tmdb))) + if meta.get("tvdb_id"): + out += ' %s\n' % ( + ' default="true"' if not tmdb_default else "", _xesc(str(meta["tvdb_id"]))) + if meta.get("imdb_id"): + out += ' %s\n' % _xesc(str(meta["imdb_id"])) + return out + + +def _genres(meta: dict) -> str: + return "".join(_t("genre", g) for g in (meta.get("genres") or []) if g) + + +def _actors(meta: dict, limit: int = 20) -> str: + out = "" + for i, c in enumerate((meta.get("cast") or [])[:limit]): + if not isinstance(c, dict) or not c.get("name"): + continue + out += " \n %s\n" % _xesc(str(c["name"])) + if c.get("character"): + out += " %s\n" % _xesc(str(c["character"])) + out += " %d\n \n" % i + return out + + +def _artwork_tags(meta: dict) -> str: + out = "" + if meta.get("poster_url"): + out += ' %s\n' % _xesc(str(meta["poster_url"])) + if meta.get("logo"): + out += ' %s\n' % _xesc(str(meta["logo"])) + if meta.get("backdrop_url"): + out += " \n %s\n \n" % _xesc(str(meta["backdrop_url"])) + return out + + +def nfo_movie(meta: dict) -> str: + """A Kodi/Jellyfin-compatible ``movie.nfo`` from a TMDB detail dict.""" + meta = meta if isinstance(meta, dict) else {} + body = ( + _t("title", meta.get("title")) + + _t("originaltitle", meta.get("original_title")) + + _t("year", _year(meta)) + + _t("plot", meta.get("overview")) + + _t("outline", meta.get("overview")) + + _t("tagline", meta.get("tagline")) + + _t("runtime", meta.get("runtime_minutes")) + + _t("mpaa", meta.get("content_rating")) + + _t("studio", meta.get("studio")) + + _t("premiered", meta.get("release_date")) + + _t("status", meta.get("status")) + + _t("rating", meta.get("rating")) + + _genres(meta) + + _uniqueids(meta, tmdb_default=True) + + _artwork_tags(meta) + + _actors(meta) + ) + return _HEAD + "\n" + body + "\n" + + +def nfo_tvshow(meta: dict) -> str: + """A Kodi/Jellyfin-compatible ``tvshow.nfo`` from a TMDB detail dict.""" + meta = meta if isinstance(meta, dict) else {} + body = ( + _t("title", meta.get("title")) + + _t("year", _year(meta)) + + _t("plot", meta.get("overview")) + + _t("outline", meta.get("overview")) + + _t("tagline", meta.get("tagline")) + + _t("runtime", meta.get("runtime_minutes")) + + _t("mpaa", meta.get("content_rating")) + + _t("studio", meta.get("network")) + + _t("premiered", meta.get("first_air_date")) + + _t("status", meta.get("status")) + + _t("rating", meta.get("rating")) + + _genres(meta) + + _uniqueids(meta, tmdb_default=True) + + _artwork_tags(meta) + + _actors(meta) + ) + return _HEAD + "\n" + body + "\n" + + +def _season_art(meta: dict) -> list: + """(url, filename) pairs for season posters — seasonNN-poster.jpg (Specials → 00).""" + out = [] + for s in (meta.get("_seasons") or []): + if not isinstance(s, dict) or not s.get("poster_url"): + continue + try: + n = int(s.get("season_number")) + except (TypeError, ValueError): + continue + out.append((s["poster_url"], "season%02d-poster.jpg" % n)) + return out + + +def plan_sidecars(scope: Any, meta: dict, settings: dict) -> dict: + """Decide what to write for one imported item. Returns + ``{"nfo": (filename, content) | None, "art": [(url, filename), ...]}`` — gated by + the ``write_nfo`` / ``save_artwork`` settings. Pure.""" + meta = meta if isinstance(meta, dict) else {} + settings = settings if isinstance(settings, dict) else {} + sc = str(scope or "").lower() + nfo = None + art = [] + + if settings.get("write_nfo"): + if sc == "movie": + nfo = ("movie.nfo", nfo_movie(meta)) + elif sc == "episode": + nfo = ("tvshow.nfo", nfo_tvshow(meta)) + + if settings.get("save_artwork"): + if meta.get("poster_url"): + art.append((meta["poster_url"], "poster.jpg")) + if meta.get("backdrop_url"): + art.append((meta["backdrop_url"], "fanart.jpg")) + if meta.get("logo"): + art.append((meta["logo"], "clearlogo.png")) + if sc == "episode": + art.extend(_season_art(meta)) + + return {"nfo": nfo, "art": art} + + +def write(folder: str, scope: Any, meta: dict, settings: dict, fs: Any) -> None: + """Write the planned sidecars into ``folder`` via the injected ``fs`` + (``list_dir``, ``makedirs``, ``write_text(path, str)``, ``save_url(url, path)``). + Idempotent: skips any file already present so upgrades/re-imports don't refetch. + Best-effort: each file is independent and a failure is swallowed.""" + import os + plan = plan_sidecars(scope, meta, settings) + if not plan["nfo"] and not plan["art"]: + return + try: + existing = {str(n).lower() for n in (fs.list_dir(folder) or [])} + except Exception: # noqa: BLE001 + existing = set() + try: + fs.makedirs(folder) + except Exception: # noqa: BLE001 + return + + if plan["nfo"]: + name, content = plan["nfo"] + if name.lower() not in existing: + try: + fs.write_text(os.path.join(folder, name), content) + except Exception: # noqa: BLE001 + pass + for url, name in plan["art"]: + if name.lower() in existing: + continue + try: + fs.save_url(url, os.path.join(folder, name)) + except Exception: # noqa: BLE001 + pass + + +def write_for(dest_path: str, scope: Any, poster_url: Any, detail: Any, + settings: dict, fs: Any) -> None: + """Resolve the sidecar folder from a finished file's path and write into it. The + folder is the movie folder (parent of the file) or, for an episode, the SHOW root + (parent of the Season folder) so show-level art/NFO land once. ``poster_url`` is the + baseline from the download row; ``detail`` (the TMDB fetch, or None) fills the rest — + so a poster still lands even when the detail fetch is unavailable.""" + import os + sc = str(scope or "").lower() + if sc == "movie": + folder = os.path.dirname(dest_path) + elif sc == "episode": + folder = os.path.dirname(os.path.dirname(dest_path)) # show root, above Season NN + else: + return + if not folder: + return + meta = {"poster_url": poster_url} if poster_url else {} + if isinstance(detail, dict): + meta.update(detail) + write(folder, sc, meta, settings, fs) + + +__all__ = ["nfo_movie", "nfo_tvshow", "plan_sidecars", "write", "write_for"] diff --git a/core/video/subtitles.py b/core/video/subtitles.py new file mode 100644 index 00000000..9ac81308 --- /dev/null +++ b/core/video/subtitles.py @@ -0,0 +1,153 @@ +"""Download external subtitle .srt files from OpenSubtitles and drop them next to the +imported video as ``