hls_generator.py

#!/usr/bin/env python3
"""
HLS Generator - Background process that converts videos to adaptive bitrate HLS.
Probes each video, generates multiple quality variants, and builds a master playlist
so clients can switch quality based on network conditions.

Usage:
  python hls_generator.py              # Process all pending, then exit
  python hls_generator.py --watch      # Continuous mode (loop every 60s)
  python hls_generator.py --workers 4  # Parallel ffmpeg processes
  python hls_generator.py --dry-run    # Show what would be processed
  python hls_generator.py --limit 10   # Process only N videos
  python hls_generator.py --single-bitrate  # Legacy stream-copy only (no transcoding)
"""

import json
import os
import sys
import time
import shutil
import argparse
import subprocess
import logging
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor, as_completed

import psycopg2

DB_URL = os.environ.get("DB_URL", "postgresql://coomer:coomer_dev@localhost:5433/coomer")
NAS_BASE = os.environ.get("NAS_BASE", "/mnt/nas/drive_2")
HLS_SEGMENT_TIME = 2  # Short segments for fast time-to-first-frame over NAS/SMB
MAX_WORKERS = 2

_OLD_GVFS_PREFIX = "/run/user/1000/gvfs/smb-share:server=unas-pro-8.local,share=drive_2"
def _resolve_path(local_path: str) -> str:
    """Resolve media path: rewrite the dead GVFS SMB prefix to the current NAS
    mount (mirrors the server's _resolve_media_path). Without this, videos whose
    local_path still carries the old GVFS prefix resolve to a non-existent file
    and get skipped forever, never gaining HLS."""
    if local_path and local_path.startswith(_OLD_GVFS_PREFIX):
        return NAS_BASE + local_path[len(_OLD_GVFS_PREFIX):]
    return local_path

# Variant definitions: (label, width, height, video_bitrate, audio_bitrate)
# video_bitrate and audio_bitrate are strings for ffmpeg args
VARIANTS = [
    {"label": "360p",  "width": 640,  "height": 360,  "vbitrate": "800k",  "abitrate": "96k",  "bandwidth": 896000},
    {"label": "720p",  "width": 1280, "height": 720,  "vbitrate": "3000k", "abitrate": "128k", "bandwidth": 3128000},
    {"label": "1080p", "width": 1920, "height": 1080, "vbitrate": "6000k", "abitrate": "192k", "bandwidth": 6192000},
]

# HEVC variants — same resolutions but ~40% lower bitrate for same quality
HEVC_VARIANTS = [
    {"label": "360p_hevc",  "width": 640,  "height": 360,  "vbitrate": "500k",  "abitrate": "96k",  "bandwidth": 596000},
    {"label": "720p_hevc",  "width": 1280, "height": 720,  "vbitrate": "1800k", "abitrate": "128k", "bandwidth": 1928000},
    {"label": "1080p_hevc", "width": 1920, "height": 1080, "vbitrate": "3500k", "abitrate": "192k", "bandwidth": 3692000},
]

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("hls_generator")


def _cleanup_dir(hls_dir: Path):
    """Remove a partially-generated HLS directory."""
    try:
        if hls_dir.exists():
            shutil.rmtree(hls_dir)
    except Exception as e:
        log.warning(f"Failed to clean up {hls_dir}: {e}")


def _probe_video(source_path: str) -> dict:
    """Get video resolution and bitrate using ffprobe."""
    cmd = [
        "ffprobe", "-v", "quiet", "-print_format", "json",
        "-show_streams", "-show_format", source_path,
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, timeout=30)
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return {}
    if result.returncode != 0:
        return {}
    try:
        data = json.loads(result.stdout)
    except (json.JSONDecodeError, ValueError):
        return {}
    video_stream = next(
        (s for s in data.get("streams", []) if s.get("codec_type") == "video"), {}
    )
    return {
        "width": int(video_stream.get("width", 0)),
        "height": int(video_stream.get("height", 0)),
        "bitrate": int(data.get("format", {}).get("bit_rate", 0)),
        "codec": video_stream.get("codec_name", ""),
        "duration": float(data.get("format", {}).get("duration", 0)),
    }


def _select_variants(source_height: int) -> list[dict]:
    """Select which transcoded variants to generate based on source resolution.

    Does not include the source (copy) variant -- that is always generated separately.
    Returns only transcoded variants whose height is strictly less than source.
    """
    if source_height <= 480:
        # Source is low-res; only the copy variant makes sense
        return []
    if source_height <= 720:
        return [v for v in VARIANTS if v["height"] <= 360]
    if source_height <= 1080:
        return [v for v in VARIANTS if v["height"] <= 720]
    # Source > 1080p: include all transcoded variants up to 1080p
    return list(VARIANTS)


def _generate_source_variant(source_path: str, variant_dir: Path, duration: float,
                              codec: str = "", width: int = 0, height: int = 0) -> bool:
    """Generate the source-quality HLS variant.

    For H.264 sources, uses stream-copy (fast).
    For HEVC/other codecs, transcodes to H.264 at source resolution for browser compatibility.
    Returns True on success.
    """
    variant_dir.mkdir(parents=True, exist_ok=True)
    playlist = variant_dir / "stream.m3u8"

    # Skip if already completed
    if playlist.exists() and playlist.stat().st_size > 0:
        return True

    needs_transcode = codec.lower() in ("hevc", "h265", "vp9", "av1")

    if needs_transcode:
        log.info(f"Source is {codec} — transcoding to H.264 (NVENC) at {width}x{height} for browser compatibility")
        # Calculate a generous bitrate to preserve quality
        vbitrate = "8000k" if height >= 1080 else "5000k" if height >= 720 else "3000k"
        vbitrate_num = int(vbitrate.replace("k", "000"))
        cmd = [
            "ffmpeg", "-y",
            "-hwaccel", "cuda", "-hwaccel_output_format", "cuda",
            "-i", source_path,
            "-c:v", "h264_nvenc",
            "-preset", "p4",
            # GOP=60 frames (~2s at 30fps). Without -g/keyint_min NVENC uses
            # default ~250 → -hls_time 2 was honoured in <8% of variants and
            # the first segment was one full GOP (~5 MB blocking start over
            # Tailscale). R2-02 #1.
            "-g", "60", "-keyint_min", "60",
            "-rc", "vbr",
            "-cq", "20",
            "-maxrate", vbitrate,
            "-bufsize", str(vbitrate_num * 2),
            "-c:a", "aac", "-b:a", "192k", "-ac", "2",
            "-muxdelay", "0", "-muxpreload", "0",
            "-hls_time", str(HLS_SEGMENT_TIME),
            "-hls_list_size", "0",
            "-hls_segment_type", "mpegts",
            "-hls_segment_filename", str(variant_dir / "seg_%04d.ts"),
            "-f", "hls",
            str(playlist),
        ]
    else:
        cmd = [
            "ffmpeg", "-y",
            "-i", source_path,
            "-c:v", "copy",
            "-c:a", "aac", "-b:a", "128k",
            "-muxdelay", "0", "-muxpreload", "0",
            "-hls_time", str(HLS_SEGMENT_TIME),
            "-hls_list_size", "0",
            "-hls_segment_type", "mpegts",
            "-hls_segment_filename", str(variant_dir / "seg_%04d.ts"),
            "-f", "hls",
            str(playlist),
        ]

    timeout = 3600
    try:
        result = subprocess.run(cmd, capture_output=True, timeout=timeout)
        if result.returncode != 0:
            stderr = result.stderr.decode(errors="replace")[-500:]
            log.error(f"ffmpeg source-copy failed: {stderr}")
            return False
        return True
    except subprocess.TimeoutExpired:
        log.error("ffmpeg source-copy timed out")
        return False
    except Exception as e:
        log.error(f"ffmpeg source-copy error: {e}")
        return False


