Add movie-keyed pool index with multi-version movies, ČSFD title bridge, subtitles and miniDLNA library scoping
This commit is contained in:
@@ -0,0 +1,421 @@
|
||||
"""ČSFD.cz scraper — a *title bridge* to TMDb.
|
||||
|
||||
Curator identifies movies by **TMDb id**. A direct TMDb search on a Czech
|
||||
distribution title is unreliable (remakes, localized names, "weird" hits).
|
||||
ČSFD, on the other hand, indexes Czech release names well **and** lists a
|
||||
film's *original* title. So the bridge is::
|
||||
|
||||
filename → ČSFD search → ČSFD film page → original title + year
|
||||
→ TMDb search → tmdbid
|
||||
|
||||
This module only scrapes ČSFD; the TMDb half lives in
|
||||
:mod:`src.core.file_manager` (:meth:`FileManager.resolve_via_csfd`). Nothing
|
||||
here feeds tags/genres — those still come from TMDb via Jellyfin/NFO.
|
||||
|
||||
ČSFD serves an **Anubis** proof-of-work wall in front of every page: a plain
|
||||
request gets a 200 with a JS challenge instead of the movie. We detect it,
|
||||
solve the SHA-256 PoW the way the bundled worker JS does, and replay the
|
||||
request through a shared :class:`requests.Session` (so a batch pays the PoW
|
||||
cost only once).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from urllib.parse import urlencode, urljoin
|
||||
|
||||
from loguru import logger
|
||||
|
||||
try:
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
_HAS_DEPS = True
|
||||
except ImportError: # pragma: no cover - exercised only without the extras
|
||||
_HAS_DEPS = False
|
||||
requests = None # type: ignore
|
||||
BeautifulSoup = None # type: ignore
|
||||
|
||||
CSFD_BASE_URL = "https://www.csfd.cz"
|
||||
CSFD_SEARCH_URL = "https://www.csfd.cz/hledat/"
|
||||
|
||||
HEADERS = {
|
||||
"User-Agent": (
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/120.0.0.0 Safari/537.36"
|
||||
),
|
||||
"Accept-Language": "cs,en;q=0.9",
|
||||
}
|
||||
|
||||
# Anubis anti-bot wall: a plain GET returns a 200 challenge page, not the movie.
|
||||
ANUBIS_CHALLENGE_MARKER = 'id="anubis_challenge"'
|
||||
ANUBIS_PASS_PATH = "/.within.website/x/cmd/anubis/api/pass-challenge"
|
||||
# Safety cap so a difficulty bump can never spin forever (difficulty 1 needs ~16).
|
||||
ANUBIS_MAX_NONCE = 50_000_000
|
||||
|
||||
|
||||
class CsfdError(RuntimeError):
|
||||
"""A ČSFD lookup could not be completed (deps missing, network, anti-bot)."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class CsfdFilm:
|
||||
"""The slice of a ČSFD film page Curator needs to reach a TMDb id."""
|
||||
|
||||
url: str
|
||||
csfd_id: Optional[int] = None
|
||||
czech_title: str = ""
|
||||
original_title: str = ""
|
||||
year: Optional[str] = None # four-digit string, or None
|
||||
|
||||
@property
|
||||
def tmdb_query(self) -> tuple[str, Optional[str]]:
|
||||
"""``(title, year)`` best suited for a TMDb search — original title first."""
|
||||
return (self.original_title or self.czech_title, self.year)
|
||||
|
||||
|
||||
def dependencies_available() -> bool:
|
||||
"""True when ``requests`` + ``beautifulsoup4`` are importable."""
|
||||
return _HAS_DEPS
|
||||
|
||||
|
||||
def _check_dependencies() -> None:
|
||||
if not _HAS_DEPS:
|
||||
raise CsfdError(
|
||||
"ČSFD vyžaduje balíčky 'requests' a 'beautifulsoup4' "
|
||||
"(poetry install je doplní)."
|
||||
)
|
||||
|
||||
|
||||
def _extract_csfd_id(url: str) -> Optional[int]:
|
||||
"""Pull the numeric film id out of a ČSFD URL, or ``None``."""
|
||||
match = re.search(r"/film/(\d+)", url)
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Anubis proof-of-work (ported verbatim from the pre-Jellyfin csfd.py)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _extract_json_blob(html: str, element_id: str):
|
||||
"""Return the parsed JSON from an Anubis ``<script id=...>`` blob, or None."""
|
||||
match = re.search(
|
||||
rf'<script id="{re.escape(element_id)}" type="application/json">(.*?)</script>',
|
||||
html,
|
||||
re.S,
|
||||
)
|
||||
if not match:
|
||||
return None
|
||||
try:
|
||||
return json.loads(match.group(1))
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
|
||||
def _solve_anubis_pow(random_data: str, difficulty: int) -> tuple[str, int, int]:
|
||||
"""Brute-force the Anubis proof-of-work.
|
||||
|
||||
Mirrors the bundled ``sha256-purejs`` worker: find the smallest ``nonce``
|
||||
such that ``sha256(random_data + str(nonce))`` has ``difficulty`` leading
|
||||
zero nibbles. Returns ``(hash_hex, nonce, elapsed_ms)``.
|
||||
"""
|
||||
full_zero_bytes = difficulty // 2
|
||||
needs_half_byte = difficulty % 2 != 0
|
||||
start = time.monotonic()
|
||||
for nonce in range(ANUBIS_MAX_NONCE):
|
||||
digest = hashlib.sha256(f"{random_data}{nonce}".encode()).digest()
|
||||
if any(digest[i] != 0 for i in range(full_zero_bytes)):
|
||||
continue
|
||||
if needs_half_byte and digest[full_zero_bytes] >> 4 != 0:
|
||||
continue
|
||||
elapsed_ms = int((time.monotonic() - start) * 1000)
|
||||
return digest.hex(), nonce, elapsed_ms
|
||||
raise CsfdError(
|
||||
f"Anubis PoW nevyřešen do {ANUBIS_MAX_NONCE} pokusů (obtížnost {difficulty})"
|
||||
)
|
||||
|
||||
|
||||
def _solve_anubis_challenge(session, html: str, url: str):
|
||||
"""Solve the Anubis challenge in ``html`` and return the real page response."""
|
||||
payload = _extract_json_blob(html, "anubis_challenge")
|
||||
if not payload:
|
||||
raise CsfdError("ČSFD anti-bot stránka bez čitelné Anubis challenge.")
|
||||
|
||||
rules = payload.get("rules", {})
|
||||
challenge = payload.get("challenge", {})
|
||||
random_data = challenge.get("randomData")
|
||||
difficulty = int(rules.get("difficulty", 1))
|
||||
if not random_data:
|
||||
raise CsfdError("Anubis challenge neobsahuje randomData.")
|
||||
|
||||
base_prefix = _extract_json_blob(html, "anubis_base_prefix") or ""
|
||||
logger.debug(f"Solving Anubis challenge (difficulty {difficulty}) for {url}")
|
||||
hash_hex, nonce, elapsed_ms = _solve_anubis_pow(random_data, difficulty)
|
||||
logger.debug(f"Anubis solved: nonce={nonce}, elapsed={elapsed_ms}ms")
|
||||
|
||||
pass_url = urljoin(CSFD_BASE_URL, f"{base_prefix}{ANUBIS_PASS_PATH}")
|
||||
response = session.get(
|
||||
pass_url,
|
||||
params={
|
||||
"id": challenge.get("id"),
|
||||
"response": hash_hex,
|
||||
"nonce": nonce,
|
||||
"redir": url,
|
||||
"elapsedTime": elapsed_ms,
|
||||
},
|
||||
headers=HEADERS,
|
||||
timeout=10,
|
||||
)
|
||||
response.raise_for_status()
|
||||
if ANUBIS_CHALLENGE_MARKER in response.text:
|
||||
raise CsfdError("ČSFD Anubis challenge se nepodařilo vyřešit (odmítnuto).")
|
||||
return response
|
||||
|
||||
|
||||
def _get_page(session, url: str):
|
||||
"""GET ``url`` through ``session``, transparently clearing an Anubis wall."""
|
||||
response = session.get(url, headers=HEADERS, timeout=10)
|
||||
response.raise_for_status()
|
||||
if ANUBIS_CHALLENGE_MARKER in response.text:
|
||||
response = _solve_anubis_challenge(session, response.text, url)
|
||||
return response
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Page parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _parse_json_ld(soup) -> tuple[str, Optional[str]]:
|
||||
"""``(name, year)`` from the film page's JSON-LD (name is the Czech title)."""
|
||||
for script in soup.find_all("script", type="application/ld+json"):
|
||||
try:
|
||||
data = json.loads(script.string or "")
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
continue
|
||||
if isinstance(data, list):
|
||||
data = next((d for d in data if d.get("@type") == "Movie"), None)
|
||||
if not isinstance(data, dict) or data.get("@type") != "Movie":
|
||||
continue
|
||||
name = (data.get("name") or "").strip()
|
||||
year: Optional[str] = None
|
||||
for key in ("dateCreated", "datePublished"):
|
||||
match = re.search(r"(19|20)\d{2}", str(data.get(key) or ""))
|
||||
if match:
|
||||
year = match.group(0)
|
||||
break
|
||||
return name, year
|
||||
return "", None
|
||||
|
||||
|
||||
def _parse_origin(soup) -> tuple[Optional[str], Optional[str]]:
|
||||
"""``(origin_country, year)`` from the ``.origin`` line.
|
||||
|
||||
ČSFD separates values with inline bullet ``<span>`` elements (no commas),
|
||||
e.g. ``USA • 1999 • 136 min``. For a co-production only the first country
|
||||
is returned — enough to match it against a name's flag.
|
||||
"""
|
||||
elem = soup.select_one(".origin")
|
||||
if not elem:
|
||||
return None, None
|
||||
raw = elem.get_text(separator="|", strip=True)
|
||||
tokens = [t.strip() for part in raw.split("|") for t in part.split(",")]
|
||||
country: Optional[str] = None
|
||||
year: Optional[str] = None
|
||||
for token in tokens:
|
||||
if not token:
|
||||
continue
|
||||
if year is None and re.fullmatch(r"(19|20)\d{2}", token):
|
||||
year = token
|
||||
continue
|
||||
if (country is None and not token[0].isdigit()
|
||||
and re.search(r"[^\W\d_]", token)):
|
||||
country = token.split("/")[0].strip()
|
||||
return country, year
|
||||
|
||||
|
||||
# ČSFD flags older Czech films "Československo" while ``.origin`` may read "Česko".
|
||||
_CZ_ORIGIN_ALIASES = {"česko", "československo"}
|
||||
|
||||
|
||||
def _parse_original_title(soup, origin_country: Optional[str]) -> Optional[str]:
|
||||
"""The film's original title from ``.film-names`` — or ``None`` to use the CZ title.
|
||||
|
||||
Each ``<li>`` carries a country/language flag (``img.flag`` ``title``) and a
|
||||
title. The original title is the entry whose flag matches the **country of
|
||||
origin** (USA → "The Shawshank Redemption"). ČSFD does **not** repeat the
|
||||
Czech title here when it equals the ``<h1>``, so a Czech film has only
|
||||
foreign/festival alt-titles in this list (``[angličtina] Cosy Dens``,
|
||||
``[Slovinsko] Samotarji``) — picking one of those would be wrong, so we
|
||||
return ``None`` and let the caller fall back to the Czech ``<h1>`` title.
|
||||
"""
|
||||
names: list[tuple[str, str]] = []
|
||||
for li in soup.select(".film-names li"):
|
||||
flag = li.select_one("img.flag")
|
||||
country = str(flag.get("title") or "").strip() if flag else ""
|
||||
for junk in li.select("span, img"):
|
||||
junk.extract()
|
||||
text = " ".join(li.get_text(" ").split()).strip(" ·")
|
||||
if text:
|
||||
names.append((country, text))
|
||||
if not names or not origin_country:
|
||||
return None
|
||||
wanted = origin_country.casefold()
|
||||
for country, text in names:
|
||||
here = country.casefold()
|
||||
if here == wanted or (wanted in _CZ_ORIGIN_ALIASES
|
||||
and here in _CZ_ORIGIN_ALIASES):
|
||||
return text
|
||||
return None
|
||||
|
||||
|
||||
def fetch_film(url: str, session=None) -> CsfdFilm:
|
||||
"""Fetch a ČSFD film page → :class:`CsfdFilm` (Czech + original title, year).
|
||||
|
||||
Pass a shared ``requests.Session`` to reuse the Anubis auth cookie across a
|
||||
batch. Raises :class:`CsfdError` on a network / anti-bot / parse failure.
|
||||
"""
|
||||
_check_dependencies()
|
||||
own_session = session is None
|
||||
if own_session:
|
||||
session = requests.Session()
|
||||
try:
|
||||
html = _get_page(session, url).text
|
||||
except requests.RequestException as exc:
|
||||
raise CsfdError(f"Nelze načíst ČSFD stránku: {exc}") from exc
|
||||
finally:
|
||||
if own_session:
|
||||
session.close()
|
||||
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
czech_title, year = _parse_json_ld(soup)
|
||||
origin_country, origin_year = _parse_origin(soup)
|
||||
year = year or origin_year
|
||||
if not czech_title:
|
||||
header = soup.select_one(".film-header-name h1")
|
||||
czech_title = header.get_text(strip=True) if header else ""
|
||||
original_title = _parse_original_title(soup, origin_country) or czech_title
|
||||
return CsfdFilm(
|
||||
url=url,
|
||||
csfd_id=_extract_csfd_id(url),
|
||||
czech_title=czech_title,
|
||||
original_title=original_title,
|
||||
year=str(year) if year else None,
|
||||
)
|
||||
|
||||
|
||||
def _films_section(soup):
|
||||
"""The ``<section>`` holding the *Filmy* results, or the whole document.
|
||||
|
||||
ČSFD's search page groups hits into ``Filmy`` / ``Seriály`` / ``Tvůrci`` /
|
||||
``Uživatelé`` sections; ``a.film-title-name`` matches in all of them, so
|
||||
without scoping a bare-title search happily returns a series episode.
|
||||
"""
|
||||
for section in soup.find_all("section"):
|
||||
header = section.select_one(".box-header h2") or section.find("h2")
|
||||
if header and header.get_text(strip=True).casefold() == "filmy":
|
||||
return section
|
||||
return soup
|
||||
|
||||
|
||||
def _is_film_href(href: str) -> bool:
|
||||
"""True for a film URL ``/film/<id>-slug/`` — a series episode has two segments
|
||||
(``/film/<series>/<episode>/``) and is rejected."""
|
||||
parts = [p for p in href.split("/") if p and p != "prehled"]
|
||||
return len(parts) == 2 and parts[0] == "film"
|
||||
|
||||
|
||||
def search_films(query: str, limit: int = 5, session=None) -> list[CsfdFilm]:
|
||||
"""Search ČSFD for ``query``; film hits only (no series/episodes), best first."""
|
||||
_check_dependencies()
|
||||
if not query.strip():
|
||||
return []
|
||||
search_url = f"{CSFD_SEARCH_URL}?{urlencode({'q': query.strip()})}"
|
||||
own_session = session is None
|
||||
if own_session:
|
||||
session = requests.Session()
|
||||
try:
|
||||
html = _get_page(session, search_url).text
|
||||
except requests.RequestException as exc:
|
||||
raise CsfdError(f"ČSFD hledání selhalo: {exc}") from exc
|
||||
finally:
|
||||
if own_session:
|
||||
session.close()
|
||||
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
films: list[CsfdFilm] = []
|
||||
for anchor in _films_section(soup).select("a.film-title-name"):
|
||||
href = str(anchor.get("href") or "")
|
||||
if not _is_film_href(href):
|
||||
continue
|
||||
films.append(CsfdFilm(
|
||||
url=urljoin(CSFD_BASE_URL, href),
|
||||
csfd_id=_extract_csfd_id(href),
|
||||
czech_title=anchor.get_text(strip=True),
|
||||
))
|
||||
if len(films) >= limit:
|
||||
break
|
||||
return films
|
||||
|
||||
|
||||
def find_csfd_url(query: str, session=None) -> Optional[str]:
|
||||
"""Return the first ČSFD film URL matching ``query``, or ``None``.
|
||||
|
||||
ČSFD's full-text search treats a trailing year as a search term (not a
|
||||
filter) and can miss on ``"Title 2006"`` while ``"Title"`` hits — so a
|
||||
yearful query with no result is retried without the year.
|
||||
"""
|
||||
if not query or not query.strip():
|
||||
return None
|
||||
query = query.strip()
|
||||
films = search_films(query, limit=1, session=session)
|
||||
if not films:
|
||||
without_year = re.sub(r"\s+(?:19|20)\d{2}$", "", query)
|
||||
if without_year != query:
|
||||
films = search_films(without_year, limit=1, session=session)
|
||||
return films[0].url if films else None
|
||||
|
||||
|
||||
# Release-name tokens that mark the end of the actual title in a filename.
|
||||
_RELEASE_MARKERS = {
|
||||
"bluray", "blu-ray", "brrip", "bdrip", "bdremux", "remux", "webrip", "web",
|
||||
"web-dl", "webdl", "hdtv", "dvdrip", "dvd", "dvd5", "dvd9", "hdrip", "cam",
|
||||
"ts", "tc", "x264", "x265", "h264", "h265", "hevc", "avc", "xvid", "divx",
|
||||
"aac", "ac3", "eac3", "dts", "dd5", "ddp5", "truehd", "atmos", "flac",
|
||||
"10bit", "8bit", "hdr", "hdr10", "dolby", "sdr", "proper", "repack",
|
||||
"extended", "unrated", "remastered", "imax", "multi", "dual", "complete",
|
||||
"internal", "limited", "uncut",
|
||||
}
|
||||
_YEAR_RE = re.compile(r"^(19|20)\d{2}$")
|
||||
_RESOLUTION_RE = re.compile(r"^\d{3,4}p$|^[24]k$", re.IGNORECASE)
|
||||
|
||||
|
||||
def clean_filename_to_query(filename: str) -> str:
|
||||
"""Turn a (possibly release-named) filename into a ČSFD search query.
|
||||
|
||||
Strips the path/extension, splits on common separators and keeps the words
|
||||
before the first release marker (year, resolution, codec, source, …). The
|
||||
detected year is appended back as a disambiguator::
|
||||
|
||||
"Matrix.1999.1080p.BluRay.x264-GROUP.mkv" -> "Matrix 1999"
|
||||
"""
|
||||
stem = Path(filename).stem
|
||||
tokens = [t for t in re.split(r"[.\s_]+", stem) if t]
|
||||
|
||||
title_words: list[str] = []
|
||||
year: Optional[str] = None
|
||||
for token in tokens:
|
||||
bare = token.strip("()[]{}")
|
||||
if _YEAR_RE.match(bare):
|
||||
year = bare
|
||||
break
|
||||
if _RESOLUTION_RE.match(bare) or bare.lower() in _RELEASE_MARKERS:
|
||||
break
|
||||
title_words.append(token)
|
||||
|
||||
title = " ".join(title_words).strip() or re.sub(r"[.\s_]+", " ", stem).strip()
|
||||
return f"{title} {year}".strip() if year else title
|
||||
+121
-15
@@ -33,15 +33,64 @@ class File:
|
||||
# free-form per-movie attributes (e.g. "collection_sort"); usable in
|
||||
# miniDLNA filename templates via name_context.
|
||||
self.attributes: dict[str, str] = {}
|
||||
# Identity key of the movie record this file belongs to (set on load /
|
||||
# first save); ``None`` only for an index-less, in-memory File.
|
||||
self._movie_key: str | None = None
|
||||
self.get_metadata()
|
||||
|
||||
def _identity_from_path(self) -> tuple[str, str, str | None, int | None, str]:
|
||||
"""``(key, title, year, tmdb_id, label)`` implied by this file's location.
|
||||
|
||||
A file inside its own ``Title (rok) [tmdbid]`` folder is one version of
|
||||
that movie — identity comes from the **folder**, and the rest of its own
|
||||
stem is the version label. A flat file carries both in its own name.
|
||||
"""
|
||||
from .movie import identity_key
|
||||
from .naming import (is_canonical_pool_stem, parse_3d_format,
|
||||
parse_pool_stem, parse_tmdb_id)
|
||||
|
||||
parent = self.file_path.parent.name
|
||||
grouped = is_canonical_pool_stem(parent)
|
||||
identity_stem = parent if grouped else self.file_path.stem
|
||||
parsed = parse_pool_stem(identity_stem)
|
||||
year = str(parsed[1]) if parsed else None
|
||||
tmdb_id = parse_tmdb_id(identity_stem)
|
||||
title = self.title or (parsed[0] if parsed else self.file_path.stem)
|
||||
|
||||
label = ""
|
||||
if grouped and self.file_path.stem.startswith(f"{parent} - "):
|
||||
label = self.file_path.stem[len(parent) + 3:]
|
||||
elif parse_3d_format(self.file_path.stem):
|
||||
label = parse_3d_format(self.file_path.stem) or ""
|
||||
return identity_key(tmdb_id, title, year), title, year, tmdb_id, label
|
||||
|
||||
def get_metadata(self) -> None:
|
||||
record = self.index.get(self.file_path) if self.index is not None else None
|
||||
if record is None:
|
||||
"""Load this file's metadata from the movie record that owns it.
|
||||
|
||||
Since the pool index became identity-keyed (see :mod:`src.core.movie`),
|
||||
a File is a *view* on one media entry of a movie record — every version
|
||||
of the same film therefore reads the same title/tags/ČSFD/… A file the
|
||||
index doesn't know yet joins the movie of the same identity when there
|
||||
is one (that's how a newly added version inherits the shared metadata
|
||||
instead of blanking it), otherwise it starts a new record.
|
||||
"""
|
||||
if self.index is None:
|
||||
self._init_new_metadata()
|
||||
self.save_metadata()
|
||||
else:
|
||||
return
|
||||
|
||||
found = self.index.find_movie_by_media(self.file_path)
|
||||
if found is not None:
|
||||
self._movie_key, record = found
|
||||
self._apply_record(record)
|
||||
return
|
||||
|
||||
self._init_new_metadata()
|
||||
key, *_ = self._identity_from_path()
|
||||
existing = self.index.get_movie(key)
|
||||
if existing is not None:
|
||||
self._movie_key = key
|
||||
self._apply_record(existing)
|
||||
self.save_metadata()
|
||||
|
||||
def _init_new_metadata(self) -> None:
|
||||
self.new = True
|
||||
@@ -101,9 +150,42 @@ class File:
|
||||
self.tags.append(tag)
|
||||
|
||||
def save_metadata(self):
|
||||
"""Persist metadata to the pool index; a no-op for an index-less File."""
|
||||
if self.index is not None:
|
||||
self.index.set(self.file_path, self._build_record())
|
||||
"""Persist metadata into this file's movie record (no-op without index).
|
||||
|
||||
Writes the shared fields onto the movie record and makes sure the
|
||||
record lists this file among its media. A file the index has never seen
|
||||
starts a new record under its own identity key.
|
||||
"""
|
||||
if self.index is None:
|
||||
return
|
||||
from .movie import ROLE_VIDEO
|
||||
|
||||
derived_key, title, year, tmdb_id, label = self._identity_from_path()
|
||||
key = self._movie_key or derived_key
|
||||
record = self.index.get_movie(key) or {}
|
||||
record.update(self._build_record())
|
||||
# The file's actual location is the truth — a move must update this.
|
||||
record["library"] = self._library_name()
|
||||
record["title"] = title
|
||||
record["year"] = year
|
||||
record["tmdb_id"] = tmdb_id
|
||||
|
||||
# Keep this file listed exactly once, with the label its name implies.
|
||||
rel = self.index._key(self.file_path)
|
||||
media = [m for m in record.get("media", []) if m.get("file") != rel]
|
||||
media.append({"role": ROLE_VIDEO, "label": label, "file": rel})
|
||||
record["media"] = media
|
||||
|
||||
self.index.set_movie(key, record)
|
||||
self._movie_key = key
|
||||
|
||||
def _library_name(self) -> str:
|
||||
"""Top-level pool folder this file sits in (its library), or ``""``."""
|
||||
if self.index is None:
|
||||
return ""
|
||||
rel = self.index._key(self.file_path)
|
||||
parts = Path(rel).parts
|
||||
return parts[0] if len(parts) > 1 else ""
|
||||
|
||||
def added_timestamp(self) -> float:
|
||||
"""Epoch seconds for 'date added' — the stored ``added`` or the file mtime.
|
||||
@@ -169,19 +251,43 @@ class File:
|
||||
self.save_metadata()
|
||||
|
||||
def delete_metadata(self) -> None:
|
||||
"""Remove this file's metadata from the pool index (no-op without one)."""
|
||||
if self.index is not None:
|
||||
self.index.delete(self.file_path)
|
||||
"""Drop this file from its movie record; the last one takes the record.
|
||||
|
||||
Removing one version of a multi-version movie leaves the movie (and its
|
||||
metadata) in place — only when its last media entry goes does the whole
|
||||
record go with it.
|
||||
"""
|
||||
if self.index is None:
|
||||
return
|
||||
found = self.index.find_movie_by_media(self.file_path)
|
||||
if found is None:
|
||||
return
|
||||
key, record = found
|
||||
rel = self.index._key(self.file_path)
|
||||
record["media"] = [m for m in record.get("media", []) if m.get("file") != rel]
|
||||
if record["media"]:
|
||||
self.index.set_movie(key, record)
|
||||
else:
|
||||
self.index.delete_movie(key)
|
||||
|
||||
def relocate(self, new_path: Path) -> None:
|
||||
"""Point this File at a new path, moving its metadata along.
|
||||
"""Point this File at a new path, moving its media entry along.
|
||||
|
||||
The physical file must already have been moved/renamed by the caller.
|
||||
Drops the metadata under the old index key and rebinds to the new path;
|
||||
call ``save_metadata()`` afterwards to write it back.
|
||||
The physical file must already have been moved/renamed by the caller;
|
||||
this only repoints the movie record's media entry (and the File), so a
|
||||
rename never detaches a file from its movie.
|
||||
"""
|
||||
if self.index is not None:
|
||||
self.index.delete(self.file_path)
|
||||
found = self.index.find_movie_by_media(self.file_path)
|
||||
if found is not None:
|
||||
key, record = found
|
||||
old_rel = self.index._key(self.file_path)
|
||||
new_rel = self.index._key(Path(new_path))
|
||||
for media in record.get("media", []):
|
||||
if media.get("file") == old_rel:
|
||||
media["file"] = new_rel
|
||||
self.index.set_movie(key, record)
|
||||
self._movie_key = key
|
||||
self.file_path = Path(new_path)
|
||||
self.filename = self.file_path.name
|
||||
|
||||
|
||||
+551
-39
@@ -7,7 +7,10 @@ from .tag_manager import TagManager
|
||||
from .pool_index import PoolIndex
|
||||
from .utils import list_files
|
||||
from .integrity import check_video_integrity, ffmpeg_available, IntegrityResult
|
||||
from .naming import canonical_pool_stem, is_valid_year, parse_pool_stem, parse_tmdb_id
|
||||
from .naming import (
|
||||
canonical_pool_stem, is_canonical_pool_stem, is_valid_year, parse_3d_format,
|
||||
parse_pool_stem, parse_tmdb_id)
|
||||
from . import subtitles
|
||||
from typing import Callable, Iterable
|
||||
from src.core.config import (
|
||||
load_global_config, save_global_config, DEFAULT_TAG_SCHEMA
|
||||
@@ -168,6 +171,118 @@ class FileManager:
|
||||
"""Names of libraries matching any of ``kinds``."""
|
||||
return [lib["name"] for lib in self.libraries if lib["kind"] in kinds]
|
||||
|
||||
@property
|
||||
def minidlna_libraries(self) -> list[str] | None:
|
||||
"""Libraries included in miniDLNA generation, or ``None`` for "all" (default).
|
||||
|
||||
Persisted in the pool index (it describes this pool, not app state) via
|
||||
*miniDLNA → "Knihovny pro generování…"*. Names no longer in
|
||||
:attr:`libraries` are dropped when read, so a renamed/removed library
|
||||
can't silently wedge the selection.
|
||||
"""
|
||||
idx = self._ensure_index()
|
||||
raw = idx.get_setting("minidlna_libraries", None) if idx is not None else None
|
||||
if raw is None:
|
||||
return None
|
||||
known = {lib["name"] for lib in self.libraries}
|
||||
return [n for n in raw if n in known]
|
||||
|
||||
def set_minidlna_libraries(self, names: list[str] | None) -> None:
|
||||
"""Restrict miniDLNA generation to ``names``; ``None`` resets to "all"."""
|
||||
self._set_setting_raw(
|
||||
"minidlna_libraries", None if names is None else list(dict.fromkeys(names)))
|
||||
|
||||
def _set_setting_raw(self, key: str, value) -> None:
|
||||
"""Persist an arbitrary pool-index setting (no legacy-default fallback)."""
|
||||
idx = self._ensure_index()
|
||||
if idx is None:
|
||||
raise RuntimeError("Pool není nastaven — není kam uložit nastavení.")
|
||||
idx.set_setting(key, value)
|
||||
|
||||
def minidlna_scope(self) -> tuple[list[File], list[str], list]:
|
||||
"""``(movies, mirrored library names, mixed-library series)`` for miniDLNA.
|
||||
|
||||
Narrowed to :attr:`minidlna_libraries` when set — movies by
|
||||
:meth:`library_of`, mirrored libraries by name, mixed-library series by
|
||||
their library — otherwise the whole pool (today's behaviour). Excluding
|
||||
a movie library here means its genre-tree links (and its share of "Tipy
|
||||
dne" / "Nově přidané") are swept as obsolete on the next generation, not
|
||||
merely skipped — the caller should say so before generating.
|
||||
|
||||
The **mirrored libraries** are ``series`` (copy-as-is) *and*
|
||||
``homevideo`` kinds — neither has genre tags to build a tree from
|
||||
(homevideo carries no TMDb id either), so both are hardcopied whole
|
||||
into ``output/<library name>`` instead of going through the tag tree.
|
||||
|
||||
A ``mixed`` library isn't itself mirrored (its top-level files are
|
||||
movies, tagged into the genre tree like any other), but any **series
|
||||
sitting in its subfolders** have no genre tags to speak of — those are
|
||||
returned separately so the caller can hardcopy them too (e.g. into
|
||||
``output/Seriály/<jejich vlastní název>``).
|
||||
"""
|
||||
selected = self.minidlna_libraries
|
||||
mixed_libs = set(self.libraries_of_kind("mixed"))
|
||||
if selected is not None:
|
||||
mixed_libs &= set(selected)
|
||||
mixed_series = (
|
||||
[s for s in self.load_series() if s.library in mixed_libs]
|
||||
if mixed_libs else [])
|
||||
|
||||
mirror_libs = self.mirror_as_is_libraries
|
||||
if selected is None:
|
||||
return list(self.filelist), mirror_libs, mixed_series
|
||||
chosen = set(selected)
|
||||
movies = [f for f in self.filelist if self.library_of(f) in chosen]
|
||||
mirrors = [name for name in mirror_libs if name in chosen]
|
||||
return movies, mirrors, mixed_series
|
||||
|
||||
def minidlna_prune_targets(
|
||||
self, mirror_libs: list[str], mixed_series: list, mixed_series_bucket: str,
|
||||
) -> tuple[list[str], set[str]]:
|
||||
"""What miniDLNA output to remove for mirrors that fell out of scope.
|
||||
|
||||
``mirror_libs`` / ``mixed_series`` come from :meth:`minidlna_scope` (the
|
||||
*current* run's selection); ``mixed_series_bucket`` is the shared output
|
||||
folder mixed-library series hardcopy into (e.g. ``"Seriály"``) — the same
|
||||
name a real ``series``/``homevideo`` library may also use, so the two
|
||||
sources are reconciled together rather than one blindly wiping the other.
|
||||
|
||||
Returns ``(stale_libraries, valid_bucket_children)``:
|
||||
|
||||
- ``stale_libraries`` — mirrored-library output folders (``output/<name>``)
|
||||
to delete **entirely**: every :attr:`mirror_as_is_libraries` name no
|
||||
longer in ``mirror_libs``, except one that shares the bucket's name
|
||||
(reconciled via the second element instead).
|
||||
- ``valid_bucket_children`` — basenames that must remain directly under
|
||||
``output/<mixed_series_bucket>`` this run: the in-scope mixed-library
|
||||
series' own folder names, plus — when a real library named
|
||||
``mixed_series_bucket`` is itself in scope — that library's current
|
||||
subfolders. Anything else found there is stale.
|
||||
"""
|
||||
in_scope = set(mirror_libs)
|
||||
stale = [name for name in self.mirror_as_is_libraries
|
||||
if name not in in_scope and name != mixed_series_bucket]
|
||||
|
||||
valid = {ser.folder.name for ser in mixed_series}
|
||||
if mixed_series_bucket in in_scope:
|
||||
pool = self.pool_dir
|
||||
src = pool / mixed_series_bucket if pool else None
|
||||
if src and src.is_dir():
|
||||
valid |= {p.name for p in src.iterdir() if p.is_dir()}
|
||||
return stale, valid
|
||||
|
||||
@property
|
||||
def mirror_as_is_libraries(self) -> list[str]:
|
||||
"""Libraries hardcopied 1:1 into miniDLNA — ``series`` and ``homevideo``.
|
||||
|
||||
Neither has genre tags to build a tree from (homevideo has no TMDb id
|
||||
either), so both mirror whole into ``output/<library name>`` instead of
|
||||
going through the tag tree. A superset of :attr:`copyasis_folders`
|
||||
(which the library organizer UI still calls "copy-as-is" and keeps
|
||||
scoped to kind ``series`` only).
|
||||
"""
|
||||
return self.libraries_of_kind("series", "homevideo")
|
||||
|
||||
@property
|
||||
def copyasis_folders(self) -> list[str]:
|
||||
"""Names of pool subfolders mirrored 1:1 (copy-as-is) — the series libraries."""
|
||||
@@ -334,7 +449,7 @@ class FileManager:
|
||||
|
||||
Returns a dict with:
|
||||
ok — False if the pool isn't set (with 'error')
|
||||
missing — index keys whose file no longer exists on disk
|
||||
missing — indexed media files that no longer exist on disk
|
||||
untracked — pool files (pool-relative) with no index entry
|
||||
index_count / disk_count — totals for the summary
|
||||
"""
|
||||
@@ -344,7 +459,14 @@ class FileManager:
|
||||
"missing": [], "untracked": []}
|
||||
|
||||
index = self.index or PoolIndex(pool)
|
||||
index_keys = set(index.records.keys())
|
||||
# Records are keyed by movie identity, so the paths to compare are the
|
||||
# media entries listed inside them.
|
||||
index_keys = {
|
||||
media["file"]
|
||||
for record in index.all_movies().values()
|
||||
for media in record.get("media", [])
|
||||
if media.get("file")
|
||||
}
|
||||
|
||||
disk_keys = set()
|
||||
for lib in self.libraries:
|
||||
@@ -518,6 +640,7 @@ class FileManager:
|
||||
move: bool = False, on_conflict: str = "suffix",
|
||||
year: int | str | None = None, tmdb_id: int | str | None = None,
|
||||
library: str | None = None, csfd_url: str | None = None,
|
||||
format_3d: str | None = None,
|
||||
) -> File | None:
|
||||
"""Bring a video file into a movie library and index its metadata.
|
||||
|
||||
@@ -527,6 +650,8 @@ class FileManager:
|
||||
``tmdb_id`` when known — the form Jellyfin matches on); without a year it
|
||||
falls back to ``Title.ext``. The GUI import always supplies a year, so a
|
||||
library built through the app is uniformly ``Title (YYYY) [tmdbid-N].ext``.
|
||||
``format_3d`` (one of :data:`naming.THREE_D_FORMATS`) appends the
|
||||
``.3D.<FORMAT>`` tag right away, when the file is already a 3D rip.
|
||||
|
||||
By default the original is **copied** (non-destructive). With ``move=True``
|
||||
the source file is moved into the pool instead, leaving nothing behind.
|
||||
@@ -555,7 +680,7 @@ class FileManager:
|
||||
source = Path(source)
|
||||
safe_title = title.strip() or source.stem
|
||||
has_year = year is not None and str(year).strip() != ""
|
||||
stem = (canonical_pool_stem(safe_title, year, tmdb_id) if has_year
|
||||
stem = (canonical_pool_stem(safe_title, year, tmdb_id, format_3d) if has_year
|
||||
else safe_title)
|
||||
target = movies / f"{stem}{source.suffix}"
|
||||
|
||||
@@ -600,7 +725,9 @@ class FileManager:
|
||||
extension is preserved; ``new_title`` is the bare name without it.
|
||||
|
||||
Raises:
|
||||
ValueError: empty name or a name containing a path separator.
|
||||
ValueError: empty name, a name containing a path separator, or
|
||||
``file_obj`` is one version of a multi-version movie (renaming
|
||||
just one file would orphan its siblings).
|
||||
FileExistsError: another pooled file already uses that name.
|
||||
"""
|
||||
new_title = new_title.strip()
|
||||
@@ -608,6 +735,10 @@ class FileManager:
|
||||
raise ValueError("Název nesmí být prázdný.")
|
||||
if "/" in new_title or "\\" in new_title:
|
||||
raise ValueError("Název nesmí obsahovat lomítka.")
|
||||
if self.is_movie_group_member(file_obj):
|
||||
raise ValueError(
|
||||
"Film s více verzemi zatím nelze takhle přejmenovat — "
|
||||
"odeber verzi nebo uprav soubory ve složce filmu ručně.")
|
||||
|
||||
old_path = file_obj.file_path
|
||||
new_path = old_path.with_name(f"{new_title}{old_path.suffix}")
|
||||
@@ -618,6 +749,7 @@ class FileManager:
|
||||
raise FileExistsError(f"Soubor „{new_path.name}“ už v poolu existuje.")
|
||||
|
||||
old_path.rename(new_path)
|
||||
subtitles.rename_subtitles_with(old_path, new_path)
|
||||
file_obj.relocate(new_path)
|
||||
file_obj.title = new_title
|
||||
file_obj.save_metadata()
|
||||
@@ -626,13 +758,15 @@ class FileManager:
|
||||
self.on_files_changed(self.filelist)
|
||||
return file_obj
|
||||
|
||||
def _canonical_name_for(self, file_obj: File) -> str | None:
|
||||
def _canonical_name_for(self, file_obj: File, *, include_3d: bool = True) -> str | None:
|
||||
"""The ``Title (YYYY) [tmdbid-N]`` stem this movie should have, or ``None``.
|
||||
|
||||
Uses the clean title and the year from the movie's metadata
|
||||
(``name_context``: cache → ``Rok`` tag → the current filename). A
|
||||
``(YYYY)`` that leaked into the title is stripped so it isn't doubled, and
|
||||
an existing ``[tmdbid-N]`` on the current filename is preserved. Returns
|
||||
an existing ``[tmdbid-N]`` / ``.3D.<FORMAT>`` on the current filename is
|
||||
preserved (``include_3d=False`` drops the 3D tag — used for a movie-group
|
||||
**folder** name, where 3D becomes a per-version label instead). Returns
|
||||
``None`` when no four-digit year is known.
|
||||
"""
|
||||
ctx = file_obj.name_context()
|
||||
@@ -644,13 +778,14 @@ class FileManager:
|
||||
title = parsed[0]
|
||||
year = year or str(parsed[1])
|
||||
|
||||
# Keep the TMDb id already in the pool filename, if any.
|
||||
# Keep the TMDb id / 3D tag already in the pool filename, if any.
|
||||
tmdb_id = parse_tmdb_id(file_obj.file_path.stem)
|
||||
format_3d = parse_3d_format(file_obj.file_path.stem) if include_3d else None
|
||||
|
||||
title = title.replace("/", "-").replace("\\", "-").strip()
|
||||
if not (title and is_valid_year(year)):
|
||||
return None
|
||||
return canonical_pool_stem(title, year, tmdb_id)
|
||||
return canonical_pool_stem(title, year, tmdb_id, format_3d)
|
||||
|
||||
def rename_to_canonical(self, file_obj: File) -> File | None:
|
||||
"""Rename a pooled movie to ``Title (YYYY).ext`` from its metadata.
|
||||
@@ -658,15 +793,20 @@ class FileManager:
|
||||
The year comes from the movie's metadata (NFO cache, falling back to a
|
||||
``Rok`` tag or the existing filename). ``file_obj.title`` keeps the title
|
||||
(without the year). Does not fire ``on_files_changed`` — the caller
|
||||
refreshes once after a batch.
|
||||
refreshes once after a batch. A movie with multiple versions is skipped
|
||||
(``None``) — renaming one version alone would orphan its siblings; use
|
||||
the version tools in *Upravit film* instead.
|
||||
|
||||
Returns:
|
||||
The file when renamed (or already canonical); ``None`` when no
|
||||
four-digit year is known, so no canonical name can be built.
|
||||
four-digit year is known (or the movie has multiple versions), so
|
||||
no canonical name can be built.
|
||||
|
||||
Raises:
|
||||
FileExistsError: another pooled file already uses that name.
|
||||
"""
|
||||
if self.is_movie_group_member(file_obj):
|
||||
return None
|
||||
new_stem = self._canonical_name_for(file_obj)
|
||||
if new_stem is None:
|
||||
return None
|
||||
@@ -680,6 +820,7 @@ class FileManager:
|
||||
|
||||
parsed = parse_pool_stem(new_stem)
|
||||
old_path.rename(new_path)
|
||||
subtitles.rename_subtitles_with(old_path, new_path)
|
||||
file_obj.relocate(new_path)
|
||||
if parsed is not None:
|
||||
file_obj.title = parsed[0]
|
||||
@@ -740,7 +881,9 @@ class FileManager:
|
||||
|
||||
Moves the file into ``pool/<new_library>/`` and re-indexes it under the
|
||||
new key. Raises:
|
||||
ValueError: the target isn't a known movie library.
|
||||
ValueError: the target isn't a known movie library, or ``file_obj``
|
||||
is one version of a multi-version movie (moving just one file
|
||||
would orphan its siblings).
|
||||
FileExistsError: a file of that name already sits in the target.
|
||||
"""
|
||||
pool = self.pool_dir
|
||||
@@ -749,6 +892,10 @@ class FileManager:
|
||||
new_library = new_library.strip()
|
||||
if new_library not in self.libraries_accepting_movies:
|
||||
raise ValueError(f"Neznámá knihovna „{new_library}“.")
|
||||
if self.is_movie_group_member(file_obj):
|
||||
raise ValueError(
|
||||
"Film s více verzemi zatím nelze takhle přesunout — "
|
||||
"odeber verzi, nebo přesuň celou složku filmu ručně.")
|
||||
|
||||
old_path = file_obj.file_path
|
||||
dest_dir = pool / new_library
|
||||
@@ -760,6 +907,7 @@ class FileManager:
|
||||
raise FileExistsError(f"V knihovně „{new_library}“ už „{old_path.name}“ je.")
|
||||
|
||||
shutil.move(str(old_path), str(new_path))
|
||||
subtitles.rename_subtitles_with(old_path, new_path)
|
||||
file_obj.relocate(new_path) # drops the old index key, rebinds to the new
|
||||
file_obj.save_metadata()
|
||||
|
||||
@@ -772,13 +920,19 @@ class FileManager:
|
||||
tmdb_id: int | str | None = None, library: str | None = None,
|
||||
date: str | None = None, attributes: dict | None = None,
|
||||
tags: list[str] | None = None, csfd_url: str | None = None,
|
||||
format_3d: str | None = None,
|
||||
) -> File:
|
||||
"""Apply all editable parameters of a movie at once (rename/move + metadata).
|
||||
|
||||
Renames the file to ``Title (rok) [tmdbid].ext`` (year permitting), moves
|
||||
it into ``library``, and sets ``title`` / ``date`` / ``attributes`` /
|
||||
``tags`` (a full ``["Category/Name", …]`` list; NFO-tag provenance is kept
|
||||
for the ones that remain). No-ops that don't change the path skip the move.
|
||||
Renames the file to ``Title (rok) [tmdbid].3D.<FORMAT>.ext`` (year/3D
|
||||
permitting), moves it into ``library``, and sets ``title`` / ``date`` /
|
||||
``attributes`` / ``tags`` (a full ``["Category/Name", …]`` list;
|
||||
NFO-tag provenance is kept for the ones that remain). No-ops that don't
|
||||
change the path skip the move.
|
||||
|
||||
``format_3d`` follows :func:`naming.canonical_pool_stem`: omitted
|
||||
(``None``) keeps whatever 3D tag the file already has, an empty string
|
||||
clears it, one of :data:`naming.THREE_D_FORMATS` sets it.
|
||||
|
||||
Raises:
|
||||
FileExistsError: the target name is already taken.
|
||||
@@ -788,18 +942,34 @@ class FileManager:
|
||||
raise RuntimeError("Pool není nastaven.")
|
||||
title = (title or "").strip() or file_obj.file_path.stem
|
||||
year_s = str(year).strip() if year not in (None, "") else ""
|
||||
stem = (canonical_pool_stem(title, year_s, tmdb_id)
|
||||
grouped = self.is_movie_group_member(file_obj)
|
||||
if format_3d is None:
|
||||
format_3d = None if grouped else parse_3d_format(file_obj.file_path.stem)
|
||||
stem = (canonical_pool_stem(title, year_s, tmdb_id, format_3d)
|
||||
if is_valid_year(year_s) else title)
|
||||
lib = ((library or "").strip()
|
||||
or self.library_of(file_obj) or self.movie_libraries[0])
|
||||
old_path = file_obj.file_path
|
||||
new_path = pool / lib / f"{stem}{old_path.suffix}"
|
||||
if new_path != old_path:
|
||||
if new_path.exists():
|
||||
raise FileExistsError(f"Soubor „{new_path.name}“ už existuje.")
|
||||
new_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(str(old_path), str(new_path))
|
||||
file_obj.relocate(new_path)
|
||||
|
||||
if grouped:
|
||||
# Identity is locked to the container folder's name; only a
|
||||
# no-op (title/year/tmdb_id/library all unchanged) is allowed
|
||||
# here — metadata-only fields below still apply normally.
|
||||
if stem != old_path.parent.name or lib != self.library_of(file_obj):
|
||||
raise ValueError(
|
||||
"Film s více verzemi zatím nelze takhle přejmenovat/přesunout "
|
||||
"(rok/název/TMDb ID/knihovna) — odeber verzi, nebo uprav "
|
||||
"soubory ve složce filmu ručně. Datum/atributy/tagy/ČSFD "
|
||||
"odkaz upravit lze.")
|
||||
else:
|
||||
new_path = pool / lib / f"{stem}{old_path.suffix}"
|
||||
if new_path != old_path:
|
||||
if new_path.exists():
|
||||
raise FileExistsError(f"Soubor „{new_path.name}“ už existuje.")
|
||||
new_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(str(old_path), str(new_path))
|
||||
subtitles.rename_subtitles_with(old_path, new_path)
|
||||
file_obj.relocate(new_path)
|
||||
|
||||
file_obj.title = title
|
||||
if date is not None:
|
||||
@@ -826,6 +996,218 @@ class FileManager:
|
||||
self.on_files_changed(self.filelist)
|
||||
return file_obj
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Movie versions (Jellyfin "multiple versions": 2D + 3D, 4K + 1080p, …)
|
||||
# ------------------------------------------------------------------
|
||||
#
|
||||
# A movie is identified by (title, year, tmdbid) — the canonical stem. As
|
||||
# long as it has exactly one video file, it stays flat:
|
||||
# pool/<Library>/Title (rok) [tmdbid].ext
|
||||
# The moment a second version is added, Curator promotes it to its own
|
||||
# folder (Jellyfin's own "multiple versions" convention — every file in it
|
||||
# must start with the folder's exact name):
|
||||
# pool/<Library>/Title (rok) [tmdbid]/
|
||||
# Title (rok) [tmdbid] - Výchozí.ext
|
||||
# Title (rok) [tmdbid] - 3D HSBS.ext
|
||||
# Shared metadata (tags, ČSFD odkaz, datum, atributy) lives on whichever
|
||||
# version is alphabetically first (Jellyfin's own "default version" rule —
|
||||
# `visible_movies` shows only that one in the table). Renaming/moving/
|
||||
# retitling a grouped movie is not supported yet (see the guards in
|
||||
# `rename_movie` / `move_to_library` / `apply_movie_edits`) — remove a
|
||||
# version (auto-demotes back to flat when one is left) or edit the files
|
||||
# directly if you need that.
|
||||
|
||||
def is_movie_group_member(self, file_obj: File) -> bool:
|
||||
"""True when ``file_obj`` is one version inside its own per-movie folder.
|
||||
|
||||
Only recognised for ``movie``-kind libraries in this first cut — a
|
||||
``mixed`` library's subfolders are always treated as series.
|
||||
"""
|
||||
pool = self.pool_dir
|
||||
lib = self.library_of(file_obj)
|
||||
if pool is None or lib is None or lib not in self.libraries_of_kind("movie"):
|
||||
return False
|
||||
parent = file_obj.file_path.parent
|
||||
return parent != pool / lib and is_canonical_pool_stem(parent.name)
|
||||
|
||||
def list_movie_versions(self, file_obj: File) -> list[File]:
|
||||
"""Every version sharing ``file_obj``'s movie-group folder, name-sorted.
|
||||
|
||||
A single-element list (just ``file_obj``) when it isn't grouped.
|
||||
"""
|
||||
if not self.is_movie_group_member(file_obj):
|
||||
return [file_obj]
|
||||
folder = file_obj.file_path.parent
|
||||
versions = []
|
||||
for p in sorted(folder.iterdir()):
|
||||
if (not p.is_file() or p.suffix.lower() in SIDECAR_SUFFIXES
|
||||
or p.name.endswith(METADATA_SUFFIXES)):
|
||||
continue
|
||||
existing = next((f for f in self.filelist if f.file_path == p), None)
|
||||
versions.append(existing or File(p, self.tagmanager, index=self.index))
|
||||
return versions
|
||||
|
||||
def primary_version(self, versions: list[File]) -> File:
|
||||
"""Which version represents the movie in the table.
|
||||
|
||||
Metadata is shared by the whole movie record now, so this is purely a
|
||||
display choice: the plain/unlabelled file first, then the one labelled
|
||||
:data:`movie.DEFAULT_VERSION_LABEL`, then alphabetically. Unrelated to
|
||||
which version Jellyfin plays by default (its own, alphabetical rule).
|
||||
"""
|
||||
from .movie import DEFAULT_VERSION_LABEL
|
||||
|
||||
def rank(f: File) -> tuple[int, str]:
|
||||
label = f._identity_from_path()[4]
|
||||
order = 0 if not label else (1 if label == DEFAULT_VERSION_LABEL else 2)
|
||||
return order, f.file_path.name.casefold()
|
||||
|
||||
return min(versions, key=rank)
|
||||
|
||||
def visible_movies(self) -> list[File]:
|
||||
"""``filelist`` collapsed to one row per movie for the table.
|
||||
|
||||
Ungrouped files pass through unchanged; a movie-group folder
|
||||
contributes only its :meth:`primary_version` — the rest are reachable
|
||||
via *Upravit film → Verze*, not as separate rows.
|
||||
"""
|
||||
seen_folders: set[Path] = set()
|
||||
result: list[File] = []
|
||||
for f in self.filelist:
|
||||
if not self.is_movie_group_member(f):
|
||||
result.append(f)
|
||||
continue
|
||||
folder = f.file_path.parent
|
||||
if folder in seen_folders:
|
||||
continue
|
||||
seen_folders.add(folder)
|
||||
result.append(self.primary_version(self.list_movie_versions(f)))
|
||||
return result
|
||||
|
||||
def add_movie_version(
|
||||
self, file_obj: File, source: Path, label: str, *,
|
||||
move: bool = False, existing_label: str = "Výchozí",
|
||||
) -> File:
|
||||
"""Add ``source`` as another playable version of ``file_obj``'s movie.
|
||||
|
||||
Promotes a flat movie to its own per-movie folder on the first extra
|
||||
version — the pre-existing file is moved in as
|
||||
``<folder> - {existing_label}.ext`` (only on this first promotion) and
|
||||
``source`` lands as ``<folder> - {label}.ext``. A movie already grouped
|
||||
just gets one more file in the same folder. Shared metadata stays
|
||||
wherever it already was; the new version starts with none of its own
|
||||
(attach subtitles / a 3D formát to it separately if needed).
|
||||
|
||||
Raises:
|
||||
RuntimeError: no pool, or ``file_obj`` has no valid title+year yet.
|
||||
ValueError: ``label`` is empty.
|
||||
FileExistsError: a version of that name already exists.
|
||||
"""
|
||||
label = (label or "").strip()
|
||||
if not label:
|
||||
raise ValueError("Popisek verze nesmí být prázdný.")
|
||||
pool = self.pool_dir
|
||||
if pool is None:
|
||||
raise RuntimeError("Pool není nastaven.")
|
||||
source = Path(source)
|
||||
|
||||
stem: str | None
|
||||
if self.is_movie_group_member(file_obj):
|
||||
folder = file_obj.file_path.parent
|
||||
stem = folder.name
|
||||
else:
|
||||
stem = self._canonical_name_for(file_obj, include_3d=False)
|
||||
if stem is None:
|
||||
raise RuntimeError(
|
||||
"Film nemá platný název/rok — nejde založit složku verzí.")
|
||||
lib = self.library_of(file_obj) or self.movie_libraries[0]
|
||||
folder = pool / lib / stem
|
||||
if folder.exists():
|
||||
raise FileExistsError(f"Složka „{stem}“ už v poolu existuje.")
|
||||
folder.mkdir(parents=True)
|
||||
old_path = file_obj.file_path
|
||||
existing_target = folder / f"{stem} - {existing_label}{old_path.suffix}"
|
||||
shutil.move(str(old_path), str(existing_target))
|
||||
subtitles.rename_subtitles_with(old_path, existing_target)
|
||||
file_obj.relocate(existing_target)
|
||||
file_obj.save_metadata()
|
||||
|
||||
target = folder / f"{stem} - {label}{source.suffix}"
|
||||
if target.exists():
|
||||
raise FileExistsError(f"Verze „{label}“ už existuje.")
|
||||
if move:
|
||||
shutil.move(str(source), str(target))
|
||||
else:
|
||||
shutil.copy2(source, target)
|
||||
new_version = File(target, self.tagmanager, index=self.index)
|
||||
new_version.added = datetime.now().isoformat(timespec="seconds")
|
||||
new_version.save_metadata()
|
||||
self.filelist.append(new_version)
|
||||
if self.on_files_changed:
|
||||
self.on_files_changed(self.filelist)
|
||||
return new_version
|
||||
|
||||
def remove_movie_version(self, file_obj: File) -> None:
|
||||
"""Remove one version (+ its subtitles) from a movie's version group.
|
||||
|
||||
When exactly one version is left afterward, the movie is demoted back
|
||||
to a flat file — moved out to the library root under the bare
|
||||
canonical stem — and the now-empty folder is removed. Metadata needs no
|
||||
rescuing: it lives on the movie record, not on any one version, so
|
||||
removing even the version shown in the table leaves it untouched.
|
||||
|
||||
Raises:
|
||||
ValueError: ``file_obj`` isn't part of a version group (nothing to
|
||||
remove it *from* — a lone movie has no "versions" to drop).
|
||||
"""
|
||||
if not self.is_movie_group_member(file_obj):
|
||||
raise ValueError("Film nemá žádné další verze k odebrání.")
|
||||
folder = file_obj.file_path.parent
|
||||
file_obj.delete_metadata()
|
||||
for sub in subtitles.find_subtitles(file_obj.file_path):
|
||||
sub.path.unlink()
|
||||
file_obj.file_path.unlink()
|
||||
if file_obj in self.filelist:
|
||||
self.filelist.remove(file_obj)
|
||||
|
||||
remaining = [
|
||||
p for p in folder.iterdir()
|
||||
if p.is_file() and p.suffix.lower() not in SIDECAR_SUFFIXES
|
||||
and not p.name.endswith(METADATA_SUFFIXES)]
|
||||
if len(remaining) == 1:
|
||||
sole = remaining[0]
|
||||
sole_file = next((f for f in self.filelist if f.file_path == sole), None)
|
||||
new_path = folder.parent / f"{folder.name}{sole.suffix}"
|
||||
sole.rename(new_path)
|
||||
subtitles.rename_subtitles_with(sole, new_path)
|
||||
if sole_file is not None:
|
||||
sole_file.relocate(new_path)
|
||||
sole_file.save_metadata()
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
|
||||
if self.on_files_changed:
|
||||
self.on_files_changed(self.filelist)
|
||||
|
||||
def list_subtitles(self, file_obj: File) -> list:
|
||||
"""External subtitle sidecars currently sitting next to a movie."""
|
||||
return subtitles.find_subtitles(file_obj.file_path)
|
||||
|
||||
def attach_subtitle(
|
||||
self, file_obj: File, source: Path, *, language: str | None = None,
|
||||
flag: str | None = None, move: bool = False,
|
||||
) -> Path:
|
||||
"""Copy (or move) a picked subtitle file next to ``file_obj``, correctly named.
|
||||
|
||||
Raises:
|
||||
ValueError: ``source`` isn't a recognised subtitle format.
|
||||
"""
|
||||
return subtitles.attach_subtitle(
|
||||
file_obj.file_path, source, language=language, flag=flag, move=move)
|
||||
|
||||
def remove_subtitle(self, subtitle_path: Path) -> None:
|
||||
"""Delete one attached subtitle file."""
|
||||
Path(subtitle_path).unlink()
|
||||
|
||||
def refresh_tags_from_tmdb(
|
||||
self, files: list[File] | None = None,
|
||||
on_progress: Callable[[int, int, File], None] | None = None,
|
||||
@@ -867,6 +1249,43 @@ class FileManager:
|
||||
return {"ok": True, "updated": updated, "without_id": without_id,
|
||||
"errors": errors}
|
||||
|
||||
def resolve_via_csfd(
|
||||
self, *, csfd_url: str | None = None, query: str | None = None,
|
||||
session: object | None = None,
|
||||
) -> dict:
|
||||
"""ČSFD → TMDb bridge for one movie.
|
||||
|
||||
With ``csfd_url`` set, reads that ČSFD film; otherwise searches ČSFD for
|
||||
``query`` (a cleaned filename) and takes the top hit. The ČSFD original
|
||||
title + year then drive a TMDb search — far more reliable than a raw
|
||||
Czech-title search. Pass a shared ``requests.Session`` to reuse the
|
||||
Anubis auth cookie across a batch.
|
||||
|
||||
Returns ``{"csfd_url", "film": CsfdFilm | None, "tmdb": TmdbResult | None}``.
|
||||
Best-effort: a network / anti-bot / no-match failure is a ``None`` in the
|
||||
result, not an exception.
|
||||
"""
|
||||
from . import csfd, tmdb
|
||||
result: dict = {"csfd_url": csfd_url, "film": None, "tmdb": None}
|
||||
try:
|
||||
url = csfd_url
|
||||
if not url and query:
|
||||
url = csfd.find_csfd_url(query, session=session)
|
||||
if not url:
|
||||
return result
|
||||
result["csfd_url"] = url
|
||||
film = csfd.fetch_film(url, session=session)
|
||||
result["film"] = film
|
||||
title, year = film.tmdb_query
|
||||
if title and tmdb.api_available():
|
||||
hits = tmdb.search_movie(title, year=year, limit=1)
|
||||
if not hits and film.czech_title and film.czech_title != title:
|
||||
hits = tmdb.search_movie(film.czech_title, year=year, limit=1)
|
||||
result["tmdb"] = hits[0] if hits else None
|
||||
except (csfd.CsfdError, tmdb.TmdbError, OSError):
|
||||
pass
|
||||
return result
|
||||
|
||||
def match_pool_to_tmdb(
|
||||
self, files: list[File] | None = None,
|
||||
on_progress: Callable[[int, int, File], None] | None = None,
|
||||
@@ -875,14 +1294,18 @@ class FileManager:
|
||||
"""Propose a TMDb id for every movie that doesn't have one yet.
|
||||
|
||||
Searches TMDb by each movie's title + year and classifies the result
|
||||
(see :func:`migrate.classify`). Movies whose filename already carries a
|
||||
(see :func:`migrate.classify`). A movie that carries a **ČSFD link**
|
||||
(``File.csfd_url``) is first resolved through ČSFD so the search runs on
|
||||
the *original* title + year — the pure Czech-title search picks the wrong
|
||||
film for remakes/localized names. Movies whose filename already carries a
|
||||
``[tmdbid-N]`` are skipped. Returns a list of ``migrate.MatchProposal``.
|
||||
"""
|
||||
from . import tmdb
|
||||
from . import csfd, tmdb
|
||||
from .migrate import MatchProposal, classify
|
||||
targets = files if files is not None else self.filelist
|
||||
proposals: list = []
|
||||
total = len(targets)
|
||||
csfd_session = None
|
||||
for i, f in enumerate(targets, 1):
|
||||
if should_cancel is not None and should_cancel():
|
||||
break
|
||||
@@ -896,14 +1319,28 @@ class FileManager:
|
||||
if parsed is not None:
|
||||
title = parsed[0]
|
||||
year = str(ctx.get("year") or "").strip() or None
|
||||
|
||||
note = ""
|
||||
if f.csfd_url and csfd.dependencies_available():
|
||||
if csfd_session is None and csfd.requests is not None:
|
||||
csfd_session = csfd.requests.Session()
|
||||
try:
|
||||
film = csfd.fetch_film(f.csfd_url, session=csfd_session)
|
||||
except csfd.CsfdError:
|
||||
film = None
|
||||
if film is not None and (film.original_title or film.czech_title):
|
||||
title = film.original_title or film.czech_title
|
||||
year = film.year or year
|
||||
note = f"přes ČSFD: {title}" + (f" ({film.year})" if film.year else "")
|
||||
|
||||
try:
|
||||
candidates = tmdb.search_movie(title, year=year, limit=5)
|
||||
status, chosen = classify(title, year, candidates)
|
||||
proposals.append(MatchProposal(
|
||||
f, f.filename, title, year, status, chosen, candidates))
|
||||
f, f.filename, title, year, status, chosen, candidates, note=note))
|
||||
except tmdb.TmdbError as exc:
|
||||
proposals.append(MatchProposal(
|
||||
f, f.filename, title, year, "error", None, [], str(exc)))
|
||||
f, f.filename, title, year, "error", None, [], str(exc), note))
|
||||
if on_progress is not None:
|
||||
on_progress(i, total, f)
|
||||
return proposals
|
||||
@@ -973,17 +1410,21 @@ class FileManager:
|
||||
self, sources: list[Path], title: str, year: int | str | None = None,
|
||||
tmdb_id: int | str | None = None, library: str | None = None,
|
||||
move: bool = False, on_conflict: str = "suffix",
|
||||
codes: dict[Path, tuple[int, int]] | None = None,
|
||||
) -> dict:
|
||||
"""Import episode files into ``<library>/Series (rok) [tmdbid]/Season NN/``.
|
||||
|
||||
Each source's season/episode is parsed from its filename and it is copied
|
||||
(or moved) in as ``Series SxxEyy.ext``. Files with no recognisable
|
||||
``SxxEyy`` are skipped and reported.
|
||||
Each source's season/episode comes from ``codes`` (an explicit
|
||||
``{path: (season, episode)}`` override set at import time) when present,
|
||||
otherwise it is parsed from the filename. It is then copied (or moved) in
|
||||
as ``Series SxxEyy.ext``. Files with neither an override nor a
|
||||
recognisable ``SxxEyy`` are skipped and reported.
|
||||
|
||||
Returns ``{imported, skipped_no_code, errors}``.
|
||||
"""
|
||||
from .series import (
|
||||
parse_episode, series_folder_stem, season_folder_name, episode_filename)
|
||||
codes = codes or {}
|
||||
pool = self.pool_dir
|
||||
if pool is None:
|
||||
raise RuntimeError("Pool není nastaven.")
|
||||
@@ -996,21 +1437,21 @@ class FileManager:
|
||||
errors: list[str] = []
|
||||
for src in sources:
|
||||
src = Path(src)
|
||||
code = parse_episode(src.name)
|
||||
code = codes.get(src) or parse_episode(src.name)
|
||||
if code is None:
|
||||
skipped_no_code.append(src.name)
|
||||
continue
|
||||
season, ep = code
|
||||
dest_dir = pool / lib / stem / season_folder_name(season)
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
target = dest_dir / episode_filename(season, ep, src.stem, src.suffix)
|
||||
target = dest_dir / episode_filename(season, ep, title, src.suffix)
|
||||
if target.exists():
|
||||
if on_conflict == "skip":
|
||||
continue
|
||||
if on_conflict == "replace":
|
||||
target.unlink()
|
||||
else: # suffix
|
||||
base = episode_filename(season, ep, src.stem, "")
|
||||
base = episode_filename(season, ep, title, "")
|
||||
counter = 1
|
||||
while target.exists():
|
||||
target = dest_dir / f"{base}_{counter}{src.suffix}"
|
||||
@@ -1027,7 +1468,7 @@ class FileManager:
|
||||
"errors": errors}
|
||||
|
||||
def rename_episodes_to_canonical(self, series) -> dict:
|
||||
"""Rename a series' episodes to ``Series SxxEyy.ext`` in the right season.
|
||||
"""Rename a series' episodes to ``SxxEyy - {series title}.ext`` in its season.
|
||||
|
||||
``series`` is a ``series.Series``. Episodes with no recognisable code are
|
||||
skipped; a target name already taken is reported as a conflict.
|
||||
@@ -1044,7 +1485,7 @@ class FileManager:
|
||||
continue
|
||||
dest_dir = series.folder / season_folder_name(ep.season)
|
||||
new_path = dest_dir / episode_filename(
|
||||
ep.season, ep.episode, ep.path.stem, ep.path.suffix)
|
||||
ep.season, ep.episode, series.title, ep.path.suffix)
|
||||
if new_path == ep.path:
|
||||
continue
|
||||
if new_path.exists():
|
||||
@@ -1052,14 +1493,64 @@ class FileManager:
|
||||
continue
|
||||
try:
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
old_name = ep.path.name
|
||||
ep.path.rename(new_path)
|
||||
old_path = ep.path
|
||||
old_path.rename(new_path)
|
||||
subtitles.rename_subtitles_with(old_path, new_path)
|
||||
ep.path = new_path
|
||||
renamed.append((old_name, new_path.name))
|
||||
renamed.append((old_path.name, new_path.name))
|
||||
except OSError as exc:
|
||||
conflicts.append((ep.path.name, str(exc)))
|
||||
return {"renamed": renamed, "skipped": skipped, "conflicts": conflicts}
|
||||
|
||||
def rename_all_series_to_canonical(
|
||||
self,
|
||||
on_progress: Callable[[int, int, str], None] | None = None,
|
||||
should_cancel: Callable[[], bool] | None = None,
|
||||
) -> dict:
|
||||
"""Canonicalise every series' episode filenames across the pool.
|
||||
|
||||
Iterates all series (``series`` + ``mixed`` libraries); per series it
|
||||
renames each episode to ``SxxEyy - {series title}.ext`` in the right
|
||||
``Season NN`` / ``Specials`` folder (:meth:`rename_episodes_to_canonical`)
|
||||
and then removes any season folder left empty by the moves. Episodes with
|
||||
no recognisable ``SxxEyy`` are left in place and listed.
|
||||
|
||||
Returns ``{series, renamed:[(series, old, new)],
|
||||
skipped_no_code:[(series, name)], conflicts:[(series, name, msg)],
|
||||
empty_dirs_removed:[str], cancelled}``.
|
||||
"""
|
||||
from .series import is_season_folder
|
||||
pool = self.pool_dir
|
||||
if pool is None:
|
||||
raise RuntimeError("Pool není nastaven.")
|
||||
all_series = self.load_series()
|
||||
total = len(all_series)
|
||||
renamed: list[tuple[str, str, str]] = []
|
||||
skipped: list[tuple[str, str]] = []
|
||||
conflicts: list[tuple[str, str, str]] = []
|
||||
empty_removed: list[str] = []
|
||||
cancelled = False
|
||||
for i, ser in enumerate(all_series, 1):
|
||||
if should_cancel is not None and should_cancel():
|
||||
cancelled = True
|
||||
break
|
||||
if on_progress is not None:
|
||||
on_progress(i, total, ser.title)
|
||||
res = self.rename_episodes_to_canonical(ser)
|
||||
renamed += [(ser.title, old, new) for old, new in res["renamed"]]
|
||||
skipped += [(ser.title, name) for name in res["skipped"]]
|
||||
conflicts += [(ser.title, name, msg) for name, msg in res["conflicts"]]
|
||||
for sub in sorted(p for p in ser.folder.iterdir() if p.is_dir()):
|
||||
if is_season_folder(sub.name) and not any(sub.iterdir()):
|
||||
try:
|
||||
sub.rmdir()
|
||||
empty_removed.append(str(sub.relative_to(pool)))
|
||||
except OSError:
|
||||
pass
|
||||
return {"series": total, "renamed": renamed, "skipped_no_code": skipped,
|
||||
"conflicts": conflicts, "empty_dirs_removed": empty_removed,
|
||||
"cancelled": cancelled}
|
||||
|
||||
def apply_series_edits(
|
||||
self, series, *, title: str, year: int | str | None = None,
|
||||
tmdb_id: int | str | None = None, library: str | None = None,
|
||||
@@ -1096,6 +1587,27 @@ class FileManager:
|
||||
else {"renamed": [], "skipped": [], "conflicts": []})
|
||||
return {"folder": new_folder, "episodes": result}
|
||||
|
||||
def remove_series(self, series) -> None:
|
||||
"""Delete a whole series folder from the pool (every season and episode).
|
||||
|
||||
Series carry no index metadata (they are scanned from disk), so removing
|
||||
one is just deleting its folder. Refuses a folder that is not inside the
|
||||
pool as a safety check.
|
||||
|
||||
Raises:
|
||||
ValueError: the series folder is outside the pool.
|
||||
"""
|
||||
pool = self.pool_dir
|
||||
if pool is None:
|
||||
raise RuntimeError("Pool není nastaven.")
|
||||
folder = Path(series.folder)
|
||||
try:
|
||||
folder.resolve().relative_to(pool.resolve())
|
||||
except ValueError:
|
||||
raise ValueError(f"Složka „{folder}“ není v poolu.")
|
||||
if folder.is_dir():
|
||||
shutil.rmtree(folder)
|
||||
|
||||
def assign_tag_to_file_objects(self, files_objs: list[File], tag):
|
||||
"""Přiřadí tag (Tag nebo 'category/name' string) ke každému souboru v seznamu."""
|
||||
for f in files_objs:
|
||||
|
||||
@@ -19,6 +19,7 @@ Example:
|
||||
"""
|
||||
import os
|
||||
import random
|
||||
import shutil
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple, Optional, Dict, Set
|
||||
@@ -341,6 +342,26 @@ class HardlinkManager:
|
||||
|
||||
return success_count, fail_count
|
||||
|
||||
def remove_mirror(self, name: str) -> bool:
|
||||
"""Remove ``output_dir/name`` entirely (file or folder tree).
|
||||
|
||||
Used to drop a whole mirrored library (or one entry of the shared
|
||||
mixed-library-series bucket) when it falls out of the miniDLNA
|
||||
generation's scope. Mirrors are pure hardlinks, so deleting the folder
|
||||
loses nothing — it's fully regenerable from the pool. ``name`` may
|
||||
contain "/" to address one entry inside a bucket folder.
|
||||
|
||||
Returns whether anything was actually removed.
|
||||
"""
|
||||
target = self.output_dir / name
|
||||
if not target.exists():
|
||||
return False
|
||||
if target.is_dir():
|
||||
shutil.rmtree(target)
|
||||
else:
|
||||
target.unlink()
|
||||
return True
|
||||
|
||||
def _fill_special_folder(
|
||||
self, chosen: List[File], subfolder: str, dry_run: bool
|
||||
) -> Tuple[int, int]:
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Readers for Curator's *pre-index* metadata layouts.
|
||||
|
||||
Before the unified pool index (`.Curator.!index`), Tagger / early Curator stored
|
||||
a movie's metadata in a **per-file sidecar** ``.{filename}.!tag`` (JSON) sitting
|
||||
next to the video. When importing a file that came from such a pool we can
|
||||
recover its **ČSFD link** (and title / year) from that sidecar — or from an old
|
||||
``.Curator.!index`` found by walking up from the file.
|
||||
|
||||
This is read-only recovery for the import dialog; nothing here writes.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Optional
|
||||
|
||||
from .pool_index import INDEX_FILENAME
|
||||
|
||||
SIDECAR_SUFFIX = ".!tag"
|
||||
_MAX_INDEX_WALK_UP = 4
|
||||
|
||||
|
||||
@dataclass
|
||||
class LegacyMeta:
|
||||
"""What an old sidecar / index record can still tell us about a movie."""
|
||||
|
||||
csfd_url: Optional[str] = None
|
||||
title: Optional[str] = None
|
||||
year: Optional[str] = None # four-digit string
|
||||
source: str = "" # "sidecar" | "index"
|
||||
|
||||
|
||||
def _record_to_meta(record: object, source: str) -> Optional[LegacyMeta]:
|
||||
"""Pull the useful fields out of a stored movie record (any schema version)."""
|
||||
if not isinstance(record, dict):
|
||||
return None
|
||||
csfd = record.get("csfd_url") or record.get("csfd_link") # new | Tagger-era key
|
||||
title = record.get("title") or None
|
||||
meta_raw = record.get("meta")
|
||||
meta = meta_raw if isinstance(meta_raw, dict) else {}
|
||||
cache_raw = record.get("csfd_cache")
|
||||
cache = cache_raw if isinstance(cache_raw, dict) else {}
|
||||
raw_year = meta.get("year") or cache.get("year")
|
||||
year = str(raw_year).strip() if raw_year is not None and str(raw_year).strip().isdigit() else None
|
||||
if not (csfd or title or year):
|
||||
return None
|
||||
return LegacyMeta(csfd_url=csfd or None, title=title, year=year, source=source)
|
||||
|
||||
|
||||
def _from_sidecar(video_path: Path) -> Optional[LegacyMeta]:
|
||||
sidecar = video_path.parent / f".{video_path.name}{SIDECAR_SUFFIX}"
|
||||
if not sidecar.is_file():
|
||||
return None
|
||||
try:
|
||||
record = json.loads(sidecar.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
return None
|
||||
return _record_to_meta(record, "sidecar")
|
||||
|
||||
|
||||
def _lookup_in_index(movies: dict, video_path: Path, index_dir: Path) -> Optional[dict]:
|
||||
"""Find the record for ``video_path`` in an index's ``movies`` map.
|
||||
|
||||
Tries the exact pool-relative key first, then falls back to a unique
|
||||
basename match (the file may have moved since the index was written).
|
||||
"""
|
||||
try:
|
||||
rel = video_path.relative_to(index_dir).as_posix()
|
||||
if rel in movies:
|
||||
return movies[rel]
|
||||
except ValueError:
|
||||
pass
|
||||
name = video_path.name
|
||||
matches = [v for k, v in movies.items() if PurePosixPath(k).name == name]
|
||||
return matches[0] if len(matches) == 1 else None
|
||||
|
||||
|
||||
def _from_legacy_index(video_path: Path) -> Optional[LegacyMeta]:
|
||||
directory = video_path.parent
|
||||
for _ in range(_MAX_INDEX_WALK_UP + 1):
|
||||
index_path = directory / INDEX_FILENAME
|
||||
if index_path.is_file():
|
||||
try:
|
||||
data = json.loads(index_path.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
return None
|
||||
movies = data.get("movies", data) if isinstance(data, dict) else None
|
||||
if not isinstance(movies, dict):
|
||||
return None
|
||||
record = _lookup_in_index(movies, video_path, directory)
|
||||
return _record_to_meta(record, "index") if record is not None else None
|
||||
if directory.parent == directory:
|
||||
break
|
||||
directory = directory.parent
|
||||
return None
|
||||
|
||||
|
||||
def read_legacy_meta(video_path: Path) -> Optional[LegacyMeta]:
|
||||
"""Recover ČSFD link / title / year from a pre-index layout, or ``None``.
|
||||
|
||||
Order: a per-file ``.{name}.!tag`` sidecar next to the video, then a
|
||||
``.Curator.!index`` found walking up (max 4 levels) from the file.
|
||||
"""
|
||||
video_path = Path(video_path)
|
||||
return _from_sidecar(video_path) or _from_legacy_index(video_path)
|
||||
@@ -27,6 +27,7 @@ class MatchProposal:
|
||||
chosen: object = None # tmdb.TmdbResult preselected for the user, or None
|
||||
candidates: list = field(default_factory=list) # list[tmdb.TmdbResult]
|
||||
error: str = ""
|
||||
note: str = "" # e.g. "přes ČSFD: The Shawshank Redemption (1994)"
|
||||
|
||||
|
||||
def normalize(text: str) -> str:
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
"""The movie object — metadata first, files second.
|
||||
|
||||
A movie is identified by ``(title, year, tmdb_id)`` and owns **its files**, not
|
||||
the other way round. That inversion is the point: the pool index used to key a
|
||||
record by a file path, so metadata was a passenger on one particular file — and
|
||||
everything about multi-version movies got awkward (which file "holds" the tags,
|
||||
why renaming a grouped movie had to be blocked, why an import could silently
|
||||
create a duplicate instead of a version).
|
||||
|
||||
Here the record is the movie; each playable file is a :class:`MediaFile` under
|
||||
it, told apart by a free-text ``label`` (``3D HSBS``, ``4K``, ``Director's
|
||||
Cut``, …). Where those files *live* is then a pure function of the record —
|
||||
:meth:`Movie.desired_layout` — so promoting to a folder, demoting back, or
|
||||
renaming the whole thing are all the same operation: recompute and move what
|
||||
doesn't match.
|
||||
|
||||
Layout (Jellyfin's own conventions — https://jellyfin.org/docs/general/server/media/movies/)::
|
||||
|
||||
one video <Library>/Title (rok) [tmdbid-N].ext
|
||||
two or more <Library>/Title (rok) [tmdbid-N]/
|
||||
Title (rok) [tmdbid-N] - Výchozí.ext
|
||||
Title (rok) [tmdbid-N].3D.HSBS.ext
|
||||
|
||||
Every file in a movie folder must start with the folder's exact name; both
|
||||
``" - label"`` and ``".3D.FMT"`` are valid separators, so a 3D file keeps the
|
||||
documented ``.3D.<FORMAT>`` spelling whether it sits flat or grouped.
|
||||
Subtitles are not media entries — they ride along with their video by filename
|
||||
stem (see :mod:`src.core.subtitles`).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .naming import THREE_D_FORMATS, canonical_pool_stem, is_valid_year
|
||||
|
||||
# Roles a media file can have. Only playable video for now — artwork stays
|
||||
# Jellyfin's job (it writes poster/fanart into the folder itself).
|
||||
ROLE_VIDEO = "video"
|
||||
|
||||
# A grouped movie needs a label on every file (all of Jellyfin's documented
|
||||
# examples label every version); this is what an unlabelled one gets.
|
||||
DEFAULT_VERSION_LABEL = "Výchozí"
|
||||
|
||||
# "HSBS" / "hsbs" / "3D HSBS" / "3d.hsbs" / "3D_HSBS" → "HSBS"
|
||||
_LABEL_3D_RE = re.compile(
|
||||
rf"^(?:3d[\s._-]*)?({'|'.join(THREE_D_FORMATS)})$", re.IGNORECASE)
|
||||
|
||||
|
||||
def normalize_3d_label(label: object) -> Optional[str]:
|
||||
"""The 3D format a version label denotes (upper-case), or ``None``.
|
||||
|
||||
Lets a plain label double as the 3D marker — ``"3D HSBS"`` typed in the
|
||||
version dialog and ``"hsbs"`` picked from a dropdown mean the same thing.
|
||||
"""
|
||||
match = _LABEL_3D_RE.match(str(label or "").strip())
|
||||
return match.group(1).upper() if match else None
|
||||
|
||||
|
||||
def media_filename(stem: str, label: str, ext: str, *, grouped: bool) -> str:
|
||||
"""Filename for one media file of a movie whose canonical stem is ``stem``.
|
||||
|
||||
- no label → ``stem.ext`` (a lone, plain video)
|
||||
- 3D label → ``stem.3D.<FORMAT>.ext`` (Jellyfin's documented 3D form)
|
||||
- anything else → ``stem - label.ext``
|
||||
|
||||
``grouped`` only decides whether an empty label is allowed: inside a movie
|
||||
folder every file gets one, so it falls back to :data:`DEFAULT_VERSION_LABEL`.
|
||||
"""
|
||||
label = (label or "").strip()
|
||||
if grouped and not label:
|
||||
label = DEFAULT_VERSION_LABEL
|
||||
if not label:
|
||||
return f"{stem}{ext}"
|
||||
fmt = normalize_3d_label(label)
|
||||
if fmt:
|
||||
return f"{stem}.3D.{fmt}{ext}"
|
||||
return f"{stem} - {label}{ext}"
|
||||
|
||||
|
||||
def identity_key(tmdb_id: object = None, title: str = "", year: object = None) -> str:
|
||||
"""The pool-index key for a movie — ``tmdb-<id>``, else a title/year slug.
|
||||
|
||||
Keying by identity (rather than by file path) is what lets a second file of
|
||||
the same film find its movie instead of becoming a separate entry.
|
||||
"""
|
||||
id_s = str(tmdb_id).strip() if tmdb_id is not None else ""
|
||||
if id_s.isdigit():
|
||||
return f"tmdb-{int(id_s)}"
|
||||
raw = f"{(title or '').strip()} {str(year or '').strip()}".strip().lower()
|
||||
# Fold diacritics so "Pelíšky" and "Pelisky" can't become two movies.
|
||||
raw = "".join(c for c in unicodedata.normalize("NFD", raw)
|
||||
if not unicodedata.combining(c))
|
||||
slug = re.sub(r"[^0-9a-z]+", "-", raw).strip("-")
|
||||
return f"name-{slug or 'bez-nazvu'}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class MediaFile:
|
||||
"""One file belonging to a movie (today always a playable video)."""
|
||||
|
||||
path: Path
|
||||
label: str = ""
|
||||
role: str = ROLE_VIDEO
|
||||
|
||||
@property
|
||||
def format_3d(self) -> Optional[str]:
|
||||
"""The 3D format this version's label denotes, if any."""
|
||||
return normalize_3d_label(self.label)
|
||||
|
||||
def to_record(self, pool_dir: Path) -> dict:
|
||||
"""Serialise for the pool index (path stored pool-relative, POSIX)."""
|
||||
try:
|
||||
rel = Path(self.path).relative_to(pool_dir).as_posix()
|
||||
except ValueError:
|
||||
rel = Path(self.path).as_posix()
|
||||
return {"role": self.role, "label": self.label, "file": rel}
|
||||
|
||||
@classmethod
|
||||
def from_record(cls, data: dict, pool_dir: Path) -> "MediaFile":
|
||||
return cls(
|
||||
path=pool_dir / str(data.get("file", "")),
|
||||
label=str(data.get("label", "") or ""),
|
||||
role=str(data.get("role") or ROLE_VIDEO),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Movie:
|
||||
"""A film in the pool: its identity, its metadata, and its media files."""
|
||||
|
||||
key: str
|
||||
title: str = ""
|
||||
year: Optional[str] = None
|
||||
tmdb_id: Optional[int] = None
|
||||
library: str = ""
|
||||
media: list[MediaFile] = field(default_factory=list)
|
||||
|
||||
# --- metadata that used to live on the "primary" File ---
|
||||
tags: list = field(default_factory=list)
|
||||
auto_tag_paths: set[str] = field(default_factory=set)
|
||||
csfd_url: Optional[str] = None
|
||||
date: Optional[str] = None
|
||||
added: Optional[str] = None
|
||||
attributes: dict[str, str] = field(default_factory=dict)
|
||||
meta: Optional[dict] = None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Identity / shape
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def videos(self) -> list[MediaFile]:
|
||||
"""Playable versions, in a stable order (labelled ones by label)."""
|
||||
return sorted(
|
||||
(m for m in self.media if m.role == ROLE_VIDEO),
|
||||
key=lambda m: m.label.casefold())
|
||||
|
||||
@property
|
||||
def is_grouped(self) -> bool:
|
||||
"""True when the movie needs its own folder (more than one version)."""
|
||||
return len(self.videos) > 1
|
||||
|
||||
@property
|
||||
def stem(self) -> str:
|
||||
"""Canonical ``Title (rok) [tmdbid-N]`` — the folder / filename base.
|
||||
|
||||
Falls back to the bare title when no four-digit year is known (a movie
|
||||
can be imported without one).
|
||||
"""
|
||||
title = (self.title or "").replace("/", "-").replace("\\", "-").strip()
|
||||
if not title:
|
||||
title = self.key
|
||||
if is_valid_year(self.year or ""):
|
||||
return canonical_pool_stem(title, self.year, self.tmdb_id)
|
||||
return title
|
||||
|
||||
@property
|
||||
def primary(self) -> Optional[MediaFile]:
|
||||
"""The version that represents the movie (first by :attr:`videos`)."""
|
||||
videos = self.videos
|
||||
return videos[0] if videos else None
|
||||
|
||||
def desired_layout(self, pool_dir: Path) -> list[tuple[MediaFile, Path]]:
|
||||
"""``(media, where it belongs)`` for every file, from the record alone.
|
||||
|
||||
The whole flat-vs-folder rule lives here: one video sits directly in
|
||||
the library, two or more move into ``<Library>/<stem>/``. Callers move
|
||||
whatever doesn't already match — which makes promotion, demotion and
|
||||
renaming the same operation.
|
||||
"""
|
||||
base = Path(pool_dir) / self.library
|
||||
grouped = self.is_grouped
|
||||
folder = base / self.stem if grouped else base
|
||||
return [
|
||||
(m, folder / media_filename(
|
||||
self.stem, m.label, Path(m.path).suffix, grouped=grouped))
|
||||
for m in self.videos
|
||||
]
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Index serialisation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def to_record(self, pool_dir: Path) -> dict:
|
||||
data = {
|
||||
"title": self.title,
|
||||
"year": self.year,
|
||||
"tmdb_id": self.tmdb_id,
|
||||
"library": self.library,
|
||||
"media": [m.to_record(pool_dir) for m in self.media],
|
||||
"tags": [getattr(t, "full_path", t) for t in self.tags],
|
||||
"auto_tags": sorted(self.auto_tag_paths),
|
||||
"csfd_url": self.csfd_url,
|
||||
"date": self.date,
|
||||
"added": self.added,
|
||||
"attributes": dict(self.attributes),
|
||||
}
|
||||
if self.meta is not None:
|
||||
data["meta"] = dict(self.meta)
|
||||
return data
|
||||
|
||||
@classmethod
|
||||
def from_record(
|
||||
cls, key: str, data: dict, pool_dir: Path, tagmanager=None
|
||||
) -> "Movie":
|
||||
movie = cls(
|
||||
key=key,
|
||||
title=str(data.get("title") or ""),
|
||||
year=(str(data["year"]) if data.get("year") else None),
|
||||
tmdb_id=(int(data["tmdb_id"]) if str(data.get("tmdb_id") or "").isdigit()
|
||||
else None),
|
||||
library=str(data.get("library") or ""),
|
||||
media=[MediaFile.from_record(m, pool_dir)
|
||||
for m in (data.get("media") or [])],
|
||||
auto_tag_paths=set(data.get("auto_tags") or []),
|
||||
csfd_url=data.get("csfd_url") or None,
|
||||
date=data.get("date") or None,
|
||||
added=data.get("added") or None,
|
||||
attributes=dict(data.get("attributes") or {}),
|
||||
meta=(dict(data["meta"]) if data.get("meta") else None),
|
||||
)
|
||||
if tagmanager is not None:
|
||||
for full_path in data.get("tags") or []:
|
||||
if "/" in str(full_path):
|
||||
category, name = str(full_path).split("/", 1)
|
||||
movie.tags.append(tagmanager.add_tag(category, name))
|
||||
return movie
|
||||
+46
-10
@@ -4,16 +4,32 @@ Every movie in the pool is named ``Title (YYYY) [tmdbid-N].ext`` — a title, a
|
||||
four-digit year in parentheses, and (when known) the TMDb id in brackets, which
|
||||
is exactly the form Jellyfin matches on. The ``[tmdbid-N]`` part is optional: a
|
||||
plain ``Title (YYYY)`` is still valid (Jellyfin then matches by name). The year
|
||||
lets two same-named films coexist and makes the filename self-describing. These
|
||||
helpers build, validate and parse the stem (the filename without its extension).
|
||||
lets two same-named films coexist and makes the filename self-describing.
|
||||
|
||||
A 3D movie additionally carries a ``.3D.<FORMAT>`` tag right before the
|
||||
extension (Jellyfin's own convention — https://jellyfin.org/docs/general/server/media/movies/):
|
||||
``Title (YYYY) [tmdbid-N].3D.HSBS.ext``. ``FORMAT`` is one of
|
||||
:data:`THREE_D_FORMATS`; Jellyfin reads it case-insensitively but Curator always
|
||||
writes it upper-case for a single canonical form.
|
||||
|
||||
These helpers build, validate and parse the stem (the filename without its
|
||||
extension).
|
||||
"""
|
||||
import re
|
||||
|
||||
# A trailing four-digit year in parentheses, then an optional ``[tmdbid-N]``. The
|
||||
# title is non-greedy so a title that itself contains parentheses keeps them:
|
||||
# "Kurz (special) (2019)" → title "Kurz (special)", year 2019.
|
||||
# Jellyfin's recognised 3D format tags (case-insensitive on read; Curator
|
||||
# always writes upper-case). hsbs/fsbs = half/full side-by-side, htab/ftab =
|
||||
# half/full top-and-bottom, mvc = Multiview Video Coding.
|
||||
THREE_D_FORMATS = ("HSBS", "FSBS", "HTAB", "FTAB", "MVC")
|
||||
|
||||
# A trailing four-digit year in parentheses, then an optional ``[tmdbid-N]`` and
|
||||
# an optional ``.3D.<FORMAT>`` tag. The title is non-greedy so a title that
|
||||
# itself contains parentheses keeps them: "Kurz (special) (2019)" → title
|
||||
# "Kurz (special)", year 2019.
|
||||
_POOL_STEM_RE = re.compile(
|
||||
r"^(?P<title>.+?) \((?P<year>\d{4})\)(?: \[tmdbid-(?P<id>\d+)\])?$")
|
||||
r"^(?P<title>.+?) \((?P<year>\d{4})\)(?: \[tmdbid-(?P<id>\d+)\])?"
|
||||
r"(?:\.3D\.(?P<format3d>hsbs|fsbs|htab|ftab|mvc))?$",
|
||||
re.IGNORECASE)
|
||||
|
||||
|
||||
def is_valid_year(year: object) -> bool:
|
||||
@@ -21,12 +37,16 @@ def is_valid_year(year: object) -> bool:
|
||||
return bool(re.fullmatch(r"\d{4}", str(year).strip()))
|
||||
|
||||
|
||||
def canonical_pool_stem(title: str, year: object, tmdb_id: object = None) -> str:
|
||||
"""Build the canonical pool stem ``Title (YYYY)`` (+ `` [tmdbid-N]`` if given).
|
||||
def canonical_pool_stem(
|
||||
title: str, year: object, tmdb_id: object = None, format_3d: object = None,
|
||||
) -> str:
|
||||
"""Build the canonical pool stem ``Title (YYYY)`` (+ `` [tmdbid-N]`` /
|
||||
``.3D.<FORMAT>`` if given).
|
||||
|
||||
Raises:
|
||||
ValueError: the title is empty, the year is not four digits, or a given
|
||||
``tmdb_id`` is not a positive integer.
|
||||
ValueError: the title is empty, the year is not four digits, a given
|
||||
``tmdb_id`` is not a positive integer, or ``format_3d`` isn't one
|
||||
of :data:`THREE_D_FORMATS`.
|
||||
"""
|
||||
title = (title or "").strip()
|
||||
year_s = str(year).strip()
|
||||
@@ -40,6 +60,13 @@ def canonical_pool_stem(title: str, year: object, tmdb_id: object = None) -> str
|
||||
if not id_s.isdigit():
|
||||
raise ValueError(f"TMDb ID musí být číslo, ne „{tmdb_id}“.")
|
||||
stem += f" [tmdbid-{id_s}]"
|
||||
if format_3d is not None and str(format_3d).strip() != "":
|
||||
fmt = str(format_3d).strip().upper()
|
||||
if fmt not in THREE_D_FORMATS:
|
||||
raise ValueError(
|
||||
f"3D formát musí být jedno z {', '.join(THREE_D_FORMATS)}, "
|
||||
f"ne „{format_3d}“.")
|
||||
stem += f".3D.{fmt}"
|
||||
return stem
|
||||
|
||||
|
||||
@@ -63,6 +90,15 @@ def parse_tmdb_id(stem: str) -> int | None:
|
||||
return int(match.group("id"))
|
||||
|
||||
|
||||
def parse_3d_format(stem: str) -> str | None:
|
||||
"""Read the ``.3D.<FORMAT>`` tag from a canonical stem (upper-cased), or
|
||||
``None`` if absent."""
|
||||
match = _POOL_STEM_RE.match((stem or "").strip())
|
||||
if not match or not match.group("format3d"):
|
||||
return None
|
||||
return match.group("format3d").upper()
|
||||
|
||||
|
||||
def is_canonical_pool_stem(stem: str) -> bool:
|
||||
"""True when ``stem`` matches the ``Title (YYYY)`` convention."""
|
||||
return _POOL_STEM_RE.match((stem or "").strip()) is not None
|
||||
|
||||
@@ -26,6 +26,7 @@ class NfoData:
|
||||
countries: list[str] = field(default_factory=list)
|
||||
rating: Optional[int] = None # 0–100 (NFO 0–10 scaled up), or None
|
||||
tmdb_id: Optional[int] = None
|
||||
csfd_url: Optional[str] = None # ČSFD film URL, if the NFO carries a csfd id
|
||||
|
||||
def field_values(self, name: str) -> list[str]:
|
||||
"""Values for a tag-schema source field (genres/countries/year/rating)."""
|
||||
@@ -78,6 +79,29 @@ def _parse_tmdb_id(root: ET.Element) -> Optional[int]:
|
||||
return int(raw.strip()) if raw and raw.strip().isdigit() else None
|
||||
|
||||
|
||||
def _parse_csfd_url(root: ET.Element) -> Optional[str]:
|
||||
"""A ČSFD film URL from the NFO, if any.
|
||||
|
||||
Czech Kodi scrapers record ČSFD as ``<uniqueid type="csfd">`` (a bare id or a
|
||||
full URL) or a legacy ``<csfdid>`` / ``<csfd>`` element. A bare id is expanded
|
||||
to ``https://www.csfd.cz/film/<id>/``.
|
||||
"""
|
||||
candidates: list[str] = []
|
||||
for uid in root.findall("uniqueid"):
|
||||
if uid.get("type") == "csfd" and uid.text and uid.text.strip():
|
||||
candidates.append(uid.text.strip())
|
||||
for tag in ("csfdid", "csfd"):
|
||||
raw = root.findtext(tag)
|
||||
if raw and raw.strip():
|
||||
candidates.append(raw.strip())
|
||||
for value in candidates:
|
||||
if value.isdigit():
|
||||
return f"https://www.csfd.cz/film/{value}/"
|
||||
if "csfd.cz" in value:
|
||||
return value if value.startswith("http") else f"https://{value.lstrip('/')}"
|
||||
return None
|
||||
|
||||
|
||||
def read_nfo(video_path: Path) -> Optional[NfoData]:
|
||||
"""Read the movie's ``.nfo`` into :class:`NfoData`, or None if absent/broken."""
|
||||
path = find_nfo(video_path)
|
||||
@@ -98,4 +122,5 @@ def read_nfo(video_path: Path) -> Optional[NfoData]:
|
||||
countries=_texts(root, "country"),
|
||||
rating=_parse_rating(root),
|
||||
tmdb_id=_parse_tmdb_id(root),
|
||||
csfd_url=_parse_csfd_url(root),
|
||||
)
|
||||
|
||||
+130
-15
@@ -10,17 +10,26 @@ The index also carries **library-level settings** (``tag_schema``,
|
||||
particular pool travels with the pool. The global config keeps only app data
|
||||
(window state, MRU) and the pointers to where the pool/output live.
|
||||
|
||||
On-disk shape::
|
||||
On-disk shape (**v2**)::
|
||||
|
||||
{"settings": {...}, "movies": {"Filmy/Matrix.mkv": {...}, ...}}
|
||||
{"version": 2,
|
||||
"settings": {...},
|
||||
"movies": {"tmdb-603": {"title": …, "media": [{"role", "label", "file"}, …]}}}
|
||||
|
||||
Records are keyed by the file's path relative to the pool root (POSIX form),
|
||||
e.g. ``"Filmy/Matrix.mkv"`` — stable and portable across machines.
|
||||
Records are keyed by the movie's **identity** (``tmdb-<id>``, or a title/year
|
||||
slug when there's no id) and own their files — see :mod:`src.core.movie`. The
|
||||
original **v1** shape keyed a record by one file's pool-relative path, which
|
||||
made metadata a passenger on that file; :meth:`PoolIndex.load` migrates such an
|
||||
index in place (backing the old file up first). Migration only rewrites
|
||||
metadata — no file on disk is moved, because a one-video movie is already in
|
||||
its v2 position.
|
||||
"""
|
||||
import json
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
INDEX_FILENAME = ".Curator.!index"
|
||||
INDEX_VERSION = 2
|
||||
|
||||
|
||||
class PoolIndex:
|
||||
@@ -42,7 +51,11 @@ class PoolIndex:
|
||||
return path.as_posix()
|
||||
|
||||
def load(self) -> None:
|
||||
"""Load the index from disk (missing/corrupt index = empty)."""
|
||||
"""Load the index from disk (missing/corrupt index = empty).
|
||||
|
||||
A v1 (path-keyed) index is migrated to v2 in place — the old file is
|
||||
copied to ``.Curator.!index.v1.bak`` first.
|
||||
"""
|
||||
if not self.index_path.exists():
|
||||
self.records = {}
|
||||
self.settings = {}
|
||||
@@ -55,13 +68,82 @@ class PoolIndex:
|
||||
except (json.JSONDecodeError, OSError):
|
||||
self.records = {}
|
||||
self.settings = {}
|
||||
return
|
||||
|
||||
if int(data.get("version", 1)) < INDEX_VERSION:
|
||||
self._migrate_v1_to_v2()
|
||||
|
||||
def _migrate_v1_to_v2(self) -> None:
|
||||
"""Re-key path-keyed records into identity-keyed movie records.
|
||||
|
||||
Each v1 record described one file; several of them can describe one
|
||||
movie (the interim multi-version layout put them in a shared folder),
|
||||
so they are merged by identity. Metadata only — nothing is moved on
|
||||
disk, since a v1 movie's files already sit where v2 wants them.
|
||||
"""
|
||||
from .movie import ROLE_VIDEO, identity_key
|
||||
from .naming import is_canonical_pool_stem, parse_pool_stem, parse_tmdb_id
|
||||
|
||||
migrated: dict[str, dict] = {}
|
||||
for rel_path, record in self.records.items():
|
||||
if not isinstance(record, dict) or "media" in record:
|
||||
continue # already v2-shaped, leave it alone
|
||||
path = Path(rel_path)
|
||||
parts = path.parts
|
||||
library = parts[0] if len(parts) > 1 else ""
|
||||
# A file inside its own "Title (rok) [tmdbid]" folder is one
|
||||
# version of that movie; its own stem carries the " - label".
|
||||
parent = path.parent.name
|
||||
grouped = len(parts) > 2 and is_canonical_pool_stem(parent)
|
||||
identity_stem = parent if grouped else path.stem
|
||||
|
||||
parsed = parse_pool_stem(identity_stem)
|
||||
title = record.get("title") or (parsed[0] if parsed else path.stem)
|
||||
year = str(parsed[1]) if parsed else None
|
||||
tmdb_id = parse_tmdb_id(identity_stem)
|
||||
label = ""
|
||||
if grouped and path.stem.startswith(f"{parent} - "):
|
||||
label = path.stem[len(parent) + 3:]
|
||||
elif grouped:
|
||||
label = path.stem[len(parent):].lstrip(". -") or ""
|
||||
|
||||
key = identity_key(tmdb_id, title, year)
|
||||
entry = {"role": ROLE_VIDEO, "label": label, "file": rel_path}
|
||||
if key in migrated:
|
||||
migrated[key]["media"].append(entry)
|
||||
continue
|
||||
migrated[key] = {
|
||||
"title": title,
|
||||
"year": year,
|
||||
"tmdb_id": tmdb_id,
|
||||
"library": library,
|
||||
"media": [entry],
|
||||
"tags": record.get("tags", []),
|
||||
"auto_tags": record.get("auto_tags", record.get("csfd_tags", [])),
|
||||
"csfd_url": record.get("csfd_url") or record.get("csfd_link"),
|
||||
"date": record.get("date"),
|
||||
"added": record.get("added"),
|
||||
"attributes": record.get("attributes", {}),
|
||||
}
|
||||
if record.get("meta"):
|
||||
migrated[key]["meta"] = record["meta"]
|
||||
|
||||
try:
|
||||
shutil.copy2(self.index_path,
|
||||
self.index_path.with_name(self.index_path.name + ".v1.bak"))
|
||||
except OSError:
|
||||
pass
|
||||
self.records = migrated
|
||||
self.save()
|
||||
|
||||
def save(self) -> None:
|
||||
"""Persist the index to disk."""
|
||||
self.pool_dir.mkdir(parents=True, exist_ok=True)
|
||||
with open(self.index_path, "w", encoding="utf-8") as f:
|
||||
json.dump(
|
||||
{"settings": self.settings, "movies": self.records},
|
||||
{"version": INDEX_VERSION,
|
||||
"settings": self.settings,
|
||||
"movies": self.records},
|
||||
f, indent=2, ensure_ascii=False,
|
||||
)
|
||||
|
||||
@@ -74,16 +156,49 @@ class PoolIndex:
|
||||
self.settings[key] = value
|
||||
self.save()
|
||||
|
||||
def get(self, file_path: Path) -> dict | None:
|
||||
"""Return the record for a file, or None if it is not indexed."""
|
||||
return self.records.get(self._key(file_path))
|
||||
# ------------------------------------------------------------------
|
||||
# Movie records (v2 — keyed by identity, own their media files)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def set(self, file_path: Path, data: dict) -> None:
|
||||
"""Upsert a record and persist the index."""
|
||||
self.records[self._key(file_path)] = data
|
||||
def get_movie(self, key: str) -> dict | None:
|
||||
"""Return the movie record under ``key``, or None."""
|
||||
return self.records.get(key)
|
||||
|
||||
def set_movie(self, key: str, data: dict) -> None:
|
||||
"""Upsert a movie record and persist the index."""
|
||||
self.records[key] = data
|
||||
self.save()
|
||||
|
||||
def delete(self, file_path: Path) -> None:
|
||||
"""Remove a record (if present) and persist the index."""
|
||||
if self.records.pop(self._key(file_path), None) is not None:
|
||||
def delete_movie(self, key: str) -> None:
|
||||
"""Remove a movie record (if present) and persist the index."""
|
||||
if self.records.pop(key, None) is not None:
|
||||
self.save()
|
||||
|
||||
def rekey_movie(self, old_key: str, new_key: str) -> None:
|
||||
"""Move a record to a new identity key (e.g. its TMDb id was set/fixed)."""
|
||||
if old_key == new_key or old_key not in self.records:
|
||||
return
|
||||
self.records[new_key] = self.records.pop(old_key)
|
||||
self.save()
|
||||
|
||||
def all_movies(self) -> dict[str, dict]:
|
||||
"""Every movie record, keyed by identity."""
|
||||
return dict(self.records)
|
||||
|
||||
def record_for(self, file_path: Path) -> dict | None:
|
||||
"""The movie record that owns ``file_path``, or None if untracked."""
|
||||
found = self.find_movie_by_media(file_path)
|
||||
return found[1] if found is not None else None
|
||||
|
||||
def find_movie_by_media(self, file_path: Path) -> tuple[str, dict] | None:
|
||||
"""``(key, record)`` of the movie owning ``file_path``, or None.
|
||||
|
||||
The reverse lookup a file needs now that records are keyed by identity
|
||||
rather than by path.
|
||||
"""
|
||||
rel = self._key(file_path)
|
||||
for key, record in self.records.items():
|
||||
for media in record.get("media", []):
|
||||
if media.get("file") == rel:
|
||||
return key, record
|
||||
return None
|
||||
|
||||
+39
-18
@@ -71,31 +71,52 @@ def parse_episode(name: str) -> Optional[tuple[int, int]]:
|
||||
return int(match.group(3)), int(match.group(4))
|
||||
|
||||
|
||||
def collect_episode_files(folder: Path) -> list[Path]:
|
||||
"""Every video file under ``folder`` (recursively), sidecars excluded, sorted.
|
||||
|
||||
Lets the episode-import dialog take a whole ``Series/Season NN/`` tree (or any
|
||||
nested layout) in one pick instead of file-by-file. Season/episode numbers
|
||||
are still read per file from its name by :func:`parse_episode`.
|
||||
"""
|
||||
folder = Path(folder)
|
||||
if not folder.is_dir():
|
||||
return []
|
||||
return sorted(
|
||||
f for f in folder.rglob("*")
|
||||
if f.is_file() and f.suffix.lower() not in _SIDECARS
|
||||
)
|
||||
|
||||
|
||||
def season_folder_name(season: int) -> str:
|
||||
"""Folder name for a season — ``Specials`` for 0, else ``Season NN``."""
|
||||
return "Specials" if season == 0 else f"Season {season:02d}"
|
||||
|
||||
|
||||
def episode_body(stem: str) -> str:
|
||||
"""The descriptive part of an episode filename, minus a leading ``SxxEyy`` code.
|
||||
|
||||
Keeps the original (release) filename so it survives into the canonical name,
|
||||
while stripping an already-applied ``SxxEyy - `` prefix so repeated renames
|
||||
don't double it. Returns ``""`` when the stem was only the code.
|
||||
"""
|
||||
s = _EPISODE_RE.sub("", stem, count=1) # drop the SxxEyy/NxNN token
|
||||
s = re.sub(r"([\s._-])[\s._-]+", r"\1", s) # collapse separator runs
|
||||
return s.strip(" ._-")
|
||||
def is_season_folder(name: str) -> bool:
|
||||
"""True for a season folder name — ``Specials`` or ``Season <n>`` (any padding)."""
|
||||
return name == "Specials" or re.fullmatch(r"Season \d{1,3}", name) is not None
|
||||
|
||||
|
||||
def episode_filename(season: int, episode: int, orig_stem: str, ext: str) -> str:
|
||||
"""Canonical episode filename ``SxxEyy - {original name}.ext``.
|
||||
# Characters not allowed in a filename on the common filesystems.
|
||||
_FS_UNSAFE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||||
|
||||
Keeps the original filename after the code; when the original was only the
|
||||
code (nothing descriptive left) it collapses to ``SxxEyy.ext``.
|
||||
|
||||
def sanitize_title(title: str) -> str:
|
||||
"""A series/episode title made safe to sit in a filename (spaces collapsed)."""
|
||||
cleaned = _FS_UNSAFE.sub("", title or "")
|
||||
return re.sub(r"\s+", " ", cleaned).strip(" .")
|
||||
|
||||
|
||||
def episode_filename(season: int, episode: int, series_title: str, ext: str) -> str:
|
||||
"""Canonical episode filename ``SxxEyy - {series title}.ext``.
|
||||
|
||||
The release string is deliberately **not** kept — identity is the ``SxxEyy``
|
||||
code, and Jellyfin fetches the episode title/metadata from TMDb (matched by
|
||||
the code + the series folder's ``[tmdbid-N]``). An empty title collapses to
|
||||
``SxxEyy.ext``.
|
||||
"""
|
||||
code = f"S{season:02d}E{episode:02d}"
|
||||
body = episode_body(orig_stem)
|
||||
body = sanitize_title(series_title)
|
||||
return f"{code} - {body}{ext}" if body else f"{code}{ext}"
|
||||
|
||||
|
||||
@@ -115,13 +136,13 @@ def series_folder_stem(title: str, year: object = None, tmdb_id: object = None)
|
||||
return f"{title} [tmdbid-{id_s}]" if id_s.isdigit() else title
|
||||
|
||||
|
||||
def is_canonical_episode(path: Path) -> bool:
|
||||
"""True when ``path`` already follows ``SxxEyy - name`` in the right season."""
|
||||
def is_canonical_episode(path: Path, series_title: str) -> bool:
|
||||
"""True when ``path`` is already ``SxxEyy - {series title}.ext`` in its season."""
|
||||
parsed = parse_episode(path.name)
|
||||
if parsed is None:
|
||||
return False
|
||||
season, episode = parsed
|
||||
expected = episode_filename(season, episode, path.stem, path.suffix)
|
||||
expected = episode_filename(season, episode, series_title, path.suffix)
|
||||
if path.name != expected:
|
||||
return False
|
||||
return path.parent.name == season_folder_name(season)
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""External subtitle sidecars — attach, discover, keep in sync on rename.
|
||||
|
||||
Curator does not mux/embed subtitles. Jellyfin plays an **external** subtitle
|
||||
file natively as long as it shares the video's filename stem, with an optional
|
||||
language code and flag suffix::
|
||||
|
||||
Title (rok) [tmdbid-N].mkv
|
||||
Title (rok) [tmdbid-N].cs.srt # language
|
||||
Title (rok) [tmdbid-N].cs.forced.srt # + a flag (forced / sdh / default)
|
||||
|
||||
Subtitle presence is a **filesystem fact**, like NFO — nothing is cached in the
|
||||
pool index. What Curator does own: copying a picked file into place with the
|
||||
right name (:func:`attach_subtitle`), and moving a subtitle along whenever its
|
||||
video gets renamed so it never silently detaches (:func:`rename_subtitles_with`).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
SUBTITLE_EXTENSIONS = {".srt", ".ass", ".ssa", ".sub", ".vtt"}
|
||||
_FLAGS = {"forced", "sdh", "default"}
|
||||
_LANG_RE = re.compile(r"[a-z]{2,3}")
|
||||
|
||||
|
||||
@dataclass
|
||||
class SubtitleFile:
|
||||
"""One subtitle sidecar found next to a video."""
|
||||
|
||||
path: Path
|
||||
language: Optional[str] = None # ISO 639-1/2 code, e.g. "cs"
|
||||
flag: Optional[str] = None # "forced" | "sdh" | "default"
|
||||
|
||||
@property
|
||||
def label(self) -> str:
|
||||
"""Human-readable ``jazyk (flag)`` for display, e.g. ``cs (forced)``."""
|
||||
text = self.language or "?"
|
||||
return f"{text} ({self.flag})" if self.flag else text
|
||||
|
||||
|
||||
def subtitle_filename(
|
||||
video_stem: str, ext: str, language: Optional[str] = None,
|
||||
flag: Optional[str] = None,
|
||||
) -> str:
|
||||
"""Canonical subtitle filename: ``{video_stem}[.language][.flag]{ext}``."""
|
||||
parts = [video_stem]
|
||||
if language:
|
||||
parts.append(language.strip().lower())
|
||||
if flag:
|
||||
parts.append(flag.strip().lower())
|
||||
return ".".join(parts) + ext.lower()
|
||||
|
||||
|
||||
def _parse_subtitle_name(
|
||||
video_stem: str, filename: str
|
||||
) -> Optional[tuple[Optional[str], Optional[str]]]:
|
||||
"""``(language, flag)`` if ``filename`` is a subtitle sidecar of ``video_stem``."""
|
||||
candidate = Path(filename)
|
||||
if candidate.suffix.lower() not in SUBTITLE_EXTENSIONS:
|
||||
return None
|
||||
stem = candidate.stem
|
||||
if stem == video_stem:
|
||||
return None, None
|
||||
if not stem.startswith(f"{video_stem}."):
|
||||
return None
|
||||
tokens = [t for t in stem[len(video_stem) + 1:].split(".") if t]
|
||||
if not tokens:
|
||||
return None, None
|
||||
language: Optional[str] = None
|
||||
flag: Optional[str] = None
|
||||
for token in tokens:
|
||||
low = token.lower()
|
||||
if low in _FLAGS:
|
||||
flag = low
|
||||
elif language is None and _LANG_RE.fullmatch(low):
|
||||
language = low
|
||||
return language, flag
|
||||
|
||||
|
||||
def find_subtitles(video_path: Path) -> list[SubtitleFile]:
|
||||
"""Every external subtitle sidecar sitting next to ``video_path``, sorted."""
|
||||
video_path = Path(video_path)
|
||||
folder = video_path.parent
|
||||
if not folder.is_dir():
|
||||
return []
|
||||
found: list[SubtitleFile] = []
|
||||
for f in sorted(folder.iterdir()):
|
||||
if not f.is_file():
|
||||
continue
|
||||
parsed = _parse_subtitle_name(video_path.stem, f.name)
|
||||
if parsed is None:
|
||||
continue
|
||||
language, flag = parsed
|
||||
found.append(SubtitleFile(f, language, flag))
|
||||
return found
|
||||
|
||||
|
||||
def attach_subtitle(
|
||||
video_path: Path, source: Path, *, language: Optional[str] = None,
|
||||
flag: Optional[str] = None, move: bool = False,
|
||||
) -> Path:
|
||||
"""Copy (or move) ``source`` next to ``video_path``, named to match it.
|
||||
|
||||
An existing subtitle already at that exact name (e.g. replacing the Czech
|
||||
track) is overwritten; nothing else next to the video is touched.
|
||||
|
||||
Raises:
|
||||
ValueError: ``source``'s extension isn't a recognised subtitle format.
|
||||
"""
|
||||
video_path = Path(video_path)
|
||||
source = Path(source)
|
||||
ext = source.suffix.lower()
|
||||
if ext not in SUBTITLE_EXTENSIONS:
|
||||
raise ValueError(f"Nepodporovaný formát titulků: {source.suffix}")
|
||||
target = video_path.parent / subtitle_filename(video_path.stem, ext, language, flag)
|
||||
if target.exists():
|
||||
target.unlink()
|
||||
if move:
|
||||
shutil.move(str(source), str(target))
|
||||
else:
|
||||
shutil.copy2(source, target)
|
||||
return target
|
||||
|
||||
|
||||
def rename_subtitles_with(old_video_path: Path, new_video_path: Path) -> list[Path]:
|
||||
"""Move every subtitle sidecar of ``old_video_path`` to match ``new_video_path``.
|
||||
|
||||
Call this wherever a video gets renamed/moved so an attached subtitle never
|
||||
silently detaches (its stem would stop matching and Jellyfin would drop it).
|
||||
No-op when the video's path didn't actually change. Returns the new paths.
|
||||
"""
|
||||
old_video_path = Path(old_video_path)
|
||||
new_video_path = Path(new_video_path)
|
||||
if old_video_path == new_video_path:
|
||||
return []
|
||||
moved: list[Path] = []
|
||||
for sub in find_subtitles(old_video_path):
|
||||
target = new_video_path.parent / subtitle_filename(
|
||||
new_video_path.stem, sub.path.suffix, sub.language, sub.flag)
|
||||
if sub.path == target:
|
||||
continue
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
if target.exists():
|
||||
target.unlink()
|
||||
sub.path.rename(target)
|
||||
moved.append(target)
|
||||
return moved
|
||||
Reference in New Issue
Block a user