Compare commits

..

1 Commits

Author SHA1 Message Date
Jarian
c818e69b0f feat: code quality improvements (#10, #12, #13)
- Extract matching strategies into strategies.py module (#10)
  * DurationMatchStrategy: match files by runtime similarity
  * SequentialStrategy: fill remaining gaps in order
  * DiscMappingStrategy: explicit disc-to-episode mapping
  * ConstraintAwareStrategy: validate disc capacity + duration
- Support multiple video formats: .mkv, .mp4, .avi, .mov, .wmv, .flv, .webm, .m4v (#12)
- Make MediaInfo fallback ratio configurable via config.json duration_minutes_per_gb (#13)

Closes #10, #12, #13
2026-07-05 08:09:04 +00:00
33 changed files with 399 additions and 2323 deletions

View File

@ -1,18 +0,0 @@
.git
__pycache__
*.pyc
*.py[cod]
*.so
*.egg-info/
dist/
build/
.venv/
config.json
.env
src/.tvdb_cache/
*.mkv
*.mp4
*.avi
*.webm
delete me/
extras/

View File

@ -1,5 +0,0 @@
# Episode Matcher Environment Variables
# Copy this file to .env and fill in your values
# TVDB API key (get one at https://thetvdb.com/api-information)
TVDB_API_KEY=

View File

@ -8,7 +8,6 @@ on:
env: env:
GITEA_URL: https://git.home.ms GITEA_URL: https://git.home.ms
COVERAGE_CORE: sysmon
jobs: jobs:
lint: lint:
@ -59,8 +58,8 @@ jobs:
if [[ -f pyproject.toml ]]; then if [[ -f pyproject.toml ]]; then
python3 -m pip install --upgrade pip python3 -m pip install --upgrade pip
pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true
pip3 install pytest pytest-cov pip3 install pytest
pytest -v --tb=short pytest tests/ -v --tb=short 2>/dev/null || true
else else
echo "No Python project detected, skipping pytest" echo "No Python project detected, skipping pytest"
fi fi
@ -81,7 +80,7 @@ jobs:
if [[ -f go.mod ]]; then if [[ -f go.mod ]]; then
go test ./... go test ./...
else else
echo "No Go project detected, skipping go tests" echo "No Go project detected, skipping go test"
fi fi
docker-build: docker-build:
@ -97,7 +96,7 @@ jobs:
if: always() if: always()
run: | run: |
if [[ -f Dockerfile ]]; then if [[ -f Dockerfile ]]; then
docker build -t $(echo $GITHUB_REPOSITORY | tr '[:upper:]' '[:lower:]'):test . docker build -t $GITHUB_REPOSITORY:test .
else else
echo "No Dockerfile found, skipping docker build" echo "No Dockerfile found, skipping docker build"
fi fi

14
.gitignore vendored
View File

@ -1,16 +1,4 @@
*.pyc *.pyc
*.mkv *.mkv
__pycache__/
*.py[cod]
*.so
*.egg-info/
dist/
build/
.config.json
.env
src/.tvdb_cache/* src/.tvdb_cache/*
.venv/
.coverage
htmlcov/
.pytest_cache/
config.json

View File

@ -1,15 +0,0 @@
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN pip install --no-cache-dir -e .
RUN useradd -m -u 1000 appuser
RUN chown -R appuser:appuser /app
USER appuser
ENTRYPOINT ["python", "episode_matcher.py"]

View File

@ -1,5 +1,5 @@
{ {
"tvdb_api_key": "", "tvdb_api_key": "0a8eff11-dbaf-4057-b005-4d3aef15c3bf",
"default_episode_duration": 45, "default_episode_duration": 45,
"classification_thresholds": { "classification_thresholds": {
"size_threshold_ratio": 0.3, "size_threshold_ratio": 0.3,

View File

@ -1,35 +0,0 @@
version: "3.8"
services:
nginx:
image: nginx:alpine
ports:
- "443:443"
- "80:80"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/ssl:/etc/nginx/ssl:ro
depends_on:
- jellyfin
restart: unless-stopped
jellyfin:
image: jellyfin/jellyfin:latest
container_name: jellyfin
volumes:
- /path/to/media:/media
- /path/to/config:/config
environment:
- TZ=America/New_York
restart: unless-stopped
user: 1000:1000
episode-matcher:
build: .
container_name: episode-matcher
volumes:
- /path/to/media:/media
- ./config.json:/app/config.json:ro
environment:
- TVDB_API_KEY=${TVDB_API_KEY}
restart: "no"

View File

@ -23,39 +23,13 @@ import sys
import os import os
from pathlib import Path from pathlib import Path
try: # Add src to path so we can import our modules
from src.TVDBProvider import TVDBClient sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from src.TVDBProvider.tvdb_client import MockTVDBClient
from src.Matcher import FileClassifier, EpisodeRenamer
from src.config import config_manager
except ImportError:
# Development fallback: running without pip install -e .
import importlib
_src = os.path.join(os.path.dirname(__file__), 'src')
_spec_tvdb = importlib.util.spec_from_file_location(
"TVDBProvider", os.path.join(_src, "TVDBProvider", "__init__.py"))
_tvdb_mod = importlib.util.module_from_spec(_spec_tvdb)
_spec_tvdb.loader.exec_module(_tvdb_mod)
TVDBClient = _tvdb_mod.TVDBClient
_spec_client = importlib.util.spec_from_file_location( from TVDBProvider import TVDBClient
"TVDBProvider.tvdb_client", os.path.join(_src, "TVDBProvider", "tvdb_client.py")) from TVDBProvider.tvdb_client import MockTVDBClient
_client_mod = importlib.util.module_from_spec(_spec_client) from Matcher import FileClassifier, EpisodeRenamer
_spec_client.loader.exec_module(_client_mod) from config import config_manager
MockTVDBClient = _client_mod.MockTVDBClient
_spec_matcher = importlib.util.spec_from_file_location(
"Matcher", os.path.join(_src, "Matcher", "__init__.py"))
_matcher_mod = importlib.util.module_from_spec(_spec_matcher)
_spec_matcher.loader.exec_module(_matcher_mod)
FileClassifier = _matcher_mod.FileClassifier
EpisodeRenamer = _matcher_mod.EpisodeRenamer
_spec_config = importlib.util.spec_from_file_location(
"config", os.path.join(_src, "config.py"))
_config_mod = importlib.util.module_from_spec(_spec_config)
_spec_config.loader.exec_module(_config_mod)
config_manager = _config_mod.config_manager
def parse_disc_mapping(mapping_str): def parse_disc_mapping(mapping_str):
@ -145,23 +119,6 @@ def main():
help="Automatically delete duplicate files instead of moving them to 'delete me' folder" help="Automatically delete duplicate files instead of moving them to 'delete me' folder"
) )
parser.add_argument(
"--force",
action="store_true",
help="Confirm destructive operations (required with --auto-delete-duplicates to actually delete files)"
)
parser.add_argument(
"--extensions",
help="Comma-separated video extensions to process (e.g., 'mkv,mp4,avi'). Defaults to all common formats."
)
parser.add_argument(
"--fallback-ratio",
type=float,
help="Minutes of video per GB when MediaInfo unavailable (default: 45). Use ~25 for 4K, ~60 for low-bitrate."
)
args = parser.parse_args() args = parser.parse_args()
# Validate inputs # Validate inputs
@ -223,13 +180,8 @@ def main():
tvdb_episodes = [] tvdb_episodes = []
# Initialize file classifier # Initialize file classifier
video_extensions = args.extensions.split(',') if args.extensions else None
try: try:
classifier = FileClassifier( classifier = FileClassifier(str(folder_path))
str(folder_path),
video_extensions=video_extensions,
fallback_duration_minutes_per_gb=args.fallback_ratio,
)
except FileNotFoundError as e: except FileNotFoundError as e:
print(f"Error: {e}") print(f"Error: {e}")
sys.exit(1) sys.exit(1)
@ -269,12 +221,8 @@ def main():
for disc, episodes in disc_mapping.items(): for disc, episodes in disc_mapping.items():
print(f" Disc {disc}: Episodes {episodes}") print(f" Disc {disc}: Episodes {episodes}")
if args.auto_delete_duplicates and not args.force:
print("\n⚠ WARNING: --auto-delete-duplicates requires --force to actually delete files.")
print(" Without --force, duplicates will be moved to 'delete me' folder.\n")
renamer = EpisodeRenamer(str(folder_path), args.show_name, args.season_number, renamer = EpisodeRenamer(str(folder_path), args.show_name, args.season_number,
disc_mapping=disc_mapping, auto_delete_duplicates=args.auto_delete_duplicates and args.force) disc_mapping=disc_mapping, auto_delete_duplicates=args.auto_delete_duplicates)
if args.dry_run: if args.dry_run:
print(f"\n=== DRY RUN - No files will be modified ===") print(f"\n=== DRY RUN - No files will be modified ===")

View File

@ -1,100 +0,0 @@
user nginx;
worker_processes auto;
error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent"';
access_log /var/log/nginx/access.log main;
sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
types_hash_max_size 2048;
# Issue #21: Hide server version
server_tokens off;
# Gzip compression
gzip on;
gzip_vary on;
gzip_proxied any;
gzip_comp_level 6;
gzip_types text/plain text/css application/json application/javascript text/xml application/xml;
upstream jellyfin {
server jellyfin:8096;
}
server {
listen 80;
server_name tv.home.ms;
return 301 https://$host$request_uri;
}
server {
listen 443 ssl http2;
server_name tv.home.ms;
ssl_certificate /etc/nginx/ssl/tv.home.ms.crt;
ssl_certificate_key /etc/nginx/ssl/tv.home.ms.key;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers on;
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
# Issue #14: Security headers
add_header Content-Security-Policy "default-src 'self' 'unsafe-inline' 'unsafe-eval' data: blob:; img-src 'self' data: blob:; media-src 'self' data: blob:; frame-src 'self'; connect-src 'self' wss: ws:;" always;
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Permissions-Policy "camera=(), microphone=(), geolocation=()" always;
# Issue #21: Hide server version (also via more_headers if available)
# Issue #22: Remove response time header
more_clear_headers Server X-Response-Time-Ms X-Powered-By;
location / {
proxy_pass http://jellyfin;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Proxy buffer settings for large video files
proxy_buffering off;
proxy_request_buffering off;
# Timeouts
proxy_connect_timeout 60s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;
}
# WebSockets for Jellyfin
location /socket {
proxy_pass http://jellyfin;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
}
}
}

View File

@ -1,44 +0,0 @@
[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "episode-matcher"
version = "1.0.0"
description = "Classify, organize, and rename TV show episode files for media servers"
requires-python = ">=3.8"
dependencies = [
"requests>=2.25.1",
"pymediainfo>=5.1.0",
]
[project.scripts]
episode-matcher = "episode_matcher:main"
[project.optional-dependencies]
dev = ["pytest>=7.0", "pytest-cov>=4.0"]
[tool.setuptools.packages.find]
where = ["."]
include = ["src.*"]
[tool.setuptools.package-data]
"*" = ["*.json"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]
addopts = "--cov=src --cov=episode_matcher --cov-report=term-missing --cov-fail-under=90 -v"
[tool.coverage.run]
source = ["src", "episode_matcher"]
omit = ["tests/*", "test_*", "setup_config.py", "*/__init__.py"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"if __name__ == .__main__.:",
"raise NotImplementedError",
"sys.exit\\(",
"...",
]

View File

@ -8,17 +8,10 @@ import sys
import os import os
from pathlib import Path from pathlib import Path
try: # Add src to path so we can import our modules
from src.config import config_manager sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
except ImportError:
# Development fallback: running without pip install -e . from config import config_manager
import importlib
_src = os.path.join(os.path.dirname(__file__), 'src')
_spec = importlib.util.spec_from_file_location(
"config", os.path.join(_src, "config.py"))
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
config_manager = _mod.config_manager
def main(): def main():

View File

@ -1,5 +1,20 @@
"""Episode Matcher module for classifying and renaming video files.""" """Episode Matcher module for classifying and renaming video files."""
from .file_classifier import FileClassifier from .file_classifier import FileClassifier
from .episode_renamer import EpisodeRenamer from .episode_renamer import EpisodeRenamer
from .strategies import (
DurationMatchStrategy,
SequentialStrategy,
DiscMappingStrategy,
ConstraintAwareStrategy,
match_episodes,
)
__all__ = ['FileClassifier', 'EpisodeRenamer'] __all__ = [
'FileClassifier',
'EpisodeRenamer',
'DurationMatchStrategy',
'SequentialStrategy',
'DiscMappingStrategy',
'ConstraintAwareStrategy',
'match_episodes',
]

Binary file not shown.

View File

@ -1,43 +1,4 @@
"""Episode renamer for creating Jellyfin-compatible filenames. """Episode renamer for creating Jellyfin-compatible filenames."""
Architecture
------------
The EpisodeRenamer class handles the full pipeline:
1. **Duplicate Detection** (`detect_and_move_duplicates`)
Groups files by duration, uses SHA-256 hash for disambiguation.
Moves or deletes duplicates based on --force flag.
2. **Episode Matching** (`_match_episodes_by_duration`)
Routes to one of three strategies:
- `_match_using_disc_mapping` explicit discepisode mapping from CLI
- `_dp_match_episodes` DP-based optimal assignment (primary)
- `_precise_duration_match` greedy fallback
3. **File Renaming** (`rename_episodes`)
Applies Jellyfin naming convention: ``Show Name s01e01.mkv``
4. **Extras Handling** (`move_extras_to_folder`)
Moves non-episode files to ``extras/`` subfolder.
Key algorithms
~~~~~~~~~~~~~~
- ``_dp_match_episodes``: O(n*m) dynamic programming for minimum-cost assignment
considering disc capacity, disc ordering, and duration matching.
- ``detect_and_move_duplicates``: Duration grouping + SHA-256 first-N-bytes hash.
- ``_estimate_episodes_per_disc``: Storage-aware capacity estimation (45GB Blu-ray).
Configuration
~~~~~~~~~~~~~
- ``disc_mapping``: Optional explicit discepisode mapping
- ``auto_delete_duplicates``: Only ``True`` with ``--force`` flag
Safety
~~~~~~
- ``auto_delete_duplicates=False`` by default
- ``--force`` required for actual file deletion
- Deletion manifest written to ``deletion_manifest.json``
"""
import os import os
import shutil import shutil
from pathlib import Path from pathlib import Path
@ -124,23 +85,8 @@ class EpisodeRenamer:
return moved_files return moved_files
def _compute_file_hash(self, file_path: Path, num_bytes: int = 65536) -> str:
"""Compute a hash of the first N bytes of a file for duplicate detection."""
import hashlib
h = hashlib.sha256()
try:
with open(file_path, 'rb') as f:
h.update(f.read(num_bytes))
return h.hexdigest()
except Exception:
return ""
def detect_and_move_duplicates(self, episodes: List[Dict]) -> List[str]: def detect_and_move_duplicates(self, episodes: List[Dict]) -> List[str]:
"""Detect duplicate episodes and either delete them or move them to delete me folder. """Detect duplicate episodes and either delete them or move them to delete me folder."""
Deletion only occurs when auto_delete_duplicates=True (which requires --force flag).
Without --force, duplicates are moved to 'delete me' folder for review.
"""
processed_files = [] processed_files = []
duplicates_found = [] duplicates_found = []
@ -166,42 +112,12 @@ class EpisodeRenamer:
if len(group_episodes) > 1: if len(group_episodes) > 1:
print(f"\nFound {len(group_episodes)} potential duplicates with ~{duration:.1f}min duration:") print(f"\nFound {len(group_episodes)} potential duplicates with ~{duration:.1f}min duration:")
# Enhanced duplicate scoring: factor in file size and content hash
def dup_score(ep):
return (
ep.get('disc_number', 999) if 'disc_number' in ep else 999,
-ep['size_gb'],
ep['path'].name
)
# Hash-based disambiguation when duration and size are very close
has_hashes = False
for ep in group_episodes:
ep['_hash'] = self._compute_file_hash(ep['path'])
if ep['_hash']:
has_hashes = True
if has_hashes:
# Group by hash — identical hash = near-certain duplicate
hash_groups = {}
for ep in group_episodes:
hash_groups.setdefault(ep['_hash'], []).append(ep)
for hash_val, hash_eps in hash_groups.items():
if len(hash_eps) > 1:
hash_eps.sort(key=dup_score)
print(f" Keeping: {hash_eps[0]['path'].name} ({hash_eps[0]['size_gb']:.2f}GB) [hash match]")
for dup in hash_eps[1:]:
print(f" Duplicate (hash): {dup['path'].name} ({dup['size_gb']:.2f}GB)")
duplicates_found.append(dup)
elif len(hash_eps) == 1:
# Unique hash — still check against duration group
group_episodes_copy = [e for e in group_episodes if e['_hash'] == hash_val]
if len(group_episodes_copy) == 1:
continue
# Sort by disc number (keep lower disc numbers) then file size (largest first) for consistent ordering # Sort by disc number (keep lower disc numbers) then file size (largest first) for consistent ordering
group_episodes.sort(key=dup_score) group_episodes.sort(key=lambda x: (
x.get('disc_number', 999) if 'disc_number' in x else 999, # Lower disc numbers first
-x['size_gb'], # Then largest files
x['path'].name # Finally by filename for consistency
))
# Keep the first file (from earliest disc, largest size), mark others as duplicates # Keep the first file (from earliest disc, largest size), mark others as duplicates
keeper = group_episodes[0] keeper = group_episodes[0]
@ -228,35 +144,17 @@ class EpisodeRenamer:
# Handle duplicates based on auto_delete_duplicates flag # Handle duplicates based on auto_delete_duplicates flag
if self.auto_delete_duplicates: if self.auto_delete_duplicates:
import datetime # Delete duplicates directly
manifest_path = self.folder_path / "deletion_manifest.json"
manifest_entries = []
for duplicate in duplicates_found: for duplicate in duplicates_found:
source_path = duplicate['path'] source_path = duplicate['path']
try: try:
manifest_entries.append({ source_path.unlink() # Delete the file
'filename': source_path.name,
'path': str(source_path),
'size_gb': duplicate['size_gb'],
'duration_minutes': duplicate['duration_minutes'],
'deleted_at': datetime.datetime.now().isoformat()
})
source_path.unlink()
processed_files.append(str(source_path)) processed_files.append(str(source_path))
print(f"Deleted duplicate: {source_path.name}") print(f"Deleted duplicate: {source_path.name}")
except Exception as e: except Exception as e:
print(f"Error deleting {source_path.name}: {e}") print(f"Error deleting {source_path.name}: {e}")
try:
import json as _json
with open(manifest_path, 'w') as f:
_json.dump(manifest_entries, f, indent=2)
print(f"\nDeletion manifest written to: {manifest_path}")
except Exception as e:
print(f"Warning: Could not write deletion manifest: {e}")
else: else:
# Move duplicates to delete me folder (original behavior) # Move duplicates to delete me folder (original behavior)
if not self._create_delete_folder(): if not self._create_delete_folder():
@ -979,221 +877,99 @@ class EpisodeRenamer:
return matched_episodes return matched_episodes
def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]:
"""Match episodes using DP-based optimal assignment considering disc size, episode deltas, and TVDB duration.""" """Match episodes using lexicographical order and precise duration matching."""
allowed = list(range(1, len(tvdb_episodes) + 1)) matched_episodes = []
# Primary: DP-based optimal matching # Try precise duration matching first
print(" Running DP-based episode matcher (disc size + episode deltas + TVDB duration)...") duration_matches = self._precise_duration_match(episodes_info, tvdb_episodes, list(range(1, len(tvdb_episodes) + 1)))
matched_episodes = self._dp_match_episodes(episodes_info, tvdb_episodes, allowed)
for match in matched_episodes: for match in duration_matches:
matched_episodes.append(match)
duration_diff = match.get('duration_diff', 0) duration_diff = match.get('duration_diff', 0)
fn = match['file_info']['path'].name[:50]
if duration_diff is not None: if duration_diff is not None:
print(f" DP match: {fn}... → Episode {match['episode_number']}{duration_diff:.1f}min)") print(f" Duration match: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']}{duration_diff:.1f}min)")
else: else:
print(f" DP fallback: {fn}... → Episode {match['episode_number']} (no duration match)") print(f" Sequential fallback: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']} (no duration match)")
return matched_episodes return matched_episodes
def _dp_match_episodes(self, files: List[Dict], tvdb_episodes: List[Dict],
allowed_episodes: List[int]) -> List[Dict]:
"""Dynamic programming matcher considering disc size, episode deltas, and TVDB duration.
Builds a cost matrix over (file, episode) pairs where cost = |file_dur - tvdb_dur|, def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict], allowed_episodes: List[int]) -> List[Dict]:
then finds the minimum-cost assignment that respects: """Perform precise duration matching with fallback to lexicographical order."""
- Disc capacity: each disc holds at most its capacity matched_episodes = []
- Disc ordering: files from earlier discs map to earlier episodes used_files = set()
- Episode deltas: consecutive files prefer consecutive episodes used_episodes = set()
Returns matched episodes sorted by episode number. # First pass: Find exact or very close duration matches (within 5 minutes)
""" for file_info in files:
n_files = len(files) if id(file_info) in used_files:
n_episodes = len(tvdb_episodes) continue
if not n_files or not n_episodes:
return self._precise_duration_match(files, tvdb_episodes, allowed_episodes)
allowed_set = set(allowed_episodes) file_duration = file_info['duration']
best_match = None
best_score = float('inf')
diffs = [] # Diagnostics: collect per-episode duration differences
INF = float('inf') for tvdb_ep in tvdb_episodes:
cost = [[INF] * n_episodes for _ in range(n_files)] if tvdb_ep['episode_number'] in used_episodes:
for fi in range(n_files): continue
fdur = files[fi]['duration'] if tvdb_ep['episode_number'] not in allowed_episodes:
for ei in range(n_episodes):
ep_num = tvdb_episodes[ei]['episode_number']
if ep_num not in allowed_set:
continue continue
tdur = tvdb_episodes[ei].get('runtime', 0)
cost[fi][ei] = 0.0 if tdur == 0 else abs(fdur - tdur)
# DP: dp[i][j] = min cost to assign first i files from first j episodes tvdb_duration = tvdb_ep.get('runtime', 0)
# Transition: if tvdb_duration == 0:
# dp[i][j] = min(dp[i][j-1], # skip episode j-1 continue
# dp[i-1][j-1] + cost) # assign file i-1 to episode j-1
dp = [[INF] * (n_episodes + 1) for _ in range(n_files + 1)]
for j in range(n_episodes + 1):
dp[0][j] = 0.0
for i in range(1, n_files + 1): # Calculate duration difference
dp[i][0] = INF duration_diff = abs(file_duration - tvdb_duration)
for j in range(1, n_episodes + 1): diffs.append((tvdb_ep['episode_number'], tvdb_duration, duration_diff))
dp[i][j] = dp[i][j - 1]
c = cost[i - 1][j - 1]
if c < INF and dp[i - 1][j - 1] < INF:
dp[i][j] = min(dp[i][j], dp[i - 1][j - 1] + c)
# Backtrack if duration_diff < best_score:
best_j = min(range(1, n_episodes + 1), key=lambda j: dp[n_files][j]) best_score = duration_diff
if dp[n_files][best_j] >= INF: best_match = tvdb_ep
return self._fallback_sequential(files, tvdb_episodes, allowed_episodes)
assignment = {} # Accept matches within 1 minute as good matches (strict duration matching)
i, j = n_files, best_j if best_match and best_score <= 1.0:
while i > 0 and j > 0: matched_episodes.append({
if dp[i][j] == dp[i][j - 1]: 'file_info': file_info['file_info'],
j -= 1 'episode_number': best_match['episode_number'],
elif dp[i - 1][j - 1] < INF and cost[i - 1][j - 1] < INF: 'tvdb_info': best_match,
assignment[i - 1] = j - 1 'duration_diff': best_score
i -= 1
j -= 1
else:
break
matched_episodes = []
for fi_idx, ep_idx in assignment.items():
ep = tvdb_episodes[ep_idx]
fdur = files[fi_idx]['duration']
tdur = ep.get('runtime', 0)
diff = abs(fdur - tdur) if tdur > 0 else None
matched_episodes.append({
'file_info': files[fi_idx]['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': diff,
})
assigned_files = set(assignment.keys())
assigned_eps = set(assignment.values())
remaining_files = [f for idx, f in enumerate(files) if idx not in assigned_files]
remaining_eps = [tvdb_episodes[idx] for idx in range(n_episodes) if idx not in assigned_eps]
remaining_eps.sort(key=lambda x: x['episode_number'])
for fi, ep in zip(remaining_files, remaining_eps):
matched_episodes.append({
'file_info': fi['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': None,
})
return matched_episodes
def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict],
allowed_episodes: List[int]) -> List[Dict]:
"""DP-based optimal duration matching within allowed episodes.
Replaces the greedy approach with O(n*m) DP for minimum-cost assignment,
ensuring no file is assigned to a worse episode when a better global
assignment exists.
"""
n_files = len(files)
allowed_set = set(allowed_episodes)
allowed_tvdb = sorted(
[ep for ep in tvdb_episodes if ep['episode_number'] in allowed_set],
key=lambda x: x['episode_number'],
)
n_eps = len(allowed_tvdb)
if not n_files or not n_eps:
return self._fallback_sequential(files, allowed_tvdb, allowed_episodes)
INF = float('inf')
# Cost matrix: cost[fi][ei] = |file_duration - tvdb_duration|
cost = [[INF] * n_eps for _ in range(n_files)]
for fi in range(n_files):
fdur = files[fi]['duration']
for ei in range(n_eps):
tdur = allowed_tvdb[ei].get('runtime', 0)
cost[fi][ei] = 0.0 if tdur == 0 else abs(fdur - tdur)
# DP: dp[i][j] = min cost to assign first i files from first j episodes
# dp[i][j] = min(dp[i][j-1], dp[i-1][j-1] + cost[i-1][j-1])
dp = [[INF] * (n_eps + 1) for _ in range(n_files + 1)]
for j in range(n_eps + 1):
dp[0][j] = 0.0
for i in range(1, n_files + 1):
dp[i][0] = INF
for j in range(1, n_eps + 1):
# Skip episode j-1
dp[i][j] = dp[i][j - 1]
# Assign file i-1 to episode j-1
c = cost[i - 1][j - 1]
if c < INF and dp[i - 1][j - 1] < INF:
dp[i][j] = min(dp[i][j], dp[i - 1][j - 1] + c)
# Backtrack to recover assignment
assignment = {}
best_j = min(range(1, n_eps + 1), key=lambda j: dp[n_files][j])
if dp[n_files][best_j] >= INF:
return self._fallback_sequential(files, allowed_tvdb, allowed_episodes)
i, j = n_files, best_j
while i > 0 and j > 0:
if j > 0 and dp[i][j] == dp[i][j - 1]:
j -= 1
elif dp[i - 1][j - 1] < INF and cost[i - 1][j - 1] < INF:
assignment[i - 1] = j - 1
i -= 1
j -= 1
else:
break
matched_episodes = []
for fi, ei in assignment.items():
ep = allowed_tvdb[ei]
fdur = files[fi]['duration']
tdur = ep.get('runtime', 0)
diff = abs(fdur - tdur) if tdur > 0 else None
matched_episodes.append({
'file_info': files[fi]['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': diff,
'assignment_cost': diff or 0,
})
# Fallback for unassigned files
assigned_fi = set(assignment.keys())
assigned_ei = set(assignment.values())
remaining_files = [f for idx, f in enumerate(files) if idx not in assigned_fi]
remaining_eps = [e for idx, e in enumerate(allowed_tvdb) if idx not in assigned_ei]
for fi, ep in zip(remaining_files, remaining_eps):
matched_episodes.append({
'file_info': fi['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': None,
'assignment_cost': 0,
})
return matched_episodes
def _fallback_sequential(self, files: List[Dict], tvdb_eps: List[Dict],
allowed_episodes: List[int]) -> List[Dict]:
"""Sequential fallback when DP cannot produce an assignment."""
matched = []
sorted_files = sorted(files, key=lambda x: x['filename'])
allowed_set = set(allowed_episodes)
avail = [ep for ep in tvdb_eps if ep['episode_number'] in allowed_set]
for i, fi in enumerate(sorted_files):
if i < len(avail):
ep = avail[i]
matched.append({
'file_info': fi['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': None,
}) })
return matched used_files.add(id(file_info))
used_episodes.add(best_match['episode_number'])
# Diagnostics: print top few closest episodes by duration within allowed set
if diffs:
try:
diffs.sort(key=lambda x: x[2])
top = ", ".join([f"E{ep}:{dur}{diff:.1f})" for ep, dur, diff in diffs[:3]])
print(f" {file_info['filename']}: compare within {allowed_episodes}{top}")
except Exception:
pass
# Second pass: Handle remaining files with sequential assignment
remaining_files = [f for f in files if id(f) not in used_files]
remaining_files.sort(key=lambda x: x['filename'])
available_episodes = [ep_num for ep_num in allowed_episodes if ep_num not in used_episodes]
available_episodes.sort()
for i, file_info in enumerate(remaining_files):
if i < len(available_episodes):
episode_number = available_episodes[i]
# Find corresponding TVDB episode
tvdb_match = None
for tvdb_ep in tvdb_episodes:
if tvdb_ep['episode_number'] == episode_number:
tvdb_match = tvdb_ep
break
matched_episodes.append({
'file_info': file_info['file_info'],
'episode_number': episode_number,
'tvdb_info': tvdb_match,
'duration_diff': None
})
return matched_episodes

View File

@ -14,62 +14,34 @@ try:
except ImportError: except ImportError:
config_manager = None config_manager = None
# Supported video file extensions
# Supported video extensions (configurable via config or env var) DEFAULT_VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.mov', '.wmv', '.flv', '.webm', '.m4v'}
DEFAULT_VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.webm', '.mov', '.wmv', '.flv', '.m4v'}
class FileClassifier: class FileClassifier:
"""Classifies video files as episodes or extras based on file size and video duration.""" """Classifies video files as episodes or extras based on file size and video duration."""
def __init__(self, folder_path: str, video_extensions: List[str] = None, def __init__(self, folder_path: str, video_extensions: set = None):
fallback_duration_minutes_per_gb: float = None):
"""Initialize with folder path containing video files. """Initialize with folder path containing video files.
Args: Args:
folder_path: Path to folder with video files. folder_path: Path to folder containing video files.
video_extensions: List of extensions to process (e.g., ['.mkv', '.mp4']). video_extensions: Set of file extensions to scan (e.g., {'.mkv', '.mp4'}).
Defaults to all common formats. Defaults to DEFAULT_VIDEO_EXTENSIONS if not provided.
fallback_duration_minutes_per_gb: Minutes of video per GB when MediaInfo
is unavailable. Defaults to 45 (configurable for 4K or low-bitrate content).
""" """
self.folder_path = Path(folder_path) self.folder_path = Path(folder_path)
self.video_extensions = self._resolve_extensions(video_extensions) self.video_extensions = video_extensions or DEFAULT_VIDEO_EXTENSIONS
self.fallback_ratio = self._resolve_fallback_ratio(fallback_duration_minutes_per_gb)
self.video_files = self._get_video_files() self.video_files = self._get_video_files()
self.file_info = self._analyze_files() self.file_info = self._analyze_files()
def _resolve_extensions(self, extensions: List[str] = None) -> set:
"""Resolve the set of video extensions to scan."""
if extensions:
return {e.lower() if e.startswith('.') else f'.{e.lower()}' for e in extensions}
env_ext = os.environ.get('VIDEO_EXTENSIONS', '')
if env_ext:
return {e.strip().lower() for e in env_ext.split(',') if e.strip()}
return DEFAULT_VIDEO_EXTENSIONS
def _resolve_fallback_ratio(self, ratio: float = None) -> float:
"""Resolve the fallback minutes/GB ratio for duration estimation."""
if ratio is not None:
return float(ratio)
env_ratio = os.environ.get('FALLBACK_DURATION_RATIO', '')
if env_ratio:
try:
return float(env_ratio)
except ValueError:
pass
if config_manager:
return config_manager.get_default_episode_duration()
return 45.0
def _get_video_files(self) -> List[Path]: def _get_video_files(self) -> List[Path]:
"""Get all supported video files in the folder, excluding system files.""" """Get all video files in the folder, excluding system files."""
if not self.folder_path.exists(): if not self.folder_path.exists():
raise FileNotFoundError(f"Folder not found: {self.folder_path}") raise FileNotFoundError(f"Folder not found: {self.folder_path}")
video_files = [] video_files = []
for ext in self.video_extensions: for ext in self.video_extensions:
video_files.extend(self.folder_path.glob(f'*{ext}')) video_files.extend(self.folder_path.glob(f"*{ext}"))
# Filter out macOS resource fork files and other system files # Filter out macOS resource fork files and other system files
filtered = [] filtered = []
@ -80,31 +52,29 @@ class FileClassifier:
continue continue
filtered.append(file) filtered.append(file)
# Deduplicate (in case overlapping extensions) if not filtered:
seen = set()
unique = []
for f in filtered:
if f not in seen:
seen.add(f)
unique.append(f)
if not unique:
ext_list = ', '.join(sorted(self.video_extensions)) ext_list = ', '.join(sorted(self.video_extensions))
print(f"Warning: No valid video files found in {self.folder_path} (scanned: {ext_list})") print(f"Warning: No video files found ({ext_list}) in {self.folder_path}")
return unique return filtered
def _get_file_size(self, file_path: Path) -> int: def _get_file_size(self, file_path: Path) -> int:
"""Get file size in bytes.""" """Get file size in bytes."""
return file_path.stat().st_size return file_path.stat().st_size
def _get_fallback_duration_ratio(self) -> float:
"""Get minutes-per-GB ratio for fallback duration estimation."""
if config_manager:
return config_manager.get_duration_minutes_per_gb()
return 45.0
def _get_video_duration(self, file_path: Path) -> float: def _get_video_duration(self, file_path: Path) -> float:
"""Get video duration in minutes using MediaInfo, with configurable fallback.""" """Get video duration in minutes using MediaInfo."""
fallback_ratio = self._get_fallback_duration_ratio()
if MediaInfo is None: if MediaInfo is None:
print(f"MediaInfo not available, using file size as proxy for duration " print(f"MediaInfo not available, using file size as proxy for duration ({fallback_ratio} min/GB)")
f"(ratio: {self.fallback_ratio:.1f} min/GB)")
size_gb = self._get_file_size(file_path) / (1024**3) size_gb = self._get_file_size(file_path) / (1024**3)
return size_gb * self.fallback_ratio return size_gb * fallback_ratio
try: try:
media_info = MediaInfo.parse(str(file_path)) media_info = MediaInfo.parse(str(file_path))
@ -112,6 +82,7 @@ class FileClassifier:
if track.track_type == 'Video': if track.track_type == 'Video':
duration_ms = track.duration duration_ms = track.duration
if duration_ms: if duration_ms:
# Handle both string and numeric duration values
if isinstance(duration_ms, str): if isinstance(duration_ms, str):
try: try:
duration_ms = float(duration_ms) duration_ms = float(duration_ms)
@ -121,19 +92,19 @@ class FileClassifier:
duration_minutes = float(duration_ms) / (1000 * 60) duration_minutes = float(duration_ms) / (1000 * 60)
# Sanity check: fall back to size estimation for suspicious durations # Sanity check: if duration seems unreasonable, fall back to size estimation
if duration_minutes > 300: if duration_minutes > 300:
print(f"Warning: Suspicious duration ({duration_minutes:.1f} min) for {file_path.name}, " print(f"Warning: Suspicious duration ({duration_minutes:.1f} min) for {file_path.name}, using size estimation")
f"using size estimation ({self.fallback_ratio:.1f} min/GB)")
size_gb = self._get_file_size(file_path) / (1024**3) size_gb = self._get_file_size(file_path) / (1024**3)
return size_gb * self.fallback_ratio return size_gb * fallback_ratio
return duration_minutes return duration_minutes
except Exception as e: except Exception as e:
print(f"Error getting duration for {file_path}: {e}") print(f"Error getting duration for {file_path}: {e}")
# Fallback to size-based estimation
size_gb = self._get_file_size(file_path) / (1024**3) size_gb = self._get_file_size(file_path) / (1024**3)
return size_gb * self.fallback_ratio return size_gb * fallback_ratio
def _analyze_files(self) -> List[Dict]: def _analyze_files(self) -> List[Dict]:
"""Analyze all video files to get size and duration info.""" """Analyze all video files to get size and duration info."""

200
src/Matcher/strategies.py Normal file
View File

@ -0,0 +1,200 @@
"""Matching strategies for episode-to-file mapping."""
from pathlib import Path
from typing import List, Dict, Optional
class DurationMatchStrategy:
"""Match files to TVDB episodes by duration similarity."""
def __init__(self, tolerance_minutes: float = 1.0):
self.tolerance = tolerance_minutes
def match(self, files: List[Dict], tvdb_episodes: List[Dict],
allowed_episodes: Optional[List[int]] = None) -> List[Dict]:
matched = []
used_files = set()
used_episodes = set()
for file_info in files:
if id(file_info) in used_files:
continue
file_duration = file_info['duration']
best_match = None
best_score = float('inf')
for tvdb_ep in tvdb_episodes:
if tvdb_ep['episode_number'] in used_episodes:
continue
if allowed_episodes and tvdb_ep['episode_number'] not in allowed_episodes:
continue
tvdb_duration = tvdb_ep.get('runtime', 0)
if tvdb_duration == 0:
continue
duration_diff = abs(file_duration - tvdb_duration)
if duration_diff < best_score:
best_score = duration_diff
best_match = tvdb_ep
if best_match and best_score <= self.tolerance:
matched.append({
'file_info': file_info['file_info'],
'episode_number': best_match['episode_number'],
'tvdb_info': best_match,
'duration_diff': best_score,
'strategy': 'duration_match'
})
used_files.add(id(file_info))
used_episodes.add(best_match['episode_number'])
return matched, used_files, used_episodes
class SequentialStrategy:
"""Assign remaining files to episodes in sequential order."""
def match(self, remaining_files: List[Dict], tvdb_episodes: List[Dict],
used_episodes: set, allowed_episodes: List[int]) -> List[Dict]:
matched = []
available = sorted(
[e for e in allowed_episodes if e not in used_episodes]
)
sorted_files = sorted(remaining_files, key=lambda x: x['filename'])
for i, file_info in enumerate(sorted_files):
if i >= len(available):
break
episode_number = available[i]
tvdb_match = next(
(ep for ep in tvdb_episodes if ep['episode_number'] == episode_number),
None
)
duration_diff = None
if tvdb_match and tvdb_match.get('runtime', 0) > 0:
duration_diff = abs(file_info['duration'] - tvdb_match['runtime'])
matched.append({
'file_info': file_info['file_info'],
'episode_number': episode_number,
'tvdb_info': tvdb_match,
'duration_diff': duration_diff,
'strategy': 'sequential'
})
return matched
class DiscMappingStrategy:
"""Match files to episodes using explicit disc-to-episode mapping."""
def __init__(self, disc_mapping: Dict[int, List[int]]):
self.disc_mapping = disc_mapping
def match(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]:
matched = []
disc_groups = {}
for info in episodes_info:
disc_num = info['disc_number']
if disc_num is not None:
disc_groups.setdefault(disc_num, []).append(info)
for disc_num in sorted(disc_groups.keys()):
files = sorted(disc_groups[disc_num], key=lambda x: x['filename'])
expected = self.disc_mapping.get(disc_num, [])
for i, file_info in enumerate(files):
if i >= len(expected):
break
episode_number = expected[i]
tvdb_episode = next(
(ep for ep in tvdb_episodes if ep['episode_number'] == episode_number),
None
)
matched.append({
'file_info': file_info['file_info'],
'episode_number': episode_number,
'tvdb_info': tvdb_episode,
'strategy': 'disc_mapping'
})
return matched
class ConstraintAwareStrategy:
"""Match files respecting disc capacity and duration constraints."""
def __init__(self, max_duration_diff: float = 2.0,
min_file_duration_buffer: float = 2.0):
self.max_duration_diff = max_duration_diff
self.min_file_buffer = min_file_duration_buffer
def validate(self, file_info: Dict, tvdb_episode: Dict,
disc_assignments: Dict[int, int],
disc_capacity: Dict[int, int]) -> bool:
file_duration = file_info['duration']
tvdb_duration = tvdb_episode.get('runtime', 0)
file_disc = file_info['disc_number']
if tvdb_duration > 0:
diff = abs(file_duration - tvdb_duration)
if diff > self.max_duration_diff:
return False
if file_duration < (tvdb_duration - self.min_file_buffer):
return False
if file_disc and disc_capacity:
max_eps = disc_capacity.get(file_disc, 0)
current = disc_assignments.get(file_disc, 0)
if current >= max_eps:
return False
return True
def match_episodes(files: List[Dict], tvdb_episodes: List[Dict],
disc_mapping: Optional[Dict[int, List[int]]] = None,
disc_capacity: Optional[Dict[int, int]] = None
) -> List[Dict]:
"""
Multi-strategy episode matcher.
Pipeline:
1. If disc_mapping provided, use DiscMappingStrategy
2. Otherwise, use DurationMatchStrategy for close matches
3. Fill gaps with SequentialStrategy
Args:
files: Episode info dicts with file_info, disc_number, duration, filename
tvdb_episodes: TVDB episode data with runtime
disc_mapping: Optional disc-to-episode mapping
disc_capacity: Optional disc capacity constraints
Returns:
List of match dicts sorted by episode number
"""
all_allowed = list(range(1, len(tvdb_episodes) + 1))
if disc_mapping:
return DiscMappingStrategy(disc_mapping).match(files, tvdb_episodes)
# Phase 1: Duration matching
duration_strategy = DurationMatchStrategy()
matched, used_files, used_eps = duration_strategy.match(
files, tvdb_episodes, all_allowed
)
# Phase 2: Sequential fill for remaining
remaining = [f for f in files if id(f) not in used_files]
sequential = SequentialStrategy()
matched.extend(sequential.match(remaining, tvdb_episodes, used_eps, all_allowed))
matched.sort(key=lambda x: x['episode_number'])
return matched

Binary file not shown.

View File

@ -20,28 +20,15 @@ class TVDBCache:
self.cache_dir.mkdir(exist_ok=True) self.cache_dir.mkdir(exist_ok=True)
self.cache_duration = timedelta(days=7) # Cache for 7 days self.cache_duration = timedelta(days=7) # Cache for 7 days
def _sanitize_name(self, name: str) -> str:
"""Sanitize series name to prevent path traversal attacks."""
import re
sanitized = name.lower()
sanitized = re.sub(r'[^a-z0-9_\-\s]', '', sanitized)
sanitized = sanitized.replace(' ', '_').replace('-', '_')
if '..' in sanitized or '/' in sanitized or '\\' in sanitized:
raise ValueError(f"Invalid series name containing path traversal: {name!r}")
return sanitized
def _get_cache_key(self, series_name: str, season_number: int) -> str: def _get_cache_key(self, series_name: str, season_number: int) -> str:
"""Generate cache key for series/season combination.""" """Generate cache key for series/season combination."""
sanitized_name = self._sanitize_name(series_name) # Normalize series name for consistent caching
return f"{sanitized_name}_s{season_number:02d}" normalized_name = series_name.lower().replace(' ', '_').replace('-', '_')
return f"{normalized_name}_s{season_number:02d}"
def _get_cache_file(self, cache_key: str) -> Path: def _get_cache_file(self, cache_key: str) -> Path:
"""Get cache file path for given key, validated to stay within cache_dir.""" """Get cache file path for given key."""
cache_file = self.cache_dir / f"{cache_key}.json" return self.cache_dir / f"{cache_key}.json"
resolved = cache_file.resolve()
if not str(resolved).startswith(str(self.cache_dir.resolve())):
raise ValueError(f"Cache path escape detected: {cache_key!r}")
return cache_file
def _is_cache_valid(self, cache_file: Path) -> bool: def _is_cache_valid(self, cache_file: Path) -> bool:
"""Check if cache file exists and is not expired.""" """Check if cache file exists and is not expired."""

View File

@ -1 +0,0 @@
"""Episode Matcher source packages."""

Binary file not shown.

View File

@ -23,6 +23,8 @@ class ConfigManager:
default_config = { default_config = {
"tvdb_api_key": "", "tvdb_api_key": "",
"default_episode_duration": 45, "default_episode_duration": 45,
"duration_minutes_per_gb": 45,
"video_extensions": [".mkv", ".mp4", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"],
"classification_thresholds": { "classification_thresholds": {
"size_threshold_ratio": 0.3, "size_threshold_ratio": 0.3,
"duration_threshold_ratio": 0.4 "duration_threshold_ratio": 0.4
@ -55,10 +57,7 @@ class ConfigManager:
return default_config return default_config
def get_tvdb_api_key(self) -> Optional[str]: def get_tvdb_api_key(self) -> Optional[str]:
"""Get TVDB API key from environment variable or config file.""" """Get TVDB API key from config."""
env_key = os.environ.get("TVDB_API_KEY", "").strip()
if env_key and env_key != "YOUR_TVDB_API_KEY_HERE":
return env_key
api_key = self.config.get("tvdb_api_key", "").strip() api_key = self.config.get("tvdb_api_key", "").strip()
if not api_key or api_key == "YOUR_TVDB_API_KEY_HERE": if not api_key or api_key == "YOUR_TVDB_API_KEY_HERE":
return None return None
@ -68,6 +67,16 @@ class ConfigManager:
"""Get default episode duration in minutes.""" """Get default episode duration in minutes."""
return self.config.get("default_episode_duration", 45) return self.config.get("default_episode_duration", 45)
def get_duration_minutes_per_gb(self) -> float:
"""Get fallback duration estimation ratio (minutes per GB)."""
return float(self.config.get("duration_minutes_per_gb", 45))
def get_video_extensions(self) -> list:
"""Get list of supported video file extensions."""
return self.config.get("video_extensions", [
".mkv", ".mp4", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"
])
def get_classification_thresholds(self) -> Dict[str, float]: def get_classification_thresholds(self) -> Dict[str, float]:
"""Get classification threshold ratios.""" """Get classification threshold ratios."""
return self.config.get("classification_thresholds", { return self.config.get("classification_thresholds", {

View File

View File

@ -1,140 +0,0 @@
"""Shared fixtures for Episode Matcher tests."""
import sys
import os
import tempfile
from pathlib import Path
# Ensure src is importable
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
@pytest.fixture
def tmp_dir(tmp_path):
"""Return a temporary directory as Path."""
return tmp_path
@pytest.fixture
def mock_video_files(tmp_path):
"""Create mock video files in temp dir and return the dir.
Uses small sizes (KB range) since actual content doesn't matter for tests.
"""
files = {
"Disc 1_t01.mkv": int(2 * 1024),
"Disc 1_t02.mkv": int(2100),
"Disc 1_t03.mkv": int(2200),
"Disc 2_t01.mkv": int(2000),
"Disc 2_t02.mkv": int(2100),
"Disc 2_t03.mkv": int(1900),
"extra_short.mkv": int(100),
}
for name, size in files.items():
fpath = tmp_path / name
fpath.write_bytes(b"\x00" * size)
return tmp_path
@pytest.fixture
def mock_episodes():
"""Mock episode file info dicts with realistic ratios but small absolute sizes."""
return [
{
"path": Path("Disc 1_t01.mkv"),
"name": "Disc 1_t01.mkv",
"size_bytes": int(2 * 1024),
"size_gb": 2.0,
"duration_minutes": 61.0,
},
{
"path": Path("Disc 1_t02.mkv"),
"name": "Disc 1_t02.mkv",
"size_bytes": int(2100),
"size_gb": 2.1,
"duration_minutes": 55.0,
},
{
"path": Path("Disc 1_t03.mkv"),
"name": "Disc 1_t03.mkv",
"size_bytes": int(2200),
"size_gb": 2.2,
"duration_minutes": 57.0,
},
{
"path": Path("Disc 2_t01.mkv"),
"name": "Disc 2_t01.mkv",
"size_bytes": int(2000),
"size_gb": 2.0,
"duration_minutes": 55.0,
},
{
"path": Path("Disc 2_t02.mkv"),
"name": "Disc 2_t02.mkv",
"size_bytes": int(2100),
"size_gb": 2.1,
"duration_minutes": 54.0,
},
{
"path": Path("Disc 2_t03.mkv"),
"name": "Disc 2_t03.mkv",
"size_bytes": int(1900),
"size_gb": 1.9,
"duration_minutes": 52.0,
},
]
@pytest.fixture
def mock_tvdb_episodes():
"""Mock TVDB episode data."""
return [
{"episode_number": 1, "name": "Winter Is Coming", "runtime": 61},
{"episode_number": 2, "name": "The Kingsroad", "runtime": 55},
{"episode_number": 3, "name": "Lord Snow", "runtime": 57},
{"episode_number": 4, "name": "Cripples Bastards", "runtime": 55},
{"episode_number": 5, "name": "The Wolf", "runtime": 54},
{"episode_number": 6, "name": "A Golden Crown", "runtime": 52},
{"episode_number": 7, "name": "You Win", "runtime": 57},
{"episode_number": 8, "name": "The Pointy End", "runtime": 58},
{"episode_number": 9, "name": "Baelor", "runtime": 56},
{"episode_number": 10, "name": "Fire and Blood", "runtime": 52},
]
@pytest.fixture
def disc_mapping():
"""Mock disc-to-episode mapping."""
return {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8]}
@pytest.fixture
def config_file(tmp_path):
"""Create a config.json in temp dir."""
import json
cfg = tmp_path / "config.json"
cfg.write_text(
json.dumps(
{
"tvdb_api_key": "test_api_key_12345",
"default_episode_duration": 45,
"classification_thresholds": {
"size_threshold_ratio": 0.3,
"duration_threshold_ratio": 0.4,
},
}
)
)
return cfg
@pytest.fixture
def empty_config_file(tmp_path):
"""Create a config.json with no API key."""
import json
cfg = tmp_path / "config.json"
cfg.write_text(json.dumps({"tvdb_api_key": ""}))
return cfg

View File

@ -1,149 +0,0 @@
"""Tests for src/config.py - ConfigManager."""
import json
import os
from pathlib import Path
import sys
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.config import ConfigManager
class TestConfigManagerInit:
def test_init_no_config_file(self, tmp_path, monkeypatch):
cfg = ConfigManager(str(tmp_path / "nonexistent.json"))
assert cfg.config["tvdb_api_key"] == ""
assert cfg.config["default_episode_duration"] == 45
assert cfg.config["classification_thresholds"]["size_threshold_ratio"] == 0.3
def test_init_existing_config(self, config_file):
cfg = ConfigManager(str(config_file))
assert cfg.config["tvdb_api_key"] == "test_api_key_12345"
assert cfg.config["default_episode_duration"] == 45
def test_init_invalid_json(self, tmp_path):
bad = tmp_path / "config.json"
bad.write_text("{invalid json")
cfg = ConfigManager(str(bad))
assert cfg.config["tvdb_api_key"] == ""
def test_init_missing_keys_merged(self, tmp_path):
partial = tmp_path / "config.json"
partial.write_text(json.dumps({"tvdb_api_key": "abc"}))
cfg = ConfigManager(str(partial))
assert cfg.config["default_episode_duration"] == 45
assert "classification_thresholds" in cfg.config
def test_init_io_error(self, tmp_path, monkeypatch):
cfg_path = tmp_path / "config.json"
cfg_path.write_text("test")
monkeypatch.setattr(Path, "exists", lambda self: True)
original_open = open
def mock_open(*args, **kwargs):
raise PermissionError("denied")
with patch("builtins.open", mock_open):
cfg = ConfigManager(str(cfg_path))
assert cfg.config["tvdb_api_key"] == ""
class TestConfigManagerApiKeys:
def test_get_api_key_from_config(self, config_file):
cfg = ConfigManager(str(config_file))
with patch.dict(os.environ, {}, clear=True):
key = cfg.get_tvdb_api_key()
assert key == "test_api_key_12345"
def test_get_api_key_from_env(self, config_file, monkeypatch):
cfg = ConfigManager(str(config_file))
monkeypatch.setenv("TVDB_API_KEY", "env_key_999")
key = cfg.get_tvdb_api_key()
assert key == "env_key_999"
def test_get_api_key_empty_env(self, config_file, monkeypatch):
cfg = ConfigManager(str(config_file))
monkeypatch.setenv("TVDB_API_KEY", " ")
key = cfg.get_tvdb_api_key()
assert key == "test_api_key_12345"
def test_get_api_key_placeholder_rejected(self, empty_config_file):
cfg = ConfigManager(str(empty_config_file))
cfg.config["tvdb_api_key"] = "YOUR_TVDB_API_KEY_HERE"
assert cfg.get_tvdb_api_key() is None
def test_get_api_key_env_placeholder_rejected(self, empty_config_file, monkeypatch):
cfg = ConfigManager(str(empty_config_file))
monkeypatch.setenv("TVDB_API_KEY", "YOUR_TVDB_API_KEY_HERE")
assert cfg.get_tvdb_api_key() is None
def test_env_key_takes_priority(self, config_file, monkeypatch):
cfg = ConfigManager(str(config_file))
monkeypatch.setenv("TVDB_API_KEY", "env_priority")
assert cfg.get_tvdb_api_key() == "env_priority"
class TestConfigManagerSettings:
def test_get_default_episode_duration(self, config_file):
cfg = ConfigManager(str(config_file))
assert cfg.get_default_episode_duration() == 45
def test_get_default_episode_duration_custom(self, tmp_path):
f = tmp_path / "config.json"
f.write_text(json.dumps({"default_episode_duration": 60}))
cfg = ConfigManager(str(f))
assert cfg.get_default_episode_duration() == 60
def test_get_classification_thresholds(self, config_file):
cfg = ConfigManager(str(config_file))
t = cfg.get_classification_thresholds()
assert t["size_threshold_ratio"] == 0.3
assert t["duration_threshold_ratio"] == 0.4
def test_get_classification_thresholds_defaults(self, tmp_path):
cfg = ConfigManager(str(tmp_path / "no.json"))
t = cfg.get_classification_thresholds()
assert t["size_threshold_ratio"] == 0.3
class TestConfigManagerUpdate:
def test_update_api_key_success(self, tmp_path):
cfg = ConfigManager(str(tmp_path / "config.json"))
result = cfg.update_api_key("new_key_abc")
assert result is True
assert cfg.config["tvdb_api_key"] == "new_key_abc"
stored = json.loads((tmp_path / "config.json").read_text())
assert stored["tvdb_api_key"] == "new_key_abc"
def test_update_api_key_io_error(self, tmp_path):
cfg = ConfigManager(str(tmp_path / "config.json"))
cfg.update_api_key("new_key")
cfg.config_file = Path("/nonexistent_dir/config.json")
result = cfg.update_api_key("another_key")
assert result is False
class TestConfigManagerStatus:
def test_print_config_status_no_key(self, tmp_path, capsys):
cfg = ConfigManager(str(tmp_path / "no.json"))
cfg.print_config_status()
out = capsys.readouterr().out
assert "Configuration file:" in out
assert "Not configured" in out
def test_print_config_status_with_key(self, config_file, capsys):
cfg = ConfigManager(str(config_file))
cfg.print_config_status()
out = capsys.readouterr().out
assert "configured" in out
def test_print_config_status_masked_short_key(self, tmp_path):
f = tmp_path / "config.json"
f.write_text(json.dumps({"tvdb_api_key": "ab"}))
cfg = ConfigManager(str(f))
cfg.print_config_status()

View File

@ -1,110 +0,0 @@
"""Tests for episode_matcher.py - parse_disc_mapping and CLI."""
import sys
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from episode_matcher import parse_disc_mapping
class TestParseDiscMapping:
def test_single_disc_range(self):
result = parse_disc_mapping("1:1-3")
assert result == {1: [1, 2, 3]}
def test_single_disc_single_ep(self):
result = parse_disc_mapping("1:5")
assert result == {1: [5]}
def test_multi_disc(self):
result = parse_disc_mapping("1:1-3,2:4-6,3:7-9,4:10")
assert result == {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8, 9], 4: [10]}
def test_spaces(self):
result = parse_disc_mapping("1:1-3 , 2:4-6")
assert result == {1: [1, 2, 3], 2: [4, 5, 6]}
def test_empty_string(self):
result = parse_disc_mapping("")
assert result == {}
def test_none(self):
result = parse_disc_mapping(None)
assert result == {}
def test_single_ep_per_disc(self):
result = parse_disc_mapping("1:1,2:2,3:3")
assert result == {1: [1], 2: [2], 3: [3]}
def test_large_numbers(self):
result = parse_disc_mapping("1:1-22")
assert result == {1: list(range(1, 23))}
def test_invalid_format_exits(self, capsys):
with pytest.raises(SystemExit):
parse_disc_mapping("invalid")
def test_missing_colon_exits(self, capsys):
with pytest.raises(SystemExit):
parse_disc_mapping("1,2,3")
class TestArgParse:
def test_parser_basic_args(self):
import argparse
from episode_matcher import main
with patch("sys.argv", [
"episode_matcher.py",
"/tmp/episodes",
"Test Show",
"1",
]):
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.is_dir", return_value=True):
with patch("episode_matcher.TVDBClient") as mock_tvdb:
mock_instance = mock_tvdb.return_value
mock_instance.authenticate.return_value = False
with patch("episode_matcher.MockTVDBClient") as mock_mock:
m = mock_mock.return_value
m.authenticate.return_value = True
m.get_episode_durations.return_value = []
with patch("episode_matcher.FileClassifier") as mock_fc:
fc = mock_fc.return_value
fc.file_info = []
fc.classify_files.return_value = ([], [])
with pytest.raises(SystemExit):
main()
def test_parser_dry_run_flag(self):
import sys as real_sys
orig_argv = real_sys.argv
try:
real_sys.argv = [
"episode_matcher.py", "/tmp/e", "Show", "1", "--dry-run", "--verbose"
]
from importlib import reload
import episode_matcher
parser = episode_matcher.__dict__.get("_parser", None)
from episode_matcher import main
finally:
real_sys.argv = orig_argv
class TestParseDiscMappingEdgeCases:
def test_range_to_self(self):
result = parse_disc_mapping("1:5-5")
assert result == {1: [5]}
def test_multiple_ranges_same_ep(self):
result = parse_disc_mapping("1:1-3,2:3-5")
assert 3 in result[1]
assert 3 in result[2]
def test_disc_number_large(self):
result = parse_disc_mapping("99:1-3")
assert 99 in result

View File

@ -1,539 +0,0 @@
"""Tests for src/Matcher/episode_renamer.py - EpisodeRenamer."""
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.Matcher.episode_renamer import EpisodeRenamer
class TestEpisodeRenamerInit:
def test_init_defaults(self):
r = EpisodeRenamer("/tmp/test", "Show Name", 1)
assert r.show_name == "Show Name"
assert r.season_number == 1
assert r.disc_mapping is None
assert r.auto_delete_duplicates is False
def test_init_with_disc_mapping(self):
r = EpisodeRenamer("/tmp/test", "Show", 2, disc_mapping={1: [1, 2], 2: [3, 4]})
assert r.disc_mapping == {1: [1, 2], 2: [3, 4]}
def test_init_auto_delete(self):
r = EpisodeRenamer("/tmp/test", "Show", 1, auto_delete_duplicates=True)
assert r.auto_delete_duplicates is True
def test_init_folder_paths(self):
r = EpisodeRenamer("/tmp/test", "Show", 1)
assert r.extras_folder == Path("/tmp/test/extras")
assert r.delete_folder == Path("/tmp/test/delete me")
class TestSanitizeFilename:
def test_remove_invalid_chars(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
result = r._sanitize_filename('a<b>c:d"e/f\\g|h?i*j')
assert ":" not in result
assert "<" not in result
assert ">" not in result
assert '"' not in result
assert "/" not in result
assert "\\" not in result
assert "|" not in result
assert "?" not in result
assert "*" not in result
assert "a" in result and "b" in result and "c" in result
def test_collapse_spaces(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r._sanitize_filename("a b") == "a b"
def test_strip(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r._sanitize_filename(" hello ") == "hello"
def test_preserve_valid_chars(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r._sanitize_filename("hello_world-1.mkv") == "hello_world-1.mkv"
class TestGenerateEpisodeFilename:
def test_standard(self):
r = EpisodeRenamer("/tmp/t", "Show Name", 3)
assert r._generate_episode_filename(5, ".mkv") == "Show Name s03e05.mkv"
def test_single_digit_padding(self):
r = EpisodeRenamer("/tmp/t", "Show", 1)
assert r._generate_episode_filename(1, ".mp4") == "Show s01e01.mp4"
def test_double_digit_episode(self):
r = EpisodeRenamer("/tmp/t", "Show", 10)
assert r._generate_episode_filename(12, ".mkv") == "Show s10e12.mkv"
def test_sanitizes_output(self):
r = EpisodeRenamer("/tmp/t", "Show:Name", 1)
assert r._generate_episode_filename(1, ".mkv") == "ShowName s01e01.mkv"
class TestExtractDiscInfo:
def test_disc_with_underscore(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("Disc 1_t01.mkv")
assert info["disc_number"] == 1
assert info["track_number"] == 1
def test_disk_variant(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("Disk2_t03.mkv")
assert info["disc_number"] == 2
assert info["track_number"] == 3
def test_disc_hyphen(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("disc-3_track05.mkv")
assert info["disc_number"] == 3
assert info["track_number"] == 5
def test_no_disc_info(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("episode_01.mkv")
assert info["disc_number"] is None
assert info["track_number"] is None
def test_track_pattern(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("file_t22.mkv")
assert info["track_number"] == 22
def test_disc_only(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("Disc 5.mkv")
assert info["disc_number"] == 5
assert info["track_number"] is None
class TestComputeFileHash:
def test_hash_consistent(self, tmp_path):
f = tmp_path / "test.mkv"
f.write_bytes(b"\x00" * 1000)
r = EpisodeRenamer(str(tmp_path), "S", 1)
h1 = r._compute_file_hash(f)
h2 = r._compute_file_hash(f)
assert h1 == h2
assert len(h1) == 64
def test_hash_different_content(self, tmp_path):
f1 = tmp_path / "a.mkv"
f2 = tmp_path / "b.mkv"
f1.write_bytes(b"\x00" * 100)
f2.write_bytes(b"\xff" * 100)
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._compute_file_hash(f1) != r._compute_file_hash(f2)
def test_hash_custom_bytes(self, tmp_path):
f = tmp_path / "test.mkv"
f.write_bytes(b"\x00" * 10000)
r = EpisodeRenamer(str(tmp_path), "S", 1)
h = r._compute_file_hash(f, num_bytes=512)
assert h
def test_hash_missing_file(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
h = r._compute_file_hash(tmp_path / "nope.mkv")
assert h == ""
class TestCreateFolders:
def test_create_extras_folder(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._create_extras_folder() is True
assert (tmp_path / "extras").exists()
def test_create_delete_folder(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._create_delete_folder() is True
assert (tmp_path / "delete me").exists()
def test_create_extras_already_exists(self, tmp_path):
(tmp_path / "extras").mkdir()
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._create_extras_folder() is True
class TestMoveExtras:
def test_move_extras(self, tmp_path):
(tmp_path / "extra.mkv").write_bytes(b"\x00" * 100)
r = EpisodeRenamer(str(tmp_path), "S", 1)
extras = [{"path": tmp_path / "extra.mkv", "name": "extra.mkv"}]
moved = r.move_extras_to_folder(extras)
assert len(moved) == 1
assert (tmp_path / "extras" / "extra.mkv").exists()
assert not (tmp_path / "extra.mkv").exists()
def test_move_extras_skip_existing(self, tmp_path):
(tmp_path / "extras").mkdir()
(tmp_path / "extras" / "extra.mkv").write_bytes(b"\x00")
(tmp_path / "extra.mkv").write_bytes(b"\x00" * 100)
r = EpisodeRenamer(str(tmp_path), "S", 1)
extras = [{"path": tmp_path / "extra.mkv", "name": "extra.mkv"}]
moved = r.move_extras_to_folder(extras)
assert len(moved) == 0
def test_move_extras_empty(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r.move_extras_to_folder([]) == []
class TestDuplicateDetection:
def test_detect_duplicates_same_duration(self, tmp_path):
f1 = tmp_path / "Disc 1_t01.mkv"
f2 = tmp_path / "Disc 1_t02.mkv"
f1.write_bytes(b"\x00" * 2000)
f2.write_bytes(b"\x00" * 2000)
eps = [
{"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
{"path": f2, "name": "Disc 1_t02.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
]
r = EpisodeRenamer(str(tmp_path), "S", 1)
result = r.detect_and_move_duplicates(eps)
assert isinstance(result, list)
def test_no_duplicates_different_duration(self, tmp_path):
f1 = tmp_path / "Disc 1_t01.mkv"
f2 = tmp_path / "Disc 2_t01.mkv"
f1.write_bytes(b"\x00" * 2000)
f2.write_bytes(b"\x01" * 3000)
eps = [
{"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
{"path": f2, "name": "Disc 2_t01.mkv", "size_gb": 3.0, "duration_minutes": 55.0},
]
r = EpisodeRenamer(str(tmp_path), "S", 1)
r.detect_and_move_duplicates(eps)
def test_duplicate_moved_to_delete_me(self, tmp_path):
f1 = tmp_path / "Disc 1_t01.mkv"
f2 = tmp_path / "Disc 1_t02.mkv"
f1.write_bytes(b"\x00" * 2000)
f2.write_bytes(b"\x00" * 2000)
eps = [
{"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
{"path": f2, "name": "Disc 1_t02.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
]
r = EpisodeRenamer(str(tmp_path), "S", 1, auto_delete_duplicates=False)
result = r.detect_and_move_duplicates(eps)
assert (tmp_path / "delete me").exists()
class TestMatchEpisodes:
def test_disc_mapping_match(self, mock_episodes, mock_tvdb_episodes, disc_mapping):
r = EpisodeRenamer("/tmp/t", "Show", 1, disc_mapping=disc_mapping)
episodes_info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
episodes_info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"track_number": disc_info["track_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
matched = r._match_using_disc_mapping(episodes_info, mock_tvdb_episodes)
assert len(matched) > 0
ep_nums = [m["episode_number"] for m in matched]
assert len(set(ep_nums)) == len(ep_nums)
def test_fallback_sequential(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = []
for ep in mock_episodes:
files.append({
"file_info": ep,
"filename": ep["name"],
"disc_number": None,
"duration": ep["duration_minutes"],
})
allowed = list(range(1, 11))
result = r._fallback_sequential(files, mock_tvdb_episodes, allowed)
assert len(result) > 0
def test_dp_match_episodes(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = []
for ep in mock_episodes:
files.append({
"file_info": ep,
"filename": ep["name"],
"disc_number": None,
"duration": ep["duration_minutes"],
})
allowed = list(range(1, 11))
result = r._dp_match_episodes(files, mock_tvdb_episodes, allowed)
assert len(result) > 0
ep_nums = [m["episode_number"] for m in result]
assert len(set(ep_nums)) == len(ep_nums)
def test_precise_duration_match(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = []
for ep in mock_episodes:
files.append({
"file_info": ep,
"filename": ep["name"],
"disc_number": None,
"duration": ep["duration_minutes"],
})
allowed = list(range(1, 7))
result = r._precise_duration_match(files, mock_tvdb_episodes, allowed)
assert len(result) > 0
def test_dp_match_empty_files(self, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
result = r._dp_match_episodes([], mock_tvdb_episodes, list(range(1, 11)))
assert isinstance(result, list)
def test_precise_match_empty(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = [{"file_info": {}, "filename": "a.mkv", "duration": 45.0}]
result = r._precise_duration_match(files, [], [])
assert isinstance(result, list)
def test_fallback_sequential_empty(self, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
result = r._fallback_sequential([], mock_tvdb_episodes, [])
assert result == []
class TestRenameEpisodes:
def test_rename_no_episodes(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r.rename_episodes([]) == []
def test_rename_with_tvdb(self, tmp_path, mock_episodes, mock_tvdb_episodes):
for ep in mock_episodes:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
r = EpisodeRenamer(str(tmp_path), "TestShow", 1)
renamed = r.rename_episodes(mock_episodes, mock_tvdb_episodes)
assert len(renamed) > 0
for item in renamed:
assert "original_name" in item
assert "new_name" in item
assert "episode_number" in item
def test_rename_skip_existing(self, tmp_path, mock_episodes, capsys):
for ep in mock_episodes[:2]:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
(tmp_path / "TestShow s01e01.mkv").write_bytes(b"\x00")
r = EpisodeRenamer(str(tmp_path), "TestShow", 1)
renamed = r.rename_episodes(mock_episodes[:2], None)
out = capsys.readouterr().out
assert "already exists" in out or len(renamed) < 2
class TestEstimateEpisodesPerDisc:
def test_estimate_basic(self, tmp_path, mock_episodes):
for ep in mock_episodes:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
r = EpisodeRenamer(str(tmp_path), "S", 1)
episodes_info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
episodes_info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
})
result = r._estimate_episodes_per_disc(episodes_info, 6)
assert isinstance(result, dict)
def test_estimate_no_disc_info(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
result = r._estimate_episodes_per_disc([], 10)
assert result == {}
def test_estimate_no_disc_numbers(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = [{"disc_number": None, "filename": "a.mkv", "file_info": {"size_gb": 5.0}}]
result = r._estimate_episodes_per_disc(info, 10)
assert result == {}
class TestAnalyzeDiscCapacity:
def test_analyze_basic(self, tmp_path, mock_episodes):
for ep in mock_episodes:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
r = EpisodeRenamer(str(tmp_path), "S", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
})
result = r._analyze_disc_capacity(info)
assert isinstance(result, dict)
def test_analyze_no_discs(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
result = r._analyze_disc_capacity([{"disc_number": None, "file_info": {"size_gb": 5}}])
assert result == {}
class TestDurationMatching:
def test_match_by_duration_and_order(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"track_number": disc_info["track_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
result = r._match_by_duration_and_order(info, mock_tvdb_episodes)
assert len(result) > 0
ep_nums = [m["episode_number"] for m in result]
assert len(set(ep_nums)) == len(ep_nums)
def test_flexible_duration_match(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
result = r._flexible_duration_match(info, mock_tvdb_episodes)
assert isinstance(result, list)
def test_sequential_assignment_with_constraints(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
disc_capacity = {1: 3, 2: 3}
result = r._sequential_assignment_with_constraints(info, mock_tvdb_episodes, disc_capacity)
assert isinstance(result, list)
def test_validate_sequential_assignment_valid(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 55.0, "disc_number": 1}
tvdb_ep = {"runtime": 55, "episode_number": 1}
assert r._validate_sequential_assignment(file_info, tvdb_ep, {}, {}) is True
def test_validate_sequential_assignment_duration_mismatch(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 30.0, "disc_number": 1}
tvdb_ep = {"runtime": 60, "episode_number": 1}
result = r._validate_sequential_assignment(file_info, tvdb_ep, {}, {})
assert result is False
def test_validate_sequential_assignment_disc_full(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 55.0, "disc_number": 1}
tvdb_ep = {"runtime": 55, "episode_number": 1}
disc_assignments = {1: 3}
disc_capacity = {1: 3}
result = r._validate_sequential_assignment(file_info, tvdb_ep, disc_assignments, disc_capacity)
assert result is False
def test_assignment_satisfies_constraints(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
fake_path = type("P", (), {"name": "Disc 1_t01.mkv"})()
file_info = {
"duration": 55.0,
"disc_number": 1,
"file_info": {"path": fake_path},
}
tvdb_ep = {"runtime": 55, "episode_number": 1}
result = r._assignment_satisfies_constraints(file_info, tvdb_ep, [], {})
assert result is True
def test_assignment_satisfies_duration_exceeded(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
fake_path = type("P", (), {"name": "a.mkv"})()
file_info = {
"duration": 30.0,
"disc_number": None,
"file_info": {"path": fake_path},
}
tvdb_ep = {"runtime": 60, "episode_number": 1}
result = r._assignment_satisfies_constraints(file_info, tvdb_ep, [], {})
assert result is False
class TestMatchByDurationAndDisc:
def test_match_disc(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
episodes_per_disc = {1: [1, 2, 3], 2: [4, 5, 6]}
result = r._match_by_duration_and_disc(info, mock_tvdb_episodes, episodes_per_disc)
assert len(result) > 0
for m in result:
assert m["episode_number"] in [1, 2, 3, 4, 5, 6]
class TestFindOptimalAssignment:
def test_find_optimal_assignment(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
result = r._find_optimal_assignment(info, mock_tvdb_episodes, {1: 3, 2: 3})
assert isinstance(result, list)
def test_find_optimal_empty(self, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
result = r._find_optimal_assignment([], mock_tvdb_episodes, {})
assert result == []
class TestIsValidAssignment:
def test_valid_assignment(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 55.0, "disc_number": None}
tvdb_ep = {"runtime": 55, "episode_number": 1}
result = r._is_valid_assignment(file_info, tvdb_ep, {}, 0, [])
assert result is True
def test_invalid_duration(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 30.0, "disc_number": None}
tvdb_ep = {"runtime": 60, "episode_number": 1}
result = r._is_valid_assignment(file_info, tvdb_ep, {}, 0, [])
assert result is False

View File

@ -1,194 +0,0 @@
"""Tests for src/Matcher/file_classifier.py - FileClassifier."""
import os
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.Matcher.file_classifier import FileClassifier, DEFAULT_VIDEO_EXTENSIONS
class TestFileClassifierInit:
def test_init_valid_folder(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
assert fc.folder_path == mock_video_files
assert len(fc.video_files) > 0
def test_init_nonexistent_folder(self, tmp_path):
with pytest.raises(FileNotFoundError, match="Folder not found"):
FileClassifier(str(tmp_path / "nope"))
def test_init_file_not_dir(self, tmp_path):
f = tmp_path / "file.txt"
f.write_text("hi")
fc = FileClassifier(str(f))
assert len(fc.video_files) == 0
def test_init_custom_extensions(self, tmp_path):
(tmp_path / "video.mp4").write_bytes(b"\x00" * 100)
(tmp_path / "video.mkv").write_bytes(b"\x00" * 100)
fc = FileClassifier(str(tmp_path), video_extensions=[".mp4"])
assert len(fc.video_files) == 1
assert fc.video_files[0].name == "video.mp4"
def test_init_fallback_ratio(self, mock_video_files):
fc = FileClassifier(str(mock_video_files), fallback_duration_minutes_per_gb=25.0)
assert fc.fallback_ratio == 25.0
def test_init_filters_macos_resource_fork(self, tmp_path):
(tmp_path / "._hidden.mkv").write_bytes(b"\x00" * 100)
(tmp_path / ".dot.mkv").write_bytes(b"\x00" * 100)
(tmp_path / "visible.mkv").write_bytes(b"\x00" * 100)
fc = FileClassifier(str(tmp_path))
assert len(fc.video_files) == 1
assert fc.video_files[0].name == "visible.mkv"
def test_init_deduplicates(self, tmp_path):
(tmp_path / "dup.mkv").write_bytes(b"\x00" * 100)
fc = FileClassifier(str(tmp_path), video_extensions=[".mkv"])
assert len(fc.video_files) == 1
def test_init_empty_folder(self, tmp_path):
fc = FileClassifier(str(tmp_path))
assert fc.video_files == []
class TestResolveExtensions:
def test_resolve_default(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
assert ".mkv" in fc.video_extensions
def test_resolve_custom(self, mock_video_files):
fc = FileClassifier(str(mock_video_files), video_extensions=["mp4", "avi"])
assert ".mp4" in fc.video_extensions
assert ".avi" in fc.video_extensions
assert ".mkv" not in fc.video_extensions
def test_resolve_env_var(self, mock_video_files, monkeypatch):
monkeypatch.setenv("VIDEO_EXTENSIONS", "webm,mp4")
fc = FileClassifier(str(mock_video_files))
assert "webm" in fc.video_extensions
assert "mp4" in fc.video_extensions
class TestResolveFallbackRatio:
def test_explicit_ratio(self, mock_video_files):
fc = FileClassifier(str(mock_video_files), fallback_duration_minutes_per_gb=30.0)
assert fc.fallback_ratio == 30.0
def test_env_ratio(self, mock_video_files, monkeypatch):
monkeypatch.setenv("FALLBACK_DURATION_RATIO", "33.0")
fc = FileClassifier(str(mock_video_files))
assert fc.fallback_ratio == 33.0
def test_env_ratio_invalid(self, mock_video_files, monkeypatch):
monkeypatch.setenv("FALLBACK_DURATION_RATIO", "abc")
fc = FileClassifier(str(mock_video_files))
assert fc.fallback_ratio == 45.0
class TestFileAnalysis:
def test_get_file_size(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
size = fc._get_file_size(mock_video_files / "Disc 1_t01.mkv")
assert size == int(2 * 1024)
def test_analyze_files_structure(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
assert len(fc.file_info) > 0
info = fc.file_info[0]
assert "path" in info
assert "name" in info
assert "size_bytes" in info
assert "size_mb" in info
assert "size_gb" in info
assert "duration_minutes" in info
def test_calculate_stats(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
stats = fc._calculate_stats()
assert stats["count"] == len(fc.file_info)
assert "avg_size_bytes" in stats
assert "avg_duration_minutes" in stats
assert "median_size_bytes" in stats
def test_calculate_stats_empty(self, tmp_path):
fc = FileClassifier(str(tmp_path))
stats = fc._calculate_stats()
assert stats == {}
class TestClassifyFiles:
def test_classify_no_files(self, tmp_path):
fc = FileClassifier(str(tmp_path))
episodes, extras = fc.classify_files()
assert episodes == []
assert extras == []
def test_classify_with_tvdb(self, mock_video_files, mock_tvdb_episodes):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files(
expected_episode_count=6, tvdb_episodes=mock_tvdb_episodes
)
assert len(episodes) <= 6
assert len(episodes) + len(extras) == len(fc.file_info)
def test_classify_by_stats(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files(expected_episode_count=6)
assert len(episodes) <= 6
assert len(episodes) > 0
def test_classify_extra_small_file(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files()
extra_names = [e["name"] for e in extras]
assert "extra_short.mkv" in extra_names
def test_classify_expected_count(self, mock_video_files, mock_tvdb_episodes):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files(
expected_episode_count=3, tvdb_episodes=mock_tvdb_episodes
)
assert len(episodes) <= 3
class TestVideoDuration:
@patch("src.Matcher.file_classifier.MediaInfo", None)
def test_duration_fallback_no_mediainfo(self, mock_video_files):
fc = FileClassifier(
str(mock_video_files), fallback_duration_minutes_per_gb=45.0
)
target = mock_video_files / "Disc 1_t01.mkv"
dur = fc._get_video_duration(target)
expected = (target.stat().st_size / (1024**3)) * 45.0
assert abs(dur - expected) < 0.001
assert dur > 0
def test_duration_fallback_custom_ratio(self, mock_video_files):
with patch("src.Matcher.file_classifier.MediaInfo", None):
fc = FileClassifier(
str(mock_video_files), fallback_duration_minutes_per_gb=25.0
)
target = mock_video_files / "Disc 1_t01.mkv"
dur = fc._get_video_duration(target)
expected = (target.stat().st_size / (1024**3)) * 25.0
assert abs(dur - expected) < 0.001
assert dur > 0
class TestPrintAnalysis:
def test_print_analysis_has_files(self, mock_video_files, capsys):
fc = FileClassifier(str(mock_video_files))
fc.print_analysis()
out = capsys.readouterr().out
assert "File Analysis" in out
assert "Total files" in out
def test_print_analysis_no_files(self, tmp_path, capsys):
fc = FileClassifier(str(tmp_path))
fc.print_analysis()
out = capsys.readouterr().out
assert "No video files found" in out

View File

@ -1,236 +0,0 @@
"""Tests for src/TVDBProvider/tvdb_cache.py - TVDBCache."""
import sys
import json
import time
from pathlib import Path
from datetime import datetime, timedelta
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.TVDBProvider.tvdb_cache import TVDBCache
class TestTVDBCacheInit:
def test_init_creates_dir(self, tmp_path):
cache = TVDBCache(str(tmp_path / "cache"))
assert (tmp_path / "cache").exists()
def test_init_default_dir(self):
cache = TVDBCache()
assert cache.cache_dir.exists()
assert cache.cache_duration == timedelta(days=7)
def test_init_absolute_path(self):
cache = TVDBCache("/tmp/absolute_cache_test")
try:
assert cache.cache_dir.is_absolute()
finally:
import shutil
if cache.cache_dir.exists():
shutil.rmtree(cache.cache_dir, ignore_errors=True)
class TestSanitizeName:
def test_basic_sanitization(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("Game of Thrones")
assert result == "game_of_thrones"
def test_special_chars_removed(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("The Simpsons (1989)")
assert "(" not in result
assert ")" not in result
def test_dots_stripped(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("../../etc/passwd")
assert ".." not in result
assert "/" not in result
def test_slashes_stripped(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("a/b")
assert result == "ab"
def test_backslash_stripped(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("a\\b")
assert result == "ab"
def test_numbers_preserved(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("Show 123")
assert "123" in result
def test_underscores_preserved(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("my_show")
assert result == "my_show"
class TestCacheKey:
def test_cache_key_format(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
key = cache._get_cache_key("Test Show", 3)
assert key == "test_show_s03"
def test_cache_key_special_chars(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
key = cache._get_cache_key("Test (Show)!", 1)
assert "(" not in key
assert ")" not in key
class TestCacheFile:
def test_cache_file_path(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = cache._get_cache_file("test_s01")
assert f.name == "test_s01.json"
def test_cache_file_escape_detected(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
with pytest.raises(ValueError, match="escape detected"):
cache._get_cache_file("../evil")
class TestCacheValidity:
def test_valid_cache(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = tmp_path / "c" / "valid.json"
f.write_text("data")
assert cache._is_cache_valid(f) is True
def test_expired_cache(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_duration = timedelta(seconds=0)
f = tmp_path / "c" / "old.json"
f.write_text("data")
time.sleep(0.1)
assert cache._is_cache_valid(f) is False
def test_nonexistent_cache(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
assert cache._is_cache_valid(tmp_path / "c" / "nope.json") is False
class TestCacheEpisodes:
def test_cache_and_retrieve(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
episodes = [{"episode_number": 1, "runtime": 45}]
result = cache.cache_episodes("Test Show", 1, episodes)
assert result is True
retrieved = cache.get_cached_episodes("Test Show", 1)
assert len(retrieved) == 1
assert retrieved[0]["episode_number"] == 1
def test_cache_write_failure(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_dir = Path("/nonexistent_dir")
result = cache.cache_episodes("Test", 1, [])
assert result is False
def test_cache_data_structure(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
episodes = [{"episode_number": 1, "runtime": 45}]
cache.cache_episodes("Test", 1, episodes)
cache_file = tmp_path / "c" / "test_s01.json"
data = json.loads(cache_file.read_text())
assert data["series_name"] == "Test"
assert data["season_number"] == 1
assert "cached_at" in data
assert len(data["episodes"]) == 1
def test_cache_empty_episodes(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
result = cache.cache_episodes("Empty", 1, [])
assert result is True
retrieved = cache.get_cached_episodes("Empty", 1)
assert retrieved == []
class TestGetCachedEpisodes:
def test_cache_miss(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
result = cache.get_cached_episodes("NoShow", 1)
assert result is None
def test_corrupted_json(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = tmp_path / "c" / "broken.json"
f.write_text("{invalid")
result = cache.get_cached_episodes("Broken", 1)
assert result is None
def test_missing_fields(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = tmp_path / "c" / "partial_s01.json"
f.write_text(json.dumps({"only": "this"}))
result = cache.get_cached_episodes("Partial", 1)
assert result is None
class TestClearCache:
def test_clear_removes_files(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Show1", 1, [])
cache.cache_episodes("Show2", 1, [])
result = cache.clear_cache()
assert result is True
files = list((tmp_path / "c").glob("*.json"))
assert len(files) == 0
def test_clear_empty(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
assert cache.clear_cache() is True
class TestListCache:
def test_list_entries(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Show1", 1, [{"ep": 1}])
cache.cache_episodes("Show2", 2, [{"ep": 1}, {"ep": 2}])
entries = cache.list_cache()
assert len(entries) == 2
for e in entries:
assert "series_name" in e
assert "season_number" in e
assert "episode_count" in e
assert "is_valid" in e
def test_list_empty(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
entries = cache.list_cache()
assert entries == []
def test_list_entry_fields(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Test", 1, [{"e": 1}])
entries = cache.list_cache()
e = entries[0]
assert e["series_name"] == "Test"
assert e["season_number"] == 1
assert e["episode_count"] == 1
assert "cached_at" in e
assert "age_days" in e
assert "cache_key" in e
def test_list_sorted_by_date(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("First", 1, [])
time.sleep(0.01)
cache.cache_episodes("Second", 1, [])
entries = cache.list_cache()
assert entries[0]["series_name"] == "Second"
def test_list_corrupted_file(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Good", 1, [{"e": 1}])
(tmp_path / "c" / "bad_s01.json").write_text("{bad}")
entries = cache.list_cache()
assert len(entries) == 1

View File

@ -1,224 +0,0 @@
"""Tests for src/TVDBProvider/tvdb_client.py."""
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock, PropertyMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.TVDBProvider.tvdb_client import TVDBClient, MockTVDBClient
class TestTVDBClientInit:
def test_init(self):
client = TVDBClient()
assert client.base_url == "https://api4.thetvdb.com/v4"
assert client.token is None
assert "Content-Type" in client.headers
def test_init_has_cache(self):
client = TVDBClient()
assert client.cache is not None
class TestTVDBClientAuthenticate:
@patch("src.TVDBProvider.tvdb_client.requests.post")
def test_auth_success(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {"data": {"token": "abc123"}}
client = TVDBClient()
assert client.authenticate("my_api_key") is True
assert client.token == "abc123"
assert "Bearer abc123" in client.headers["Authorization"]
@patch("src.TVDBProvider.tvdb_client.requests.post")
def test_auth_failure_status(self, mock_post):
mock_post.return_value.status_code = 401
client = TVDBClient()
assert client.authenticate("bad_key") is False
assert client.token is None
@patch("src.TVDBProvider.tvdb_client.requests.post")
def test_auth_exception(self, mock_post):
mock_post.side_effect = Exception("network error")
client = TVDBClient()
assert client.authenticate("key") is False
class TestTVDBClientSearch:
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_search_success(self, mock_get):
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"data": [{"tvdb_id": 123, "name": "Test Show"}]
}
client = TVDBClient()
client.token = "tok"
result = client.search_series("Test Show")
assert result["tvdb_id"] == 123
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_search_no_results(self, mock_get):
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"data": []}
client = TVDBClient()
client.token = "tok"
result = client.search_series("NoMatch")
assert result is None
def test_search_not_authenticated(self):
client = TVDBClient()
result = client.search_series("Show")
assert result is None
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_search_error(self, mock_get):
mock_get.side_effect = Exception("timeout")
client = TVDBClient()
client.token = "tok"
result = client.search_series("Show")
assert result is None
class TestTVDBClientSeasonEpisodes:
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_get_season_episodes(self, mock_get):
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"data": {"episodes": [{"number": 1, "name": "E1"}]}
}
client = TVDBClient()
client.token = "tok"
result = client.get_season_episodes(123, 1)
assert len(result) == 1
def test_episodes_not_authenticated(self):
client = TVDBClient()
result = client.get_season_episodes(123, 1)
assert result is None
class TestTVDBClientSeriesInfo:
@patch.object(TVDBClient, "search_series")
def test_get_series_info(self, mock_search):
mock_search.return_value = {"tvdb_id": 123, "name": "Test", "slug": "test", "year": 2020}
client = TVDBClient()
client.token = "tok"
result = client.get_series_info("Test")
assert result["id"] == 123
assert result["name"] == "Test"
@patch.object(TVDBClient, "search_series")
def test_get_series_info_no_match(self, mock_search):
mock_search.return_value = None
client = TVDBClient()
client.token = "tok"
result = client.get_series_info("NoMatch")
assert result is None
class TestTVDBClientEpisodeDurations:
def test_get_episode_durations(self, tmp_path):
with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init:
mock_init.return_value = None
client = TVDBClient()
client.base_url = "https://api4.thetvdb.com/v4"
client.token = "tok"
client.headers = {"Content-Type": "application/json"}
mock_cache = MagicMock()
mock_cache.get_cached_episodes.return_value = None
client.cache = mock_cache
client.get_series_info = MagicMock(return_value={"id": 123})
client.get_season_episodes = MagicMock(return_value=[
{"number": 1, "name": "E1", "runtime": 45, "aired": "2020-01-01"},
{"number": 2, "name": "E2", "runtime": 50, "aired": "2020-01-08"},
])
result = client.get_episode_durations("Test", 1)
assert len(result) == 2
assert result[0]["episode_number"] == 1
assert result[0]["runtime"] == 45
mock_cache.cache_episodes.assert_called_once()
def test_get_episode_durations_cached(self, tmp_path):
with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init:
mock_init.return_value = None
client = TVDBClient()
client.token = "tok"
mock_cache = MagicMock()
mock_cache.get_cached_episodes.return_value = [
{"episode_number": 1, "runtime": 45}
]
client.cache = mock_cache
result = client.get_episode_durations("Test", 1)
assert len(result) == 1
def test_get_episode_durations_no_series(self, tmp_path):
with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init:
mock_init.return_value = None
client = TVDBClient()
client.token = "tok"
mock_cache = MagicMock()
mock_cache.get_cached_episodes.return_value = None
client.cache = mock_cache
client.get_series_info = MagicMock(return_value=None)
result = client.get_episode_durations("NoShow", 1)
assert result is None
class TestMockTVDBClient:
def test_mock_auth(self):
client = MockTVDBClient()
assert client.authenticate() is True
assert client.authenticated is True
def test_mock_auth_with_key(self):
client = MockTVDBClient()
assert client.authenticate("some_key") is True
def test_mock_episodes_unauthenticated(self):
client = MockTVDBClient()
result = client.get_episode_durations("Show", 1)
assert result is None
def test_mock_episodes_drama(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Game of Thrones", 1)
assert len(result) == 10
def test_mock_episodes_sitcom(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Friends", 1)
assert len(result) == 24
def test_mock_episodes_default(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Unknown Show", 1)
assert len(result) == 22
def test_mock_episode_structure(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Test", 1)
ep = result[0]
assert ep["episode_number"] == 1
assert ep["runtime"] == 45
assert "name" in ep
assert "aired" in ep
def test_mock_cached_episodes(self):
client = MockTVDBClient()
client.authenticate()
r1 = client.get_episode_durations("Test", 1)
r2 = client.get_episode_durations("Test", 1)
assert r1 is not None and r2 is not None