def _generate_transcoded_variant(
    source_path: str,
    variant_dir: Path,
    width: int,
    height: int,
    vbitrate: str,
    abitrate: str,
    duration: float,
) -> bool:
    """Generate a transcoded HLS variant at the specified resolution. Returns True on success."""
    variant_dir.mkdir(parents=True, exist_ok=True)
    playlist = variant_dir / "stream.m3u8"

    # Skip if already completed
    if playlist.exists() and playlist.stat().st_size > 0:
        return True

    # Parse numeric bitrate for bufsize calculation
    vbitrate_num = int(vbitrate.replace("k", "000").replace("M", "000000"))

    cmd = [
        "ffmpeg", "-y",
        "-hwaccel", "cuda", "-hwaccel_output_format", "cuda",
        "-i", source_path,
        "-vf", f"scale_cuda={width}:{height}:force_original_aspect_ratio=decrease:force_divisible_by=2",
        "-c:v", "h264_nvenc",
        "-preset", "p4",
        # GOP=60 frames (~2s at 30fps) — see R2-02 #1 note in the h264 path above.
        "-g", "60", "-keyint_min", "60",
        "-rc", "vbr",
        "-cq", "23",
        "-maxrate", vbitrate,
        "-bufsize", str(vbitrate_num * 2),
        "-c:a", "aac",
        "-b:a", abitrate,
        "-ac", "2",
        "-muxdelay", "0", "-muxpreload", "0",
        "-hls_time", str(HLS_SEGMENT_TIME),
        "-hls_list_size", "0",
        "-hls_segment_type", "mpegts",
        "-hls_segment_filename", str(variant_dir / "seg_%04d.ts"),
        "-f", "hls",
        str(playlist),
    ]

    timeout = max(3600, int(duration * 4)) if duration > 0 else 3600
    try:
        result = subprocess.run(cmd, capture_output=True, timeout=timeout)
        if result.returncode != 0:
            stderr = result.stderr.decode(errors="replace")[-500:]
            log.error(f"ffmpeg transcode {width}x{height} failed: {stderr}")
            return False
        return True
    except subprocess.TimeoutExpired:
        log.error(f"ffmpeg transcode {width}x{height} timed out after {timeout}s")
        return False
    except Exception as e:
        log.error(f"ffmpeg transcode {width}x{height} error: {e}")
        return False


def _generate_hevc_variant(
    source_path: str,
    variant_dir: Path,
    width: int,
    height: int,
    vbitrate: str,
    abitrate: str,
    duration: float,
) -> bool:
    """Generate an HEVC/H.265 HLS variant using NVENC. Returns True on success."""
    variant_dir.mkdir(parents=True, exist_ok=True)
    playlist = variant_dir / "stream.m3u8"

    if playlist.exists() and playlist.stat().st_size > 0:
        return True

    vbitrate_num = int(vbitrate.replace("k", "000").replace("M", "000000"))

    cmd = [
        "ffmpeg", "-y",
        "-hwaccel", "cuda", "-hwaccel_output_format", "cuda",
        "-i", source_path,
        "-vf", f"scale_cuda={width}:{height}:force_original_aspect_ratio=decrease:force_divisible_by=2",
        "-c:v", "hevc_nvenc",
        "-preset", "p4",
        # GOP=60 frames (~2s at 30fps) — see R2-02 #1 note in the h264 path above.
        "-g", "60", "-keyint_min", "60",
        "-rc", "vbr",
        "-cq", "23",
        "-maxrate", vbitrate,
        "-bufsize", str(vbitrate_num * 2),
        "-tag:v", "hvc1",  # Required for Safari/iOS HEVC HLS
        "-c:a", "aac",
        "-b:a", abitrate,
        "-ac", "2",
        "-muxdelay", "0", "-muxpreload", "0",
        "-hls_time", str(HLS_SEGMENT_TIME),
        "-hls_list_size", "0",
        "-hls_segment_type", "mpegts",
        "-hls_segment_filename", str(variant_dir / "seg_%04d.ts"),
        "-f", "hls",
        str(playlist),
    ]

    timeout = max(3600, int(duration * 4)) if duration > 0 else 3600
    try:
        result = subprocess.run(cmd, capture_output=True, timeout=timeout)
        if result.returncode != 0:
            stderr = result.stderr.decode(errors="replace")[-500:]
            log.error(f"ffmpeg HEVC transcode {width}x{height} failed: {stderr}")
            return False
        return True
    except subprocess.TimeoutExpired:
        log.error(f"ffmpeg HEVC transcode {width}x{height} timed out")
        return False
    except Exception as e:
        log.error(f"ffmpeg HEVC transcode {width}x{height} error: {e}")
        return False


def _select_hevc_variants(source_height: int) -> list[dict]:
    """Select HEVC variants based on source resolution.

    Always includes the source resolution as HEVC — that's the main benefit
    (same quality at ~40% lower bitrate). Lower resolutions are optional.
    """
    if source_height <= 480:
        return [v for v in HEVC_VARIANTS if v["height"] <= source_height]
    if source_height <= 720:
        return [v for v in HEVC_VARIANTS if v["height"] <= 720]
    if source_height <= 1080:
        return [v for v in HEVC_VARIANTS if v["height"] <= 1080]
    return list(HEVC_VARIANTS)


