From b5025bb5f8074f57cdb06eca36647643d2c3c892 Mon Sep 17 00:00:00 2001 From: Francwa Date: Thu, 14 May 2026 16:09:33 +0200 Subject: [PATCH] refactor(resolve_destination): factor shared series-folder resolution + DTO base MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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). --- .../filesystem/resolve_destination.py | 283 +++++++----------- 1 file changed, 109 insertions(+), 174 deletions(-) diff --git a/alfred/application/filesystem/resolve_destination.py b/alfred/application/filesystem/resolve_destination.py index 9230419..2afa16c 100644 --- a/alfred/application/filesystem/resolve_destination.py +++ b/alfred/application/filesystem/resolve_destination.py @@ -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,27 +48,68 @@ 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. - status: str # "ok" | "needs_clarification" | "error" + 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(). + """ - # 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 + status: str # "ok" | "needs_clarification" | "error" # needs_clarification question: 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,38 +150,19 @@ 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 + library_file: str | None = None # full path to destination .mkv series_folder_name: str | None = None season_folder_name: str | None = None 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 [] - ) - 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, - ) + resolved = _resolve_series_folder(tv_root, tmdb_title, tmdb_year, computed_name, confirmed_folder) + if isinstance(resolved, _Clarification): + return ResolvedSeasonDestination( + status="needs_clarification", + 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 [] - ) - 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, - ) + resolved = _resolve_series_folder(tv_root, tmdb_title, tmdb_year, computed_name, confirmed_folder) + if isinstance(resolved, _Clarification): + return ResolvedEpisodeDestination( + status="needs_clarification", + 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 [] - ) - 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, - ) + resolved = _resolve_series_folder(tv_root, tmdb_title, tmdb_year, computed_name, confirmed_folder) + if isinstance(resolved, _Clarification): + return ResolvedSeriesDestination( + status="needs_clarification", + question=resolved.question, + options=resolved.options, + ) + series_folder_name, is_new = resolved series_path = tv_root / series_folder_name return ResolvedSeriesDestination(