fix(tv_shows): correct PACK vs EPISODIC classification model

The Phase 4 walker + rescan logic classified seasons by parser
output (does the filename carry Exx?), but PACK vs EPISODIC is a
structural distinction:

* PACK = season folder with N flat SxxEyy videos directly inside
* EPISODIC = season folder with N subfolders, each holding one video

Changes:
* walker.py: descends two levels under show_root and classifies
  each season folder by FS structure. SeasonFolder now carries
  mode: ReleaseMode | None. Mixed layouts (flat + subfolders) and
  EPISODIC subfolders with >1 video log a warning and report
  mode=None.
* rescan.py: trusts walker.mode; drops the bogus 'single un-
  numbered video → PACK with empty episodes' branch. A season
  with no parseable episodes is now skipped with a warning.
* Tests rewritten against the real model: PACK with flat numbered
  files, EPISODIC with one-video-per-subfolder, malformed mixed
  layout skipped, single-un-numbered-file skipped.

Suite: 1237 → 1245 passing.
This commit is contained in:
2026-05-25 21:37:34 +02:00
parent fe9857aaed
commit 97dc799a26
4 changed files with 416 additions and 120 deletions
+25 -38
View File
@@ -15,27 +15,18 @@ a single source of truth for parsing / probing rules. The orchestrator
just translates per-file :class:`InspectedResult` into release just translates per-file :class:`InspectedResult` into release
aggregate construction. aggregate construction.
PACK vs EPISODIC detection PACK vs EPISODIC
--------------------------- ----------------
Detection lives in this layer (until the TMDB-driven season-tracker Classification is done by the walker, by inspecting the season
arrives). The current rule: folder's filesystem structure (flat videos → PACK, subfolders →
EPISODIC). See :mod:`alfred.application.tv_shows.walker`. The
orchestrator trusts ``season_folder.mode`` and never re-derives.
* A season folder containing exactly one video whose parser yields Files whose parser yields ``season is None`` or ``episode is None``
``season is not None`` and ``episode is None`` → **PACK** with an are logged and skipped — a real PACK or EPISODIC file always carries
empty ``episodes`` tuple. We record the season's mode + folder, but both. Mixed-season folders (two different ``Sxx`` numbers in the
we cannot fill the episode slot map without TMDB's same directory) are skipped with a warning.
``episode_count`` — that's Phase 5's job. The file is still on
disk; the next TMDB sync repairs the slot map.
* Otherwise → **EPISODIC**: every file with a valid ``(season,
episode)`` becomes an :class:`EpisodeRelease`. Multi-episode files
(``S01E01E02``) are recorded once with a wide ``EpisodeRange``;
Phase 4 only handles the single-episode case (the parser does not
yet expose ``episode_end`` on ``ParsedRelease``).
Files that fall outside both rules (no season parsed, mixed season
numbers in a folder, etc.) are logged and skipped — the walker doesn't
raise on corrupt input, and neither does the orchestrator.
TMDB TMDB
---- ----
@@ -124,6 +115,11 @@ def _ingest_season(
kb: ReleaseKnowledge, kb: ReleaseKnowledge,
prober: MediaProber, prober: MediaProber,
) -> SeasonRelease | None: ) -> SeasonRelease | None:
if season_folder.mode is None:
# Walker already logged the reason (empty / malformed mix /
# multi-video subfolder). Just skip.
return None
if not season_folder.video_files: if not season_folder.video_files:
_LOG.warning( _LOG.warning(
"rescan_show: season folder %s contains no video file — skipping", "rescan_show: season folder %s contains no video file — skipping",
@@ -131,8 +127,7 @@ def _ingest_season(
) )
return None return None
# Inspect every video first; we need the whole batch to decide # Inspect every video to extract season + episode numbers.
# PACK vs EPISODIC before assembling the SeasonRelease.
inspected = [] inspected = []
for video_path in season_folder.video_files: for video_path in season_folder.video_files:
result = inspect_release(video_path.name, video_path, kb, prober) result = inspect_release(video_path.name, video_path, kb, prober)
@@ -157,22 +152,6 @@ def _ingest_season(
season_number = SeasonNumber(season_numbers.pop()) season_number = SeasonNumber(season_numbers.pop())
folder_name = season_folder.season_dir.name folder_name = season_folder.season_dir.name
# Single video, no episode → PACK with empty episodes. We can't
# synthesize an EpisodeRange without TMDB's episode_count; the
# Phase 5 sync repairs the slot map. Track info from the PACK
# file is intentionally not persisted here — re-derivable on the
# next rescan after the sync fills the range.
if len(inspected) == 1 and inspected[0][1].parsed.episode is None:
return SeasonRelease(
season_number=season_number,
folder=folder_name,
mode=ReleaseMode.PACK,
episodes=(),
)
# EPISODIC: every file with a parseable episode number becomes an
# EpisodeRelease. Files without an episode number are logged and
# dropped (a mixed PACK/EPISODIC folder is malformed).
episodes: list[EpisodeRelease] = [] episodes: list[EpisodeRelease] = []
for video_path, result in inspected: for video_path, result in inspected:
if result.parsed.episode is None: if result.parsed.episode is None:
@@ -189,10 +168,18 @@ def _ingest_season(
media_info=result.media_info, media_info=result.media_info,
) )
) )
if not episodes:
_LOG.warning(
"rescan_show: no parseable episodes in %s — skipping",
season_folder.season_dir,
)
return None
return SeasonRelease( return SeasonRelease(
season_number=season_number, season_number=season_number,
folder=folder_name, folder=folder_name,
mode=ReleaseMode.EPISODIC, mode=season_folder.mode,
episodes=tuple(episodes), episodes=tuple(episodes),
) )
+135 -25
View File
@@ -1,11 +1,13 @@
"""Show tree walker — minimal filesystem traversal of a TV show folder. """Show tree walker — minimal filesystem traversal of a TV show folder.
The walker is intentionally dumb: it lists season folders and the The walker is intentionally dumb: it lists season folders, classifies
video files inside them, nothing more. It does not parse release each one as PACK or EPISODIC by **inspecting its filesystem
names, run ffprobe, or classify subtitle files. All of that structure**, and hands the orchestrator a flat list of video files
intelligence lives in the existing release pipeline per season. It does not parse release names, run ffprobe, or
(``inspect_release`` + downstream services); the walker just hands classify subtitle files. All of that intelligence lives in the
the orchestrator the paths to feed into that pipeline. existing release pipeline (``inspect_release`` + downstream
services); the walker just hands the orchestrator the paths to feed
into that pipeline.
Folder convention Folder convention
----------------- -----------------
@@ -13,32 +15,47 @@ Folder convention
Inside an Alfred-managed library, a show root looks like:: Inside an Alfred-managed library, a show root looks like::
Foundation/ Foundation/
Foundation.S01.1080p.WEB-DL.x265-GROUP/ ← season folder Foundation.S01.1080p.WEB-DL.x265-GROUP/ ← PACK season
Foundation.S01E01.1080p.WEB-DL.x265.mkv ← episode (EPISODIC) Foundation.S01E01.1080p.WEB-DL.x265.mkv ← flat video
Foundation.S01E02.1080p.WEB-DL.x265.mkv Foundation.S01E02.1080p.WEB-DL.x265.mkv
Foundation.S02.Complete.1080p.WEB-DL-GROUP/ ← season folder ...
Foundation.S02.Complete.1080p.WEB-DL.mkv ← PACK (single video) Foundation.S02/ ← EPISODIC season
Foundation.S02E01.1080p.WEB-DL.x265-GROUP/ ← episode subfolder
Foundation.S02E01.1080p.WEB-DL.x265-GROUP.mkv
Foundation.S02E02.1080p.WEB-DL.x265-OTHER/
Foundation.S02E02.1080p.WEB-DL.x265-OTHER.mkv
The walker recognizes a season folder by a ``Sxx`` token anywhere in The walker recognizes a season folder by a ``Sxx`` token anywhere in
its name (case-insensitive). It does **not** care about Plex-style its name (case-insensitive). It does **not** care about Plex-style
names (``Season 01``, ``Specials``) — the Alfred library uses names (``Season 01``, ``Specials``) — the Alfred library uses
release-style folder names only. release-style folder names only.
PACK vs EPISODIC detection happens **above** the walker, by the PACK vs EPISODIC is a **structural distinction**, not a naming one:
orchestrator looking at how many video files came back and whether
they carry ``Exx`` tokens. The walker itself reports every video * **PACK** — season folder contains N flat video files. No
file it sees, untouched. subfolders.
* **EPISODIC** — season folder contains N subfolders, each holding
exactly one video.
A season folder that mixes the two layouts (some flat videos AND
some subfolders) is malformed: the walker reports
``mode=None`` and an empty ``video_files`` tuple so the
orchestrator can warn and skip it.
""" """
from __future__ import annotations from __future__ import annotations
import logging
import re import re
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from alfred.domain.release.ports import ReleaseKnowledge from alfred.domain.release.ports import ReleaseKnowledge
from alfred.domain.releases.value_objects import ReleaseMode
from alfred.domain.shared.ports import FilesystemScanner from alfred.domain.shared.ports import FilesystemScanner
_LOG = logging.getLogger(__name__)
# Matches any ``Sxx`` token (1-2 digits) bounded by non-alphanumerics. # Matches any ``Sxx`` token (1-2 digits) bounded by non-alphanumerics.
# Examples that match: ``Foundation.S01.1080p`` , ``S2.Pack`` , ``BBC.s10.bluray``. # Examples that match: ``Foundation.S01.1080p`` , ``S2.Pack`` , ``BBC.s10.bluray``.
# Examples that don't: ``Sample`` , ``Soundtrack`` , ``2024.S0E1`` (no S+digits boundary). # Examples that don't: ``Sample`` , ``Soundtrack`` , ``2024.S0E1`` (no S+digits boundary).
@@ -47,9 +64,21 @@ _SEASON_TOKEN_RE = re.compile(r"(?<![A-Za-z0-9])s(\d{1,2})(?![A-Za-z0-9])", re.I
@dataclass(frozen=True) @dataclass(frozen=True)
class SeasonFolder: class SeasonFolder:
"""One season folder discovered inside a show root.""" """One season folder discovered inside a show root.
``mode`` is set by the walker from the FS structure:
* :attr:`ReleaseMode.PACK` — ``video_files`` lists the season
folder's flat videos.
* :attr:`ReleaseMode.EPISODIC` — ``video_files`` lists each
episode subfolder's single video.
* ``None`` — the folder is empty, malformed (mixed layout), or
otherwise unclassifiable. ``video_files`` is empty. The
orchestrator decides whether to warn/skip.
"""
season_dir: Path season_dir: Path
mode: ReleaseMode | None
video_files: tuple[Path, ...] video_files: tuple[Path, ...]
@@ -73,26 +102,107 @@ def walk_show(
* lists direct children of ``show_root``, * lists direct children of ``show_root``,
* keeps the directories whose name contains a ``Sxx`` token, * keeps the directories whose name contains a ``Sxx`` token,
* lists each season directory's video files (one level deep — * classifies each season folder as PACK / EPISODIC / unknown by
no recursion into sub-sub-folders), inspecting its direct children (videos vs subfolders),
* for EPISODIC, descends one extra level into each episode
subfolder to collect its single video,
* sorts season folders by name and video files by name within * sorts season folders by name and video files by name within
each folder. each folder.
Empty / unreadable directories surface as a ``SeasonFolder`` with The walker never raises — empty / unreadable / malformed
an empty ``video_files`` tuple. The orchestrator decides how to directories surface as a ``SeasonFolder`` with ``mode=None`` and
react (skip, warn, fail) — the walker never raises. an empty ``video_files`` tuple.
""" """
video_exts = {ext.lower() for ext in kb.video_extensions} video_exts = {ext.lower() for ext in kb.video_extensions}
season_folders: list[SeasonFolder] = [] season_folders: list[SeasonFolder] = []
for entry in scanner.scan_dir(show_root): for entry in scanner.scan_dir(show_root):
if not entry.is_dir or not _SEASON_TOKEN_RE.search(entry.name): if not entry.is_dir or not _SEASON_TOKEN_RE.search(entry.name):
continue continue
videos = tuple( season_folders.append(
child.path _classify_season(entry.path, scanner=scanner, video_exts=video_exts)
for child in scanner.scan_dir(entry.path)
if child.is_file and child.suffix.lower() in video_exts
) )
season_folders.append(SeasonFolder(season_dir=entry.path, video_files=videos))
return ShowTree( return ShowTree(
show_root=show_root, season_folders=tuple(season_folders) show_root=show_root, season_folders=tuple(season_folders)
) )
# --------------------------------------------------------------------------- #
# Season-folder classification #
# --------------------------------------------------------------------------- #
def _classify_season(
season_dir: Path,
*,
scanner: FilesystemScanner,
video_exts: set[str],
) -> SeasonFolder:
"""Inspect one season folder and decide PACK / EPISODIC / unknown.
Looks only at direct children. For EPISODIC, descends one extra
level into each subfolder to collect its single video. Mixed
layouts (flat videos + subfolders) are reported as ``mode=None``
so the orchestrator can skip them with a warning.
"""
flat_videos: list[Path] = []
subdirs: list[Path] = []
for child in scanner.scan_dir(season_dir):
if child.is_file and child.suffix.lower() in video_exts:
flat_videos.append(child.path)
elif child.is_dir:
subdirs.append(child.path)
# Anything else (non-video files like .nfo, .srt at the season
# root) is ignored — it doesn't affect classification.
has_flat = bool(flat_videos)
has_subdirs = bool(subdirs)
if has_flat and has_subdirs:
_LOG.warning(
"walker: season folder %s mixes flat videos and subfolders — "
"malformed layout, skipping",
season_dir,
)
return SeasonFolder(season_dir=season_dir, mode=None, video_files=())
if has_flat:
return SeasonFolder(
season_dir=season_dir,
mode=ReleaseMode.PACK,
video_files=tuple(sorted(flat_videos)),
)
if has_subdirs:
episode_videos: list[Path] = []
for sub in sorted(subdirs):
videos_in_sub = [
child.path
for child in scanner.scan_dir(sub)
if child.is_file and child.suffix.lower() in video_exts
]
if len(videos_in_sub) == 0:
_LOG.warning(
"walker: episode subfolder %s contains no video — skipping",
sub,
)
continue
if len(videos_in_sub) > 1:
_LOG.warning(
"walker: episode subfolder %s contains %d videos — "
"malformed, skipping season %s",
sub,
len(videos_in_sub),
season_dir,
)
return SeasonFolder(
season_dir=season_dir, mode=None, video_files=()
)
episode_videos.append(videos_in_sub[0])
return SeasonFolder(
season_dir=season_dir,
mode=ReleaseMode.EPISODIC,
video_files=tuple(episode_videos),
)
# No flat videos, no subdirs → empty season folder.
return SeasonFolder(season_dir=season_dir, mode=None, video_files=())
+115 -30
View File
@@ -85,31 +85,41 @@ def _default_info() -> MediaInfo:
def _make_foundation_library( def _make_foundation_library(
root: Path, root: Path,
*, *,
episodic_episodes: tuple[int, ...] = (1, 2, 3), pack_episodes: tuple[int, ...] = (1, 2, 3),
include_pack_s2: bool = True, episodic_episodes: tuple[int, ...] = (1, 2),
) -> Path: ) -> Path:
"""Build a fake Foundation show folder under ``root``. """Build a fake Foundation show folder under ``root``.
Layout:: Layout::
root/Foundation/ root/Foundation/
Foundation.S01.1080p.x265-ELiTE/ ← EPISODIC Foundation.S01.1080p.x265-ELiTE/ ← PACK
Foundation.S01E01.1080p.x265-ELiTE.mkv Foundation.S01E01.1080p.x265-ELiTE.mkv (flat)
Foundation.S01E02.1080p.x265-ELiTE.mkv Foundation.S01E02.1080p.x265-ELiTE.mkv
Foundation.S01E03.1080p.x265-ELiTE.mkv Foundation.S01E03.1080p.x265-ELiTE.mkv
Foundation.S02.1080p.x265-ELiTE/ ← PACK Foundation.S02/ ← EPISODIC
Foundation.S02.1080p.x265-ELiTE.mkv Foundation.S02E01.1080p.x265-ELiTE/ (subfolder)
Foundation.S02E01.1080p.x265-ELiTE.mkv
Foundation.S02E02.1080p.x265-ELiTE/
Foundation.S02E02.1080p.x265-ELiTE.mkv
``pack_episodes`` and ``episodic_episodes`` can each be set to
empty tuples to omit the corresponding season.
""" """
show_root = root / "Foundation" show_root = root / "Foundation"
show_root.mkdir() show_root.mkdir()
if pack_episodes:
s1 = show_root / "Foundation.S01.1080p.x265-ELiTE" s1 = show_root / "Foundation.S01.1080p.x265-ELiTE"
s1.mkdir() s1.mkdir()
for n in episodic_episodes: for n in pack_episodes:
(s1 / f"Foundation.S01E{n:02d}.1080p.x265-ELiTE.mkv").write_bytes(b"") (s1 / f"Foundation.S01E{n:02d}.1080p.x265-ELiTE.mkv").write_bytes(b"")
if include_pack_s2: if episodic_episodes:
s2 = show_root / "Foundation.S02.1080p.x265-ELiTE" s2 = show_root / "Foundation.S02"
s2.mkdir() s2.mkdir()
(s2 / "Foundation.S02.1080p.x265-ELiTE.mkv").write_bytes(b"") for n in episodic_episodes:
sub = s2 / f"Foundation.S02E{n:02d}.1080p.x265-ELiTE"
sub.mkdir()
(sub / f"Foundation.S02E{n:02d}.1080p.x265-ELiTE.mkv").write_bytes(b"")
return show_root return show_root
@@ -127,7 +137,7 @@ def _library_with_show(tmp_path: Path) -> tuple[Path, Path]:
class TestHappyPath: class TestHappyPath:
def test_builds_release_with_episodic_and_pack_seasons(self, tmp_path): def test_builds_release_with_pack_and_episodic_seasons(self, tmp_path):
library, show_root = _library_with_show(tmp_path) library, show_root = _library_with_show(tmp_path)
repo = DotAlfredSeriesReleaseRepository(library) repo = DotAlfredSeriesReleaseRepository(library)
prober = _StubProber() prober = _StubProber()
@@ -150,9 +160,9 @@ class TestHappyPath:
s2 = release.get_season(SeasonNumber(2)) s2 = release.get_season(SeasonNumber(2))
assert s1 is not None and s2 is not None assert s1 is not None and s2 is not None
# S01 EPISODIC: three single-episode releases, each carrying # S01 PACK: three single-episode releases, each carrying its
# its own TrackProfile. # own TrackProfile.
assert s1.mode is ReleaseMode.EPISODIC assert s1.mode is ReleaseMode.PACK
assert s1.episode_count() == 3 assert s1.episode_count() == 3
for ep in s1.episodes: for ep in s1.episodes:
assert ep.episodes.is_single() assert ep.episodes.is_single()
@@ -160,12 +170,12 @@ class TestHappyPath:
assert ep.tracks.audio_tracks[0].language == "eng" assert ep.tracks.audio_tracks[0].language == "eng"
assert len(ep.tracks.subtitle_tracks) == 1 assert len(ep.tracks.subtitle_tracks) == 1
# S02 PACK: empty episodes tuple. Phase 5 (TMDB sync) will # S02 EPISODIC: two single-episode releases, each in its own
# populate the slot map once episode_count is known. # subfolder.
assert s2.mode is ReleaseMode.PACK assert s2.mode is ReleaseMode.EPISODIC
assert s2.episodes == () assert s2.episode_count() == 2
def test_episode_paths_are_relative_to_show_root(self, tmp_path): def test_pack_episode_paths_relative_and_flat(self, tmp_path):
library, show_root = _library_with_show(tmp_path) library, show_root = _library_with_show(tmp_path)
repo = DotAlfredSeriesReleaseRepository(library) repo = DotAlfredSeriesReleaseRepository(library)
release = rescan_show( release = rescan_show(
@@ -181,11 +191,33 @@ class TestHappyPath:
assert s1 is not None assert s1 is not None
for ep in s1.episodes: for ep in s1.episodes:
path_str = str(ep.file_path) path_str = str(ep.file_path)
# Must NOT be absolute and must start with the season folder.
assert not Path(path_str).is_absolute() assert not Path(path_str).is_absolute()
# PACK files sit directly inside the season folder.
assert path_str.startswith("Foundation.S01.1080p.x265-ELiTE/") assert path_str.startswith("Foundation.S01.1080p.x265-ELiTE/")
assert path_str.count("/") == 1
def test_season_folder_name_recorded(self, tmp_path): def test_episodic_episode_paths_descend_into_subfolders(self, tmp_path):
library, show_root = _library_with_show(tmp_path)
repo = DotAlfredSeriesReleaseRepository(library)
release = rescan_show(
show_root,
tmdb_id=_TMDB_ID,
imdb_id=_IMDB_ID,
series_repo=repo,
scanner=_SCANNER,
prober=_StubProber(),
kb=_KB,
)
s2 = release.get_season(SeasonNumber(2))
assert s2 is not None
for ep in s2.episodes:
path_str = str(ep.file_path)
assert not Path(path_str).is_absolute()
# EPISODIC: season/episode_subdir/video.mkv → 2 separators.
assert path_str.startswith("Foundation.S02/")
assert path_str.count("/") == 2
def test_season_folder_names_recorded(self, tmp_path):
library, show_root = _library_with_show(tmp_path) library, show_root = _library_with_show(tmp_path)
repo = DotAlfredSeriesReleaseRepository(library) repo = DotAlfredSeriesReleaseRepository(library)
release = rescan_show( release = rescan_show(
@@ -201,7 +233,7 @@ class TestHappyPath:
s2 = release.get_season(SeasonNumber(2)) s2 = release.get_season(SeasonNumber(2))
assert s1 is not None and s2 is not None assert s1 is not None and s2 is not None
assert s1.folder == "Foundation.S01.1080p.x265-ELiTE" assert s1.folder == "Foundation.S01.1080p.x265-ELiTE"
assert s2.folder == "Foundation.S02.1080p.x265-ELiTE" assert s2.folder == "Foundation.S02"
def test_persists_sidecar_on_disk(self, tmp_path): def test_persists_sidecar_on_disk(self, tmp_path):
library, show_root = _library_with_show(tmp_path) library, show_root = _library_with_show(tmp_path)
@@ -216,7 +248,6 @@ class TestHappyPath:
kb=_KB, kb=_KB,
) )
assert (show_root / SIDECAR_FILENAME).is_file() assert (show_root / SIDECAR_FILENAME).is_file()
# Round-trip via the repo.
recovered = repo.find_by_tmdb_id(_TMDB_ID) recovered = repo.find_by_tmdb_id(_TMDB_ID)
assert recovered is not None assert recovered is not None
assert len(recovered.seasons) == 2 assert len(recovered.seasons) == 2
@@ -234,8 +265,8 @@ class TestHappyPath:
prober=prober, prober=prober,
kb=_KB, kb=_KB,
) )
# 3 episodes + 1 pack video = 4 probes. # 3 PACK files + 2 EPISODIC files = 5 probes.
assert len(prober.calls) == 4 assert len(prober.calls) == 5
def test_imdb_id_optional(self, tmp_path): def test_imdb_id_optional(self, tmp_path):
library, show_root = _library_with_show(tmp_path) library, show_root = _library_with_show(tmp_path)
@@ -274,12 +305,35 @@ class TestEdgeCases:
assert release.seasons == () assert release.seasons == ()
assert (show_root / SIDECAR_FILENAME).is_file() assert (show_root / SIDECAR_FILENAME).is_file()
def test_season_folder_without_videos_is_skipped(self, tmp_path, caplog): def test_empty_season_folder_is_skipped(self, tmp_path):
library = tmp_path / "library" library = tmp_path / "library"
library.mkdir() library.mkdir()
show_root = library / "Foundation" show_root = library / "Foundation"
show_root.mkdir() show_root.mkdir()
(show_root / "Foundation.S01.WEB").mkdir() # empty season folder (show_root / "Foundation.S01.WEB").mkdir() # empty
repo = DotAlfredSeriesReleaseRepository(library)
release = rescan_show(
show_root,
tmdb_id=_TMDB_ID,
series_repo=repo,
scanner=_SCANNER,
prober=_StubProber(),
kb=_KB,
)
assert release.seasons == ()
def test_malformed_mixed_season_is_skipped(self, tmp_path, caplog):
library = tmp_path / "library"
library.mkdir()
show_root = library / "Foundation"
show_root.mkdir()
season = show_root / "Foundation.S01.Mixed"
season.mkdir()
# Flat video + subfolder → malformed.
(season / "Foundation.S01E01.mkv").write_bytes(b"")
sub = season / "Foundation.S01E02-RG"
sub.mkdir()
(sub / "Foundation.S01E02-RG.mkv").write_bytes(b"")
repo = DotAlfredSeriesReleaseRepository(library) repo = DotAlfredSeriesReleaseRepository(library)
with caplog.at_level("WARNING"): with caplog.at_level("WARNING"):
release = rescan_show( release = rescan_show(
@@ -291,16 +345,47 @@ class TestEdgeCases:
kb=_KB, kb=_KB,
) )
assert release.seasons == () assert release.seasons == ()
assert any("no video file" in r.message for r in caplog.records) assert any(
"mixes flat videos and subfolders" in r.message
for r in caplog.records
)
def test_single_unnumbered_pack_file_is_skipped(self, tmp_path, caplog):
# A season folder with a single video whose name has no Exx
# marker: the season number is parsed but the episode is not,
# so the file is skipped and (no other parseable files →)
# the whole season is skipped.
library = tmp_path / "library"
library.mkdir()
show_root = library / "Foundation"
show_root.mkdir()
season = show_root / "Foundation.S01.Complete"
season.mkdir()
(season / "Foundation.S01.Complete.1080p.mkv").write_bytes(b"")
repo = DotAlfredSeriesReleaseRepository(library)
with caplog.at_level("WARNING"):
release = rescan_show(
show_root,
tmdb_id=_TMDB_ID,
series_repo=repo,
scanner=_SCANNER,
prober=_StubProber(),
kb=_KB,
)
assert release.seasons == ()
assert any(
"no parseable episodes" in r.message or
"no episode number parsed" in r.message
for r in caplog.records
)
def test_prober_returning_none_still_produces_episodes(self, tmp_path): def test_prober_returning_none_still_produces_episodes(self, tmp_path):
library = tmp_path / "library" library = tmp_path / "library"
library.mkdir() library.mkdir()
show_root = _make_foundation_library( show_root = _make_foundation_library(
library, episodic_episodes=(1,), include_pack_s2=False library, pack_episodes=(1,), episodic_episodes=()
) )
repo = DotAlfredSeriesReleaseRepository(library) repo = DotAlfredSeriesReleaseRepository(library)
# Prober returns None — inspect_release skips enrichment, tracks empty.
release = rescan_show( release = rescan_show(
show_root, show_root,
tmdb_id=_TMDB_ID, tmdb_id=_TMDB_ID,
+138 -24
View File
@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
from alfred.application.tv_shows.walker import walk_show from alfred.application.tv_shows.walker import walk_show
from alfred.domain.releases.value_objects import ReleaseMode
from alfred.infrastructure.filesystem.scanner import PathlibFilesystemScanner from alfred.infrastructure.filesystem.scanner import PathlibFilesystemScanner
from alfred.infrastructure.knowledge.release_kb import YamlReleaseKnowledge from alfred.infrastructure.knowledge.release_kb import YamlReleaseKnowledge
@@ -10,10 +11,15 @@ _KB = YamlReleaseKnowledge()
_SCANNER = PathlibFilesystemScanner() _SCANNER = PathlibFilesystemScanner()
def _episode(dir_path, name: str) -> None: def _file(dir_path, name: str) -> None:
(dir_path / name).write_bytes(b"") (dir_path / name).write_bytes(b"")
# ════════════════════════════════════════════════════════════════════════════
# Season-folder detection (by Sxx token in folder name)
# ════════════════════════════════════════════════════════════════════════════
class TestSeasonFolderDetection: class TestSeasonFolderDetection:
def test_keeps_folders_with_season_token(self, tmp_path): def test_keeps_folders_with_season_token(self, tmp_path):
show = tmp_path / "Foundation" show = tmp_path / "Foundation"
@@ -66,54 +72,162 @@ class TestSeasonFolderDetection:
] ]
class TestVideoFileCollection: # ════════════════════════════════════════════════════════════════════════════
def test_keeps_only_video_extensions(self, tmp_path): # PACK classification (flat numbered videos inside season folder)
# ════════════════════════════════════════════════════════════════════════════
class TestPackClassification:
def test_flat_videos_yield_pack(self, tmp_path):
show = tmp_path / "Foundation"
show.mkdir()
season = show / "Foundation.S01.1080p.WEB-DL.x265-GROUP"
season.mkdir()
_file(season, "Foundation.S01E01.1080p.WEB-DL.x265-GROUP.mkv")
_file(season, "Foundation.S01E02.1080p.WEB-DL.x265-GROUP.mkv")
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert folder.mode is ReleaseMode.PACK
assert [p.name for p in folder.video_files] == [
"Foundation.S01E01.1080p.WEB-DL.x265-GROUP.mkv",
"Foundation.S01E02.1080p.WEB-DL.x265-GROUP.mkv",
]
def test_pack_keeps_only_video_extensions(self, tmp_path):
show = tmp_path / "Foundation" show = tmp_path / "Foundation"
show.mkdir() show.mkdir()
season = show / "Foundation.S01.WEB" season = show / "Foundation.S01.WEB"
season.mkdir() season.mkdir()
_episode(season, "Foundation.S01E01.mkv") _file(season, "Foundation.S01E01.mkv")
_episode(season, "Foundation.S01E02.mp4") _file(season, "Foundation.S01E02.mp4")
_episode(season, "Foundation.S01E01.eng.srt") # not a video _file(season, "Foundation.S01E01.eng.srt") # not a video
_episode(season, "info.nfo") # not a video _file(season, "info.nfo") # not a video
tree = walk_show(show, scanner=_SCANNER, kb=_KB) tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders (folder,) = tree.season_folders
assert folder.mode is ReleaseMode.PACK
suffixes = sorted(p.suffix for p in folder.video_files) suffixes = sorted(p.suffix for p in folder.video_files)
assert suffixes == [".mkv", ".mp4"] assert suffixes == [".mkv", ".mp4"]
def test_empty_season_folder_surfaces(self, tmp_path):
# ════════════════════════════════════════════════════════════════════════════
# EPISODIC classification (subfolders, one video each)
# ════════════════════════════════════════════════════════════════════════════
class TestEpisodicClassification:
def test_subfolders_yield_episodic(self, tmp_path):
show = tmp_path / "Foundation"
show.mkdir()
season = show / "Foundation.S02"
season.mkdir()
for n in (1, 2):
sub = season / f"Foundation.S02E{n:02d}.1080p.x265-GROUP"
sub.mkdir()
_file(sub, f"Foundation.S02E{n:02d}.1080p.x265-GROUP.mkv")
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert folder.mode is ReleaseMode.EPISODIC
assert [p.name for p in folder.video_files] == [
"Foundation.S02E01.1080p.x265-GROUP.mkv",
"Foundation.S02E02.1080p.x265-GROUP.mkv",
]
def test_episodic_paths_descend_into_subfolders(self, tmp_path):
show = tmp_path / "Foundation"
show.mkdir()
season = show / "Foundation.S02"
season.mkdir()
sub = season / "Foundation.S02E01-RG"
sub.mkdir()
_file(sub, "Foundation.S02E01-RG.mkv")
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert folder.video_files == (sub / "Foundation.S02E01-RG.mkv",)
def test_episodic_skips_empty_subfolder(self, tmp_path, caplog):
show = tmp_path / "Foundation"
show.mkdir()
season = show / "Foundation.S02"
season.mkdir()
(season / "Foundation.S02E01-RG").mkdir() # empty subfolder
good_sub = season / "Foundation.S02E02-RG"
good_sub.mkdir()
_file(good_sub, "Foundation.S02E02-RG.mkv")
with caplog.at_level("WARNING"):
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert folder.mode is ReleaseMode.EPISODIC
assert len(folder.video_files) == 1
assert any("no video" in r.message for r in caplog.records)
def test_episodic_with_two_videos_in_subfolder_skips_season(
self, tmp_path, caplog
):
show = tmp_path / "Foundation"
show.mkdir()
season = show / "Foundation.S02"
season.mkdir()
sub = season / "Foundation.S02E01-RG"
sub.mkdir()
_file(sub, "Foundation.S02E01-RG.mkv")
_file(sub, "Foundation.S02E01-RG-extras.mkv") # second video!
with caplog.at_level("WARNING"):
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert folder.mode is None
assert folder.video_files == ()
assert any("malformed" in r.message for r in caplog.records)
# ════════════════════════════════════════════════════════════════════════════
# Malformed and empty seasons
# ════════════════════════════════════════════════════════════════════════════
class TestMalformedAndEmpty:
def test_mixed_flat_and_subfolders_is_malformed(self, tmp_path, caplog):
show = tmp_path / "Foundation"
show.mkdir()
season = show / "Foundation.S01.Mixed"
season.mkdir()
_file(season, "Foundation.S01E01.mkv") # flat video
sub = season / "Foundation.S01E02-RG" # AND a subfolder
sub.mkdir()
_file(sub, "Foundation.S01E02-RG.mkv")
with caplog.at_level("WARNING"):
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert folder.mode is None
assert folder.video_files == ()
assert any("mixes flat videos and subfolders" in r.message for r in caplog.records)
def test_empty_season_folder_yields_unknown_mode(self, tmp_path):
show = tmp_path / "Foundation" show = tmp_path / "Foundation"
show.mkdir() show.mkdir()
(show / "Foundation.S01.WEB").mkdir() (show / "Foundation.S01.WEB").mkdir()
tree = walk_show(show, scanner=_SCANNER, kb=_KB) tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders (folder,) = tree.season_folders
assert folder.mode is None
assert folder.video_files == () assert folder.video_files == ()
def test_single_pack_video_is_kept(self, tmp_path): def test_season_with_only_non_video_files_is_unknown(self, tmp_path):
show = tmp_path / "Foundation" # No videos, no subfolders → unknown. The .nfo doesn't count.
show.mkdir()
season = show / "Foundation.S01.Complete.WEB"
season.mkdir()
_episode(season, "Foundation.S01.Complete.WEB.mkv")
tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders
assert [p.name for p in folder.video_files] == [
"Foundation.S01.Complete.WEB.mkv"
]
def test_does_not_recurse_into_sub_subfolders(self, tmp_path):
show = tmp_path / "Foundation" show = tmp_path / "Foundation"
show.mkdir() show.mkdir()
season = show / "Foundation.S01.WEB" season = show / "Foundation.S01.WEB"
season.mkdir() season.mkdir()
nested = season / "Extras" _file(season, "info.nfo")
nested.mkdir()
_episode(nested, "Foundation.S01E01.mkv")
tree = walk_show(show, scanner=_SCANNER, kb=_KB) tree = walk_show(show, scanner=_SCANNER, kb=_KB)
(folder,) = tree.season_folders (folder,) = tree.season_folders
assert folder.mode is None
assert folder.video_files == () assert folder.video_files == ()
# ════════════════════════════════════════════════════════════════════════════
# Edge cases on the show root itself
# ════════════════════════════════════════════════════════════════════════════
class TestEdgeCases: class TestEdgeCases:
def test_empty_show_root(self, tmp_path): def test_empty_show_root(self, tmp_path):
show = tmp_path / "Empty" show = tmp_path / "Empty"