def _write_master_playlist(master_path: Path, entries: list[dict]):
    """Write (or overwrite) the master m3u8 playlist with variant entries.

    Each entry: {"dir_name": "v0_360p", "bandwidth": 896000, "resolution": "640x360", "codecs": "..."}
    Supports both h264 and HEVC variants with proper CODECS tags for browser selection.
    """
    lines = ["#EXTM3U", "#EXT-X-VERSION:3", ""]
    for entry in entries:
        codecs = entry.get("codecs", "")
        stream_inf = f'#EXT-X-STREAM-INF:BANDWIDTH={entry["bandwidth"]},RESOLUTION={entry["resolution"]}'
        if codecs:
            stream_inf += f',CODECS="{codecs}"'
        lines.append(stream_inf)
        lines.append(f'{entry["dir_name"]}/stream.m3u8')
        lines.append("")
    master_path.write_text("\n".join(lines))


def generate_hls(
    media_id: int,
    source_path: str,
    force: bool = False,
    single_bitrate: bool = False,
) -> str | None:
    """Generate adaptive bitrate HLS for a single video.

    Produces a master playlist with multiple quality variants.
    The source (copy) variant is generated first so the video becomes
    playable immediately; transcoded variants are added progressively.

    If single_bitrate is True, falls back to legacy stream-copy-only behavior.

    Returns hls_dir path on success, None on failure.
    """
    # Always write HLS to NAS for centralized storage
    HLS_OUTPUT_BASE = Path(NAS_BASE) / "onlyfans" / ".hls_output"
    hls_dir = HLS_OUTPUT_BASE / str(media_id)
    master_playlist = hls_dir / "stream.m3u8"

    if force and hls_dir.exists():
        _cleanup_dir(hls_dir)

    hls_dir.mkdir(parents=True, exist_ok=True)

    # Skip if already generated (recovery from interrupted run)
    if master_playlist.exists() and master_playlist.stat().st_size > 0:
        log.info(f"Skipping media {media_id}: HLS already exists at {hls_dir}")
        return str(hls_dir)

    # --- Legacy single-bitrate mode ---
    if single_bitrate:
        return _generate_hls_single(media_id, source_path, hls_dir, master_playlist)

    # --- Adaptive bitrate mode ---

    # Step 1: Probe the source video
    probe = _probe_video(source_path)
    if not probe or probe["width"] == 0:
        log.warning(
            f"Media {media_id}: probe failed, falling back to single-bitrate"
        )
        return _generate_hls_single(media_id, source_path, hls_dir, master_playlist)

    src_w = probe["width"]
    src_h = probe["height"]
    src_bitrate = probe["bitrate"]
    duration = probe["duration"]
    duration_str = f"{int(duration // 60)}m{int(duration % 60)}s" if duration > 0 else "unknown"
    bitrate_mbps = f"{src_bitrate / 1_000_000:.1f} Mbps" if src_bitrate else "unknown bitrate"

    # Step 2: Select which transcoded variants to generate
    transcoded_variants = _select_variants(src_h)
    hevc_variants = _select_hevc_variants(src_h)
    variant_labels = [v["label"] for v in transcoded_variants] + [v["label"] for v in hevc_variants] + ["source"]

    log.info(
        f"Media {media_id}: {src_w}x{src_h} @ {bitrate_mbps}, {duration_str} "
        f"- generating {', '.join(variant_labels)} variants"
    )

    # Step 3: Generate source (copy) variant FIRST -- fast, makes video playable immediately
    source_idx = len(transcoded_variants)  # source goes last in numbering
    source_dir_name = f"v{source_idx}_source"
    source_variant_dir = hls_dir / source_dir_name

    src_codec = probe.get("codec", "")
    if not _generate_source_variant(source_path, source_variant_dir, duration,
                                     codec=src_codec, width=src_w, height=src_h):
        log.error(f"Media {media_id}: source variant failed, aborting")
        _cleanup_dir(hls_dir)
        return None

    # Step 4: Write initial master playlist with just the source variant
    source_resolution = f"{src_w}x{src_h}"
    source_bandwidth = src_bitrate if src_bitrate > 0 else 15_000_000
    playlist_entries = [
        {
            "dir_name": source_dir_name,
            "bandwidth": source_bandwidth,
            "resolution": source_resolution,
            "codecs": "avc1.640028,mp4a.40.2",
        }
    ]
    _write_master_playlist(master_playlist, playlist_entries)
    log.info(f"Media {media_id}: source variant done, video is now playable")

    # Step 5: Generate h264 transcoded variants (NVENC GPU)
    for idx, variant in enumerate(transcoded_variants):
        dir_name = f"v{idx}_{variant['label']}"
        variant_dir = hls_dir / dir_name

        ok = _generate_transcoded_variant(
            source_path=source_path,
            variant_dir=variant_dir,
            width=variant["width"],
            height=variant["height"],
            vbitrate=variant["vbitrate"],
            abitrate=variant["abitrate"],
            duration=duration,
        )

        if ok:
            playlist_entries.insert(len(playlist_entries) - 1, {
                "dir_name": dir_name,
                "bandwidth": variant["bandwidth"],
                "resolution": f"{variant['width']}x{variant['height']}",
                "codecs": "avc1.640028,mp4a.40.2",
            })
            _write_master_playlist(master_playlist, playlist_entries)
            log.info(f"Media {media_id}: {variant['label']} variant done")
        else:
            log.warning(f"Media {media_id}: {variant['label']} transcode failed, continuing")
            if variant_dir.exists():
                shutil.rmtree(variant_dir, ignore_errors=True)

    # Step 6: Generate HEVC variants (NVENC GPU) — ~40% lower bitrate for same quality
    for idx, variant in enumerate(hevc_variants):
        dir_name = f"h{idx}_{variant['label']}"
        variant_dir = hls_dir / dir_name

        ok = _generate_hevc_variant(
            source_path=source_path,
            variant_dir=variant_dir,
            width=variant["width"],
            height=variant["height"],
            vbitrate=variant["vbitrate"],
            abitrate=variant["abitrate"],
            duration=duration,
        )

        if ok:
            playlist_entries.append({
                "dir_name": dir_name,
                "bandwidth": variant["bandwidth"],
                "resolution": f"{variant['width']}x{variant['height']}",
                "codecs": "hvc1.1.6.L120.90,mp4a.40.2",
            })
            _write_master_playlist(master_playlist, playlist_entries)
            log.info(f"Media {media_id}: {variant['label']} HEVC variant done")
        else:
            log.warning(f"Media {media_id}: {variant['label']} HEVC failed, skipping")
            if variant_dir.exists():
                shutil.rmtree(variant_dir, ignore_errors=True)

    return str(hls_dir)


