Honza/Curator
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rework Curator into a Jellyfin-first library manager

Browse Source
This commit is contained in:
Honza
2026-08-31 18:34:03 +02:00
parent 1c78ca9013
commit 6a2baf243f
25 changed files with 2420 additions and 1502 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/config.py
+3 -2
View File
@@ -26,8 +26,9 @@ GLOBAL_CONFIG_FILE = Path(__file__).parent.parent.parent / ".Curator.!gtag"
# they are derived from ČSFD, and how they map to the Filmotéka folder tree.
# Each entry:
# category — tag category name (e.g. "Žánr")
# csfd_field — CSFDMovie attribute feeding it (genres / year / countries /
# rating / directors / actors), or None for a user-only category
# csfd_field — NFO source field feeding it (genres / year / countries /
# rating), or None for a user-only category. (Key name kept for
# index compatibility; the data now comes from the NFO.)
# transform — named value transform: None (str) or "decade_band" (rating)
# filmoteka_root — folder under the output: "" = at the root (e.g. genres),
# "Dle X" = grouping folder, None = filterable but no folders
src/core/csfd.py
-682
View File
@@ -1,682 +0,0 @@
"""
CSFD.cz scraper module for fetching movie information.
This module provides functionality to fetch movie data from CSFD.cz (Czech-Slovak Film Database).
"""
from __future__ import annotations
import re
import json
import time
import hashlib
from dataclasses import dataclass, field
from typing import Optional, TYPE_CHECKING
from urllib.parse import urljoin
from loguru import logger
try:
import requests
from bs4 import BeautifulSoup
HAS_DEPENDENCIES = True
except ImportError:
HAS_DEPENDENCIES = False
requests = None # type: ignore
BeautifulSoup = None # type: ignore
if TYPE_CHECKING:
from bs4 import BeautifulSoup
CSFD_BASE_URL = "https://www.csfd.cz"
CSFD_SEARCH_URL = "https://www.csfd.cz/hledat/"
# User agent to avoid being blocked
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 is the proof-of-work anti-bot wall ČSFD now puts in front of every page.
# A plain request gets a 200 with a JS challenge page (title "Ujišťujeme se, že
# nejste robot!") instead of the movie, so JSON-LD/genres/year all parse empty.
# We detect that page, solve the PoW the way the bundled worker JS does, and
# replay the request through the same session to obtain the auth cookie.
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
# Keep only the top-billed cast from a movie's actor list.
MAX_ACTORS = 10
@dataclass
class CSFDMovie:
"""Represents movie data from CSFD.cz"""
title: str
url: str
year: Optional[int] = None
genres: list[str] = field(default_factory=list)
directors: list[str] = field(default_factory=list)
actors: list[str] = field(default_factory=list)
rating: Optional[int] = None # Percentage 0-100
rating_count: Optional[int] = None
duration: Optional[int] = None # Minutes
# A movie can be a co-production, so the origin is a list of countries
# (ČSFD writes them slash-separated, e.g. "Japonsko / USA").
countries: list[str] = field(default_factory=list)
poster_url: Optional[str] = None
plot: Optional[str] = None
csfd_id: Optional[int] = None
def to_dict(self) -> dict:
"""Serialize to a plain dict for storage in .!tag cache."""
return {
"title": self.title,
"url": self.url,
"year": self.year,
"genres": self.genres,
"directors": self.directors,
"actors": self.actors,
"rating": self.rating,
"rating_count": self.rating_count,
"duration": self.duration,
"countries": self.countries,
"poster_url": self.poster_url,
"plot": self.plot,
"csfd_id": self.csfd_id,
}
@classmethod
def from_dict(cls, data: dict) -> "CSFDMovie":
"""Deserialize from a plain dict (e.g. loaded from .!tag cache)."""
countries = data.get("countries")
if countries is None:
# Legacy cache stored a single "country" string (possibly slash-joined)
countries = _split_countries(data.get("country"))
return cls(
title=data.get("title", ""),
url=data.get("url", ""),
year=data.get("year"),
genres=data.get("genres", []),
directors=data.get("directors", []),
actors=data.get("actors", []),
rating=data.get("rating"),
rating_count=data.get("rating_count"),
duration=data.get("duration"),
countries=countries,
poster_url=data.get("poster_url"),
plot=data.get("plot"),
csfd_id=data.get("csfd_id"),
)
def __str__(self) -> str:
parts = [self.title]
if self.year:
parts[0] += f" ({self.year})"
if self.rating is not None:
parts.append(f"Hodnocení: {self.rating}%")
if self.genres:
parts.append(f"Žánr: {', '.join(self.genres)}")
if self.countries:
parts.append(f"Země původu: {', '.join(self.countries)}")
if self.directors:
parts.append(f"Režie: {', '.join(self.directors)}")
return " | ".join(parts)
def rating_band(rating: int) -> str:
"""Bucket a 0–100 ČSFD rating into a ten-point band label (e.g. "80–89 %").
The top bucket spans 90–100 % so a perfect 100 still lands in a band.
"""
low = min((rating // 10) * 10, 90)
high = 100 if low == 90 else low + 9
return f"{low}–{high} %"
def _decade_band(value) -> str:
"""``decade_band`` transform: pull the leading integer and bucket it."""
match = re.search(r"\d+", str(value))
return rating_band(int(match.group())) if match else str(value)
# Named value transforms usable from the tag schema (config-driven, extensible).
# These are applied when building Filmotéka FOLDER names — the tag itself keeps
# the exact value (e.g. rating 90 → tag "Hodnocení/90", folder "…/90–100 %").
TAG_TRANSFORMS = {
None: str,
"identity": str,
"decade_band": _decade_band,
}
def apply_transform(value: str, transform: Optional[str]) -> str:
"""Apply a named tag-schema transform to a value, returning a string."""
fn = TAG_TRANSFORMS.get(transform, str)
try:
return str(fn(value))
except (ValueError, TypeError, AttributeError):
return str(value)
def csfd_field_values(movie: "CSFDMovie", field: str) -> list[str]:
"""Exact tag values for a CSFDMovie attribute (scalar or list), as strings.
No transform is applied here — tags carry the precise value; any grouping
transform happens later, when the Filmotéka folder names are built.
Unknown field / missing value yields ``[]``.
"""
raw = getattr(movie, field, None)
if raw is None or raw == "":
return []
items = list(raw) if isinstance(raw, (list, tuple)) else [raw]
values = [str(item).strip() for item in items if item is not None and item != ""]
return [v for v in values if v]
def _split_countries(text: Optional[str]) -> list[str]:
"""Split a ČSFD origin country string into individual countries.
ČSFD writes co-productions slash-separated, e.g. ``"Japonsko / USA"`` →
``["Japonsko", "USA"]``. ``None``/empty yields an empty list.
"""
if not text:
return []
return [part.strip() for part in text.split("/") if part.strip()]
def _check_dependencies():
"""Check if required dependencies are installed."""
if not HAS_DEPENDENCIES:
raise ImportError(
"CSFD module requires 'requests' and 'beautifulsoup4' packages. "
"Install them with: pip install requests beautifulsoup4"
)
def _extract_csfd_id(url: str) -> Optional[int]:
"""Extract CSFD movie ID from URL."""
match = re.search(r"/film/(\d+)", url)
return int(match.group(1)) if match else None
def _parse_duration(duration_str: str) -> Optional[int]:
"""Parse ISO 8601 duration (PT97M) to minutes."""
match = re.search(r"PT(\d+)M", duration_str)
return int(match.group(1)) if match else None
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 ValueError(
f"Anubis PoW unsolved within {ANUBIS_MAX_NONCE} attempts (difficulty {difficulty})"
)
def _solve_anubis_challenge(session, html: str, url: str):
"""Solve the Anubis challenge in ``html`` and return the real page response.
Posts the proof-of-work back to the pass-challenge endpoint through
``session`` (which stores the resulting auth cookie) and follows the
redirect to the originally requested page.
"""
payload = _extract_json_blob(html, "anubis_challenge")
if not payload:
raise ValueError("Č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 ValueError("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 ValueError("Č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
def fetch_movie(url: str, session=None) -> CSFDMovie:
"""
Fetch movie information from CSFD.cz URL.
Args:
url: Full CSFD.cz movie URL (e.g., https://www.csfd.cz/film/9423-pane-vy-jste-vdova/)
session: Optional ``requests.Session`` to reuse (keeps the Anubis auth
cookie across calls so only the first fetch pays the PoW cost).
Returns:
CSFDMovie object with extracted data
Raises:
ImportError: If required dependencies are not installed
requests.RequestException: If network request fails
ValueError: If URL is invalid or page cannot be parsed
"""
_check_dependencies()
own_session = session is None
if own_session:
session = requests.Session()
try:
response = _get_page(session, url)
finally:
if own_session:
session.close()
soup = BeautifulSoup(response.text, "html.parser")
# Try to extract JSON-LD structured data first (most reliable)
movie_data = _extract_json_ld(soup)
# Extract additional data from HTML
movie_data["url"] = url
movie_data["csfd_id"] = _extract_csfd_id(url)
# Get rating from HTML if not in JSON-LD
if movie_data.get("rating") is None:
movie_data["rating"] = _extract_rating(soup)
# Get poster URL
if movie_data.get("poster_url") is None:
movie_data["poster_url"] = _extract_poster(soup)
# Get plot summary
if movie_data.get("plot") is None:
movie_data["plot"] = _extract_plot(soup)
# Get countries and year from origin info
origin_info = _extract_origin_info(soup)
if origin_info:
if not movie_data.get("countries"):
movie_data["countries"] = origin_info.get("countries", [])
if movie_data.get("year") is None:
movie_data["year"] = origin_info.get("year")
if movie_data.get("duration") is None:
movie_data["duration"] = origin_info.get("duration")
# Get genres from HTML if not in JSON-LD
if not movie_data.get("genres"):
movie_data["genres"] = _extract_genres(soup)
# Keep only the leading cast (ČSFD lists them in billing order)
movie_data["actors"] = movie_data.get("actors", [])[:MAX_ACTORS]
return CSFDMovie(**movie_data)
def _extract_json_ld(soup: BeautifulSoup) -> dict:
"""Extract movie data from JSON-LD structured data."""
data = {
"title": "",
"year": None,
"genres": [],
"directors": [],
"actors": [],
"rating": None,
"rating_count": None,
"duration": None,
"countries": [],
"poster_url": None,
"plot": None,
}
# Find JSON-LD script
script_tags = soup.find_all("script", type="application/ld+json")
for script in script_tags:
try:
json_data = json.loads(script.string)
# Handle both single object and array
if isinstance(json_data, list):
for item in json_data:
if item.get("@type") == "Movie":
json_data = item
break
else:
continue
if json_data.get("@type") != "Movie":
continue
# Title
data["title"] = json_data.get("name", "")
# Genres
genre = json_data.get("genre", [])
if isinstance(genre, str):
data["genres"] = [genre]
else:
data["genres"] = list(genre)
# Directors
directors = json_data.get("director", [])
if isinstance(directors, dict):
directors = [directors]
data["directors"] = [d.get("name", "") for d in directors if d.get("name")]
# Actors
actors = json_data.get("actor", [])
if isinstance(actors, dict):
actors = [actors]
data["actors"] = [a.get("name", "") for a in actors if a.get("name")]
# Rating
agg_rating = json_data.get("aggregateRating", {})
if agg_rating:
rating_value = agg_rating.get("ratingValue")
if rating_value is not None:
data["rating"] = round(float(rating_value))
data["rating_count"] = agg_rating.get("ratingCount")
# Duration
duration_str = json_data.get("duration", "")
if duration_str:
data["duration"] = _parse_duration(duration_str)
# Year (CSFD exposes it via dateCreated / datePublished)
for date_key in ("dateCreated", "datePublished"):
date_val = json_data.get(date_key)
if date_val:
year_match = re.search(r"(19\d{2}|20\d{2})", str(date_val))
if year_match:
data["year"] = int(year_match.group(1))
break
# Poster
image = json_data.get("image")
if image:
if isinstance(image, str):
data["poster_url"] = image
elif isinstance(image, dict):
data["poster_url"] = image.get("url")
# Description
data["plot"] = json_data.get("description")
break # Found movie data
except (json.JSONDecodeError, KeyError, TypeError):
continue
return data
def _extract_rating(soup: BeautifulSoup) -> Optional[int]:
"""Extract rating percentage from HTML."""
# Look for rating box
rating_elem = soup.select_one(".film-rating-average")
if rating_elem:
text = rating_elem.get_text(strip=True)
match = re.search(r"(\d+)%", text)
if match:
return int(match.group(1))
return None
def _extract_poster(soup: BeautifulSoup) -> Optional[str]:
"""Extract poster image URL from HTML."""
# Look for poster image
poster = soup.select_one(".film-poster img")
if poster:
src = poster.get("src") or poster.get("data-src")
if src:
if src.startswith("//"):
return "https:" + src
return src
return None
def _extract_plot(soup: BeautifulSoup) -> Optional[str]:
"""Extract plot summary from HTML."""
# Look for plot/description section
plot_elem = soup.select_one(".plot-full p")
if plot_elem:
return plot_elem.get_text(strip=True)
# Alternative: shorter plot
plot_elem = soup.select_one(".plot-preview p")
if plot_elem:
return plot_elem.get_text(strip=True)
return None
def _extract_genres(soup: BeautifulSoup) -> list[str]:
"""Extract genres from HTML."""
genres = []
genre_links = soup.select(".genres a")
for link in genre_links:
genre = link.get_text(strip=True)
if genre:
genres.append(genre)
return genres
def _extract_origin_info(soup: BeautifulSoup) -> dict:
"""Extract countries, year, duration from the origin info line.
CSFD separates the values with inline bullet ``<span>`` elements (no commas),
so ``get_text(strip=True)`` would glue them together (e.g. "USA1999136 min").
We tokenize on those inline boundaries (and on commas, for the older format)
before extracting each field. The country segment of a co-production is
slash-separated (e.g. "USA / Velká Británie") and is split into a list.
"""
info: dict = {}
origin_elem = soup.select_one(".origin")
if not origin_elem:
return info
# Split on inline element boundaries, then also on commas (older format).
raw = origin_elem.get_text(separator="|", strip=True)
tokens = [t.strip() for part in raw.split("|") for t in part.split(",")]
tokens = [t for t in tokens if t]
for token in tokens:
if "year" not in info and re.fullmatch(r"(19\d{2}|20\d{2})", token):
info["year"] = int(token)
continue
if "duration" not in info:
duration_match = re.search(r"(\d+)\s*min", token)
if duration_match:
info["duration"] = int(duration_match.group(1))
continue
# Countries: first alphabetic token that doesn't start with a digit;
# may list several slash-separated countries for a co-production.
if "countries" not in info and not token[0].isdigit() and re.search(r"[^\W\d_]", token):
info["countries"] = _split_countries(token)
return info
def search_movies(query: str, limit: int = 10, session=None) -> list[CSFDMovie]:
"""
Search for movies on CSFD.cz.
Args:
query: Search query string
limit: Maximum number of results to return
session: Optional ``requests.Session`` to reuse (keeps the Anubis auth
cookie across calls so only the first lookup pays the PoW cost).
Returns:
List of CSFDMovie objects with basic info (title, url, year)
"""
_check_dependencies()
search_url = f"{CSFD_SEARCH_URL}?q={requests.utils.quote(query)}"
own_session = session is None
if own_session:
session = requests.Session()
try:
response = _get_page(session, search_url)
finally:
if own_session:
session.close()
soup = BeautifulSoup(response.text, "html.parser")
results = []
# Find movie results
movie_items = soup.select(".film-title-name, .search-result-item a[href*='/film/']")
for item in movie_items[:limit]:
href = item.get("href", "")
if "/film/" not in href:
continue
title = item.get_text(strip=True)
url = urljoin(CSFD_BASE_URL, href)
# Try to get year from sibling/parent
year = None
parent = item.find_parent(class_="article-content")
if parent:
year_elem = parent.select_one(".info")
if year_elem:
year_match = re.search(r"\((\d{4})\)", year_elem.get_text())
if year_match:
year = int(year_match.group(1))
results.append(CSFDMovie(
title=title,
url=url,
year=year,
csfd_id=_extract_csfd_id(url)
))
return results
def fetch_movie_by_id(csfd_id: int) -> CSFDMovie:
"""
Fetch movie by CSFD ID.
Args:
csfd_id: CSFD movie ID number
Returns:
CSFDMovie object with full data
"""
url = f"{CSFD_BASE_URL}/film/{csfd_id}/"
return fetch_movie(url)
# 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. Example::
"Matrix.1999.1080p.BluRay.x264-GROUP.mkv" -> "Matrix 1999"
"""
from pathlib import Path
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
# also stop at a release group glued with a dash (e.g. "x264-GROUP")
title_words.append(token)
# If nothing survived (title started with a marker), fall back to the stem.
title = " ".join(title_words).strip() or re.sub(r"[.\s_]+", " ", stem).strip()
return f"{title} {year}".strip() if year else title
def find_csfd_url(query: str, session=None) -> Optional[str]:
"""Return the first ČSFD film URL matching a query, or None.
Thin wrapper over :func:`search_movies` that takes the top result. Pass a
shared ``session`` to reuse the Anubis auth cookie across several lookups.
"""
if not query.strip():
return None
results = search_movies(query, limit=1, session=session)
return results[0].url if results else None
src/core/file.py
+43 -79
View File
@@ -1,10 +1,6 @@
from pathlib import Path
from .tag import Tag
# Bump this when the csfd_cache schema changes to force re-fetch on next open.
# v2: country (str) → countries (list[str]) for co-productions.
CSFD_CACHE_VERSION = 2
class File:
def __init__(self, file_path: Path, tagmanager=None, index=None) -> None:
@@ -22,17 +18,17 @@ class File:
self.date: str | None = None
# movie-library fields
self.title: str | None = None
self.csfd_link: str | None = None
# ISO timestamp of when the movie was added to the pool (set on import);
# None for older/unknown → falls back to the file's ctime.
self.added: str | None = None
# Cached CSFD data — avoids re-fetching on every open
self.csfd_cache: dict | None = None
# full_paths of tags that came from ČSFD (vs. user-added); only these are
# regenerated when the ČSFD link changes — user tags are preserved.
self.csfd_tag_paths: set[str] = set()
# Cached metadata used for folder grouping (year, rating) — filled from
# the movie's NFO (which Jellyfin writes).
self.meta: dict | None = None
# full_paths of tags derived automatically from the NFO (vs. user-added);
# only these are regenerated on refresh — user tags are preserved.
self.auto_tag_paths: set[str] = set()
# free-form per-movie attributes (e.g. "collection_sort"); usable in
# Filmotéka filename templates via name_context.
# miniDLNA filename templates via name_context.
self.attributes: dict[str, str] = {}
self.get_metadata()
@@ -50,9 +46,8 @@ class File:
self.tags = []
self.date = None
self.title = None
self.csfd_link = None
self.csfd_cache = None
self.csfd_tag_paths = set()
self.meta = None
self.auto_tag_paths = set()
self.attributes = {}
self.added = None
@@ -62,17 +57,16 @@ class File:
"ignored": self.ignored,
# ukládáme full_path tagů
"tags": [tag.full_path if isinstance(tag, Tag) else tag for tag in self.tags],
# which of the tags came from ČSFD (provenance)
"csfd_tags": sorted(self.csfd_tag_paths),
# which of the tags were derived automatically (from the NFO)
"auto_tags": sorted(self.auto_tag_paths),
# date může být None
"date": self.date,
"title": self.title,
"csfd_link": self.csfd_link,
"attributes": self.attributes,
"added": self.added,
}
if self.csfd_cache is not None:
data["csfd_cache"] = {"version": CSFD_CACHE_VERSION, **self.csfd_cache}
if self.meta is not None:
data["meta"] = dict(self.meta)
return data
def _apply_record(self, data: dict) -> None:
@@ -81,16 +75,15 @@ class File:
self.tags = []
self.date = data.get("date", None)
self.title = data.get("title", None)
self.csfd_link = data.get("csfd_link", None)
# Legacy records have no provenance → treated as all-user (empty set)
self.csfd_tag_paths = set(data.get("csfd_tags", []))
# provenance of auto (NFO-derived) tags; legacy key "csfd_tags" honored
self.auto_tag_paths = set(data.get("auto_tags", data.get("csfd_tags", [])))
self.attributes = dict(data.get("attributes", {}))
self.added = data.get("added", None)
raw_cache = data.get("csfd_cache")
if raw_cache and raw_cache.get("version") == CSFD_CACHE_VERSION:
self.csfd_cache = {k: v for k, v in raw_cache.items() if k != "version"}
else:
self.csfd_cache = None
raw_meta = data.get("meta")
if raw_meta is None: # legacy: pull year/rating out of the old csfd cache
old = data.get("csfd_cache") or {}
raw_meta = {"year": old.get("year"), "rating": old.get("rating")} if old else None
self.meta = dict(raw_meta) if raw_meta else None
if not self.tagmanager:
return
@@ -126,13 +119,13 @@ class File:
return 0.0
def name_context(self) -> dict:
"""Fields for a Filmotéka filename template (see config tag schema).
"""Fields for a miniDLNA filename template (see config tag schema).
Provides title / year / rating / ext / stem / filename from the movie's
metadata (year & rating come from the ČSFD cache, year falling back to a
``Rok`` tag). Missing values render as an empty string.
metadata (year & rating come from the NFO cache, year falling back to a
``Rok`` tag, then the canonical filename). Missing values render empty.
"""
cache = self.csfd_cache or {}
cache = self.meta or {}
year = cache.get("year")
if year is None:
for t in self.tags:
@@ -195,78 +188,49 @@ class File:
self.date = date_str
self.save_metadata()
def set_csfd_link(self, url: str | None) -> None:
"""Nastaví CSFD odkaz nebo None. Invaliduje cache při změně odkazu."""
new_url = url if url else None
if new_url != self.csfd_link:
self.csfd_cache = None # odkaz se změnil — stará cache je neplatná
self.csfd_link = new_url
self.save_metadata()
def apply_nfo(self, nfo, schema: list[dict] | None = None) -> dict:
"""Regenerate the auto (NFO-derived) tags from the movie's NFO.
def get_cached_movie(self):
"""Vrátí CSFDMovie z cache, nebo None. Nevyžaduje síť."""
if self.csfd_cache is None:
return None
try:
from .csfd import CSFDMovie
return CSFDMovie.from_dict(self.csfd_cache)
except Exception:
return None
def apply_csfd_tags(self, schema: list[dict] | None = None) -> dict:
"""Načte data z ČSFD a přegeneruje **jen ČSFD tagy** podle tag schématu.
``schema`` je seznam definic kategorií (viz ``DEFAULT_TAG_SCHEMA``); pro
každý záznam s ``csfd_field`` vytvoří tagy z příslušného pole filmu
(s volitelným ``transform``). Tagy přidané uživatelem (mimo provenance
ČSFD) zůstávají nedotčené — proto změna ČSFD odkazu nepřepíše vlastní tagy.
Režie/herci se stále cachují, tag se z nich tvoří jen je-li v schématu.
Reads genres/year/countries/rating from ``nfo`` (``nfo.NfoData``) per the
tag schema's source fields and (re)creates those tags; user-added tags are
kept. Also caches year/rating (for folder grouping) and fills the title
when missing. ``nfo`` of ``None`` (no NFO yet) is a no-op.
Returns:
dict s klíči 'success', 'movie'/'error', 'tags_added'
dict with 'success' and 'tags_added'.
"""
if not self.csfd_link:
return {"success": False, "error": "CSFD odkaz není nastaven", "tags_added": []}
if nfo is None:
return {"success": False, "error": "NFO nenalezeno", "tags_added": []}
if schema is None:
from .config import DEFAULT_TAG_SCHEMA
schema = DEFAULT_TAG_SCHEMA
try:
from .csfd import fetch_movie
movie = fetch_movie(self.csfd_link)
self.csfd_cache = movie.to_dict()
except ImportError as e:
return {"success": False, "error": f"Chybí závislosti: {e}", "tags_added": []}
except Exception as e:
return {"success": False, "error": f"Chyba při načítání CSFD: {e}", "tags_added": []}
self.meta = {"year": nfo.year, "rating": nfo.rating}
from .csfd import csfd_field_values
# Drop previous ČSFD-sourced tags; keep user-added ones.
self.tags = [t for t in self.tags if t.full_path not in self.csfd_tag_paths]
self.csfd_tag_paths = set()
# Drop previous auto (NFO) tags; keep user-added ones.
self.tags = [t for t in self.tags if t.full_path not in self.auto_tag_paths]
self.auto_tag_paths = set()
tags_added: list[str] = []
for entry in schema:
field = entry.get("csfd_field")
field = entry.get("csfd_field") # schema key = source field name
if not field:
continue # user-only category
category = entry["category"]
for value in csfd_field_values(movie, field):
for value in nfo.field_values(field):
tag_obj = (self.tagmanager.add_tag(category, value)
if self.tagmanager else Tag(category, value))
if tag_obj not in self.tags:
self.tags.append(tag_obj)
tags_added.append(tag_obj.full_path)
self.csfd_tag_paths.add(tag_obj.full_path)
self.auto_tag_paths.add(tag_obj.full_path)
# Use the CSFD title if we don't have one yet
if movie.title and not self.title:
self.title = movie.title
if nfo.title and not self.title:
self.title = nfo.title
self.save_metadata()
return {"success": True, "movie": movie, "tags_added": tags_added}
return {"success": True, "tags_added": tags_added}
def add_tag(self, tag):
# tag může být Tag nebo string
src/core/file_manager.py
+308 -40
View File
@@ -7,7 +7,7 @@ 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
from .naming import canonical_pool_stem, is_valid_year, parse_pool_stem, parse_tmdb_id
from typing import Callable, Iterable
from src.core.config import (
load_global_config, save_global_config, DEFAULT_TAG_SCHEMA
@@ -21,6 +21,9 @@ POOL_SERIES = "Seriály"
_LIBRARY_SETTINGS_DEFAULTS = {
"tag_schema": DEFAULT_TAG_SCHEMA,
"copyasis_folders": [POOL_SERIES],
# Top-level pool folders that hold movies — each is a Jellyfin library
# (physical partition). "Filmy" is the default; users add Kids/Dokumenty/…
"movie_libraries": [POOL_MOVIES],
}
# Curator metadata files that must never be treated as content. The legacy
@@ -28,6 +31,13 @@ _LIBRARY_SETTINGS_DEFAULTS = {
# stray ones are still skipped so they never leak into the Filmotéka.
METADATA_SUFFIXES = (".!tag", ".!ftag", ".!gtag", ".!index")
# Sidecars Jellyfin writes next to a movie (NFO + artwork/subtitles). These are
# not movies and must be skipped when scanning a library.
SIDECAR_SUFFIXES = (
".nfo", ".jpg", ".jpeg", ".png", ".webp", ".bmp", ".gif", ".tbn",
".srt", ".ass", ".ssa", ".sub", ".idx", ".vtt",
)
class FileManager:
def __init__(self, tagmanager: TagManager):
@@ -117,6 +127,27 @@ class FileManager:
cleaned = [n.strip() for n in names if n.strip()]
self._set_setting("copyasis_folders", cleaned)
@property
def movie_libraries(self) -> list[str]:
"""Top-level pool folders holding movies (each a Jellyfin library)."""
libs = self._get_setting("movie_libraries")
return libs if libs else [POOL_MOVIES]
def set_movie_libraries(self, names: list[str]) -> None:
"""Set the movie-library list (never empty — falls back to Filmy)."""
cleaned = [n.strip() for n in names if n.strip()]
self._set_setting("movie_libraries", cleaned or [POOL_MOVIES])
def library_of(self, file_obj: File) -> str | None:
"""Which library (top-level pool folder) a pooled movie lives in."""
pool = self.pool_dir
if not pool:
return None
try:
return file_obj.file_path.relative_to(pool).parts[0]
except (ValueError, IndexError):
return None
@property
def tag_schema(self) -> list[dict]:
"""Tag categories + ČSFD/Filmotéka rules (see config.DEFAULT_TAG_SCHEMA)."""
@@ -164,12 +195,13 @@ class FileManager:
return Path(value) if value else None
def set_pool_dir(self, pool_dir: Path) -> None:
"""Set the managed pool root and create its top-level folders."""
"""Set the managed pool root and create every library's top-level folder."""
pool_dir = Path(pool_dir)
(pool_dir / POOL_MOVIES).mkdir(parents=True, exist_ok=True)
(pool_dir / POOL_SERIES).mkdir(parents=True, exist_ok=True)
self.global_config["pool_dir"] = str(pool_dir)
save_global_config(self.global_config)
self._open_index(pool_dir) # settings now readable through the index
for name in self.movie_libraries + self.copyasis_folders:
(pool_dir / name).mkdir(parents=True, exist_ok=True)
def set_filmoteka_dir(self, filmoteka_dir: Path) -> None:
"""Set the Filmotéka output folder (generated hardlink tree)."""
@@ -177,22 +209,33 @@ class FileManager:
save_global_config(self.global_config)
def load_pool_movies(self) -> None:
"""Reload the movie list from pool/Filmy using the unified pool index."""
"""Reload the movie list from every movie library, via the pool index.
Scans each configured movie library (``movie_libraries``) recursively —
a movie may sit directly in the library or inside its own per-movie
folder. Curator's own metadata and the sidecars Jellyfin writes (NFO,
artwork, subtitles) are skipped.
"""
self.filelist = []
movies = self.movies_dir
pool = self.pool_dir
if not (movies and movies.is_dir() and pool):
if not pool:
return
self._open_index(pool)
for each in list_files(movies):
if each.name.endswith(METADATA_SUFFIXES):
for library in self.movie_libraries:
lib_dir = pool / library
if not lib_dir.is_dir():
continue
file_obj = File(each, self.tagmanager, index=self.index)
self.filelist.append(file_obj)
for each in list_files(lib_dir):
if each.name.endswith(METADATA_SUFFIXES):
continue
if each.suffix.lower() in SIDECAR_SUFFIXES:
continue
file_obj = File(each, self.tagmanager, index=self.index)
self.filelist.append(file_obj)
def check_data_consistency(self) -> dict:
"""Compare the pool index against the actual files in pool/Filmy.
"""Compare the pool index against the actual files in the movie libraries.
Detects drift such as a movie file deleted directly (metadata left behind)
or a file dropped into the pool without importing (no metadata).
@@ -204,8 +247,7 @@ class FileManager:
index_count / disk_count — totals for the summary
"""
pool = self.pool_dir
movies = self.movies_dir
if not (pool and movies and movies.is_dir()):
if not (pool and pool.is_dir()):
return {"ok": False, "error": "Pool není nastaven.",
"missing": [], "untracked": []}
@@ -213,10 +255,16 @@ class FileManager:
index_keys = set(index.records.keys())
disk_keys = set()
for f in list_files(movies):
if f.name.endswith(METADATA_SUFFIXES):
for library in self.movie_libraries:
lib_dir = pool / library
if not lib_dir.is_dir():
continue
disk_keys.add(f.relative_to(pool).as_posix())
for f in list_files(lib_dir):
if f.name.endswith(METADATA_SUFFIXES):
continue
if f.suffix.lower() in SIDECAR_SUFFIXES:
continue
disk_keys.add(f.relative_to(pool).as_posix())
missing = sorted(k for k in index_keys if not (pool / k).is_file())
untracked = sorted(disk_keys - index_keys)
@@ -228,6 +276,51 @@ class FileManager:
"disk_count": len(disk_keys),
}
def verify_against_jellyfin(self, jf_movies: list) -> dict:
"""Compare the pool against what Jellyfin currently sees.
Matches each pooled movie to a Jellyfin item by **TMDb id** (from the
filename), falling back to (title, year). Reports what is out of sync.
``jf_movies`` is a list of ``jellyfin.JellyfinMovie``. Returns a dict:
matched — how many pooled movies Jellyfin has
pool_only — pooled filenames Jellyfin does not see (rescan/rematch)
jellyfin_only — Jellyfin titles with no matching pooled movie
pool_total / jellyfin_total — totals for the summary
"""
by_tmdb = {m.tmdb_id: m for m in jf_movies if m.tmdb_id}
by_name = {(m.title.strip().lower(), m.year): m for m in jf_movies if m.title}
matched_ids: set = set()
pool_only: list[str] = []
for f in self.filelist:
tmdb_id = parse_tmdb_id(f.file_path.stem)
ctx = f.name_context()
title = str(f.title or ctx.get("title") or "").strip().lower()
year_str = str(ctx.get("year") or "")
year = int(year_str) if year_str.isdigit() else None
match = None
if tmdb_id is not None and tmdb_id in by_tmdb:
match = by_tmdb[tmdb_id]
elif (title, year) in by_name:
match = by_name[(title, year)]
if match is not None:
matched_ids.add(match.jellyfin_id)
else:
pool_only.append(f.filename)
jellyfin_only = sorted(
m.title for m in jf_movies if m.jellyfin_id not in matched_ids)
return {
"matched": len(self.filelist) - len(pool_only),
"pool_only": sorted(pool_only),
"jellyfin_only": jellyfin_only,
"pool_total": len(self.filelist),
"jellyfin_total": len(jf_movies),
}
def scan_video_integrity(
self,
files: list[File],
@@ -279,7 +372,7 @@ class FileManager:
"""
files = self.filelist
total_size = 0
with_csfd = 0
with_nfo = 0
untagged = 0
ratings: list[int] = []
categories: dict[str, Counter] = {}
@@ -289,21 +382,21 @@ class FileManager:
total_size += f.file_path.stat().st_size
except OSError:
pass
if f.csfd_link:
with_csfd += 1
if f.meta is not None:
with_nfo += 1
if not f.tags:
untagged += 1
for tag in f.tags:
categories.setdefault(tag.category, Counter())[tag.name] += 1
rating = (f.csfd_cache or {}).get("rating")
rating = (f.meta or {}).get("rating")
if isinstance(rating, (int, float)):
ratings.append(rating)
return {
"count": len(files),
"total_size": total_size,
"with_csfd": with_csfd,
"without_csfd": len(files) - with_csfd,
"with_nfo": with_nfo,
"without_nfo": len(files) - with_nfo,
"untagged": untagged,
"avg_rating": round(sum(ratings) / len(ratings), 1) if ratings else None,
"categories": {
@@ -325,16 +418,19 @@ class FileManager:
self.filelist.remove(file_obj)
def import_movie(
self, source: Path, title: str, csfd_link: str | None = None,
self, source: Path, title: str,
move: bool = False, on_conflict: str = "suffix",
year: int | str | None = None,
year: int | str | None = None, tmdb_id: int | str | None = None,
library: str | None = None,
) -> File | None:
"""Bring a video file into pool/Filmy and index its metadata.
"""Bring a video file into a movie library and index its metadata.
The pooled filename follows the naming convention ``Title (YYYY).ext``
when a ``year`` is given (validated to four digits); without a year it
Lands in ``pool/<library>/`` (``library`` defaults to the first movie
library, "Filmy"). The pooled filename follows the naming convention
``Title (YYYY) [tmdbid-N].ext`` when a ``year`` is given (and a
``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).ext``.
library built through the app is uniformly ``Title (YYYY) [tmdbid-N].ext``.
By default the original is **copied** (non-destructive). With ``move=True``
the source file is moved into the pool instead, leaving nothing behind.
@@ -350,10 +446,11 @@ class FileManager:
Raises:
ValueError: a ``year`` was given but is not a four-digit number.
"""
movies = self.movies_dir
pool = self.pool_dir
if movies is None or pool is None:
if pool is None:
raise RuntimeError("Pool není nastaven.")
lib = (library or "").strip() or self.movie_libraries[0]
movies = pool / lib
movies.mkdir(parents=True, exist_ok=True)
if self.index is None:
@@ -362,7 +459,8 @@ 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) if has_year else safe_title
stem = (canonical_pool_stem(safe_title, year, tmdb_id) if has_year
else safe_title)
target = movies / f"{stem}{source.suffix}"
existing = self.pooled_with_stem(stem)
@@ -389,7 +487,6 @@ class FileManager:
file_obj = File(target, self.tagmanager, index=self.index)
file_obj.title = safe_title
file_obj.csfd_link = csfd_link or None
file_obj.added = datetime.now().isoformat(timespec="seconds")
file_obj.save_metadata()
@@ -433,12 +530,13 @@ class FileManager:
return file_obj
def _canonical_name_for(self, file_obj: File) -> str | None:
"""The ``Title (YYYY)`` stem this movie should have, or ``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``: ČSFD cache → ``Rok`` tag → the current filename). A
``(YYYY)`` that leaked into the title is stripped so it isn't doubled.
Returns ``None`` when no four-digit year is known.
(``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
``None`` when no four-digit year is known.
"""
ctx = file_obj.name_context()
title = (file_obj.title or ctx.get("title") or file_obj.file_path.stem).strip()
@@ -449,16 +547,19 @@ class FileManager:
title = parsed[0]
year = year or str(parsed[1])
# Keep the TMDb id already in the pool filename, if any.
tmdb_id = parse_tmdb_id(file_obj.file_path.stem)
title = title.replace("/", "-").replace("\\", "-").strip()
if not (title and is_valid_year(year)):
return None
return canonical_pool_stem(title, year)
return canonical_pool_stem(title, year, tmdb_id)
def rename_to_canonical(self, file_obj: File) -> File | None:
"""Rename a pooled movie to ``Title (YYYY).ext`` from its metadata.
The year comes from the movie's ČSFD metadata (falling back to a ``Rok``
tag or the existing filename). ``file_obj.title`` keeps the clean title
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.
@@ -537,6 +638,173 @@ class FileManager:
"conflicts": conflicts, "unchanged": unchanged,
"cancelled": cancelled}
def move_to_library(self, file_obj: File, new_library: str) -> File:
"""Move a pooled movie into another library (top-level pool folder).
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.
FileExistsError: a file of that name already sits in the target.
"""
pool = self.pool_dir
if pool is None:
raise RuntimeError("Pool není nastaven.")
new_library = new_library.strip()
if new_library not in self.movie_libraries:
raise ValueError(f"Neznámá knihovna „{new_library}“.")
old_path = file_obj.file_path
dest_dir = pool / new_library
dest_dir.mkdir(parents=True, exist_ok=True)
new_path = dest_dir / old_path.name
if new_path == old_path:
return file_obj # already there
if new_path.exists():
raise FileExistsError(f"V knihovně „{new_library}“ už „{old_path.name}“ je.")
shutil.move(str(old_path), str(new_path))
file_obj.relocate(new_path) # drops the old index key, rebinds to the new
file_obj.save_metadata()
if self.on_files_changed:
self.on_files_changed(self.filelist)
return file_obj
def refresh_tags_from_nfo(self, files: list[File] | None = None) -> dict:
"""(Re)derive each movie's auto tags from its NFO (written by Jellyfin).
Reads the ``.nfo`` next to each movie and regenerates the genre / year /
country / rating tags per the tag schema (`File.apply_nfo`); user-added
tags are preserved. Movies without an NFO yet are left untouched.
Returns ``{updated, without_nfo}``.
"""
from .nfo import read_nfo
targets = files if files is not None else self.filelist
schema = self.tag_schema
updated, without_nfo = 0, 0
for f in targets:
nfo = read_nfo(f.file_path)
if nfo is None:
without_nfo += 1
continue
f.apply_nfo(nfo, schema)
updated += 1
if updated and self.on_files_changed:
self.on_files_changed(self.filelist)
return {"updated": updated, "without_nfo": without_nfo}
# ------------------------------------------------------------------
# Series (Jellyfin TV layout: Series/Season NN/Series SxxEyy.ext)
# ------------------------------------------------------------------
@property
def series_libraries(self) -> list[str]:
"""Pool folders that hold series (the copy-as-is mirrors, e.g. Seriály)."""
return self.copyasis_folders
def load_series(self) -> list:
"""Scan every series library into a list of ``series.Series``."""
from .series import scan_series
pool = self.pool_dir
if not pool:
return []
result: list = []
for library in self.series_libraries:
result.extend(scan_series(pool / library, library))
return result
def import_episodes(
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",
) -> 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.
Returns ``{imported, skipped_no_code, errors}``.
"""
from .series import (
parse_episode, series_folder_stem, season_folder_name, episode_filename)
pool = self.pool_dir
if pool is None:
raise RuntimeError("Pool není nastaven.")
lib = ((library or "").strip()
or (self.series_libraries[0] if self.series_libraries else POOL_SERIES))
stem = series_folder_stem(title, year, tmdb_id)
imported = 0
skipped_no_code: list[str] = []
errors: list[str] = []
for src in sources:
src = Path(src)
code = 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(title, season, ep, src.suffix)
if target.exists():
if on_conflict == "skip":
continue
if on_conflict == "replace":
target.unlink()
else: # suffix
base = episode_filename(title, season, ep, "")
counter = 1
while target.exists():
target = dest_dir / f"{base}_{counter}{src.suffix}"
counter += 1
try:
if move:
shutil.move(str(src), str(target))
else:
shutil.copy2(src, target)
imported += 1
except OSError as exc:
errors.append(f"{src.name}: {exc}")
return {"imported": imported, "skipped_no_code": skipped_no_code,
"errors": errors}
def rename_episodes_to_canonical(self, series) -> dict:
"""Rename a series' episodes to ``Series SxxEyy.ext`` in the right season.
``series`` is a ``series.Series``. Episodes with no recognisable code are
skipped; a target name already taken is reported as a conflict.
Returns ``{renamed, skipped, conflicts}``.
"""
from .series import season_folder_name, episode_filename
renamed: list[tuple[str, str]] = []
skipped: list[str] = []
conflicts: list[tuple[str, str]] = []
for ep in series.episodes:
if ep.season is None or ep.episode is None:
skipped.append(ep.path.name)
continue
dest_dir = series.folder / season_folder_name(ep.season)
new_path = dest_dir / episode_filename(
series.title, ep.season, ep.episode, ep.path.suffix)
if new_path == ep.path:
continue
if new_path.exists():
conflicts.append((ep.path.name, f"„{new_path.name}“ už existuje"))
continue
try:
dest_dir.mkdir(parents=True, exist_ok=True)
old_name = ep.path.name
ep.path.rename(new_path)
ep.path = new_path
renamed.append((old_name, new_path.name))
except OSError as exc:
conflicts.append((ep.path.name, str(exc)))
return {"renamed": renamed, "skipped": skipped, "conflicts": conflicts}
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:
src/core/hardlink_manager.py
+1 -1
View File
@@ -175,7 +175,7 @@ class HardlinkManager:
band) only happens here, when naming the folder.
"""
if transforms and tag.category in transforms:
from .csfd import apply_transform
from .transforms import apply_transform
return apply_transform(tag.name, transforms[tag.category])
return tag.name
src/core/jellyfin.py
+105
View File
@@ -0,0 +1,105 @@
"""Jellyfin server API client — thin, hybrid integration.
The NFO + artwork on disk stay authoritative (Jellyfin writes and reads them);
this client only does what the disk can't: **trigger a library scan** and **read
back what Jellyfin currently sees** (to verify the pool against it). Connection
details come from ``.env`` — ``JELLYFIN_URL`` and ``JELLYFIN_API_KEY`` (created in
the Jellyfin dashboard). Everything degrades gracefully when they are missing.
"""
import os
from dataclasses import dataclass
from typing import Optional
import requests
try: # the values can also come from the real environment
from dotenv import load_dotenv
load_dotenv()
except Exception: # noqa: BLE001
pass
_TIMEOUT = 30
@dataclass
class JellyfinMovie:
"""A movie as Jellyfin currently sees it (for verification)."""
jellyfin_id: str
title: str
year: Optional[int]
tmdb_id: Optional[int]
path: Optional[str]
class JellyfinError(RuntimeError):
"""A Jellyfin request could not be completed (no config, network, status)."""
def base_url() -> Optional[str]:
"""The configured Jellyfin base URL (no trailing slash), or None."""
url = os.getenv("JELLYFIN_URL")
return url.strip().rstrip("/") if url and url.strip() else None
def api_key() -> Optional[str]:
"""The configured Jellyfin API key, or None."""
key = os.getenv("JELLYFIN_API_KEY")
return key.strip() if key and key.strip() else None
def api_available() -> bool:
"""True when both the Jellyfin URL and API key are configured."""
return bool(base_url() and api_key())
def _request(method: str, path: str, params: Optional[dict] = None,
timeout: Optional[int] = None) -> requests.Response:
url, key = base_url(), api_key()
if not (url and key):
raise JellyfinError("JELLYFIN_URL / JELLYFIN_API_KEY nejsou nastaveny (.env).")
try:
resp = requests.request(
method, f"{url}{path}", headers={"X-Emby-Token": key},
params=params, timeout=timeout or _TIMEOUT)
except requests.RequestException as exc:
raise JellyfinError(f"Jellyfin nedostupný: {exc}") from exc
if resp.status_code in (401, 403):
raise JellyfinError("Neplatný JELLYFIN_API_KEY.")
if not resp.ok:
raise JellyfinError(f"Jellyfin odpověděl HTTP {resp.status_code}.")
return resp
def server_version(timeout: Optional[int] = None) -> str:
"""Return the Jellyfin server version (a quick connectivity check)."""
data = _request("GET", "/System/Info", timeout=timeout).json()
return str(data.get("Version") or "?")
def scan_all() -> None:
"""Trigger a scan of all Jellyfin libraries (picks up on-disk changes)."""
_request("POST", "/Library/Refresh")
def fetch_movies() -> list[JellyfinMovie]:
"""Read every movie Jellyfin currently has, with its TMDb id and path."""
data = _request("GET", "/Items", params={
"IncludeItemTypes": "Movie",
"Recursive": "true",
"Fields": "ProviderIds,Path,ProductionYear",
"EnableTotalRecordCount": "false",
}).json()
movies: list[JellyfinMovie] = []
for item in data.get("Items", []):
provider = item.get("ProviderIds") or {}
raw_tmdb = provider.get("Tmdb") or provider.get("tmdb")
tmdb_id = int(raw_tmdb) if raw_tmdb and str(raw_tmdb).isdigit() else None
movies.append(JellyfinMovie(
jellyfin_id=str(item.get("Id")),
title=item.get("Name") or "",
year=item.get("ProductionYear"),
tmdb_id=tmdb_id,
path=item.get("Path"),
))
return movies
src/core/naming.py
+34 -14
View File
@@ -1,18 +1,19 @@
"""Pool file naming convention.
Every movie in the pool is named ``Title (YYYY).ext`` — a title followed by a
four-digit year in parentheses. The year lets two films of the same name live
side by side and makes the filename self-describing (the year can be recovered
without any metadata). These helpers build, validate and parse that stem (the
filename without its extension).
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).
"""
import re
# A trailing four-digit year in parentheses. The title is non-greedy so a title
# that itself contains parentheses keeps them: "Já, robot (2004)" but also
# 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.
_POOL_STEM_RE = re.compile(r"^(?P<title>.+?) \((?P<year>\d{4})\)$")
_FOUR_DIGITS = re.compile(r"\d{4}")
_POOL_STEM_RE = re.compile(
r"^(?P<title>.+?) \((?P<year>\d{4})\)(?: \[tmdbid-(?P<id>\d+)\])?$")
def is_valid_year(year: object) -> bool:
@@ -20,11 +21,12 @@ 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) -> str:
"""Build the canonical pool stem ``Title (YYYY)``.
def canonical_pool_stem(title: str, year: object, tmdb_id: object = None) -> str:
"""Build the canonical pool stem ``Title (YYYY)`` (+ `` [tmdbid-N]`` if given).
Raises:
ValueError: the title is empty or the year is not four digits.
ValueError: the title is empty, the year is not four digits, or a given
``tmdb_id`` is not a positive integer.
"""
title = (title or "").strip()
year_s = str(year).strip()
@@ -32,17 +34,35 @@ def canonical_pool_stem(title: str, year: object) -> str:
raise ValueError("Název filmu nesmí být prázdný.")
if not re.fullmatch(r"\d{4}", year_s):
raise ValueError(f"Rok musí být čtyřmístné číslo, ne „{year}“.")
return f"{title} ({year_s})"
stem = f"{title} ({year_s})"
if tmdb_id is not None and str(tmdb_id).strip() != "":
id_s = str(tmdb_id).strip()
if not id_s.isdigit():
raise ValueError(f"TMDb ID musí být číslo, ne „{tmdb_id}“.")
stem += f" [tmdbid-{id_s}]"
return stem
def parse_pool_stem(stem: str) -> tuple[str, int] | None:
"""Split a canonical stem into ``(title, year)``; ``None`` if it isn't one."""
"""Split a canonical stem into ``(title, year)``; ``None`` if it isn't one.
A trailing ``[tmdbid-N]`` is tolerated and dropped (use :func:`parse_tmdb_id`
to read it), so the returned title is always the clean title.
"""
match = _POOL_STEM_RE.match((stem or "").strip())
if not match:
return None
return match.group("title").strip(), int(match.group("year"))
def parse_tmdb_id(stem: str) -> int | None:
"""Read the ``[tmdbid-N]`` from a canonical stem, or ``None`` if absent."""
match = _POOL_STEM_RE.match((stem or "").strip())
if not match or not match.group("id"):
return None
return int(match.group("id"))
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
src/core/nfo.py
+101
View File
@@ -0,0 +1,101 @@
"""NFO reader — Curator's metadata source after ČSFD.
Jellyfin (configured to save metadata into the media folders) writes a Kodi-style
``.nfo`` next to each movie plus artwork. Curator only **reads** it — to drive the
miniDLNA genre tree and to fill the year/rating used by folder grouping. It never
writes NFO (Jellyfin owns that).
For a flat pool file ``Title (rok) [tmdbid].mkv`` the sidecar is
``Title (rok) [tmdbid].nfo``; a per-movie folder may instead hold ``movie.nfo``.
Both are found. Missing or unparsable NFO → ``None`` (the movie simply has no
genres until Jellyfin has scanned it).
"""
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
@dataclass
class NfoData:
"""The bits of a movie ``.nfo`` Curator uses."""
title: Optional[str] = None
year: Optional[str] = None # four-digit string
genres: list[str] = field(default_factory=list)
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
def field_values(self, name: str) -> list[str]:
"""Values for a tag-schema source field (genres/countries/year/rating)."""
if name == "genres":
return list(self.genres)
if name == "countries":
return list(self.countries)
if name == "year":
return [self.year] if self.year else []
if name == "rating":
return [str(self.rating)] if self.rating is not None else []
return []
def find_nfo(video_path: Path) -> Optional[Path]:
"""Locate the ``.nfo`` for a movie: sibling ``<stem>.nfo`` or ``movie.nfo``."""
video_path = Path(video_path)
sibling = video_path.with_suffix(".nfo")
if sibling.is_file():
return sibling
movie_nfo = video_path.parent / "movie.nfo"
if movie_nfo.is_file():
return movie_nfo
return None
def _texts(root: ET.Element, tag: str) -> list[str]:
return [e.text.strip() for e in root.findall(tag) if e.text and e.text.strip()]
def _parse_rating(root: ET.Element) -> Optional[int]:
"""Read a 0–10 NFO rating and scale it to 0–100 (to match the old bands)."""
raw = root.findtext("rating")
if raw is None:
value = root.find("ratings/rating/value")
raw = value.text if value is not None else None
if not raw:
return None
try:
return round(float(raw.strip().replace(",", ".")) * 10)
except ValueError:
return None
def _parse_tmdb_id(root: ET.Element) -> Optional[int]:
for uid in root.findall("uniqueid"):
if uid.get("type") == "tmdb" and uid.text and uid.text.strip().isdigit():
return int(uid.text.strip())
raw = root.findtext("tmdbid")
return int(raw.strip()) if raw and raw.strip().isdigit() else 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)
if path is None:
return None
try:
root = ET.parse(path).getroot()
except (ET.ParseError, OSError):
return None
year = root.findtext("year")
year = year.strip() if year and year.strip().isdigit() else None
title = root.findtext("title")
return NfoData(
title=title.strip() if title and title.strip() else None,
year=year,
genres=_texts(root, "genre"),
countries=_texts(root, "country"),
rating=_parse_rating(root),
tmdb_id=_parse_tmdb_id(root),
)
src/core/series.py
+139
View File
@@ -0,0 +1,139 @@
"""Series naming + scanning for the Jellyfin TV layout.
Jellyfin wants series laid out as::
Seriály/Series (year) [tmdbid-N]/
tvshow.nfo (Jellyfin writes it)
Season 01/
Series (year) S01E01.ext
Specials/ (season 0)
Episodes are matched by ``SxxEyy``. These helpers parse the episode number out of
a raw filename, build the canonical episode/season/series names, and scan a
series library into a ``Series → Season → Episode`` tree with a naming status.
"""
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from .naming import canonical_pool_stem, parse_pool_stem, parse_tmdb_id
# SxxEyy / s1e2 / 1x02 — the episode marker in a filename. First match wins.
_EPISODE_RE = re.compile(r"(?:[Ss](\d{1,2})[\s._-]*[Ee](\d{1,3}))|(?:(\d{1,2})x(\d{1,3}))")
# Files next to an episode that are not video (NFO / artwork / subtitles).
_SIDECARS = {
".nfo", ".jpg", ".jpeg", ".png", ".webp", ".bmp", ".gif", ".tbn",
".srt", ".ass", ".ssa", ".sub", ".idx", ".vtt",
}
@dataclass
class Episode:
path: Path
season: Optional[int]
episode: Optional[int]
@property
def code(self) -> str:
if self.season is None or self.episode is None:
return "?"
return f"S{self.season:02d}E{self.episode:02d}"
@dataclass
class Series:
folder: Path
title: str
year: Optional[str]
tmdb_id: Optional[int]
library: str
episodes: list[Episode] = field(default_factory=list)
def seasons(self) -> dict[int, list[Episode]]:
"""Episodes grouped by season number (``-1`` = unrecognised)."""
grouped: dict[int, list[Episode]] = {}
for ep in self.episodes:
grouped.setdefault(ep.season if ep.season is not None else -1, []).append(ep)
for eps in grouped.values():
eps.sort(key=lambda e: (e.episode if e.episode is not None else 1_000_000))
return dict(sorted(grouped.items()))
def parse_episode(name: str) -> Optional[tuple[int, int]]:
"""Extract ``(season, episode)`` from a filename, or ``None``."""
match = _EPISODE_RE.search(name)
if not match:
return None
if match.group(1) is not None:
return int(match.group(1)), int(match.group(2))
return int(match.group(3)), int(match.group(4))
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_filename(series_title: str, season: int, episode: int, ext: str) -> str:
"""Canonical episode filename ``Series SxxEyy.ext``."""
return f"{series_title} S{season:02d}E{episode:02d}{ext}"
def series_folder_stem(title: str, year: object = None, tmdb_id: object = None) -> str:
"""Series folder name: ``Title (year) [tmdbid-N]`` (year/id optional).
Reuses the movie convention when a year is present; otherwise falls back to
``Title [tmdbid-N]`` or just ``Title`` so a year-less series still works.
"""
title = (title or "").strip()
if not title:
raise ValueError("Název seriálu nesmí být prázdný.")
year_s = str(year).strip() if year is not None else ""
if re.fullmatch(r"\d{4}", year_s):
return canonical_pool_stem(title, year_s, tmdb_id)
id_s = str(tmdb_id).strip() if tmdb_id is not None else ""
return f"{title} [tmdbid-{id_s}]" if id_s.isdigit() else title
def is_canonical_episode(path: Path, series_title: str) -> bool:
"""True when ``path`` already follows ``Series SxxEyy`` in the right season."""
parsed = parse_episode(path.name)
if parsed is None:
return False
season, episode = parsed
expected = episode_filename(series_title, season, episode, path.suffix)
if path.name != expected:
return False
return path.parent.name == season_folder_name(season)
def scan_series(lib_dir: Path, library: str) -> list[Series]:
"""Scan one series library into a list of :class:`Series`.
Each immediate subfolder of ``lib_dir`` is a series; its name yields the
title/year/tmdbid. Video files anywhere below it are episodes (season/episode
parsed from the filename). Non-video sidecars are skipped.
"""
lib_dir = Path(lib_dir)
if not lib_dir.is_dir():
return []
series_list: list[Series] = []
for folder in sorted(p for p in lib_dir.iterdir() if p.is_dir()):
parsed = parse_pool_stem(folder.name)
if parsed is not None:
title, year = parsed[0], str(parsed[1])
else:
title, year = folder.name, None
tmdb_id = parse_tmdb_id(folder.name)
episodes: list[Episode] = []
for f in sorted(folder.rglob("*")):
if not f.is_file() or f.suffix.lower() in _SIDECARS:
continue
code = parse_episode(f.name)
episodes.append(Episode(
f, code[0] if code else None, code[1] if code else None))
series_list.append(Series(folder, title, year, tmdb_id, library, episodes))
return series_list
src/core/tmdb.py
+137
View File
@@ -0,0 +1,137 @@
"""TMDb (The Movie Database) client — Curator's identity source.
Replaces ČSFD. Its only job is to turn a **TMDb id** (or a title/year search)
into a canonical **title + year** for naming the pool file. The rich metadata
and artwork are fetched by **Jellyfin itself** — Curator does not need them here.
Needs a free API key in ``TMDB_API_KEY`` (loaded from ``.env``). Everything
degrades gracefully when the key or the network is missing: :func:`api_available`
reports readiness and the lookups raise :class:`TmdbError` with a clear message.
"""
import os
from dataclasses import dataclass
from typing import Optional
import requests
try: # optional — the key can also come from the real environment
from dotenv import load_dotenv
load_dotenv()
except Exception: # noqa: BLE001 — dotenv is a convenience, never required
pass
_BASE = "https://api.themoviedb.org/3"
_LANG = os.getenv("TMDB_LANGUAGE", "cs-CZ")
_TIMEOUT = 15
@dataclass
class TmdbResult:
"""A resolved TMDb entry — just enough to name the pool file."""
tmdb_id: int
title: str
year: Optional[str] # four-digit string, or None when TMDb has no date
kind: str # "movie" | "tv"
class TmdbError(RuntimeError):
"""A TMDb request could not be completed (no key, network, bad status)."""
def api_key() -> Optional[str]:
"""The configured TMDb API key, or None."""
key = os.getenv("TMDB_API_KEY")
return key.strip() if key and key.strip() else None
def api_available() -> bool:
"""True when a TMDb API key is configured."""
return api_key() is not None
def _get(path: str, params: dict) -> dict:
key = api_key()
if not key:
raise TmdbError("TMDB_API_KEY není nastaven (.env).")
query = {"api_key": key, "language": _LANG, **params}
try:
resp = requests.get(f"{_BASE}{path}", params=query, timeout=_TIMEOUT)
except requests.RequestException as exc:
raise TmdbError(f"Síťová chyba TMDb: {exc}") from exc
if resp.status_code == 401:
raise TmdbError("Neplatný TMDB_API_KEY.")
if resp.status_code == 404:
raise TmdbError("TMDb ID nenalezeno.")
if not resp.ok:
raise TmdbError(f"TMDb odpověděl HTTP {resp.status_code}.")
return resp.json()
def _year_of(date_str: Optional[str]) -> Optional[str]:
if date_str and len(date_str) >= 4 and date_str[:4].isdigit():
return date_str[:4]
return None
def lookup_movie(tmdb_id: int) -> Optional[TmdbResult]:
"""Resolve a movie by its TMDb id → title + year, or None if it has no title."""
data = _get(f"/movie/{int(tmdb_id)}", {})
title = data.get("title") or data.get("original_title")
if not title:
return None
return TmdbResult(int(data["id"]), title, _year_of(data.get("release_date")), "movie")
def lookup_tv(tmdb_id: int) -> Optional[TmdbResult]:
"""Resolve a series by its TMDb id → name + first-air year, or None."""
data = _get(f"/tv/{int(tmdb_id)}", {})
title = data.get("name") or data.get("original_name")
if not title:
return None
return TmdbResult(int(data["id"]), title, _year_of(data.get("first_air_date")), "tv")
def lookup(tmdb_id: int, kind: str = "movie") -> Optional[TmdbResult]:
"""Resolve a TMDb id as a movie (default) or a series (``kind="tv"``)."""
return lookup_tv(tmdb_id) if kind == "tv" else lookup_movie(tmdb_id)
def search_movie(
query: str, year: Optional[str] = None, limit: int = 5
) -> list[TmdbResult]:
"""Search movies by title (optionally narrowed by year); best matches first."""
if not query.strip():
return []
params: dict = {"query": query.strip()}
if year:
params["year"] = year
data = _get("/search/movie", params)
results: list[TmdbResult] = []
for item in (data.get("results") or [])[:limit]:
title = item.get("title") or item.get("original_title")
if title and item.get("id"):
results.append(
TmdbResult(int(item["id"]), title,
_year_of(item.get("release_date")), "movie"))
return results
def search_tv(
query: str, year: Optional[str] = None, limit: int = 5
) -> list[TmdbResult]:
"""Search series by name (optionally narrowed by first-air year)."""
if not query.strip():
return []
params: dict = {"query": query.strip()}
if year:
params["first_air_date_year"] = year
data = _get("/search/tv", params)
results: list[TmdbResult] = []
for item in (data.get("results") or [])[:limit]:
title = item.get("name") or item.get("original_name")
if title and item.get("id"):
results.append(
TmdbResult(int(item["id"]), title,
_year_of(item.get("first_air_date")), "tv"))
return results
src/core/transforms.py
+41
View File
@@ -0,0 +1,41 @@
"""Tag-schema value transforms for miniDLNA folder names.
Applied when building a folder name from a tag value — the tag keeps its exact
value (e.g. rating 87 → tag "Hodnocení/87"), while the folder groups it
(→ "…/80–89 %"). Config-driven and extensible via ``TAG_TRANSFORMS``.
"""
import re
from typing import Optional
def rating_band(rating: int) -> str:
"""Bucket a 0–100 rating into a ten-point band label (e.g. "80–89 %").
The top bucket spans 90–100 % so a perfect 100 still lands in a band.
"""
low = min((rating // 10) * 10, 90)
high = 100 if low == 90 else low + 9
return f"{low}–{high} %"
def _decade_band(value) -> str:
"""``decade_band`` transform: pull the leading integer and bucket it."""
match = re.search(r"\d+", str(value))
return rating_band(int(match.group())) if match else str(value)
# Named value transforms usable from the tag schema (config-driven, extensible).
TAG_TRANSFORMS = {
None: str,
"identity": str,
"decade_band": _decade_band,
}
def apply_transform(value: str, transform: Optional[str]) -> str:
"""Apply a named tag-schema transform to a value, returning a string."""
fn = TAG_TRANSFORMS.get(transform, str)
try:
return str(fn(value))
except (ValueError, TypeError, AttributeError):
return str(value)