francwa/alfred
1
0
Fork 0
Code Issues 1 Pull Requests 8 Actions Packages Projects Releases Wiki Activity

refactor(domain): identity-based equality + dedup track helpers

Browse Source 
Two related DDD fixes for Movie and Episode entities:

- Identity equality: @dataclass(eq=False) with custom __eq__/__hash__.
  Movie is identified by imdb_id, Episode by (season, episode) within
  the TVShow aggregate. Auto-generated field-by-field equality was
  incorrectly making two Movie instances with the same imdb_id but
  different audio_tracks compare unequal — breaks dedup/caching.

- MediaWithTracks mixin: the 5 audio/subtitle helpers
  (has_audio_in / audio_languages / has_subtitles_in / has_forced_subs /
  subtitle_languages) were duplicated verbatim between Movie and Episode.
  Extracted to shared/media/tracks_mixin.py; both entities now inherit.

Bonus: dropped the object.__setattr__ coercion dance in Movie.__post_init__
— the class isn't frozen so plain assignment is the right call.
This commit is contained in:
Francwa
2026-05-19 14:17:47 +02:00
parent 7cd24f3a31
commit 481eeb5afd
4 changed files with 120 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alfred/domain/movies/entities.py
+20 -41
View File
@@ -3,13 +3,13 @@
from dataclasses import dataclass, field
from datetime import datetime
from ..shared.media import AudioTrack, SubtitleTrack, track_lang_matches
from ..shared.value_objects import FilePath, FileSize, ImdbId, Language
from ..shared.media import AudioTrack, MediaWithTracks, SubtitleTrack
from ..shared.value_objects import FilePath, FileSize, ImdbId
from .value_objects import MovieTitle, Quality, ReleaseYear
@dataclass
class Movie:
@dataclass(eq=False)
class Movie(MediaWithTracks):
"""
Movie aggregate root for the movies domain.
@@ -20,6 +20,10 @@ class Movie:
Track helpers follow the same "C+" contract as ``Episode``: pass a
``Language`` for cross-format matching, or a ``str`` for case-insensitive
direct comparison.
Equality is identity-based: two ``Movie`` instances are equal iff they
share the same ``imdb_id``, regardless of file/track contents. This is
the DDD aggregate invariant — the aggregate is identified by its root id.
"""
imdb_id: ImdbId
@@ -38,7 +42,7 @@ class Movie:
# Ensure ImdbId is actually an ImdbId instance
if not isinstance(self.imdb_id, ImdbId):
if isinstance(self.imdb_id, str):
object.__setattr__(self, "imdb_id", ImdbId(self.imdb_id))
self.imdb_id = ImdbId(self.imdb_id)
else:
raise ValueError(
f"imdb_id must be ImdbId or str, got {type(self.imdb_id)}"
@@ -47,12 +51,20 @@ class Movie:
# Ensure MovieTitle is actually a MovieTitle instance
if not isinstance(self.title, MovieTitle):
if isinstance(self.title, str):
object.__setattr__(self, "title", MovieTitle(self.title))
self.title = MovieTitle(self.title)
else:
raise ValueError(
f"title must be MovieTitle or str, got {type(self.title)}"
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Movie):
return NotImplemented
return self.imdb_id == other.imdb_id
def __hash__(self) -> int:
return hash(self.imdb_id)
def has_file(self) -> bool:
"""Check if the movie has an associated file."""
return self.file_path is not None and self.file_path.exists()
@@ -61,41 +73,8 @@ class Movie:
"""Check if the movie is downloaded (has a file)."""
return self.has_file()
# ── Audio helpers ──────────────────────────────────────────────────────
def has_audio_in(self, lang: str | Language) -> bool:
"""True if at least one audio track is in the given language."""
return any(track_lang_matches(t.language, lang) for t in self.audio_tracks)
def audio_languages(self) -> list[str]:
"""Unique audio languages across all tracks, in track order."""
seen: set[str] = set()
result: list[str] = []
for t in self.audio_tracks:
if t.language and t.language not in seen:
seen.add(t.language)
result.append(t.language)
return result
# ── Subtitle helpers ───────────────────────────────────────────────────
def has_subtitles_in(self, lang: str | Language) -> bool:
"""True if at least one subtitle track is in the given language."""
return any(track_lang_matches(t.language, lang) for t in self.subtitle_tracks)
def has_forced_subs(self) -> bool:
"""True if at least one subtitle track is flagged as forced."""
return any(t.is_forced for t in self.subtitle_tracks)
def subtitle_languages(self) -> list[str]:
"""Unique subtitle languages across all tracks, in track order."""
seen: set[str] = set()
result: list[str] = []
for t in self.subtitle_tracks:
if t.language and t.language not in seen:
seen.add(t.language)
result.append(t.language)
return result
# Track helpers (has_audio_in / audio_languages / has_subtitles_in /
# has_forced_subs / subtitle_languages) come from MediaWithTracks.
def get_folder_name(self) -> str:
"""
alfred/domain/shared/media/__init__.py
+2
View File
@@ -8,11 +8,13 @@ from .audio import AudioTrack
from .info import MediaInfo
from .matching import track_lang_matches
from .subtitle import SubtitleTrack
from .tracks_mixin import MediaWithTracks
from .video import VideoTrack
__all__ = [
"AudioTrack",
"MediaInfo",
"MediaWithTracks",
"SubtitleTrack",
"VideoTrack",
"track_lang_matches",
alfred/domain/shared/media/tracks_mixin.py
+77
View File
@@ -0,0 +1,77 @@
"""Mixin shared by entities that carry audio + subtitle tracks.
Both ``Movie`` and ``Episode`` carry a ``list[AudioTrack]`` plus a
``list[SubtitleTrack]`` and answer the same 5 queries about them (language
presence, unique languages, forced flag). Keep that behavior in one place so a
fix in one is a fix in both.
The mixin is plain Python (no dataclass machinery) so it composes cleanly with
``@dataclass`` entities — it only reads ``self.audio_tracks`` and
``self.subtitle_tracks`` which the host class provides as fields.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from ..value_objects import Language
from .matching import track_lang_matches
if TYPE_CHECKING:
from .audio import AudioTrack
from .subtitle import SubtitleTrack
class MediaWithTracks:
"""
Mixin providing audio/subtitle helpers for entities with track collections.
Hosts must expose two attributes:
* ``audio_tracks: list[AudioTrack]``
* ``subtitle_tracks: list[SubtitleTrack]``
The helpers follow the "C+" matching contract: pass a :class:`Language`
for cross-format matching, or a ``str`` for case-insensitive comparison.
"""
# These attributes are provided by the host entity (Movie, Episode, …).
# Declared here only for type-checkers and to make the contract explicit.
audio_tracks: list["AudioTrack"]
subtitle_tracks: list["SubtitleTrack"]
# ── Audio helpers ──────────────────────────────────────────────────────
def has_audio_in(self, lang: str | Language) -> bool:
"""True if at least one audio track is in the given language."""
return any(track_lang_matches(t.language, lang) for t in self.audio_tracks)
def audio_languages(self) -> list[str]:
"""Unique audio languages across all tracks, in track order."""
seen: set[str] = set()
result: list[str] = []
for t in self.audio_tracks:
if t.language and t.language not in seen:
seen.add(t.language)
result.append(t.language)
return result
# ── Subtitle helpers ───────────────────────────────────────────────────
def has_subtitles_in(self, lang: str | Language) -> bool:
"""True if at least one subtitle track is in the given language."""
return any(track_lang_matches(t.language, lang) for t in self.subtitle_tracks)
def has_forced_subs(self) -> bool:
"""True if at least one subtitle track is flagged as forced."""
return any(t.is_forced for t in self.subtitle_tracks)
def subtitle_languages(self) -> list[str]:
"""Unique subtitle languages across all tracks, in track order."""
seen: set[str] = set()
result: list[str] = []
for t in self.subtitle_tracks:
if t.language and t.language not in seen:
seen.add(t.language)
result.append(t.language)
return result
alfred/domain/tv_shows/entities.py
+21 -39
View File
@@ -28,12 +28,11 @@ from __future__ import annotations
import re
from dataclasses import dataclass, field
from ..shared.media import AudioTrack, SubtitleTrack, track_lang_matches
from ..shared.media import AudioTrack, MediaWithTracks, SubtitleTrack
from ..shared.value_objects import (
FilePath,
FileSize,
ImdbId,
Language,
to_dot_folder_name,
)
from .value_objects import (
@@ -48,8 +47,8 @@ from .value_objects import (
# ════════════════════════════════════════════════════════════════════════════
@dataclass
class Episode:
@dataclass(eq=False)
class Episode(MediaWithTracks):
"""
A single episode of a TV show — leaf of the TVShow aggregate.
@@ -57,6 +56,11 @@ class Episode:
(audio + subtitle). Track lists are populated by the ffprobe + subtitle
scan pipeline; they may be empty when the episode is known but not yet
scanned, or when no file is downloaded yet.
Equality is identity-based within the aggregate: two ``Episode`` instances
are equal iff they share the same ``(season_number, episode_number)``,
regardless of title/file/track contents. The root TVShow guarantees
cross-show uniqueness.
"""
season_number: SeasonNumber
@@ -76,6 +80,17 @@ class Episode:
if isinstance(self.episode_number, int):
self.episode_number = EpisodeNumber(self.episode_number)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Episode):
return NotImplemented
return (
self.season_number == other.season_number
and self.episode_number == other.episode_number
)
def __hash__(self) -> int:
return hash((self.season_number, self.episode_number))
# ── File presence ──────────────────────────────────────────────────────
def has_file(self) -> bool:
@@ -86,41 +101,8 @@ class Episode:
"""Alias of ``has_file()`` — reads better in collection-status contexts."""
return self.has_file()
# ── Audio helpers ──────────────────────────────────────────────────────
def has_audio_in(self, lang: str | Language) -> bool:
"""True if at least one audio track is in the given language."""
return any(track_lang_matches(t.language, lang) for t in self.audio_tracks)
def audio_languages(self) -> list[str]:
"""Unique audio languages across all tracks, in track order."""
seen: set[str] = set()
result: list[str] = []
for t in self.audio_tracks:
if t.language and t.language not in seen:
seen.add(t.language)
result.append(t.language)
return result
# ── Subtitle helpers ───────────────────────────────────────────────────
def has_subtitles_in(self, lang: str | Language) -> bool:
"""True if at least one subtitle track is in the given language."""
return any(track_lang_matches(t.language, lang) for t in self.subtitle_tracks)
def has_forced_subs(self) -> bool:
"""True if at least one subtitle track is flagged as forced."""
return any(t.is_forced for t in self.subtitle_tracks)
def subtitle_languages(self) -> list[str]:
"""Unique subtitle languages across all tracks, in track order."""
seen: set[str] = set()
result: list[str] = []
for t in self.subtitle_tracks:
if t.language and t.language not in seen:
seen.add(t.language)
result.append(t.language)
return result
# Track helpers (has_audio_in / audio_languages / has_subtitles_in /
# has_forced_subs / subtitle_languages) come from MediaWithTracks.
# ── Naming ─────────────────────────────────────────────────────────────