def _generate_hls_single(
    media_id: int, source_path: str, hls_dir: Path, playlist: Path
) -> str | None:
    """Legacy single-bitrate HLS generation (stream copy only)."""
    cmd = [
        "ffmpeg", "-y",
        "-i", source_path,
        "-c:v", "copy",
        "-c:a", "copy",
        "-hls_time", str(HLS_SEGMENT_TIME),
        "-hls_list_size", "0",
        "-hls_segment_type", "mpegts",
        "-hls_segment_filename", str(hls_dir / "seg_%04d.ts"),
        "-f", "hls",
        str(playlist),
    ]

    try:
        result = subprocess.run(cmd, capture_output=True, timeout=3600)
        if result.returncode != 0:
            stderr = result.stderr.decode(errors="replace")[-500:]
            log.error(f"ffmpeg failed for media {media_id}: {stderr}")
            _cleanup_dir(hls_dir)
            return None
        return str(hls_dir)
    except subprocess.TimeoutExpired:
        log.error(f"ffmpeg timed out for media {media_id}")
        _cleanup_dir(hls_dir)
        return None
    except Exception as e:
        log.error(f"Unexpected error for media {media_id}: {e}")
        _cleanup_dir(hls_dir)
        return None


def process_pending(
    conn,
    max_workers: int,
    limit: int = 0,
    dry_run: bool = False,
    force: bool = False,
    single_bitrate: bool = False,
):
    """Fetch all pending videos and process them."""
    cur = conn.cursor()
    if force:
        sql = """
            SELECT id, local_path, file_size
            FROM media_files
            WHERE file_type = 'video' AND local_path IS NOT NULL
            ORDER BY file_size ASC
        """
    else:
        sql = """
            SELECT id, local_path, file_size
            FROM media_files
            WHERE file_type = 'video' AND hls_path IS NULL AND local_path IS NOT NULL
            ORDER BY file_size ASC
        """
    if limit:
        sql += f" LIMIT {limit}"
    cur.execute(sql)
    pending = cur.fetchall()

    total_size = sum(row[2] or 0 for row in pending)
    mode_label = "single-bitrate" if single_bitrate else "adaptive bitrate"
    log.info(
        f"Found {len(pending)} videos pending HLS generation "
        f"({total_size / 1024 / 1024 / 1024:.1f} GB) [{mode_label}]"
    )

    if dry_run:
        for media_id, local_path, file_size in pending[:20]:
            local_path = _resolve_path(local_path)
            exists = "OK" if os.path.isfile(local_path) else "MISSING"
            size_mb = (file_size or 0) / 1024 / 1024
            log.info(f"  [{exists}] media {media_id}: {size_mb:.1f} MB - {os.path.basename(local_path)}")
        if len(pending) > 20:
            log.info(f"  ... and {len(pending) - 20} more")
        return

    processed = 0
    failed = 0
    skipped = 0
    start_time = time.time()

    with ProcessPoolExecutor(max_workers=max_workers) as pool:
        futures = {}
        for media_id, local_path, file_size in pending:
            local_path = _resolve_path(local_path)
            if not os.path.isfile(local_path):
                log.warning(f"Skipping media {media_id}: file not found at {local_path}")
                skipped += 1
                continue
            fut = pool.submit(generate_hls, media_id, local_path, force, single_bitrate)
            futures[fut] = (media_id, file_size)

        total_to_process = len(futures)
        for fut in as_completed(futures):
            media_id, file_size = futures[fut]
            try:
                hls_dir = fut.result()
            except Exception as e:
                log.error(f"[FAIL] media {media_id}: {e}")
                failed += 1
                continue

            if hls_dir:
                cur.execute(
                    "UPDATE media_files SET hls_path = %s WHERE id = %s",
                    (hls_dir, media_id)
                )
                conn.commit()
                processed += 1
                elapsed = time.time() - start_time
                rate = processed / elapsed if elapsed > 0 else 0
                remaining = (total_to_process - processed - failed) / rate if rate > 0 else 0
                size_mb = (file_size or 0) / 1024 / 1024
                log.info(
                    f"[{processed + failed}/{total_to_process}] media {media_id} "
                    f"({size_mb:.1f} MB) -> OK | "
                    f"ETA: {remaining / 60:.0f}m"
                )
            else:
                failed += 1
                log.error(f"[{processed + failed}/{total_to_process}] media {media_id} -> FAILED")

    elapsed = time.time() - start_time
    log.info(
        f"Done: {processed} processed, {failed} failed, {skipped} skipped "
        f"in {elapsed / 60:.1f} minutes"
    )


def main():
    parser = argparse.ArgumentParser(description="Generate HLS streams from video files")
    parser.add_argument("--watch", action="store_true", help="Continuous mode (loop every 60s)")
    parser.add_argument("--workers", type=int, default=MAX_WORKERS, help="Parallel ffmpeg processes")
    parser.add_argument("--limit", type=int, default=0, help="Max videos to process (0=all)")
    parser.add_argument("--dry-run", action="store_true", help="Show what would be processed")
    parser.add_argument("--interval", type=int, default=60, help="Watch mode poll interval (seconds)")
    parser.add_argument("--force", action="store_true", help="Regenerate all HLS (delete and recreate)")
    parser.add_argument("--db-url", default=DB_URL, help="Database URL")
    parser.add_argument(
        "--single-bitrate",
        action="store_true",
        help="Legacy mode: stream-copy only, no adaptive bitrate transcoding",
    )
    args = parser.parse_args()

    # Check NAS is mounted
    if not os.path.isdir(NAS_BASE):
        log.error(f"NAS not mounted at {NAS_BASE}")
        sys.exit(1)

    # Check ffmpeg
    try:
        subprocess.run(["ffmpeg", "-version"], capture_output=True, check=True)
    except (FileNotFoundError, subprocess.CalledProcessError):
        log.error("ffmpeg not found. Install it first.")
        sys.exit(1)

    conn = psycopg2.connect(args.db_url)
    conn.autocommit = False

    try:
        if args.watch:
            log.info(f"Watch mode: polling every {args.interval}s with {args.workers} workers")
            while True:
                process_pending(
                    conn, args.workers, args.limit, args.dry_run,
                    args.force, args.single_bitrate,
                )
                log.info(f"Sleeping {args.interval}s...")
                time.sleep(args.interval)
        else:
            process_pending(
                conn, args.workers, args.limit, args.dry_run,
                args.force, args.single_bitrate,
            )
    except KeyboardInterrupt:
        log.info("Interrupted by user")
    finally:
        conn.close()


if __name__ == "__main__":
    main()