francwa
9556bf9e08
DDD-pure cleanup — entities and value objects no longer query the world
at read time.
FilePath: drop .exists() / .is_file() / .is_dir(). The VO is now a
pure address; ask the injected FilesystemScanner for live state.
Movie: drop .has_file() / .is_downloaded(). Invariant: when the
application sets file_path, it has already constated the file
exists; downstream readers trust the snapshot.
Episode: same — drop .has_file() / .is_downloaded().
SubtitlePlacer: drop the pre-check .exists() calls. The placer now
attempts os.link() and reports FileNotFoundError / FileExistsError
as skip reasons. Removes a TOCTOU race as a bonus.
Tests adjusted: the FilePath VO method tests are gone (the methods are
gone), test_has_file_false_when_no_path replaced by a plain assertion
on file_path is None. Placer tests are unchanged — the skip-reason
strings ('not found', 'already exists') match the new try/except paths.
The 'snapshot value objects' pattern (ProbedMediaInfo, TmdbMovieInfo)
that this cleanup enables is documented in refactor_domain_io.md, to
be applied when a future use case actually needs richer metadata —
not now, no speculative VOs.
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
"""SubtitlePlacer — hard-links matched subtitle tracks next to the destination video."""
|
|
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from ..entities import SubtitleCandidate
|
|
from ..value_objects import SubtitleType
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _build_dest_name(track: SubtitleCandidate, video_stem: str) -> str:
|
|
"""
|
|
Build the destination filename for a subtitle track.
|
|
|
|
Format: {video_stem}.{lang}.{ext}
|
|
{video_stem}.{lang}.sdh.{ext}
|
|
{video_stem}.{lang}.forced.{ext}
|
|
"""
|
|
if not track.language or not track.format:
|
|
raise ValueError("Cannot compute destination name: language or format missing")
|
|
|
|
ext = track.format.extensions[0].lstrip(".")
|
|
parts = [video_stem, track.language.code]
|
|
if track.subtitle_type == SubtitleType.SDH:
|
|
parts.append("sdh")
|
|
elif track.subtitle_type == SubtitleType.FORCED:
|
|
parts.append("forced")
|
|
return ".".join(parts) + "." + ext
|
|
|
|
|
|
@dataclass
|
|
class PlacedTrack:
|
|
source: Path
|
|
destination: Path
|
|
filename: str
|
|
|
|
|
|
@dataclass
|
|
class PlaceResult:
|
|
placed: list[PlacedTrack]
|
|
skipped: list[tuple[SubtitleCandidate, str]] # (track, reason)
|
|
|
|
@property
|
|
def placed_count(self) -> int:
|
|
return len(self.placed)
|
|
|
|
@property
|
|
def skipped_count(self) -> int:
|
|
return len(self.skipped)
|
|
|
|
|
|
class SubtitlePlacer:
|
|
"""
|
|
Hard-links matched SubtitleCandidate files next to a destination video.
|
|
|
|
Uses the same hard-link strategy as FileManager.copy_file:
|
|
instant, no data duplication, qBittorrent keeps seeding.
|
|
|
|
Embedded tracks are skipped — nothing to place on disk.
|
|
"""
|
|
|
|
def place(
|
|
self,
|
|
tracks: list[SubtitleCandidate],
|
|
destination_video: Path,
|
|
) -> PlaceResult:
|
|
placed: list[PlacedTrack] = []
|
|
skipped: list[tuple[SubtitleCandidate, str]] = []
|
|
|
|
dest_dir = destination_video.parent
|
|
|
|
for track in tracks:
|
|
if track.is_embedded:
|
|
logger.debug(f"SubtitlePlacer: skip embedded track ({track.language})")
|
|
skipped.append((track, "embedded — no file to place"))
|
|
continue
|
|
|
|
if not track.file_path:
|
|
skipped.append((track, "source file not set"))
|
|
continue
|
|
|
|
try:
|
|
dest_name = _build_dest_name(track, destination_video.stem)
|
|
except ValueError as e:
|
|
skipped.append((track, str(e)))
|
|
continue
|
|
|
|
dest_path = dest_dir / dest_name
|
|
|
|
try:
|
|
os.link(track.file_path, dest_path)
|
|
placed.append(
|
|
PlacedTrack(
|
|
source=track.file_path,
|
|
destination=dest_path,
|
|
filename=dest_name,
|
|
)
|
|
)
|
|
logger.info(f"SubtitlePlacer: placed {dest_name}")
|
|
except FileNotFoundError:
|
|
skipped.append((track, "source file not found"))
|
|
except FileExistsError:
|
|
logger.debug(f"SubtitlePlacer: skip {dest_name} — already exists")
|
|
skipped.append((track, "destination already exists"))
|
|
except OSError as e:
|
|
logger.warning(f"SubtitlePlacer: failed to place {dest_name}: {e}")
|
|
skipped.append((track, str(e)))
|
|
|
|
logger.info(
|
|
f"SubtitlePlacer: {len(placed)} placed, {len(skipped)} skipped "
|
|
f"for {destination_video.name}"
|
|
)
|
|
return PlaceResult(placed=placed, skipped=skipped)
|