commit a8b247a83fba440c36386672f079a6149fd54632 Author: Dan Head Date: Wed Jan 7 17:43:09 2026 +0000 Initial commit: TVH2Jellyfin application A Docker-based Python application that transfers completed recordings from TVHeadend to Jellyfin media storage via SCP. Features: - Monitors TVHeadend for completed recordings via JSON API - Transfers files via SCP with SSH key authentication - Organizes files by show name - Deletes recordings from TVHeadend after successful transfer - Tracks transferred files to avoid duplicates - Dry run mode for testing without making changes diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ed09e6c --- /dev/null +++ b/.env.example @@ -0,0 +1,45 @@ +# TVHeadend server configuration +TVH_HOST=192.168.1.100 +TVH_PORT=9981 +TVH_USER=admin +TVH_PASS=your_tvheadend_password + +# TVHeadend SSH user (for SCP file transfer, defaults to TVH_USER) +TVH_SSH_USER=admin +# SSH port (defaults to 22) +TVH_SSH_PORT=22 + +# Path to SSH private key on the Docker host (for volume mount) +SSH_KEY_HOST_PATH=./id_rsa + +# Path to Jellyfin media library on the Docker host +JELLYFIN_MEDIA_PATH=/mnt/jellyfin/tv + +# Application settings +POLL_INTERVAL=300 +LOG_LEVEL=INFO + +# Dry run mode - set to true to test without transferring/deleting +DRY_RUN=false + +# TVDB configuration (optional - for episode lookup) +# Get your API key from https://thetvdb.com/api-information +TVDB_API_KEY= +TVDB_PIN= +TVDB_ENABLED=true +TVDB_CACHE_TTL=86400 +# Optional JSON file mapping TVHeadend show names to TVDB names +# TVDB_SHOW_MAPPING=/app/data/show_mapping.json + +# TVMaze configuration (optional - fallback for episode lookup) +# No API key required - public API +TVMAZE_ENABLED=true +TVMAZE_CACHE_TTL=86400 + +# Jellyfin API configuration (optional - for library scan triggers) +# Get your API key from Jellyfin Admin Dashboard > API Keys +JELLYFIN_URL=http://localhost:8096 +JELLYFIN_API_KEY= +# Optional: scan only a specific library (get ID from get_libraries()) +# JELLYFIN_LIBRARY_ID= +JELLYFIN_ENABLED=true diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..56995a5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,50 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual environments +venv/ +ENV/ +env/ +.venv/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# Environment +.env +.env.local + +# Application data +data/ + +# SSH keys (never commit these!) +*.pem +id_rsa +id_ed25519 diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..415d2b8 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,42 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Build & Development Commands + +```bash +# Install for development +pip install -e ".[dev]" + +# Run the application +python -m tvh2jellyfin.main + +# Run tests +pytest + +# Run single test +pytest tests/test_file.py::test_function + +# Lint +ruff check src/ + +# Format +ruff format src/ +``` + +## Architecture + +This application transfers TV recordings from a remote TVHeadend server to a local Jellyfin media library. + +**Data Flow:** +1. `tvheadend.py` - Queries TVHeadend JSON API for completed recordings +2. `transfer.py` - Downloads files via SCP from TVHeadend server to local destination +3. `organizer.py` - Generates destination paths organized by show name +4. `state.py` - Tracks transferred recordings (JSON file) to avoid duplicates +5. `main.py` - Orchestrates the sync loop with scheduling + +**Key Design Decisions:** +- SCP pulls files FROM TVHeadend (source is remote, destination is local Docker volume) +- TVHeadend API used for both discovery (grid_finished) and cleanup (remove after transfer) +- Dry run mode skips SSH connection and all mutations +- State persisted as JSON for Docker volume compatibility diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..a111cde --- /dev/null +++ b/Dockerfile @@ -0,0 +1,31 @@ +FROM python:3.12-slim + +WORKDIR /app + +# Install system dependencies for paramiko +RUN apt-get update && apt-get install -y --no-install-recommends \ + openssh-client \ + && rm -rf /var/lib/apt/lists/* + +# Copy requirements first for better caching +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application code +COPY src/ ./src/ +COPY pyproject.toml . + +# Install the application +RUN pip install --no-cache-dir . + +# Create data directory for state persistence +RUN mkdir -p /app/data + +# Create .ssh directory for keys (owned by runtime user 1000:1000) +RUN mkdir -p /app/.ssh && chmod 700 /app/.ssh && chown 1000:1000 /app/.ssh + +ENV PYTHONUNBUFFERED=1 +ENV STATE_FILE=/app/data/state.json +ENV SSH_KEY_PATH=/app/.ssh/id_rsa + +CMD ["python", "-m", "tvh2jellyfin.main"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..7c00a9c --- /dev/null +++ b/README.md @@ -0,0 +1,211 @@ +# TVH2Jellyfin + +Transfer completed recordings from TVHeadend to Jellyfin media storage. + +## Features + +- Monitors TVHeadend for completed recordings via JSON API +- Pulls files from TVHeadend server via SCP +- Writes to local Jellyfin media directory (Docker volume) +- Organizes files by show name: `{dest_path}/{show_name}/{episode}.ts` +- Deletes recordings from TVHeadend after successful transfer +- Tracks transferred files to avoid duplicates +- Runs as a Docker container +- Dry run mode for testing without making changes + +## Prerequisites + +1. SSH key access from the Docker host to the TVHeadend server +2. TVHeadend API access (username/password) +3. Docker and Docker Compose + +## Setup + +### 1. Generate SSH Key (if needed) + +```bash +ssh-keygen -t rsa -b 4096 -f ~/.ssh/tvh2jellyfin +ssh-copy-id -i ~/.ssh/tvh2jellyfin.pub user@tvheadend-server +``` + +### 2. Configure Environment + +Copy the example environment file and edit with your settings: + +```bash +cp .env.example .env +``` + +Edit `.env` with your configuration: + +```bash +# TVHeadend server +TVH_HOST=192.168.1.100 +TVH_PORT=9981 +TVH_USER=admin +TVH_PASS=your_password + +# SSH user on TVHeadend server (defaults to TVH_USER) +TVH_SSH_USER=admin + +# SSH key path on Docker host +SSH_KEY_PATH=~/.ssh/tvh2jellyfin + +# Jellyfin media library path on Docker host +JELLYFIN_MEDIA_PATH=/mnt/jellyfin/tv + +# Poll interval in seconds (default: 300 = 5 minutes) +POLL_INTERVAL=300 +``` + +### 3. Build and Run + +```bash +docker compose up -d +``` + +### 4. View Logs + +```bash +docker compose logs -f +``` + +## Configuration Options + +| Variable | Description | Default | +|----------|-------------|---------| +| `TVH_HOST` | TVHeadend server hostname/IP | `localhost` | +| `TVH_PORT` | TVHeadend HTTP port | `9981` | +| `TVH_USER` | TVHeadend API username | (required) | +| `TVH_PASS` | TVHeadend API password | (required) | +| `TVH_SSH_USER` | SSH username for TVHeadend server | `TVH_USER` | +| `SSH_KEY_PATH` | Path to SSH private key on Docker host | `~/.ssh/id_rsa` | +| `JELLYFIN_MEDIA_PATH` | Jellyfin media path on Docker host | `/mnt/jellyfin/tv` | +| `POLL_INTERVAL` | Seconds between checks | `300` | +| `LOG_LEVEL` | Logging level | `INFO` | +| `DRY_RUN` | Test mode (no transfers/deletions) | `false` | + +## How It Works + +``` +┌─────────────────────┐ ┌─────────────────────┐ +│ TVHeadend Server │ │ Docker Host │ +│ │ │ │ +│ ┌───────────────┐ │ HTTP API │ ┌───────────────┐ │ +│ │ TVHeadend │◄─┼────────────────────┼──│ tvh2jellyfin │ │ +│ │ API (:9981) │ │ │ │ container │ │ +│ └───────────────┘ │ │ └───────┬───────┘ │ +│ │ │ │ │ +│ ┌───────────────┐ │ SCP (SSH) │ ▼ │ +│ │ Recordings │──┼────────────────────┼──►┌─────────────┐ │ +│ │ /recordings │ │ │ │ /media/tv │ │ +│ └───────────────┘ │ │ │ (Jellyfin) │ │ +│ │ │ └─────────────┘ │ +└─────────────────────┘ └─────────────────────┘ +``` + +1. Container queries TVHeadend API for completed recordings +2. Downloads files via SCP from TVHeadend server +3. Writes to local Jellyfin media volume +4. Removes recording from TVHeadend via API + +## File Organization + +Recordings are organized by show name: + +``` +/mnt/jellyfin/tv/ +├── Show Name/ +│ ├── Show Name - Episode Title.ts +│ └── Show Name - 2024-01-15 2000.ts (if no episode title) +└── Another Show/ + └── Another Show - Episode.ts +``` + +## Dry Run Mode + +Test what would happen without actually transferring or deleting anything: + +```bash +DRY_RUN=true docker compose up +``` + +In dry run mode: +- Connects to TVHeadend API and lists completed recordings +- Shows what files would be transferred and where +- Shows which recordings have already been transferred +- Does NOT connect via SSH +- Does NOT transfer any files +- Does NOT delete recordings from TVHeadend +- Does NOT update the state file +- Exits after a single check (no polling loop) + +Example output: +``` +[DRY RUN] Starting sync cycle... +[DRY RUN] Found 2 new recording(s) to transfer +[DRY RUN] Would process: The News - Evening Edition +[DRY RUN] Source: /recordings/The News - Evening Edition.ts +[DRY RUN] Destination: /media/tv/The News/The News - Evening Edition.ts +[DRY RUN] Size: 1250.5 MB +[DRY RUN] Would delete from TVHeadend after transfer +``` + +## Development + +### Install Dependencies + +```bash +python -m venv venv +source venv/bin/activate +pip install -e ".[dev]" +``` + +### Run Tests + +```bash +pytest +``` + +### Run Locally + +```bash +export TVH_HOST=... +export TVH_USER=... +export TVH_PASS=... +export TVH_SSH_USER=... +export DEST_PATH=/path/to/destination +python -m tvh2jellyfin.main +``` + +## Troubleshooting + +### SSH Connection Issues + +Ensure the SSH key has correct permissions: +```bash +chmod 600 ~/.ssh/tvh2jellyfin +``` + +Test SSH connection manually: +```bash +ssh -i ~/.ssh/tvh2jellyfin user@tvheadend-server +``` + +### TVHeadend API Issues + +Test API access: +```bash +curl -u admin:password http://tvheadend:9981/api/serverinfo +``` + +### Permission Issues + +Ensure the Jellyfin media directory is writable: +```bash +touch /mnt/jellyfin/tv/.test && rm /mnt/jellyfin/tv/.test +``` + +### View Transfer History + +The state file (`data/state.json`) contains all transfer records. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..90e143a --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,28 @@ +services: + tvh2jellyfin: + user: "1000:1000" + build: . + container_name: tvh2jellyfin + environment: + # TVHeadend connection (API) + - TVH_HOST=${TVH_HOST:-192.168.1.100} + - TVH_PORT=${TVH_PORT:-9981} + - TVH_USER=${TVH_USER} + - TVH_PASS=${TVH_PASS} + # TVHeadend SSH access (for file transfer) + - TVH_SSH_USER=${TVH_SSH_USER:-${TVH_USER}} + - SSH_KEY_PATH=/app/.ssh/id_rsa + # Local destination path (inside container) + - DEST_PATH=/media/tv + # Application settings + - POLL_INTERVAL=${POLL_INTERVAL:-300} + - LOG_LEVEL=${LOG_LEVEL:-INFO} + - DRY_RUN=${DRY_RUN:-false} + volumes: + # State persistence + - ./data:/app/data + # SSH key for SCP from TVHeadend (host path, read-only) + - ${SSH_KEY_HOST_PATH:-~/.ssh/id_rsa}:/app/.ssh/id_rsa:ro + # Jellyfin media library (destination) + - ${JELLYFIN_MEDIA_PATH:-/mnt/jellyfin/tv}:/media/tv + restart: unless-stopped diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..acdd55f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,44 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "tvh2jellyfin" +version = "0.1.0" +description = "Transfer recordings from TVHeadend to Jellyfin" +readme = "README.md" +requires-python = ">=3.10" +license = {text = "MIT"} +authors = [ + {name = "Dan"} +] +dependencies = [ + "requests>=2.31.0", + "paramiko>=3.4.0", + "schedule>=1.2.0", + "python-dotenv>=1.0.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-cov>=4.1.0", + "ruff>=0.1.0", +] + +[project.scripts] +tvh2jellyfin = "tvh2jellyfin.main:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "N", "W"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1dd61df --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +requests>=2.31.0 +paramiko>=3.4.0 +schedule>=1.2.0 +python-dotenv>=1.0.0 diff --git a/src/tvh2jellyfin/__init__.py b/src/tvh2jellyfin/__init__.py new file mode 100644 index 0000000..9e77845 --- /dev/null +++ b/src/tvh2jellyfin/__init__.py @@ -0,0 +1,3 @@ +"""TVH2Jellyfin - Transfer recordings from TVHeadend to Jellyfin.""" + +__version__ = "0.1.0" diff --git a/src/tvh2jellyfin/config.py b/src/tvh2jellyfin/config.py new file mode 100644 index 0000000..bd219bc --- /dev/null +++ b/src/tvh2jellyfin/config.py @@ -0,0 +1,127 @@ +"""Configuration management for TVH2Jellyfin.""" + +import os +from dataclasses import dataclass +from pathlib import Path + +from dotenv import load_dotenv + +from .jellyfin import JellyfinConfig +from .tvdb import TVDBConfig +from .tvmaze import TVMazeConfig + + +@dataclass +class TVHeadendConfig: + """TVHeadend server configuration for API and SSH access.""" + + host: str + port: int + username: str + password: str + ssh_user: str + ssh_key_path: str + ssh_port: int = 22 + + @property + def base_url(self) -> str: + return f"http://{self.host}:{self.port}" + + +@dataclass +class Config: + """Application configuration.""" + + tvheadend: TVHeadendConfig + dest_path: str + poll_interval: int + state_file: str + log_level: str + dry_run: bool + tvdb: TVDBConfig | None + tvmaze: TVMazeConfig + jellyfin: JellyfinConfig | None + + @classmethod + def from_env(cls) -> "Config": + """Load configuration from environment variables.""" + load_dotenv() + + tvh_host = os.environ.get("TVH_HOST", "localhost") + + tvheadend = TVHeadendConfig( + host=tvh_host, + port=int(os.environ.get("TVH_PORT", "9981")), + username=os.environ.get("TVH_USER", ""), + password=os.environ.get("TVH_PASS", ""), + ssh_user=os.environ.get("TVH_SSH_USER", os.environ.get("TVH_USER", "")), + ssh_key_path=os.environ.get("SSH_KEY_PATH", str(Path.home() / ".ssh" / "id_rsa")), + ssh_port=int(os.environ.get("TVH_SSH_PORT", "22")), + ) + + # TVDB configuration (optional) + tvdb_api_key = os.environ.get("TVDB_API_KEY", "") + tvdb: TVDBConfig | None = None + if tvdb_api_key: + tvdb = TVDBConfig( + api_key=tvdb_api_key, + pin=os.environ.get("TVDB_PIN", ""), + cache_ttl=int(os.environ.get("TVDB_CACHE_TTL", "86400")), + show_mapping_file=os.environ.get("TVDB_SHOW_MAPPING"), + enabled=os.environ.get("TVDB_ENABLED", "true").lower() + in ("true", "1", "yes"), + ) + + # TVmaze configuration (always available as fallback) + tvmaze = TVMazeConfig( + cache_ttl=int(os.environ.get("TVMAZE_CACHE_TTL", "86400")), + enabled=os.environ.get("TVMAZE_ENABLED", "true").lower() + in ("true", "1", "yes"), + ) + + # Jellyfin configuration (optional) + jellyfin_url = os.environ.get("JELLYFIN_URL", "") + jellyfin_api_key = os.environ.get("JELLYFIN_API_KEY", "") + jellyfin: JellyfinConfig | None = None + if jellyfin_url and jellyfin_api_key: + jellyfin = JellyfinConfig( + url=jellyfin_url, + api_key=jellyfin_api_key, + library_id=os.environ.get("JELLYFIN_LIBRARY_ID") or None, + enabled=os.environ.get("JELLYFIN_ENABLED", "true").lower() + in ("true", "1", "yes"), + ) + + return cls( + tvheadend=tvheadend, + dest_path=os.environ.get("DEST_PATH", "/media/tv"), + poll_interval=int(os.environ.get("POLL_INTERVAL", "300")), + state_file=os.environ.get("STATE_FILE", "/app/data/state.json"), + log_level=os.environ.get("LOG_LEVEL", "INFO"), + dry_run=os.environ.get("DRY_RUN", "").lower() in ("true", "1", "yes"), + tvdb=tvdb, + tvmaze=tvmaze, + jellyfin=jellyfin, + ) + + def validate(self) -> list[str]: + """Validate configuration and return list of errors.""" + errors = [] + + if not self.tvheadend.host: + errors.append("TVH_HOST is required") + if not self.tvheadend.username: + errors.append("TVH_USER is required") + if not self.tvheadend.password: + errors.append("TVH_PASS is required") + + # SSH and destination validation only needed when not in dry run mode + if not self.dry_run: + if not self.tvheadend.ssh_user: + errors.append("TVH_SSH_USER is required") + if not Path(self.tvheadend.ssh_key_path).exists(): + errors.append(f"SSH key not found at {self.tvheadend.ssh_key_path}") + if not self.dest_path: + errors.append("DEST_PATH is required") + + return errors diff --git a/src/tvh2jellyfin/episode_resolver.py b/src/tvh2jellyfin/episode_resolver.py new file mode 100644 index 0000000..f94ad72 --- /dev/null +++ b/src/tvh2jellyfin/episode_resolver.py @@ -0,0 +1,99 @@ +"""Episode resolver that orchestrates lookups across multiple sources.""" + +import logging + +from .tvdb import EpisodeInfo, TVDBClient, TVDBConfig +from .tvmaze import TVMazeClient, TVMazeConfig + +logger = logging.getLogger(__name__) + + +class EpisodeResolver: + """ + Resolves episode information using multiple sources. + + Tries sources in order: + 1. TheTVDB (primary) + 2. TVmaze (fallback) + """ + + def __init__( + self, + tvdb_client: TVDBClient | None = None, + tvmaze_client: TVMazeClient | None = None, + ): + self.tvdb_client = tvdb_client + self.tvmaze_client = tvmaze_client + + @classmethod + def create( + cls, + tvdb_config: TVDBConfig | None = None, + tvmaze_config: TVMazeConfig | None = None, + ) -> "EpisodeResolver": + """ + Create an EpisodeResolver with configured clients. + + Args: + tvdb_config: TVDB configuration (optional) + tvmaze_config: TVmaze configuration (optional, defaults to enabled) + + Returns: + Configured EpisodeResolver instance + """ + tvdb_client = None + tvmaze_client = None + + if tvdb_config and tvdb_config.enabled and tvdb_config.api_key: + tvdb_client = TVDBClient(tvdb_config) + logger.info("TVDB client initialized") + + # TVmaze is always available as fallback (no API key required) + if tvmaze_config is None: + tvmaze_config = TVMazeConfig() + + if tvmaze_config.enabled: + tvmaze_client = TVMazeClient(tvmaze_config) + logger.info("TVmaze client initialized (fallback)") + + return cls(tvdb_client=tvdb_client, tvmaze_client=tvmaze_client) + + def lookup( + self, + show_name: str, + broadcast_timestamp: int, + ) -> tuple[EpisodeInfo | None, str]: + """ + Look up episode information from available sources. + + Args: + show_name: Show name from TVHeadend recording + broadcast_timestamp: Unix timestamp of broadcast start + + Returns: + Tuple of (EpisodeInfo or None, source name) + Source name is one of: "tvdb", "tvmaze", "none" + """ + # Try TVDB first + if self.tvdb_client: + try: + episode = self.tvdb_client.lookup_episode(show_name, broadcast_timestamp) + if episode: + return episode, "tvdb" + except Exception as e: + logger.warning(f"TVDB lookup error: {e}") + + # Fall back to TVmaze + if self.tvmaze_client: + try: + episode = self.tvmaze_client.lookup_episode(show_name, broadcast_timestamp) + if episode: + return episode, "tvmaze" + except Exception as e: + logger.warning(f"TVmaze lookup error: {e}") + + return None, "none" + + def is_available(self) -> bool: + """Check if at least one lookup source is available.""" + return self.tvdb_client is not None or self.tvmaze_client is not None diff --git a/src/tvh2jellyfin/jellyfin.py b/src/tvh2jellyfin/jellyfin.py new file mode 100644 index 0000000..1ff1299 --- /dev/null +++ b/src/tvh2jellyfin/jellyfin.py @@ -0,0 +1,105 @@ +"""Jellyfin API client for triggering library scans.""" + +import logging +from dataclasses import dataclass + +import requests + +logger = logging.getLogger(__name__) + + +@dataclass +class JellyfinConfig: + """Jellyfin server configuration.""" + + url: str + api_key: str + library_id: str | None = None # Optional: scan specific library only + enabled: bool = True + + +class JellyfinClient: + """Client for Jellyfin API to trigger library scans.""" + + def __init__(self, config: JellyfinConfig): + self.config = config + self.session = requests.Session() + self.session.headers.update({ + "X-MediaBrowser-Token": config.api_key, + "Content-Type": "application/json", + }) + # Normalize URL (remove trailing slash) + self.base_url = config.url.rstrip("/") + + def _api_post(self, endpoint: str) -> bool: + """Make POST request to Jellyfin API.""" + url = f"{self.base_url}{endpoint}" + + try: + response = self.session.post(url, timeout=30) + response.raise_for_status() + return True + except requests.exceptions.RequestException as e: + logger.error(f"Jellyfin API call failed: {endpoint} - {e}") + return False + + def _api_get(self, endpoint: str) -> dict | None: + """Make GET request to Jellyfin API.""" + url = f"{self.base_url}{endpoint}" + + try: + response = self.session.get(url, timeout=30) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + logger.error(f"Jellyfin API call failed: {endpoint} - {e}") + return None + + def test_connection(self) -> bool: + """Test connection to Jellyfin server.""" + try: + data = self._api_get("/System/Info") + if data: + server_name = data.get("ServerName", "Unknown") + version = data.get("Version", "Unknown") + logger.info(f"Connected to Jellyfin: {server_name} v{version}") + return True + return False + except Exception as e: + logger.error(f"Failed to connect to Jellyfin: {e}") + return False + + def trigger_library_scan(self) -> bool: + """ + Trigger a library scan in Jellyfin. + + If a specific library_id is configured, only that library is scanned. + Otherwise, triggers a full library refresh. + """ + if self.config.library_id: + # Scan specific library + endpoint = f"/Items/{self.config.library_id}/Refresh" + logger.info(f"Triggering Jellyfin library scan for library: {self.config.library_id}") + else: + # Full library refresh + endpoint = "/Library/Refresh" + logger.info("Triggering full Jellyfin library scan") + + success = self._api_post(endpoint) + + if success: + logger.info("Jellyfin library scan triggered successfully") + else: + logger.warning("Failed to trigger Jellyfin library scan") + + return success + + def get_libraries(self) -> list[dict]: + """Get list of media libraries (useful for finding library IDs).""" + data = self._api_get("/Library/VirtualFolders") + if data: + return [ + {"name": lib.get("Name"), "id": lib.get("ItemId")} + for lib in data + ] + return [] diff --git a/src/tvh2jellyfin/main.py b/src/tvh2jellyfin/main.py new file mode 100644 index 0000000..62e863a --- /dev/null +++ b/src/tvh2jellyfin/main.py @@ -0,0 +1,311 @@ +"""Main application entry point for TVH2Jellyfin.""" + +import logging +import signal +import sys +import time + +import schedule + +from .config import Config +from .episode_resolver import EpisodeResolver +from .jellyfin import JellyfinClient +from .organizer import generate_destination_path_with_episode +from .state import StateManager, TransferRecord +from .transfer import TransferError, TransferService +from .tvheadend import Recording, TVHeadendClient + +logger = logging.getLogger(__name__) + +# Global flag for graceful shutdown +shutdown_requested = False + + +def setup_logging(level: str) -> None: + """Configure logging for the application.""" + logging.basicConfig( + level=getattr(logging, level.upper(), logging.INFO), + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + handlers=[logging.StreamHandler(sys.stdout)], + ) + + +def signal_handler(signum: int, frame) -> None: + """Handle shutdown signals gracefully.""" + global shutdown_requested + logger.info(f"Received signal {signum}, initiating shutdown...") + shutdown_requested = True + + +def process_recording( + recording: Recording, + tvh_client: TVHeadendClient, + transfer_service: TransferService | None, + state_manager: StateManager, + dest_base_path: str, + episode_resolver: EpisodeResolver | None = None, + dry_run: bool = False, +) -> bool: + """ + Process a single recording: transfer, verify, and cleanup. + + Returns True if successful, False otherwise. + """ + # Look up episode information + episode_info = None + source_name = "none" + if episode_resolver: + episode_info, source_name = episode_resolver.lookup( + recording.title, recording.start + ) + + # Generate destination path using episode info if available + dest_dir, dest_filename = generate_destination_path_with_episode( + recording, dest_base_path, episode_info + ) + full_dest_path = f"{dest_dir}/{dest_filename}" + + if dry_run: + logger.info(f"[DRY RUN] Would process: {recording.display_name}") + logger.info(f"[DRY RUN] Source: {recording.filename}") + logger.info(f"[DRY RUN] Destination: {full_dest_path}") + if episode_info: + logger.info( + f"[DRY RUN] Episode: {episode_info.series_name} " + f"{episode_info.formatted_episode} (via {source_name})" + ) + else: + logger.info("[DRY RUN] Episode: Not found (using fallback naming)") + logger.info(f"[DRY RUN] Size: {recording.filesize / 1024 / 1024:.1f} MB") + logger.info("[DRY RUN] Would delete from TVHeadend after transfer") + return True + + logger.info(f"Processing: {recording.display_name}") + + if transfer_service is None: + logger.error("Transfer service not initialized") + return False + + try: + # Transfer the file + transfer_service.transfer_file( + source_path=recording.filename, + dest_dir=dest_dir, + dest_filename=dest_filename, + expected_size=recording.filesize, + ) + + # Verify the transfer + if not transfer_service.verify_transfer(full_dest_path, recording.filesize): + logger.error(f"Transfer verification failed for {recording.display_name}") + transfer_service.delete_local_file(full_dest_path) + return False + + # Remove from TVHeadend + if not tvh_client.remove_recording(recording.uuid): + logger.warning( + f"Transfer succeeded but failed to remove from TVHeadend: {recording.uuid}" + ) + # Continue anyway - the file is transferred + + # Record successful transfer + record = TransferRecord.create( + uuid=recording.uuid, + title=recording.display_name, + source_path=recording.filename, + dest_path=full_dest_path, + filesize=recording.filesize, + ) + state_manager.mark_transferred(record) + + logger.info(f"Successfully processed: {recording.display_name}") + return True + + except TransferError as e: + logger.error(f"Transfer failed for {recording.display_name}: {e}") + return False + + +def run_sync( + tvh_client: TVHeadendClient, + transfer_service: TransferService | None, + state_manager: StateManager, + dest_base_path: str, + episode_resolver: EpisodeResolver | None = None, + jellyfin_client: JellyfinClient | None = None, + dry_run: bool = False, +) -> None: + """Run a single sync cycle.""" + prefix = "[DRY RUN] " if dry_run else "" + logger.info(f"{prefix}Starting sync cycle...") + + try: + # Get finished recordings from TVHeadend + recordings = tvh_client.get_finished_recordings() + + if not recordings: + logger.info(f"{prefix}No completed recordings found") + return + + # Filter out already transferred (skip in dry run to show all) + if dry_run: + new_recordings = recordings + already_transferred = [r for r in recordings if state_manager.is_transferred(r.uuid)] + if already_transferred: + logger.info(f"{prefix}Already transferred ({len(already_transferred)}):") + for r in already_transferred: + logger.info(f"{prefix} - {r.display_name}") + new_recordings = [r for r in recordings if not state_manager.is_transferred(r.uuid)] + else: + new_recordings = [r for r in recordings if not state_manager.is_transferred(r.uuid)] + + if not new_recordings: + logger.info(f"{prefix}No new recordings to transfer") + return + + logger.info(f"{prefix}Found {len(new_recordings)} new recording(s) to transfer") + + # Process each recording + success_count = 0 + for recording in new_recordings: + if shutdown_requested: + logger.info("Shutdown requested, stopping sync") + break + + if process_recording( + recording, + tvh_client, + transfer_service, + state_manager, + dest_base_path, + episode_resolver=episode_resolver, + dry_run=dry_run, + ): + success_count += 1 + + logger.info(f"{prefix}Sync cycle complete: {success_count}/{len(new_recordings)} processed") + + # Trigger Jellyfin library scan if any transfers succeeded + if success_count > 0 and jellyfin_client and not dry_run: + jellyfin_client.trigger_library_scan() + + except Exception as e: + logger.error(f"Sync cycle failed: {e}") + + +def main() -> None: + """Main entry point.""" + global shutdown_requested + + # Load configuration + config = Config.from_env() + setup_logging(config.log_level) + + if config.dry_run: + logger.info("TVH2Jellyfin starting in DRY RUN mode...") + logger.info("No files will be transferred, renamed, or deleted") + else: + logger.info("TVH2Jellyfin starting...") + + # Validate configuration + errors = config.validate() + if errors: + for error in errors: + logger.error(f"Configuration error: {error}") + sys.exit(1) + + # Initialize components + tvh_client = TVHeadendClient(config.tvheadend) + transfer_service: TransferService | None = None + if not config.dry_run: + transfer_service = TransferService(config.tvheadend, config.dest_path) + state_manager = StateManager(config.state_file) + + # Initialize episode resolver for TVDB/TVmaze lookups + episode_resolver = EpisodeResolver.create( + tvdb_config=config.tvdb, + tvmaze_config=config.tvmaze, + ) + if episode_resolver.is_available(): + if config.tvdb and config.tvdb.enabled: + logger.info("Episode lookup enabled: TVDB (primary) + TVmaze (fallback)") + else: + logger.info("Episode lookup enabled: TVmaze only") + else: + logger.warning("Episode lookup disabled: no sources configured") + + # Initialize Jellyfin client for library scan triggers + jellyfin_client: JellyfinClient | None = None + if config.jellyfin and config.jellyfin.enabled: + jellyfin_client = JellyfinClient(config.jellyfin) + logger.info("Jellyfin integration enabled: library scan will trigger after transfers") + else: + logger.info("Jellyfin integration disabled") + + # Test connections + logger.info("Testing connections...") + + if not tvh_client.test_connection(): + logger.error("Failed to connect to TVHeadend") + sys.exit(1) + + if not config.dry_run: + if transfer_service: + if not transfer_service.test_connection(): + logger.error("Failed to connect to TVHeadend via SSH") + sys.exit(1) + if not transfer_service.test_destination(): + logger.error("Destination path not accessible") + sys.exit(1) + if jellyfin_client: + if not jellyfin_client.test_connection(): + logger.error("Failed to connect to Jellyfin") + sys.exit(1) + logger.info("All connections verified successfully") + else: + logger.info("TVHeadend API verified (skipping SSH/destination/Jellyfin in dry run mode)") + + # Set up signal handlers + signal.signal(signal.SIGTERM, signal_handler) + signal.signal(signal.SIGINT, signal_handler) + + # Create sync function with dependencies + def sync_job(): + run_sync( + tvh_client, + transfer_service, + state_manager, + config.dest_path, + episode_resolver=episode_resolver, + jellyfin_client=jellyfin_client, + dry_run=config.dry_run, + ) + + # Schedule regular syncs + schedule.every(config.poll_interval).seconds.do(sync_job) + + # Run initial sync + logger.info("Running initial sync...") + sync_job() + + # In dry run mode, exit after single run + if config.dry_run: + logger.info("Dry run complete") + return + + # Main loop + logger.info(f"Scheduling syncs every {config.poll_interval} seconds") + + while not shutdown_requested: + schedule.run_pending() + time.sleep(1) + + # Cleanup + logger.info("Shutting down...") + if transfer_service: + transfer_service.close() + logger.info("TVH2Jellyfin stopped") + + +if __name__ == "__main__": + main() diff --git a/src/tvh2jellyfin/organizer.py b/src/tvh2jellyfin/organizer.py new file mode 100644 index 0000000..d6969e6 --- /dev/null +++ b/src/tvh2jellyfin/organizer.py @@ -0,0 +1,139 @@ +"""File organization and naming logic for Jellyfin destination.""" + +from __future__ import annotations + +import re +import unicodedata +from pathlib import Path +from typing import TYPE_CHECKING + +from .tvheadend import Recording + +if TYPE_CHECKING: + from .tvdb import EpisodeInfo + + +def sanitize_filename(name: str) -> str: + """ + Sanitize a string to be safe for use as a filename. + + - Removes or replaces characters that are invalid in filenames + - Normalizes unicode characters + - Replaces multiple spaces/underscores with single ones + """ + # Normalize unicode characters + name = unicodedata.normalize("NFKD", name) + + # Replace characters that are problematic in filenames + # Keep alphanumeric, spaces, hyphens, underscores, and periods + name = re.sub(r'[<>:"/\\|?*]', "", name) + + # Replace multiple spaces or underscores with single space + name = re.sub(r"[\s_]+", " ", name) + + # Strip leading/trailing whitespace and periods + name = name.strip(" .") + + # Ensure we have something left + if not name: + name = "Unknown" + + return name + + +def sanitize_directory_name(name: str) -> str: + """ + Sanitize a string to be safe for use as a directory name. + + Similar to sanitize_filename but also handles trailing dots/spaces + which can cause issues on some filesystems. + """ + name = sanitize_filename(name) + + # Remove trailing dots and spaces (problematic on Windows/NTFS) + name = name.rstrip(". ") + + return name if name else "Unknown" + + +def generate_destination_path( + recording: Recording, + base_path: str, +) -> tuple[str, str]: + """ + Generate the destination directory and filename for a recording. + + Organizes files by show name: + {base_path}/{show_name}/{episode_title}.{ext} + + Returns: + tuple of (directory_path, filename) + """ + # Get the show name (title) for the directory + show_name = sanitize_directory_name(recording.title) + + # Build the episode filename + source_path = Path(recording.filename) + extension = source_path.suffix.lower() or ".ts" + + # Create a descriptive filename + if recording.subtitle: + # Has episode info: "Show Name - Episode Title.ts" + episode_name = f"{recording.title} - {recording.subtitle}" + else: + # No episode info, include timestamp to make unique + from datetime import datetime + + timestamp = datetime.fromtimestamp(recording.start).strftime("%Y-%m-%d %H%M") + episode_name = f"{recording.title} - {timestamp}" + + filename = sanitize_filename(episode_name) + extension + + # Build full destination directory + dest_dir = str(Path(base_path) / show_name) + + return dest_dir, filename + + +def get_full_destination_path(recording: Recording, base_path: str) -> str: + """Get the complete destination path including filename.""" + dest_dir, filename = generate_destination_path(recording, base_path) + return str(Path(dest_dir) / filename) + + +def generate_destination_path_with_episode( + recording: Recording, + base_path: str, + episode_info: EpisodeInfo | None = None, +) -> tuple[str, str]: + """ + Generate the destination directory and filename for a recording. + + Uses episode information from TVDB/TVmaze when available for proper + Jellyfin naming format: {show_name}/{show_name} - SxxExx.{ext} + + Falls back to the original naming logic when episode info is not available. + + Args: + recording: The recording to process + base_path: Base destination path + episode_info: Episode information from TVDB/TVmaze lookup (optional) + + Returns: + tuple of (directory_path, filename) + """ + # Get file extension from source + source_path = Path(recording.filename) + extension = source_path.suffix.lower() or ".ts" + + if episode_info: + # Use TVDB/TVmaze canonical name and episode info + show_name = sanitize_directory_name(episode_info.series_name) + episode_name = f"{episode_info.series_name} - {episode_info.formatted_episode}" + filename = sanitize_filename(episode_name) + extension + season_folder = f"Season {episode_info.season:02d}" + dest_dir = str(Path(base_path) / show_name / season_folder) + return dest_dir, filename + + # Fall back to original logic when no episode info available + return generate_destination_path(recording, base_path) diff --git a/src/tvh2jellyfin/state.py b/src/tvh2jellyfin/state.py new file mode 100644 index 0000000..d9fec18 --- /dev/null +++ b/src/tvh2jellyfin/state.py @@ -0,0 +1,114 @@ +"""State management for tracking transferred recordings.""" + +import json +import logging +from dataclasses import asdict, dataclass +from datetime import datetime +from pathlib import Path + +logger = logging.getLogger(__name__) + + +@dataclass +class TransferRecord: + """Record of a transferred file.""" + + uuid: str + title: str + source_path: str + dest_path: str + filesize: int + transferred_at: str + + @classmethod + def create( + cls, + uuid: str, + title: str, + source_path: str, + dest_path: str, + filesize: int, + ) -> "TransferRecord": + """Create a new transfer record with current timestamp.""" + return cls( + uuid=uuid, + title=title, + source_path=source_path, + dest_path=dest_path, + filesize=filesize, + transferred_at=datetime.now().isoformat(), + ) + + +class StateManager: + """Manages persistent state of transferred recordings.""" + + def __init__(self, state_file: str): + self.state_file = Path(state_file) + self._state: dict = {"transferred": {}, "version": 1} + self._load() + + def _load(self) -> None: + """Load state from disk.""" + if self.state_file.exists(): + try: + with open(self.state_file) as f: + self._state = json.load(f) + logger.info(f"Loaded state with {len(self._state.get('transferred', {}))} records") + except json.JSONDecodeError as e: + logger.error(f"Failed to parse state file: {e}") + self._state = {"transferred": {}, "version": 1} + else: + logger.info("No existing state file, starting fresh") + + def _save(self) -> None: + """Save state to disk.""" + # Ensure parent directory exists + self.state_file.parent.mkdir(parents=True, exist_ok=True) + + try: + with open(self.state_file, "w") as f: + json.dump(self._state, f, indent=2) + logger.debug("State saved") + except IOError as e: + logger.error(f"Failed to save state: {e}") + + def is_transferred(self, uuid: str) -> bool: + """Check if a recording has already been transferred.""" + return uuid in self._state.get("transferred", {}) + + def mark_transferred(self, record: TransferRecord) -> None: + """Mark a recording as successfully transferred.""" + if "transferred" not in self._state: + self._state["transferred"] = {} + + self._state["transferred"][record.uuid] = asdict(record) + self._save() + logger.info(f"Marked as transferred: {record.title} ({record.uuid})") + + def get_transfer_record(self, uuid: str) -> TransferRecord | None: + """Get the transfer record for a specific UUID.""" + data = self._state.get("transferred", {}).get(uuid) + if data: + return TransferRecord(**data) + return None + + def get_all_transfers(self) -> list[TransferRecord]: + """Get all transfer records.""" + records = [] + for data in self._state.get("transferred", {}).values(): + records.append(TransferRecord(**data)) + return records + + def get_transfer_count(self) -> int: + """Get the number of transferred recordings.""" + return len(self._state.get("transferred", {})) + + def remove_transfer(self, uuid: str) -> bool: + """Remove a transfer record (e.g., if cleanup needed).""" + if uuid in self._state.get("transferred", {}): + del self._state["transferred"][uuid] + self._save() + logger.info(f"Removed transfer record: {uuid}") + return True + return False diff --git a/src/tvh2jellyfin/transfer.py b/src/tvh2jellyfin/transfer.py new file mode 100644 index 0000000..5213d37 --- /dev/null +++ b/src/tvh2jellyfin/transfer.py @@ -0,0 +1,211 @@ +"""Transfer service - SCP from TVHeadend to local destination.""" + +import logging +import time +from pathlib import Path + +import paramiko + +from .config import TVHeadendConfig + +logger = logging.getLogger(__name__) + + +class TransferError(Exception): + """Raised when a file transfer fails.""" + + pass + + +class TransferService: + """Service for transferring files from TVHeadend via SCP to local destination.""" + + def __init__(self, tvh_config: TVHeadendConfig, dest_path: str): + self.tvh_config = tvh_config + self.dest_path = Path(dest_path) + self.client: paramiko.SSHClient | None = None + self.sftp: paramiko.SFTPClient | None = None + + def connect(self) -> None: + """Establish SSH connection to TVHeadend server.""" + self.client = paramiko.SSHClient() + self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + + try: + key_path = Path(self.tvh_config.ssh_key_path).expanduser() + private_key = paramiko.RSAKey.from_private_key_file(str(key_path)) + + self.client.connect( + hostname=self.tvh_config.host, + port=self.tvh_config.ssh_port, + username=self.tvh_config.ssh_user, + pkey=private_key, + timeout=30, + ) + self.sftp = self.client.open_sftp() + logger.info(f"Connected to TVHeadend at {self.tvh_config.host}") + except paramiko.AuthenticationException as e: + logger.error(f"SSH authentication failed: {e}") + raise TransferError(f"SSH authentication failed: {e}") + except paramiko.SSHException as e: + logger.error(f"SSH connection failed: {e}") + raise TransferError(f"SSH connection failed: {e}") + except FileNotFoundError: + logger.error(f"SSH key not found: {self.tvh_config.ssh_key_path}") + raise TransferError(f"SSH key not found: {self.tvh_config.ssh_key_path}") + + def close(self) -> None: + """Close SSH connection.""" + if self.sftp: + self.sftp.close() + self.sftp = None + if self.client: + self.client.close() + self.client = None + logger.debug("SSH connection closed") + + def ensure_connected(self) -> None: + """Ensure we have an active connection.""" + if not self.sftp: + self.connect() + + def _ensure_local_directory(self, directory: Path) -> None: + """Create local directory and any missing parents.""" + try: + directory.mkdir(parents=True, exist_ok=True) + logger.debug(f"Ensured local directory exists: {directory}") + except OSError as e: + raise TransferError(f"Failed to create local directory {directory}: {e}") + + def transfer_file( + self, + source_path: str, + dest_dir: str, + dest_filename: str, + expected_size: int | None = None, + ) -> str: + """ + Transfer a file from TVHeadend to local destination. + + Args: + source_path: Full path to source file on TVHeadend server + dest_dir: Local destination directory + dest_filename: Filename for destination + expected_size: Expected file size for verification + + Returns: + Full path to transferred file on local filesystem + """ + self.ensure_connected() + + if not self.sftp: + raise TransferError("Not connected") + + # Ensure local destination directory exists + local_dest_dir = Path(dest_dir) + self._ensure_local_directory(local_dest_dir) + + local_dest_path = local_dest_dir / dest_filename + + logger.info(f"Transferring: {self.tvh_config.host}:{source_path} -> {local_dest_path}") + + # Get remote file size if not provided + if expected_size is None: + try: + file_stat = self.sftp.stat(source_path) + expected_size = file_stat.st_size + except IOError as e: + raise TransferError(f"Cannot access remote file: {e}") + + start_time = time.time() + + try: + # Download file from TVHeadend to local destination + self.sftp.get(source_path, str(local_dest_path)) + + elapsed = time.time() - start_time + speed_mbps = (expected_size / 1024 / 1024) / elapsed if elapsed > 0 else 0 + logger.info( + f"Transfer complete: {expected_size / 1024 / 1024:.1f} MB " + f"in {elapsed:.1f}s ({speed_mbps:.1f} MB/s)" + ) + + except IOError as e: + raise TransferError(f"Transfer failed: {e}") + + return str(local_dest_path) + + def verify_transfer(self, local_path: str, expected_size: int) -> bool: + """ + Verify a transferred file exists locally and has correct size. + + Args: + local_path: Path to file on local filesystem + expected_size: Expected file size in bytes + + Returns: + True if verification passes + """ + path = Path(local_path) + + if not path.exists(): + logger.error(f"Transferred file not found: {local_path}") + return False + + actual_size = path.stat().st_size + + if actual_size != expected_size: + logger.error( + f"Size mismatch for {local_path}: expected {expected_size}, got {actual_size}" + ) + return False + + logger.info(f"Verified transfer: {local_path} ({actual_size} bytes)") + return True + + def delete_local_file(self, local_path: str) -> bool: + """Delete a local file (for cleanup on failed verification).""" + path = Path(local_path) + + try: + if path.exists(): + path.unlink() + logger.info(f"Deleted local file: {local_path}") + return True + except OSError as e: + logger.error(f"Failed to delete local file {local_path}: {e}") + return False + + def test_connection(self) -> bool: + """Test SSH connection to TVHeadend server.""" + try: + self.connect() + logger.info(f"Successfully connected to TVHeadend via SSH") + return True + except TransferError: + return False + finally: + self.close() + + def test_destination(self) -> bool: + """Test that local destination path is accessible and writable.""" + try: + if not self.dest_path.exists(): + logger.warning(f"Destination path does not exist, will create: {self.dest_path}") + + # Try to create and write to a test file + self._ensure_local_directory(self.dest_path) + test_file = self.dest_path / ".tvh2jellyfin_test" + try: + test_file.touch() + test_file.unlink() + except OSError as e: + logger.error(f"Destination path is not writable: {e}") + return False + + logger.info(f"Destination path verified: {self.dest_path}") + return True + + except Exception as e: + logger.error(f"Failed to verify destination: {e}") + return False diff --git a/src/tvh2jellyfin/tvdb.py b/src/tvh2jellyfin/tvdb.py new file mode 100644 index 0000000..99c5060 --- /dev/null +++ b/src/tvh2jellyfin/tvdb.py @@ -0,0 +1,284 @@ +"""TheTVDB API client for episode information lookup.""" + +import json +import logging +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +import requests + +logger = logging.getLogger(__name__) + + +@dataclass +class EpisodeInfo: + """Episode information from TVDB lookup.""" + + series_id: int + series_name: str + season: int + episode: int + episode_name: str + air_date: str # YYYY-MM-DD format + + @property + def formatted_episode(self) -> str: + """Return episode in Jellyfin format: SxxExx.""" + return f"S{self.season:02d}E{self.episode:02d}" + + +@dataclass +class TVDBConfig: + """TheTVDB API configuration.""" + + api_key: str + pin: str = "" + cache_ttl: int = 86400 # 24 hours default + show_mapping_file: str | None = None + enabled: bool = True + + +class TVDBClient: + """Client for TheTVDB API v4.""" + + BASE_URL = "https://api4.thetvdb.com/v4" + + def __init__(self, config: TVDBConfig): + self.config = config + self.session = requests.Session() + self._token: str | None = None + self._token_expires: float = 0 + # Cache: name -> (series_id, canonical_name, timestamp) + self._series_cache: dict[str, tuple[int, str, float]] = {} + # Cache: cache_key -> (EpisodeInfo | None, timestamp) + self._episode_cache: dict[str, tuple[EpisodeInfo | None, float]] = {} + # Show name mapping: tvh_name -> tvdb_name + self._show_mapping: dict[str, str] = {} + self._load_show_mapping() + + def _load_show_mapping(self) -> None: + """Load show name mapping from JSON file if configured.""" + if not self.config.show_mapping_file: + return + + mapping_path = Path(self.config.show_mapping_file) + if not mapping_path.exists(): + logger.warning(f"Show mapping file not found: {mapping_path}") + return + + try: + with open(mapping_path) as f: + self._show_mapping = json.load(f) + logger.info(f"Loaded {len(self._show_mapping)} show name mappings") + except (json.JSONDecodeError, OSError) as e: + logger.error(f"Failed to load show mapping file: {e}") + + def _authenticate(self) -> None: + """Get JWT token from TVDB API.""" + url = f"{self.BASE_URL}/login" + payload = {"apikey": self.config.api_key} + if self.config.pin: + payload["pin"] = self.config.pin + + try: + response = self.session.post(url, json=payload, timeout=30) + response.raise_for_status() + data = response.json() + self._token = data["data"]["token"] + # Token is valid for 30 days, but refresh more often for safety + self._token_expires = time.time() + 86400 # 24 hours + logger.debug("TVDB authentication successful") + except requests.exceptions.RequestException as e: + logger.error(f"TVDB authentication failed: {e}") + raise + + def _ensure_authenticated(self) -> None: + """Ensure we have a valid token.""" + if self._token is None or time.time() >= self._token_expires: + self._authenticate() + + def _api_get(self, endpoint: str, params: dict | None = None) -> dict: + """Make authenticated GET request to TVDB API.""" + self._ensure_authenticated() + + url = f"{self.BASE_URL}{endpoint}" + headers = {"Authorization": f"Bearer {self._token}"} + + try: + response = self.session.get(url, headers=headers, params=params, timeout=30) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + logger.error(f"TVDB API call failed: {endpoint} - {e}") + raise + + def _is_cache_valid(self, timestamp: float) -> bool: + """Check if a cached entry is still valid.""" + return time.time() - timestamp < self.config.cache_ttl + + def search_series(self, name: str) -> tuple[int, str] | None: + """ + Search for a series by name, return (series_id, canonical_name). + + Uses cache to minimize API calls. + """ + # Check show mapping first + mapped_name = self._show_mapping.get(name, name) + + # Check cache + cache_key = mapped_name.lower() + if cache_key in self._series_cache: + series_id, canonical_name, timestamp = self._series_cache[cache_key] + if self._is_cache_valid(timestamp): + return series_id, canonical_name + + # Search via API + try: + data = self._api_get("/search", {"query": mapped_name, "type": "series"}) + results = data.get("data", []) + + if not results: + logger.warning(f"No series found for: {name}") + return None + + # Use first result (best match) + series = results[0] + series_id = series["tvdb_id"] + canonical_name = series.get("name", name) + + # Cache the result + self._series_cache[cache_key] = (series_id, canonical_name, time.time()) + logger.debug(f"Found series: {canonical_name} (ID: {series_id})") + + return series_id, canonical_name + + except requests.exceptions.RequestException: + return None + + def get_episodes_by_airdate( + self, series_id: int, air_date: str + ) -> list[EpisodeInfo]: + """ + Get episodes for a series that aired on a specific date. + + Args: + series_id: TVDB series ID + air_date: Date in YYYY-MM-DD format + + Returns: + List of matching episodes (usually 1, but could be multiple) + """ + # Check cache + cache_key = f"{series_id}:{air_date}" + if cache_key in self._episode_cache: + cached_info, timestamp = self._episode_cache[cache_key] + if self._is_cache_valid(timestamp): + return [cached_info] if cached_info else [] + + try: + # Get all episodes for the series + # TVDB v4 doesn't have direct airdate filter, so we need to paginate + episodes: list[EpisodeInfo] = [] + page = 0 + + while True: + data = self._api_get( + f"/series/{series_id}/episodes/default", + {"page": page} + ) + + for ep in data.get("data", {}).get("episodes", []): + ep_aired = ep.get("aired", "") + if ep_aired == air_date: + info = EpisodeInfo( + series_id=series_id, + series_name=data.get("data", {}).get("name", "Unknown"), + season=ep.get("seasonNumber", 0), + episode=ep.get("number", 0), + episode_name=ep.get("name", ""), + air_date=air_date, + ) + episodes.append(info) + + # Check for more pages + links = data.get("links", {}) + if links.get("next"): + page += 1 + else: + break + + # Cache the first result (or None if no matches) + first_ep = episodes[0] if episodes else None + self._episode_cache[cache_key] = (first_ep, time.time()) + + return episodes + + except requests.exceptions.RequestException: + return [] + + def lookup_episode( + self, + show_name: str, + broadcast_timestamp: int, + ) -> EpisodeInfo | None: + """ + Main entry point: look up episode info by show name and broadcast time. + + Args: + show_name: Show name from TVHeadend recording + broadcast_timestamp: Unix timestamp of broadcast start + + Returns: + EpisodeInfo if found, None otherwise + """ + # Convert timestamp to air date + air_date = datetime.fromtimestamp(broadcast_timestamp).strftime("%Y-%m-%d") + + # Check episode cache first (using show name as part of key) + cache_key = f"{show_name.lower()}:{air_date}" + if cache_key in self._episode_cache: + cached_info, timestamp = self._episode_cache[cache_key] + if self._is_cache_valid(timestamp): + return cached_info + + # Search for series + series_result = self.search_series(show_name) + if not series_result: + # Cache negative result + self._episode_cache[cache_key] = (None, time.time()) + return None + + series_id, canonical_name = series_result + + # Find episodes for the air date + episodes = self.get_episodes_by_airdate(series_id, air_date) + + if not episodes: + logger.debug(f"No episode found for {show_name} on {air_date}") + # Cache negative result + self._episode_cache[cache_key] = (None, time.time()) + return None + + # Return first match + episode = episodes[0] + + # Update episode with canonical series name + episode = EpisodeInfo( + series_id=episode.series_id, + series_name=canonical_name, + season=episode.season, + episode=episode.episode, + episode_name=episode.episode_name, + air_date=episode.air_date, + ) + + # Cache by show name too + self._episode_cache[cache_key] = (episode, time.time()) + + logger.info( + f"TVDB: {show_name} ({air_date}) -> {canonical_name} {episode.formatted_episode}" + ) + + return episode diff --git a/src/tvh2jellyfin/tvheadend.py b/src/tvh2jellyfin/tvheadend.py new file mode 100644 index 0000000..1f5bb8f --- /dev/null +++ b/src/tvh2jellyfin/tvheadend.py @@ -0,0 +1,139 @@ +"""TVHeadend API client for fetching and managing recordings.""" + +import logging +from dataclasses import dataclass +from typing import Any + +import requests +from requests.auth import HTTPDigestAuth + +from .config import TVHeadendConfig + +logger = logging.getLogger(__name__) + + +@dataclass +class Recording: + """Represents a TVHeadend recording.""" + + uuid: str + title: str + subtitle: str + filename: str + filesize: int + channel: str + start: int + stop: int + status: str + + @property + def display_name(self) -> str: + """Human-readable name for the recording.""" + if self.subtitle: + return f"{self.title} - {self.subtitle}" + return self.title + + @classmethod + def from_api_response(cls, data: dict[str, Any]) -> "Recording": + """Create a Recording from TVHeadend API response data.""" + return cls( + uuid=data.get("uuid", ""), + title=data.get("disp_title", data.get("title", "Unknown")), + subtitle=data.get("disp_subtitle", data.get("subtitle", "")), + filename=data.get("filename", ""), + filesize=data.get("filesize", 0), + channel=data.get("channelname", ""), + start=data.get("start", 0), + stop=data.get("stop", 0), + status=data.get("status", ""), + ) + + +class TVHeadendClient: + """Client for interacting with TVHeadend JSON API.""" + + def __init__(self, config: TVHeadendConfig): + self.config = config + self._create_session() + + def _create_session(self) -> None: + """Create a new session with fresh authentication.""" + self.session = requests.Session() + self.session.auth = HTTPDigestAuth(self.config.username, self.config.password) + + def _api_call(self, endpoint: str, params: dict | None = None) -> dict[str, Any]: + """Make an API call to TVHeadend.""" + url = f"{self.config.base_url}/api/{endpoint}" + try: + response = self.session.get(url, params=params, timeout=30) + if response.status_code == 401: + # Digest auth nonce may have expired, refresh session and retry + logger.debug("Got 401, refreshing session and retrying") + self._create_session() + response = self.session.get(url, params=params, timeout=30) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + logger.error(f"API call failed: {endpoint} - {e}") + raise + + def _api_post(self, endpoint: str, data: dict | None = None) -> dict[str, Any]: + """Make a POST API call to TVHeadend.""" + url = f"{self.config.base_url}/api/{endpoint}" + try: + response = self.session.post(url, data=data, timeout=30) + if response.status_code == 401: + # Digest auth nonce may have expired, refresh session and retry + logger.debug("Got 401, refreshing session and retrying") + self._create_session() + response = self.session.post(url, data=data, timeout=30) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + logger.error(f"API POST call failed: {endpoint} - {e}") + raise + + def get_finished_recordings(self) -> list[Recording]: + """Fetch all finished recordings from TVHeadend.""" + params = { + "limit": 1000, # Get all recordings + } + data = self._api_call("dvr/entry/grid_finished", params) + + recordings = [] + for entry in data.get("entries", []): + if entry.get("status") == "Completed OK": + recordings.append(Recording.from_api_response(entry)) + + logger.info(f"Found {len(recordings)} completed recordings") + return recordings + + def get_recording_details(self, uuid: str) -> Recording | None: + """Get details for a specific recording.""" + params = {"uuid": uuid} + data = self._api_call("dvr/entry/grid", params) + + entries = data.get("entries", []) + if entries: + return Recording.from_api_response(entries[0]) + return None + + def remove_recording(self, uuid: str) -> bool: + """Remove a recording from TVHeadend (deletes file and entry).""" + try: + self._api_post("dvr/entry/remove", {"uuid": uuid}) + logger.info(f"Removed recording from TVHeadend: {uuid}") + return True + except requests.exceptions.RequestException: + logger.error(f"Failed to remove recording: {uuid}") + return False + + def test_connection(self) -> bool: + """Test connection to TVHeadend server.""" + try: + self._api_call("serverinfo") + logger.info(f"Successfully connected to TVHeadend at {self.config.base_url}") + return True + except requests.exceptions.RequestException as e: + logger.error(f"Failed to connect to TVHeadend: {e}") + return False diff --git a/src/tvh2jellyfin/tvmaze.py b/src/tvh2jellyfin/tvmaze.py new file mode 100644 index 0000000..6c321a1 --- /dev/null +++ b/src/tvh2jellyfin/tvmaze.py @@ -0,0 +1,181 @@ +"""TVmaze API client for episode information lookup (fallback).""" + +import logging +import time +from dataclasses import dataclass +from datetime import datetime + +import requests + +from .tvdb import EpisodeInfo + +logger = logging.getLogger(__name__) + + +@dataclass +class TVMazeConfig: + """TVmaze API configuration.""" + + cache_ttl: int = 86400 # 24 hours default + enabled: bool = True + + +class TVMazeClient: + """Client for TVmaze API (no authentication required).""" + + BASE_URL = "https://api.tvmaze.com" + + def __init__(self, config: TVMazeConfig | None = None): + self.config = config or TVMazeConfig() + self.session = requests.Session() + # Cache: name -> (show_id, show_name, timestamp) + self._series_cache: dict[str, tuple[int, str, float]] = {} + # Cache: cache_key -> (EpisodeInfo | None, timestamp) + self._episode_cache: dict[str, tuple[EpisodeInfo | None, float]] = {} + + def _is_cache_valid(self, timestamp: float) -> bool: + """Check if a cached entry is still valid.""" + return time.time() - timestamp < self.config.cache_ttl + + def search_series(self, name: str) -> tuple[int, str] | None: + """ + Search for a series by name, return (show_id, show_name). + + Uses cache to minimize API calls. + """ + # Check cache + cache_key = name.lower() + if cache_key in self._series_cache: + show_id, show_name, timestamp = self._series_cache[cache_key] + if self._is_cache_valid(timestamp): + return show_id, show_name + + # Search via API + try: + url = f"{self.BASE_URL}/singlesearch/shows" + response = self.session.get(url, params={"q": name}, timeout=30) + + if response.status_code == 404: + logger.debug(f"TVmaze: No series found for: {name}") + return None + + response.raise_for_status() + data = response.json() + + show_id = data["id"] + show_name = data.get("name", name) + + # Cache the result + self._series_cache[cache_key] = (show_id, show_name, time.time()) + logger.debug(f"TVmaze: Found series: {show_name} (ID: {show_id})") + + return show_id, show_name + + except requests.exceptions.RequestException as e: + logger.error(f"TVmaze search failed: {e}") + return None + + def get_episode_by_date( + self, show_id: int, air_date: str, show_name: str + ) -> EpisodeInfo | None: + """ + Get episode for a show that aired on a specific date. + + Args: + show_id: TVmaze show ID + air_date: Date in YYYY-MM-DD format + show_name: Show name (for building EpisodeInfo) + + Returns: + EpisodeInfo if found, None otherwise + """ + # Check cache + cache_key = f"{show_id}:{air_date}" + if cache_key in self._episode_cache: + cached_info, timestamp = self._episode_cache[cache_key] + if self._is_cache_valid(timestamp): + return cached_info + + try: + # TVmaze has a direct endpoint for episodes by date + url = f"{self.BASE_URL}/shows/{show_id}/episodesbydate" + response = self.session.get(url, params={"date": air_date}, timeout=30) + + if response.status_code == 404: + logger.debug(f"TVmaze: No episode found for show {show_id} on {air_date}") + self._episode_cache[cache_key] = (None, time.time()) + return None + + response.raise_for_status() + episodes = response.json() + + if not episodes: + self._episode_cache[cache_key] = (None, time.time()) + return None + + # Use first episode if multiple on same day + ep = episodes[0] + info = EpisodeInfo( + series_id=show_id, + series_name=show_name, + season=ep.get("season", 0), + episode=ep.get("number", 0), + episode_name=ep.get("name", ""), + air_date=air_date, + ) + + self._episode_cache[cache_key] = (info, time.time()) + return info + + except requests.exceptions.RequestException as e: + logger.error(f"TVmaze episode lookup failed: {e}") + return None + + def lookup_episode( + self, + show_name: str, + broadcast_timestamp: int, + ) -> EpisodeInfo | None: + """ + Main entry point: look up episode info by show name and broadcast time. + + Args: + show_name: Show name from TVHeadend recording + broadcast_timestamp: Unix timestamp of broadcast start + + Returns: + EpisodeInfo if found, None otherwise + """ + # Convert timestamp to air date + air_date = datetime.fromtimestamp(broadcast_timestamp).strftime("%Y-%m-%d") + + # Check episode cache first (using show name as part of key) + cache_key = f"tvmaze:{show_name.lower()}:{air_date}" + if cache_key in self._episode_cache: + cached_info, timestamp = self._episode_cache[cache_key] + if self._is_cache_valid(timestamp): + return cached_info + + # Search for series + series_result = self.search_series(show_name) + if not series_result: + self._episode_cache[cache_key] = (None, time.time()) + return None + + show_id, canonical_name = series_result + + # Find episode for the air date + episode = self.get_episode_by_date(show_id, air_date, canonical_name) + + if not episode: + self._episode_cache[cache_key] = (None, time.time()) + return None + + # Cache by show name too + self._episode_cache[cache_key] = (episode, time.time()) + + logger.info( + f"TVmaze: {show_name} ({air_date}) -> {canonical_name} {episode.formatted_episode}" + ) + + return episode diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_episode_resolver.py b/tests/test_episode_resolver.py new file mode 100644 index 0000000..0f7b7a3 --- /dev/null +++ b/tests/test_episode_resolver.py @@ -0,0 +1,117 @@ +"""Tests for episode resolver.""" + +from unittest.mock import Mock + +import pytest + +from tvh2jellyfin.episode_resolver import EpisodeResolver +from tvh2jellyfin.tvdb import EpisodeInfo + + +class TestEpisodeResolver: + """Tests for EpisodeResolver.""" + + @pytest.fixture + def mock_episode_info(self): + return EpisodeInfo( + series_id=456, + series_name="Test Show", + season=1, + episode=5, + episode_name="Pilot", + air_date="2024-01-15", + ) + + def test_lookup_tvdb_success(self, mock_episode_info): + """Test successful lookup from TVDB.""" + mock_tvdb = Mock() + mock_tvdb.lookup_episode.return_value = mock_episode_info + + resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=None) + result, source = resolver.lookup("Test Show", 1705312800) + + assert result is not None + assert result.season == 1 + assert result.episode == 5 + assert source == "tvdb" + + def test_lookup_tvmaze_fallback(self, mock_episode_info): + """Test fallback to TVmaze when TVDB fails.""" + mock_tvdb = Mock() + mock_tvdb.lookup_episode.return_value = None + + mock_tvmaze = Mock() + mock_tvmaze.lookup_episode.return_value = mock_episode_info + + resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=mock_tvmaze) + result, source = resolver.lookup("Test Show", 1705312800) + + assert result is not None + assert source == "tvmaze" + + def test_lookup_tvdb_exception_fallback(self, mock_episode_info): + """Test fallback to TVmaze when TVDB raises exception.""" + mock_tvdb = Mock() + mock_tvdb.lookup_episode.side_effect = Exception("API Error") + + mock_tvmaze = Mock() + mock_tvmaze.lookup_episode.return_value = mock_episode_info + + resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=mock_tvmaze) + result, source = resolver.lookup("Test Show", 1705312800) + + assert result is not None + assert source == "tvmaze" + + def test_lookup_both_fail(self): + """Test when both sources fail.""" + mock_tvdb = Mock() + mock_tvdb.lookup_episode.return_value = None + + mock_tvmaze = Mock() + mock_tvmaze.lookup_episode.return_value = None + + resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=mock_tvmaze) + result, source = resolver.lookup("Unknown Show", 1705312800) + + assert result is None + assert source == "none" + + def test_lookup_no_clients(self): + """Test when no clients are available.""" + resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=None) + result, source = resolver.lookup("Test Show", 1705312800) + + assert result is None + assert source == "none" + + def test_lookup_tvmaze_only(self, mock_episode_info): + """Test with only TVmaze client.""" + mock_tvmaze = Mock() + mock_tvmaze.lookup_episode.return_value = mock_episode_info + + resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=mock_tvmaze) + result, source = resolver.lookup("Test Show", 1705312800) + + assert result is not None + assert source == "tvmaze" + + def test_is_available_with_tvdb(self): + """Test is_available with TVDB client.""" + resolver = EpisodeResolver(tvdb_client=Mock(), tvmaze_client=None) + assert resolver.is_available() is True + + def test_is_available_with_tvmaze(self): + """Test is_available with TVmaze client.""" + resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=Mock()) + assert resolver.is_available() is True + + def test_is_available_with_both(self): + """Test is_available with both clients.""" + resolver = EpisodeResolver(tvdb_client=Mock(), tvmaze_client=Mock()) + assert resolver.is_available() is True + + def test_is_available_with_none(self): + """Test is_available with no clients.""" + resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=None) + assert resolver.is_available() is False diff --git a/tests/test_jellyfin.py b/tests/test_jellyfin.py new file mode 100644 index 0000000..308b617 --- /dev/null +++ b/tests/test_jellyfin.py @@ -0,0 +1,150 @@ +"""Tests for Jellyfin client.""" + +from unittest.mock import Mock, patch + +import pytest +import requests + +from tvh2jellyfin.jellyfin import JellyfinClient, JellyfinConfig + + +class TestJellyfinConfig: + """Tests for JellyfinConfig dataclass.""" + + def test_config_defaults(self): + """Test default values.""" + config = JellyfinConfig( + url="http://localhost:8096", + api_key="test_key", + ) + assert config.library_id is None + assert config.enabled is True + + def test_config_with_library_id(self): + """Test config with specific library ID.""" + config = JellyfinConfig( + url="http://localhost:8096", + api_key="test_key", + library_id="abc123", + ) + assert config.library_id == "abc123" + + +class TestJellyfinClient: + """Tests for Jellyfin API client.""" + + @pytest.fixture + def config(self): + return JellyfinConfig( + url="http://localhost:8096", + api_key="test_api_key", + ) + + @pytest.fixture + def config_with_library(self): + return JellyfinConfig( + url="http://localhost:8096", + api_key="test_api_key", + library_id="lib123", + ) + + @pytest.fixture + def client(self, config): + return JellyfinClient(config) + + @pytest.fixture + def client_with_library(self, config_with_library): + return JellyfinClient(config_with_library) + + def test_url_normalization(self): + """Test that trailing slashes are removed from URL.""" + config = JellyfinConfig( + url="http://localhost:8096/", + api_key="test_key", + ) + client = JellyfinClient(config) + assert client.base_url == "http://localhost:8096" + + def test_headers_set(self, client): + """Test that auth headers are set correctly.""" + assert client.session.headers["X-MediaBrowser-Token"] == "test_api_key" + assert client.session.headers["Content-Type"] == "application/json" + + @patch("requests.Session.get") + def test_test_connection_success(self, mock_get, client): + """Test successful connection test.""" + mock_get.return_value.json.return_value = { + "ServerName": "Test Server", + "Version": "10.8.0", + } + mock_get.return_value.raise_for_status = Mock() + + result = client.test_connection() + + assert result is True + mock_get.assert_called_once() + assert "/System/Info" in mock_get.call_args[0][0] + + @patch("requests.Session.get") + def test_test_connection_failure(self, mock_get, client): + """Test connection test failure.""" + mock_get.side_effect = requests.exceptions.ConnectionError("Connection refused") + + result = client.test_connection() + + assert result is False + + @patch("requests.Session.post") + def test_trigger_library_scan_full(self, mock_post, client): + """Test triggering full library scan.""" + mock_post.return_value.raise_for_status = Mock() + + result = client.trigger_library_scan() + + assert result is True + mock_post.assert_called_once() + assert "/Library/Refresh" in mock_post.call_args[0][0] + + @patch("requests.Session.post") + def test_trigger_library_scan_specific_library(self, mock_post, client_with_library): + """Test triggering scan for specific library.""" + mock_post.return_value.raise_for_status = Mock() + + result = client_with_library.trigger_library_scan() + + assert result is True + mock_post.assert_called_once() + assert "/Items/lib123/Refresh" in mock_post.call_args[0][0] + + @patch("requests.Session.post") + def test_trigger_library_scan_failure(self, mock_post, client): + """Test library scan trigger failure.""" + mock_post.side_effect = requests.exceptions.RequestException("API error") + + result = client.trigger_library_scan() + + assert result is False + + @patch("requests.Session.get") + def test_get_libraries_success(self, mock_get, client): + """Test getting list of libraries.""" + mock_get.return_value.json.return_value = [ + {"Name": "TV Shows", "ItemId": "tv123"}, + {"Name": "Movies", "ItemId": "mov456"}, + ] + mock_get.return_value.raise_for_status = Mock() + + result = client.get_libraries() + + assert len(result) == 2 + assert result[0] == {"name": "TV Shows", "id": "tv123"} + assert result[1] == {"name": "Movies", "id": "mov456"} + + @patch("requests.Session.get") + def test_get_libraries_failure(self, mock_get, client): + """Test getting libraries when API fails.""" + mock_get.side_effect = requests.exceptions.RequestException("API error") + + result = client.get_libraries() + + assert result == [] diff --git a/tests/test_organizer.py b/tests/test_organizer.py new file mode 100644 index 0000000..7c0fd39 --- /dev/null +++ b/tests/test_organizer.py @@ -0,0 +1,216 @@ +"""Tests for organizer module.""" + +import pytest + +from tvh2jellyfin.organizer import ( + generate_destination_path, + generate_destination_path_with_episode, + sanitize_directory_name, + sanitize_filename, +) +from tvh2jellyfin.tvdb import EpisodeInfo +from tvh2jellyfin.tvheadend import Recording + + +class TestSanitizeFilename: + """Tests for filename sanitization.""" + + def test_removes_invalid_characters(self): + assert sanitize_filename('Test: "Show"') == "Test Show" + # Multiple spaces get normalized to single space + assert sanitize_filename("Show <1> | Part 2") == "Show 1 Part 2" + + def test_normalizes_spaces(self): + assert sanitize_filename("Test Show") == "Test Show" + assert sanitize_filename("Test_Show") == "Test Show" + + def test_strips_whitespace(self): + assert sanitize_filename(" Test Show ") == "Test Show" + + def test_empty_returns_unknown(self): + assert sanitize_filename("") == "Unknown" + assert sanitize_filename(" ") == "Unknown" + + +class TestSanitizeDirectoryName: + """Tests for directory name sanitization.""" + + def test_removes_trailing_dots(self): + assert sanitize_directory_name("Test Show.") == "Test Show" + assert sanitize_directory_name("Test Show...") == "Test Show" + + def test_handles_empty(self): + assert sanitize_directory_name("") == "Unknown" + + +class TestGenerateDestinationPath: + """Tests for destination path generation.""" + + @pytest.fixture + def recording_with_subtitle(self): + return Recording( + uuid="123", + title="Test Show", + subtitle="Episode Title", + filename="/recordings/test.ts", + filesize=1000000, + channel="BBC One", + start=1705312800, + stop=1705316400, + status="Completed OK", + ) + + @pytest.fixture + def recording_without_subtitle(self): + return Recording( + uuid="456", + title="Another Show", + subtitle="", + filename="/recordings/another.ts", + filesize=2000000, + channel="ITV", + start=1705312800, + stop=1705316400, + status="Completed OK", + ) + + def test_with_subtitle(self, recording_with_subtitle): + dest_dir, filename = generate_destination_path( + recording_with_subtitle, "/media/tv" + ) + assert dest_dir == "/media/tv/Test Show" + assert filename == "Test Show - Episode Title.ts" + + def test_without_subtitle(self, recording_without_subtitle): + dest_dir, filename = generate_destination_path( + recording_without_subtitle, "/media/tv" + ) + assert dest_dir == "/media/tv/Another Show" + # Should include timestamp + assert filename.startswith("Another Show - ") + assert filename.endswith(".ts") + + +class TestGenerateDestinationPathWithEpisode: + """Tests for destination path generation with episode info.""" + + @pytest.fixture + def recording(self): + return Recording( + uuid="123", + title="Coronation Street", + subtitle="", + filename="/recordings/Coronation-Street/Coronation Street2026-01-06.ts", + filesize=1000000, + channel="ITV", + start=1767744000, # 2026-01-06 + stop=1767747600, + status="Completed OK", + ) + + @pytest.fixture + def episode_info(self): + return EpisodeInfo( + series_id=456, + series_name="Coronation Street", + season=67, + episode=4, + episode_name="Episode 4", + air_date="2026-01-06", + ) + + def test_with_episode_info(self, recording, episode_info): + """Test Jellyfin naming format with episode info.""" + dest_dir, filename = generate_destination_path_with_episode( + recording, "/media/tv", episode_info + ) + assert dest_dir == "/media/tv/Coronation Street/Season 67" + assert filename == "Coronation Street - S67E04.ts" + + def test_without_episode_info(self, recording): + """Test fallback when episode info not available.""" + dest_dir, filename = generate_destination_path_with_episode( + recording, "/media/tv", None + ) + assert dest_dir == "/media/tv/Coronation Street" + # Should use fallback logic (timestamp) + assert filename.startswith("Coronation Street - ") + assert filename.endswith(".ts") + + def test_uses_canonical_series_name(self, recording): + """Test that canonical series name from TVDB is used.""" + episode_info = EpisodeInfo( + series_id=456, + series_name="Doctor Who (2005)", # TVDB canonical name + season=14, + episode=1, + episode_name="The Church on Ruby Road", + air_date="2023-12-25", + ) + + # Recording might have different name + recording_with_diff_name = Recording( + uuid="789", + title="Doctor Who", # TVHeadend name + subtitle="", + filename="/recordings/doctor-who.ts", + filesize=3000000, + channel="BBC One", + start=1703512800, + stop=1703516400, + status="Completed OK", + ) + + dest_dir, filename = generate_destination_path_with_episode( + recording_with_diff_name, "/media/tv", episode_info + ) + + # Should use TVDB canonical name with season folder + assert dest_dir == "/media/tv/Doctor Who (2005)/Season 14" + assert filename == "Doctor Who (2005) - S14E01.ts" + + def test_large_episode_numbers(self, recording): + """Test with season/episode numbers > 99.""" + episode_info = EpisodeInfo( + series_id=456, + series_name="Long Running Show", + season=67, + episode=125, + episode_name="Episode", + air_date="2026-01-06", + ) + + dest_dir, filename = generate_destination_path_with_episode( + recording, "/media/tv", episode_info + ) + + assert filename == "Long Running Show - S67E125.ts" + + def test_preserves_file_extension(self, recording): + """Test that file extension is preserved.""" + recording_mkv = Recording( + uuid="123", + title="Test Show", + subtitle="", + filename="/recordings/test.mkv", + filesize=1000000, + channel="BBC", + start=1705312800, + stop=1705316400, + status="Completed OK", + ) + + episode_info = EpisodeInfo( + series_id=456, + series_name="Test Show", + season=1, + episode=1, + episode_name="Pilot", + air_date="2024-01-15", + ) + + _, filename = generate_destination_path_with_episode( + recording_mkv, "/media/tv", episode_info + ) + + assert filename.endswith(".mkv") diff --git a/tests/test_tvdb.py b/tests/test_tvdb.py new file mode 100644 index 0000000..a22bfce --- /dev/null +++ b/tests/test_tvdb.py @@ -0,0 +1,190 @@ +"""Tests for TheTVDB client.""" + +from unittest.mock import Mock, patch + +import pytest + +from tvh2jellyfin.tvdb import EpisodeInfo, TVDBClient, TVDBConfig + + +class TestEpisodeInfo: + """Tests for EpisodeInfo dataclass.""" + + def test_formatted_episode_single_digit(self): + """Test SxxExx formatting with single digit numbers.""" + info = EpisodeInfo( + series_id=123, + series_name="Test Show", + season=1, + episode=5, + episode_name="Pilot", + air_date="2024-01-15", + ) + assert info.formatted_episode == "S01E05" + + def test_formatted_episode_double_digit(self): + """Test SxxExx formatting with double digit numbers.""" + info = EpisodeInfo( + series_id=123, + series_name="Test Show", + season=12, + episode=25, + episode_name="Episode", + air_date="2024-01-15", + ) + assert info.formatted_episode == "S12E25" + + def test_formatted_episode_large_numbers(self): + """Test formatting with season/episode > 99.""" + info = EpisodeInfo( + series_id=123, + series_name="Long Running Show", + season=67, + episode=125, + episode_name="Episode", + air_date="2024-01-15", + ) + assert info.formatted_episode == "S67E125" + + +class TestTVDBClient: + """Tests for TVDB API client.""" + + @pytest.fixture + def config(self): + return TVDBConfig( + api_key="test_key", + pin="", + cache_ttl=3600, + show_mapping_file=None, + enabled=True, + ) + + @pytest.fixture + def client(self, config): + with patch.object(TVDBClient, "_authenticate"): + return TVDBClient(config) + + def test_cache_valid(self, client): + """Test cache validity check.""" + import time + + # Recent timestamp should be valid + assert client._is_cache_valid(time.time() - 100) + + # Old timestamp should be invalid + assert not client._is_cache_valid(time.time() - 7200) + + @patch("requests.Session.post") + def test_authenticate(self, mock_post, config): + """Test authentication flow.""" + mock_post.return_value.json.return_value = {"data": {"token": "test_token"}} + mock_post.return_value.raise_for_status = Mock() + + client = TVDBClient(config) + client._authenticate() + + assert client._token == "test_token" + mock_post.assert_called_once() + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_search_series_success(self, mock_post, mock_get, config): + """Test successful series search.""" + mock_post.return_value.json.return_value = {"data": {"token": "test_token"}} + mock_post.return_value.raise_for_status = Mock() + + mock_get.return_value.json.return_value = { + "data": [{"tvdb_id": 456, "name": "The Test Show"}] + } + mock_get.return_value.raise_for_status = Mock() + + client = TVDBClient(config) + result = client.search_series("Test Show") + + assert result is not None + assert result[0] == 456 + assert result[1] == "The Test Show" + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_search_series_not_found(self, mock_post, mock_get, config): + """Test series search with no results.""" + mock_post.return_value.json.return_value = {"data": {"token": "test_token"}} + mock_post.return_value.raise_for_status = Mock() + + mock_get.return_value.json.return_value = {"data": []} + mock_get.return_value.raise_for_status = Mock() + + client = TVDBClient(config) + result = client.search_series("Nonexistent Show") + + assert result is None + + @patch("tvh2jellyfin.tvdb.requests.Session") + def test_lookup_episode_success(self, mock_session_class, config): + """Test successful episode lookup.""" + from datetime import datetime + + mock_session = Mock() + mock_session_class.return_value = mock_session + + # Use a timestamp and derive the expected air date from it + timestamp = 1767744000 + expected_air_date = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d") + + # Mock authentication + mock_session.post.return_value.json.return_value = {"data": {"token": "test_token"}} + mock_session.post.return_value.raise_for_status = Mock() + + # Mock search and episode responses + mock_session.get.return_value.raise_for_status = Mock() + mock_session.get.return_value.json.side_effect = [ + {"data": [{"tvdb_id": 456, "name": "Coronation Street"}]}, + { + "data": { + "name": "Coronation Street", + "episodes": [ + { + "seasonNumber": 67, + "number": 4, + "name": "Episode 4", + "aired": expected_air_date, + } + ], + }, + "links": {}, + }, + ] + + client = TVDBClient(config) + result = client.lookup_episode("Coronation Street", timestamp) + + assert result is not None + assert result.season == 67 + assert result.episode == 4 + assert result.formatted_episode == "S67E04" + + def test_show_mapping_file(self, tmp_path, config): + """Test show name mapping from file.""" + mapping_file = tmp_path / "mapping.json" + mapping_file.write_text('{"Doctor Who": "Doctor Who (2005)"}') + + config.show_mapping_file = str(mapping_file) + + with patch.object(TVDBClient, "_authenticate"): + client = TVDBClient(config) + + assert client._show_mapping.get("Doctor Who") == "Doctor Who (2005)" + + def test_caching_prevents_duplicate_calls(self, client): + """Test that cache prevents duplicate API calls.""" + import time + + # Manually populate cache + client._series_cache["test show"] = (123, "Test Show", time.time()) + + # Search should use cache, not make API call + result = client.search_series("test show") + + assert result == (123, "Test Show") diff --git a/tests/test_tvmaze.py b/tests/test_tvmaze.py new file mode 100644 index 0000000..8a4fe9f --- /dev/null +++ b/tests/test_tvmaze.py @@ -0,0 +1,121 @@ +"""Tests for TVmaze client.""" + +from unittest.mock import Mock, patch + +import pytest + +from tvh2jellyfin.tvmaze import TVMazeClient, TVMazeConfig + + +class TestTVMazeClient: + """Tests for TVmaze API client.""" + + @pytest.fixture + def config(self): + return TVMazeConfig(cache_ttl=3600, enabled=True) + + @pytest.fixture + def client(self, config): + return TVMazeClient(config) + + @patch("requests.Session.get") + def test_search_series_success(self, mock_get, client): + """Test successful series search.""" + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = { + "id": 789, + "name": "Coronation Street", + } + mock_get.return_value.raise_for_status = Mock() + + result = client.search_series("Coronation Street") + + assert result is not None + assert result[0] == 789 + assert result[1] == "Coronation Street" + + @patch("requests.Session.get") + def test_search_series_not_found(self, mock_get, client): + """Test series search with 404 response.""" + mock_get.return_value.status_code = 404 + + result = client.search_series("Nonexistent Show") + + assert result is None + + @patch("requests.Session.get") + def test_get_episode_by_date_success(self, mock_get, client): + """Test successful episode lookup by date.""" + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = [ + {"season": 67, "number": 4, "name": "Episode 4"} + ] + mock_get.return_value.raise_for_status = Mock() + + result = client.get_episode_by_date(789, "2026-01-06", "Coronation Street") + + assert result is not None + assert result.season == 67 + assert result.episode == 4 + assert result.series_name == "Coronation Street" + + @patch("requests.Session.get") + def test_get_episode_by_date_not_found(self, mock_get, client): + """Test episode lookup with no results.""" + mock_get.return_value.status_code = 404 + + result = client.get_episode_by_date(789, "2026-01-01", "Coronation Street") + + assert result is None + + @patch("requests.Session.get") + def test_lookup_episode_full_flow(self, mock_get, client): + """Test full episode lookup flow.""" + # Set up responses for search and episode lookup + mock_get.return_value.status_code = 200 + mock_get.return_value.json.side_effect = [ + {"id": 789, "name": "Coronation Street"}, # Search response + [{"season": 67, "number": 4, "name": "Episode 4"}], # Episode response + ] + mock_get.return_value.raise_for_status = Mock() + + # Timestamp for 2026-01-06 + result = client.lookup_episode("Coronation Street", 1767744000) + + assert result is not None + assert result.series_name == "Coronation Street" + assert result.formatted_episode == "S67E04" + + def test_caching_series(self, client): + """Test that series cache prevents duplicate calls.""" + import time + + # Manually populate cache + client._series_cache["coronation street"] = (789, "Coronation Street", time.time()) + + # Should use cache + result = client.search_series("coronation street") + + assert result == (789, "Coronation Street") + + def test_caching_episodes(self, client): + """Test that episode cache prevents duplicate calls.""" + import time + + from tvh2jellyfin.tvdb import EpisodeInfo + + cached_episode = EpisodeInfo( + series_id=789, + series_name="Coronation Street", + season=67, + episode=4, + episode_name="Episode 4", + air_date="2026-01-06", + ) + client._episode_cache["789:2026-01-06"] = (cached_episode, time.time()) + + # Should use cache + result = client.get_episode_by_date(789, "2026-01-06", "Coronation Street") + + assert result is not None + assert result.season == 67