refactor(resolve_destination): factor shared series-folder resolution + DTO base

- New _Clarification sentinel and _resolve_series_folder() helper —
  the three TV use cases now share one matching/clarification path
  instead of triplicating the same if/elif/else block.
- New _ResolvedDestinationBase carrying status/question/options/error/
  message plus a _base_dict() helper; the four concrete DTOs only
  declare their own ok-state fields and a slim to_dict().
- No behaviour change: same outputs for ok/needs_clarification/error
  cases (verified by import + DTO smoke tests).
This commit is contained in:
2026-05-14 16:09:33 +02:00
parent e45465d52d
commit b5025bb5f8
@@ -29,9 +29,7 @@ def _sanitize(text: str) -> str:
return _WIN_FORBIDDEN.sub("", text)
def _find_existing_tvshow_folders(
tv_root: Path, tmdb_title: str, tmdb_year: int
) -> list[str]:
def _find_existing_tvshow_folders(tv_root: Path, tmdb_title: str, tmdb_year: int) -> list[str]:
"""Return folder names in tv_root that match title + year prefix."""
if not tv_root.exists():
return []
@@ -50,28 +48,69 @@ def _get_tv_root() -> Path | None:
return Path(tv_root) if tv_root else None
# ---------------------------------------------------------------------------
# Internal sentinel + series-folder resolver (shared by the 3 TV use cases)
# ---------------------------------------------------------------------------
@dataclass
class _Clarification:
"""Module-private sentinel signalling that user input is needed."""
question: str
options: list[str]
def _resolve_series_folder(
tv_root: Path,
tmdb_title: str,
tmdb_year: int,
computed_name: str,
confirmed_folder: str | None,
) -> tuple[str, bool] | _Clarification:
"""
Resolve which series folder to use.
Returns:
(folder_name, is_new) if resolved unambiguously,
_Clarification(question, options) if the caller must ask the user.
"""
if confirmed_folder:
return confirmed_folder, not (tv_root / confirmed_folder).exists()
existing = _find_existing_tvshow_folders(tv_root, tmdb_title, tmdb_year)
if not existing:
return computed_name, True
if len(existing) == 1 and existing[0] == computed_name:
return existing[0], False
options = existing + ([computed_name] if computed_name not in existing else [])
return _Clarification(
question=(
f"Un dossier série existe déjà pour '{tmdb_title}' "
f"mais son nom diffère du nom calculé ({computed_name}). "
f"Lequel utiliser ?"
),
options=options,
)
# ---------------------------------------------------------------------------
# DTOs
# ---------------------------------------------------------------------------
@dataclass
class ResolvedSeasonDestination:
"""Paths for a season pack — folder move, no individual file paths."""
class _ResolvedDestinationBase:
"""
Shared shape across all resolution DTOs.
Holds the status flag and the fields used in non-ok states
(error / needs_clarification). Subclasses add their own ok-state fields
and a to_dict() that delegates the non-ok cases via _base_dict().
"""
status: str # "ok" | "needs_clarification" | "error"
# ok
series_folder: str | None = (
None # /tv_shows/A.Knight.of.the.Seven.Kingdoms.2024.1080p.WEBRip.x265-KONTRAST
)
season_folder: str | None = (
None # .../A.Knight.of.the.Seven.Kingdoms.S01.1080p.WEBRip.x265-KONTRAST
)
series_folder_name: str | None = None
season_folder_name: str | None = None
is_new_series_folder: bool = False
# needs_clarification
question: str | None = None
options: list[str] | None = None
@@ -80,16 +119,27 @@ class ResolvedSeasonDestination:
error: str | None = None
message: str | None = None
def to_dict(self) -> dict:
def _base_dict(self) -> dict | None:
"""Return the dict for error/needs_clarification, or None for ok."""
if self.status == "error":
return {"status": self.status, "error": self.error, "message": self.message}
if self.status == "needs_clarification":
return {
"status": self.status,
"question": self.question,
"options": self.options or [],
}
return {
return {"status": self.status, "question": self.question, "options": self.options or []}
return None
@dataclass
class ResolvedSeasonDestination(_ResolvedDestinationBase):
"""Paths for a season pack — folder move, no individual file paths."""
series_folder: str | None = None
season_folder: str | None = None
series_folder_name: str | None = None
season_folder_name: str | None = None
is_new_series_folder: bool = False
def to_dict(self) -> dict:
return self._base_dict() or {
"status": self.status,
"series_folder": self.series_folder,
"season_folder": self.season_folder,
@@ -100,12 +150,9 @@ class ResolvedSeasonDestination:
@dataclass
class ResolvedEpisodeDestination:
class ResolvedEpisodeDestination(_ResolvedDestinationBase):
"""Paths for a single episode — file move."""
status: str
# ok
series_folder: str | None = None
season_folder: str | None = None
library_file: str | None = None # full path to destination .mkv
@@ -114,24 +161,8 @@ class ResolvedEpisodeDestination:
filename: str | None = None
is_new_series_folder: bool = False
# needs_clarification
question: str | None = None
options: list[str] | None = None
# error
error: str | None = None
message: str | None = None
def to_dict(self) -> dict:
if self.status == "error":
return {"status": self.status, "error": self.error, "message": self.message}
if self.status == "needs_clarification":
return {
"status": self.status,
"question": self.question,
"options": self.options or [],
}
return {
return self._base_dict() or {
"status": self.status,
"series_folder": self.series_folder,
"season_folder": self.season_folder,
@@ -144,36 +175,17 @@ class ResolvedEpisodeDestination:
@dataclass
class ResolvedMovieDestination:
class ResolvedMovieDestination(_ResolvedDestinationBase):
"""Paths for a movie — file move."""
status: str
# ok
movie_folder: str | None = None
library_file: str | None = None
movie_folder_name: str | None = None
filename: str | None = None
is_new_folder: bool = False
# needs_clarification
question: str | None = None
options: list[str] | None = None
# error
error: str | None = None
message: str | None = None
def to_dict(self) -> dict:
if self.status == "error":
return {"status": self.status, "error": self.error, "message": self.message}
if self.status == "needs_clarification":
return {
"status": self.status,
"question": self.question,
"options": self.options or [],
}
return {
return self._base_dict() or {
"status": self.status,
"movie_folder": self.movie_folder,
"library_file": self.library_file,
@@ -184,34 +196,15 @@ class ResolvedMovieDestination:
@dataclass
class ResolvedSeriesDestination:
class ResolvedSeriesDestination(_ResolvedDestinationBase):
"""Paths for a complete multi-season series pack — folder move."""
status: str
# ok
series_folder: str | None = None
series_folder_name: str | None = None
is_new_series_folder: bool = False
# needs_clarification
question: str | None = None
options: list[str] | None = None
# error
error: str | None = None
message: str | None = None
def to_dict(self) -> dict:
if self.status == "error":
return {"status": self.status, "error": self.error, "message": self.message}
if self.status == "needs_clarification":
return {
"status": self.status,
"question": self.question,
"options": self.options or [],
}
return {
return self._base_dict() or {
"status": self.status,
"series_folder": self.series_folder,
"series_folder_name": self.series_folder_name,
@@ -223,7 +216,6 @@ class ResolvedSeriesDestination:
# Use cases
# ---------------------------------------------------------------------------
def resolve_season_destination(
release_name: str,
tmdb_title: str,
@@ -239,42 +231,22 @@ def resolve_season_destination(
tv_root = _get_tv_root()
if not tv_root:
return ResolvedSeasonDestination(
status="error",
error="library_not_set",
status="error", error="library_not_set",
message="TV show library path is not configured.",
)
parsed = parse_release(release_name)
computed_name = _sanitize(parsed.show_folder_name(tmdb_title, tmdb_year))
if confirmed_folder:
series_folder_name = confirmed_folder
is_new = not (tv_root / confirmed_folder).exists()
else:
existing = _find_existing_tvshow_folders(tv_root, tmdb_title, tmdb_year)
if len(existing) == 0:
series_folder_name = computed_name
is_new = True
elif len(existing) == 1 and existing[0] == computed_name:
# Exact match — use it
series_folder_name = existing[0]
is_new = False
else:
# One folder with a different name, or multiple — ask user
options = existing + (
[computed_name] if computed_name not in existing else []
)
resolved = _resolve_series_folder(tv_root, tmdb_title, tmdb_year, computed_name, confirmed_folder)
if isinstance(resolved, _Clarification):
return ResolvedSeasonDestination(
status="needs_clarification",
question=(
f"Un dossier série existe déjà pour '{tmdb_title}' "
f"mais son nom diffère du nom calculé ({computed_name}). "
f"Lequel utiliser ?"
),
options=options,
question=resolved.question,
options=resolved.options,
)
series_folder_name, is_new = resolved
season_folder_name = parsed.season_folder_name()
series_path = tv_root / series_folder_name
season_path = series_path / season_folder_name
@@ -305,8 +277,7 @@ def resolve_episode_destination(
tv_root = _get_tv_root()
if not tv_root:
return ResolvedEpisodeDestination(
status="error",
error="library_not_set",
status="error", error="library_not_set",
message="TV show library path is not configured.",
)
@@ -314,32 +285,15 @@ def resolve_episode_destination(
ext = Path(source_file).suffix
computed_name = _sanitize(parsed.show_folder_name(tmdb_title, tmdb_year))
if confirmed_folder:
series_folder_name = confirmed_folder
is_new = not (tv_root / confirmed_folder).exists()
else:
existing = _find_existing_tvshow_folders(tv_root, tmdb_title, tmdb_year)
if len(existing) == 0:
series_folder_name = computed_name
is_new = True
elif len(existing) == 1 and existing[0] == computed_name:
series_folder_name = existing[0]
is_new = False
else:
options = existing + (
[computed_name] if computed_name not in existing else []
)
resolved = _resolve_series_folder(tv_root, tmdb_title, tmdb_year, computed_name, confirmed_folder)
if isinstance(resolved, _Clarification):
return ResolvedEpisodeDestination(
status="needs_clarification",
question=(
f"Un dossier série existe déjà pour '{tmdb_title}' "
f"mais son nom diffère du nom calculé ({computed_name}). "
f"Lequel utiliser ?"
),
options=options,
question=resolved.question,
options=resolved.options,
)
series_folder_name, is_new = resolved
season_folder_name = parsed.season_folder_name()
filename = _sanitize(parsed.episode_filename(tmdb_episode_title, ext))
@@ -374,8 +328,7 @@ def resolve_movie_destination(
movies_root = memory.ltm.library_paths.get("movie")
if not movies_root:
return ResolvedMovieDestination(
status="error",
error="library_not_set",
status="error", error="library_not_set",
message="Movie library path is not configured.",
)
@@ -412,40 +365,22 @@ def resolve_series_destination(
tv_root = _get_tv_root()
if not tv_root:
return ResolvedSeriesDestination(
status="error",
error="library_not_set",
status="error", error="library_not_set",
message="TV show library path is not configured.",
)
parsed = parse_release(release_name)
computed_name = _sanitize(parsed.show_folder_name(tmdb_title, tmdb_year))
if confirmed_folder:
series_folder_name = confirmed_folder
is_new = not (tv_root / confirmed_folder).exists()
else:
existing = _find_existing_tvshow_folders(tv_root, tmdb_title, tmdb_year)
if len(existing) == 0:
series_folder_name = computed_name
is_new = True
elif len(existing) == 1 and existing[0] == computed_name:
series_folder_name = existing[0]
is_new = False
else:
options = existing + (
[computed_name] if computed_name not in existing else []
)
resolved = _resolve_series_folder(tv_root, tmdb_title, tmdb_year, computed_name, confirmed_folder)
if isinstance(resolved, _Clarification):
return ResolvedSeriesDestination(
status="needs_clarification",
question=(
f"Un dossier série existe déjà pour '{tmdb_title}' "
f"mais son nom diffère du nom calculé ({computed_name}). "
f"Lequel utiliser ?"
),
options=options,
question=resolved.question,
options=resolved.options,
)
series_folder_name, is_new = resolved
series_path = tv_root / series_folder_name
return ResolvedSeriesDestination(