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

Add edit dialogs, library organizer, and TMDb pool migration

Browse Source
This commit is contained in:
Honza
2026-09-08 07:37:40 +02:00
parent 39712d7283
commit 79c9e34a5d
10 changed files with 1371 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/file_manager.py
+279 -29
View File
@@ -17,12 +17,23 @@ from src.core.config import (
POOL_MOVIES = "Filmy"
POOL_SERIES = "Seriály"
# A pool library (top-level folder) has one of these kinds:
# movie — single-file movies, TMDb naming (Title (rok) [tmdbid].ext)
# series — series/season/episode tree, mirrored as-is (Jellyfin Shows)
# mixed — both movies and series in one folder (Jellyfin "Mixed"; wiring TBD)
# homevideo — files without TMDb (home videos; Jellyfin "Home videos"; wiring TBD)
LIBRARY_KINDS = ("movie", "series", "mixed", "homevideo")
_DEFAULT_LIBRARIES = [
{"name": POOL_MOVIES, "kind": "movie"},
{"name": POOL_SERIES, "kind": "series"},
]
# Library-level settings that live in the pool index (not the global config).
# ``libraries`` is the source of truth; the legacy ``movie_libraries`` /
# ``copyasis_folders`` are kept only to synthesise ``libraries`` for old pools.
_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],
}
@@ -117,26 +128,100 @@ class FileManager:
raise RuntimeError("Pool není nastaven — není kam uložit nastavení.")
idx.set_setting(key, value)
@property
def libraries(self) -> list[dict]:
"""All pool libraries as ``[{"name", "kind"}, …]`` (source of truth).
For pools predating the unified model this is synthesised from the legacy
``movie_libraries`` (→ kind ``movie``) + ``copyasis_folders`` (→ ``series``).
"""
idx = self._ensure_index()
raw = idx.get_setting("libraries", None) if idx is not None else None
if raw is None:
raw = (
[{"name": n, "kind": "movie"} for n in self._get_setting("movie_libraries")]
+ [{"name": n, "kind": "series"} for n in self._get_setting("copyasis_folders")]
)
out: list[dict] = []
seen: set[str] = set()
for spec in raw:
name = (spec.get("name") or "").strip()
kind = spec.get("kind") if spec.get("kind") in LIBRARY_KINDS else "movie"
if name and name.lower() not in seen:
out.append({"name": name, "kind": kind})
seen.add(name.lower())
return out or list(_DEFAULT_LIBRARIES)
def set_libraries(self, specs: list[dict]) -> None:
"""Replace the library list (list of ``{"name", "kind"}``) and persist it."""
cleaned: list[dict] = []
seen: set[str] = set()
for spec in specs:
name = (spec.get("name") or "").strip()
kind = spec.get("kind") if spec.get("kind") in LIBRARY_KINDS else "movie"
if name and name.lower() not in seen:
cleaned.append({"name": name, "kind": kind})
seen.add(name.lower())
self._set_setting("libraries", cleaned or list(_DEFAULT_LIBRARIES))
def libraries_of_kind(self, *kinds: str) -> list[str]:
"""Names of libraries matching any of ``kinds``."""
return [lib["name"] for lib in self.libraries if lib["kind"] in kinds]
@property
def copyasis_folders(self) -> list[str]:
"""Names of pool subfolders mirrored 1:1 (copy-as-is) into the output."""
return self._get_setting("copyasis_folders")
"""Names of pool subfolders mirrored 1:1 (copy-as-is) — the series libraries."""
return self.libraries_of_kind("series")
def set_copyasis_folders(self, names: list[str]) -> None:
"""Set the copy-as-is subfolder list and persist it."""
"""Set the series (copy-as-is) libraries, keeping the other kinds."""
cleaned = [n.strip() for n in names if n.strip()]
self._set_setting("copyasis_folders", cleaned)
others = [lib for lib in self.libraries if lib["kind"] != "series"]
self.set_libraries(others + [{"name": n, "kind": "series"} for n in 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]
"""Pool folders holding single-file movies (kind ``movie``)."""
return self.libraries_of_kind("movie") or [POOL_MOVIES]
@property
def libraries_accepting_movies(self) -> list[str]:
"""Libraries a movie can live in / be assigned to — kind movie or mixed."""
return self.libraries_of_kind("movie", "mixed") or [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])
"""Set the movie libraries, keeping the other kinds; never empty."""
cleaned = [n.strip() for n in names if n.strip()] or [POOL_MOVIES]
others = [lib for lib in self.libraries if lib["kind"] != "movie"]
self.set_libraries([{"name": n, "kind": "movie"} for n in cleaned] + others)
def orphan_pool_folders(self) -> list[tuple[str, bool]]:
"""Top-level pool folders that aren't a configured library.
Returns ``(name, is_empty)`` pairs. Only these are candidates for cleanup;
non-empty ones are reported but never auto-deleted (they may hold media).
"""
pool = self.pool_dir
if not (pool and pool.is_dir()):
return []
known = {lib["name"] for lib in self.libraries}
orphans: list[tuple[str, bool]] = []
for entry in sorted(pool.iterdir()):
if entry.is_dir() and entry.name not in known:
orphans.append((entry.name, not any(entry.iterdir())))
return orphans
def remove_empty_pool_folder(self, name: str) -> bool:
"""Delete an **empty** orphan folder in the pool root. Returns True if removed."""
pool = self.pool_dir
if not pool:
return False
known = {lib["name"] for lib in self.libraries}
target = pool / name
if (target.is_dir() and name not in known and not any(target.iterdir())):
target.rmdir()
return True
return False
def library_of(self, file_obj: File) -> str | None:
"""Which library (top-level pool folder) a pooled movie lives in."""
@@ -200,8 +285,8 @@ class FileManager:
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)
for lib in self.libraries:
(pool_dir / lib["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)."""
@@ -222,17 +307,24 @@ class FileManager:
return
self._open_index(pool)
for library in self.movie_libraries:
lib_dir = pool / library
for lib in self.libraries:
if lib["kind"] not in ("movie", "mixed"):
continue
lib_dir = pool / lib["name"]
if not lib_dir.is_dir():
continue
for each in list_files(lib_dir):
# A mixed library keeps movies at the top level and series in
# subfolders (Season NN) → only its top-level files are movies.
if lib["kind"] == "mixed":
candidates = [p for p in sorted(lib_dir.iterdir()) if p.is_file()]
else:
candidates = list_files(lib_dir)
for each in candidates:
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)
self.filelist.append(File(each, self.tagmanager, index=self.index))
def check_data_consistency(self) -> dict:
"""Compare the pool index against the actual files in the movie libraries.
@@ -255,11 +347,15 @@ class FileManager:
index_keys = set(index.records.keys())
disk_keys = set()
for library in self.movie_libraries:
lib_dir = pool / library
for lib in self.libraries:
if lib["kind"] not in ("movie", "mixed"):
continue
lib_dir = pool / lib["name"]
if not lib_dir.is_dir():
continue
for f in list_files(lib_dir):
files = ([p for p in lib_dir.iterdir() if p.is_file()]
if lib["kind"] == "mixed" else list_files(lib_dir))
for f in files:
if f.name.endswith(METADATA_SUFFIXES):
continue
if f.suffix.lower() in SIDECAR_SUFFIXES:
@@ -421,7 +517,7 @@ class FileManager:
self, source: Path, title: str,
move: bool = False, on_conflict: str = "suffix",
year: int | str | None = None, tmdb_id: int | str | None = None,
library: str | None = None,
library: str | None = None, csfd_url: str | None = None,
) -> File | None:
"""Bring a video file into a movie library and index its metadata.
@@ -487,6 +583,7 @@ class FileManager:
file_obj = File(target, self.tagmanager, index=self.index)
file_obj.title = safe_title
file_obj.csfd_url = csfd_url or None
file_obj.added = datetime.now().isoformat(timespec="seconds")
file_obj.save_metadata()
@@ -650,7 +747,7 @@ class FileManager:
if pool is None:
raise RuntimeError("Pool není nastaven.")
new_library = new_library.strip()
if new_library not in self.movie_libraries:
if new_library not in self.libraries_accepting_movies:
raise ValueError(f"Neznámá knihovna „{new_library}“.")
old_path = file_obj.file_path
@@ -670,6 +767,123 @@ class FileManager:
self.on_files_changed(self.filelist)
return file_obj
def apply_movie_edits(
self, file_obj: File, *, title: str, year: int | str | None = None,
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,
) -> 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.
Raises:
FileExistsError: the target name is already taken.
"""
pool = self.pool_dir
if pool is None:
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)
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)
file_obj.title = title
if date is not None:
file_obj.date = date or None
if csfd_url is not None:
file_obj.csfd_url = csfd_url or None
if attributes is not None:
file_obj.attributes = dict(attributes)
if tags is not None:
new_tags: list = []
seen: set[str] = set()
for full_path in tags:
full_path = full_path.strip()
if "/" not in full_path or full_path in seen:
continue
cat, name = full_path.split("/", 1)
new_tags.append(self.tagmanager.add_tag(cat.strip(), name.strip()))
seen.add(full_path)
file_obj.tags = new_tags
file_obj.auto_tag_paths = {p for p in file_obj.auto_tag_paths if p in seen}
file_obj.save_metadata()
if self.on_files_changed:
self.on_files_changed(self.filelist)
return file_obj
def match_pool_to_tmdb(
self, files: list[File] | None = None,
on_progress: Callable[[int, int, File], None] | None = None,
should_cancel: Callable[[], bool] | None = None,
) -> list:
"""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
``[tmdbid-N]`` are skipped. Returns a list of ``migrate.MatchProposal``.
"""
from . import tmdb
from .migrate import MatchProposal, classify
targets = files if files is not None else self.filelist
proposals: list = []
total = len(targets)
for i, f in enumerate(targets, 1):
if should_cancel is not None and should_cancel():
break
if parse_tmdb_id(f.file_path.stem) is not None:
if on_progress is not None:
on_progress(i, total, f)
continue # already migrated
ctx = f.name_context()
title = (f.title or ctx.get("title") or f.file_path.stem).strip()
parsed = parse_pool_stem(title)
if parsed is not None:
title = parsed[0]
year = str(ctx.get("year") or "").strip() or None
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))
except tmdb.TmdbError as exc:
proposals.append(MatchProposal(
f, f.filename, title, year, "error", None, [], str(exc)))
if on_progress is not None:
on_progress(i, total, f)
return proposals
def assign_tmdb_id(
self, file_obj: File, tmdb_id: int | str, year: int | str | None = None
) -> File:
"""Attach a TMDb id to a movie and rename it to the canonical convention.
``year`` (typically from the TMDb match) fills a missing local year so the
file becomes ``Title (rok) [tmdbid].ext``. Title/library/tags/date are kept.
"""
ctx = file_obj.name_context()
title = (file_obj.title or ctx.get("title") or file_obj.file_path.stem).strip()
parsed = parse_pool_stem(title)
if parsed is not None:
title = parsed[0]
year_s = str(year).strip() if year not in (None, "") else str(ctx.get("year") or "")
return self.apply_movie_edits(
file_obj, title=title, year=year_s or None, tmdb_id=tmdb_id)
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).
@@ -700,8 +914,8 @@ class FileManager:
@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
"""Pool folders that can hold series — kind ``series`` or ``mixed``."""
return self.libraries_of_kind("series", "mixed")
def load_series(self) -> list:
"""Scan every series library into a list of ``series.Series``."""
@@ -748,14 +962,14 @@ class FileManager:
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)
target = dest_dir / episode_filename(season, ep, src.stem, src.suffix)
if target.exists():
if on_conflict == "skip":
continue
if on_conflict == "replace":
target.unlink()
else: # suffix
base = episode_filename(title, season, ep, "")
base = episode_filename(season, ep, src.stem, "")
counter = 1
while target.exists():
target = dest_dir / f"{base}_{counter}{src.suffix}"
@@ -789,7 +1003,7 @@ class FileManager:
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)
ep.season, ep.episode, ep.path.stem, ep.path.suffix)
if new_path == ep.path:
continue
if new_path.exists():
@@ -805,6 +1019,42 @@ class FileManager:
conflicts.append((ep.path.name, str(exc)))
return {"renamed": renamed, "skipped": skipped, "conflicts": conflicts}
def apply_series_edits(
self, series, *, title: str, year: int | str | None = None,
tmdb_id: int | str | None = None, library: str | None = None,
) -> dict:
"""Rename/move a series folder and re-title its episodes to match.
Renames the series folder to ``Title (rok) [tmdbid]`` (moving it into
``library`` if changed), then re-derives the series and renames every
episode to ``NewTitle SxxEyy.ext`` in the right Season folder.
Returns ``{folder, episodes}`` (``episodes`` = rename_episodes result).
Raises:
FileExistsError: the target series folder already exists.
"""
from .series import scan_series, series_folder_stem
pool = self.pool_dir
if pool is None:
raise RuntimeError("Pool není nastaven.")
title = (title or "").strip() or series.title
new_stem = series_folder_stem(title, year or None, tmdb_id or None)
lib = ((library or "").strip() or series.library)
old_folder = series.folder
new_folder = pool / lib / new_stem
if new_folder != old_folder:
if new_folder.exists():
raise FileExistsError(f"Složka „{new_stem}“ už existuje.")
new_folder.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(old_folder), str(new_folder))
target = next((s for s in scan_series(new_folder.parent, lib)
if s.folder == new_folder), None)
result = (self.rename_episodes_to_canonical(target) if target
else {"renamed": [], "skipped": [], "conflicts": []})
return {"folder": new_folder, "episodes": result}
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: