Dispatcharr/apps/channels/tasks.py

# apps/channels/tasks.py
import logging
import os
import select
import re
import requests
import time
import json
import shutil
import subprocess
import signal
import threading
from collections import deque
from zoneinfo import ZoneInfo
from datetime import datetime, timedelta
import gc

from celery import shared_task
from celery.signals import worker_shutting_down
from django.utils.text import slugify
from rapidfuzz import fuzz

from apps.channels.models import Channel
from apps.epg.models import EPGData
from core.models import CoreSettings
from core.utils import acquire_task_lock, release_task_lock

from django.db import OperationalError, close_old_connections
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
import tempfile
from urllib.parse import quote

logger = logging.getLogger(__name__)


# Flag set by the Celery `worker_shutting_down` signal so in-flight
# `run_recording` tasks can distinguish a graceful worker shutdown (e.g.
# `docker stop`, container update) from the natural end of a recording.
# When set, the post-FFmpeg path leaves the HLS working directory intact
# and marks the recording "interrupted" so `recover_recordings_on_startup`
# can resume it on the next boot via the existing path-reuse logic.
_DVR_SHUTTING_DOWN = False


@worker_shutting_down.connect
def _dvr_mark_shutting_down(**_kwargs):
    global _DVR_SHUTTING_DOWN
    _DVR_SHUTTING_DOWN = True


_url_validation_cache = {}
_URL_CACHE_TTL = 300  # seconds


def _validate_url(url, timeout=4):
    """Validate that an HTTP(S) URL is reachable via HEAD request.
    Returns True for non-HTTP URLs (skip validation) or 2xx/3xx responses.
    Results are cached per-worker for 5 minutes to avoid redundant requests
    when multiple recordings reference the same dead URL.
    """
    if not url or not isinstance(url, str):
        return False
    if not url.startswith(("http://", "https://")):
        return True

    now = time.monotonic()
    cached = _url_validation_cache.get(url)
    if cached is not None and (now - cached[1]) < _URL_CACHE_TTL:
        return cached[0]

    try:
        resp = requests.head(url, timeout=timeout, allow_redirects=True)
        if resp.status_code == 405:
            # Server doesn't support HEAD; fall back to ranged GET
            resp = requests.get(
                url, timeout=timeout, allow_redirects=True,
                headers={"Range": "bytes=0-0"}, stream=True,
            )
            resp.close()
        result = resp.status_code < 400
    except Exception:
        result = False

    _url_validation_cache[url] = (result, now)

    # Evict expired entries when cache grows large
    if len(_url_validation_cache) > 512:
        cutoff = now - _URL_CACHE_TTL
        expired = [k for k, v in _url_validation_cache.items() if v[1] < cutoff]
        for k in expired:
            del _url_validation_cache[k]

    return result


def _pick_best_image_from_epg_props(epg_props):
    """Select the highest-quality poster/cover image from EPG custom_properties."""
    try:
        images = epg_props.get("images") or []
        if not isinstance(images, list):
            return None
        size_order = {"xxl": 6, "xl": 5, "l": 4, "m": 3, "s": 2, "xs": 1}
        def score(img):
            t = (img.get("type") or "").lower()
            size = (img.get("size") or "").lower()
            return (2 if t in ("poster", "cover") else 1, size_order.get(size, 0))
        best = None
        for im in images:
            if not isinstance(im, dict):
                continue
            url = im.get("url")
            if not url:
                continue
            if best is None or score(im) > score(best):
                best = im
        return best.get("url") if best else None
    except Exception:
        return None


def _match_epg_program_by_timeslot(channel_epg_data, rec_start, rec_end):
    """Find an EPG program that covers at least 80% of the recording window.

    Queries all programs overlapping the recording, calculates overlap for
    each, and returns the best match only if it covers >= 80% of the
    recording duration.  Recordings spanning multiple programs with no
    dominant show return None (displayed as "Custom Recording").
    Returns a dict with id, title, sub_title, and description, or None.
    """
    if not channel_epg_data or not rec_start or not rec_end:
        return None
    try:
        candidates = channel_epg_data.programs.filter(
            start_time__lt=rec_end,
            end_time__gt=rec_start,
        ).only("id", "title", "sub_title", "description", "start_time", "end_time")

        rec_duration = (rec_end - rec_start).total_seconds()
        if rec_duration <= 0:
            return None

        best = None
        best_overlap = 0
        for prog in candidates:
            overlap_start = max(rec_start, prog.start_time)
            overlap_end = min(rec_end, prog.end_time)
            overlap = (overlap_end - overlap_start).total_seconds()
            if overlap > best_overlap:
                best_overlap = overlap
                best = prog

        if best and (best_overlap / rec_duration) >= 0.8:
            return {
                "id": best.id,
                "title": best.title or "",
                "sub_title": best.sub_title or "",
                "description": best.description or "",
            }
    except Exception:
        pass
    return None


def _db_retry(fn, max_retries=3, base_interval=1, label="DB operation"):
    """Execute fn() with exponential backoff retry on transient DB errors.

    Follows the same backoff pattern as RedisClient.get_client().
    Resets stale connections between attempts so the ORM reconnects.
    """
    for attempt in range(max_retries):
        try:
            return fn()
        except OperationalError:
            if attempt + 1 >= max_retries:
                raise
            wait = base_interval * (2 ** attempt)
            logger.warning(
                f"{label}: failed, retrying in {wait}s "
                f"({attempt + 1}/{max_retries})..."
            )
            close_old_connections()
            time.sleep(wait)


# PostgreSQL btree index has a limit of ~2704 bytes (1/3 of 8KB page size)
# We use 2000 as a safe maximum to account for multibyte characters
def validate_logo_url(logo_url, max_length=2000):
    """
    Fast validation for logo URLs during bulk creation.
    Returns None if URL is too long (would exceed PostgreSQL btree index limit),
    original URL otherwise.

    PostgreSQL btree indexes have a maximum size of ~2704 bytes. URLs longer than
    this cannot be indexed and would cause database errors. These are typically
    base64-encoded images embedded in URLs.
    """
    if logo_url and len(logo_url) > max_length:
        logger.warning(f"Logo URL too long ({len(logo_url)} > {max_length}), skipping: {logo_url[:100]}...")
        return None
    return logo_url

def send_epg_matching_progress(total_channels, matched_channels, current_channel_name="", stage="matching"):
    """
    Send EPG matching progress via WebSocket
    """
    try:
        channel_layer = get_channel_layer()
        if channel_layer:
            progress_data = {
                'type': 'epg_matching_progress',
                'total': total_channels,
                'matched': len(matched_channels) if isinstance(matched_channels, list) else matched_channels,
                'remaining': total_channels - (len(matched_channels) if isinstance(matched_channels, list) else matched_channels),
                'current_channel': current_channel_name,
                'stage': stage,
                'progress_percent': round((len(matched_channels) if isinstance(matched_channels, list) else matched_channels) / total_channels * 100, 1) if total_channels > 0 else 0
            }

            async_to_sync(channel_layer.group_send)(
                "updates",
                {
                    "type": "update",
                    "data": {
                        "type": "epg_matching_progress",
                        **progress_data
                    }
                }
            )
    except Exception as e:
        logger.warning(f"Failed to send EPG matching progress: {e}")

# Lazy loading for ML models - only imported/loaded when needed
_ml_model_cache = {
    'sentence_transformer': None
}

def get_sentence_transformer():
    """Lazy load the sentence transformer model only when needed"""
    if _ml_model_cache['sentence_transformer'] is None:
        try:
            from sentence_transformers import SentenceTransformer
            from sentence_transformers import util

            model_name = "sentence-transformers/all-MiniLM-L6-v2"
            cache_dir = "/data/models"

            # Check environment variable to disable downloads
            disable_downloads = os.environ.get('DISABLE_ML_DOWNLOADS', 'false').lower() == 'true'

            if disable_downloads:
                # Check if model exists before attempting to load
                hf_model_path = os.path.join(cache_dir, f"models--{model_name.replace('/', '--')}")
                if not os.path.exists(hf_model_path):
                    logger.warning("ML model not found and downloads disabled (DISABLE_ML_DOWNLOADS=true). Skipping ML matching.")
                    return None, None

            # Ensure cache directory exists
            os.makedirs(cache_dir, exist_ok=True)

            # Let sentence-transformers handle all cache detection and management
            logger.info(f"Loading sentence transformer model (cache: {cache_dir})")
            _ml_model_cache['sentence_transformer'] = SentenceTransformer(
                model_name,
                cache_folder=cache_dir
            )

            return _ml_model_cache['sentence_transformer'], util
        except ImportError:
            logger.warning("sentence-transformers not available - ML-enhanced matching disabled")
            return None, None
        except Exception as e:
            logger.error(f"Failed to load sentence transformer: {e}")
            return None, None
    else:
        from sentence_transformers import util
        return _ml_model_cache['sentence_transformer'], util

# ML matching thresholds (same as original script)
BEST_FUZZY_THRESHOLD = 85
LOWER_FUZZY_THRESHOLD = 40
EMBED_SIM_THRESHOLD = 0.65

# Words we remove to help with fuzzy + embedding matching
COMMON_EXTRANEOUS_WORDS = [
    "tv", "channel", "network", "television",
    "east", "west", "hd", "uhd", "24/7",
    "1080p", "720p", "540p", "480p",
    "film", "movie", "movies"
]

def normalize_name(name: str) -> str:
    """
    A more aggressive normalization that:
      - Removes user-configured prefixes/suffixes/custom strings (only if mode is 'advanced')
      - Lowercases
      - Removes bracketed/parenthesized text
      - Removes punctuation
      - Strips extraneous words
      - Collapses extra spaces
    """
    if not name:
        return ""

    # Load user-configured EPG matching rules (fail gracefully)
    prefixes = []
    suffixes = []
    custom_strings = []

    try:
        from core.models import CoreSettings
        settings = CoreSettings.get_epg_settings()

        # Check if user has enabled advanced mode
        mode = settings.get("epg_match_mode", "default")

        # Only use custom settings if mode is 'advanced'
        if mode == "advanced":
            prefixes = settings.get("epg_match_ignore_prefixes", [])
            suffixes = settings.get("epg_match_ignore_suffixes", [])
            custom_strings = settings.get("epg_match_ignore_custom", [])

            # Ensure we have lists
            if not isinstance(prefixes, list):
                prefixes = []
            if not isinstance(suffixes, list):
                suffixes = []
            if not isinstance(custom_strings, list):
                custom_strings = []

    except Exception as e:
        # Settings unavailable or error - continue with empty lists (graceful degradation)
        logger.debug(f"Could not load EPG matching settings: {e}")
        prefixes = []
        suffixes = []
        custom_strings = []

    result = name

    # Step 1: Remove prefixes (from START only - exact string match)
    for prefix in prefixes:
        # Skip empty or non-string entries
        if not prefix or not isinstance(prefix, str):
            continue
        # Exact match at start
        if result.startswith(prefix):
            result = result[len(prefix):]
            break  # Only remove first matching prefix

    # Step 2: Remove suffixes (from END only - exact string match)
    for suffix in suffixes:
        # Skip empty or non-string entries
        if not suffix or not isinstance(suffix, str):
            continue
        # Exact match at end
        if result.endswith(suffix):
            result = result[:-len(suffix)]
            break  # Only remove first matching suffix

    # Step 3: Remove custom strings (from ANYWHERE - exact string match)
    for custom in custom_strings:
        # Skip empty or non-string entries
        if not custom or not isinstance(custom, str):
            continue
        try:
            # Exact string removal (replace with empty string)
            result = result.replace(custom, "")
        except Exception as e:
            # If removal fails for any reason, skip this entry
            logger.debug(f"Failed to remove custom string '{custom}': {e}")
            continue

    # Step 4: Existing normalization logic (unchanged)
    norm = result.lower()
    norm = re.sub(r"\[.*?\]", "", norm)

    # Extract and preserve important call signs from parentheses before removing them
    # This captures call signs like (KVLY), (KING), (KARE), etc.
    call_sign_match = re.search(r"\(([A-Z]{3,5})\)", name)
    preserved_call_sign = ""
    if call_sign_match:
        preserved_call_sign = " " + call_sign_match.group(1).lower()

    # Now remove all parentheses content
    norm = re.sub(r"\(.*?\)", "", norm)

    # Add back the preserved call sign
    norm = norm + preserved_call_sign

    norm = re.sub(r"[^\w\s]", "", norm)
    tokens = norm.split()
    tokens = [t for t in tokens if t not in COMMON_EXTRANEOUS_WORDS]
    norm = " ".join(tokens).strip()
    return norm

def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True, send_progress=True):
    """
    EPG matching logic that finds the best EPG matches for channels using
    multiple matching strategies including fuzzy matching and ML models.

    Automatically uses conservative thresholds for bulk matching (multiple channels)
    to avoid bad matches that create user cleanup work, and aggressive thresholds
    for single channel matching where users specifically requested a match attempt.
    """
    channels_to_update = []
    matched_channels = []
    total_channels = len(channels_data)

    # Send initial progress
    if send_progress:
        send_epg_matching_progress(total_channels, 0, stage="starting")

    # Try to get ML models if requested (but don't load yet - lazy loading)
    st_model, util = None, None
    epg_embeddings = None
    ml_available = use_ml

    # Automatically determine matching strategy based on number of channels
    is_bulk_matching = len(channels_data) > 1

    # Adjust matching thresholds based on operation type
    if is_bulk_matching:
        # Conservative thresholds for bulk matching to avoid creating cleanup work
        FUZZY_HIGH_CONFIDENCE = 90      # Only very high fuzzy scores
        FUZZY_MEDIUM_CONFIDENCE = 70    # Higher threshold for ML enhancement
        ML_HIGH_CONFIDENCE = 0.75       # Higher ML confidence required
        ML_LAST_RESORT = 0.65          # More conservative last resort
        FUZZY_LAST_RESORT_MIN = 50     # Higher fuzzy minimum for last resort
        logger.info(f"Using conservative thresholds for bulk matching ({total_channels} channels)")
    else:
        # More aggressive thresholds for single channel matching (user requested specific match)
        FUZZY_HIGH_CONFIDENCE = 85      # Original threshold
        FUZZY_MEDIUM_CONFIDENCE = 40    # Original threshold
        ML_HIGH_CONFIDENCE = 0.65       # Original threshold
        ML_LAST_RESORT = 0.50          # Original desperate threshold
        FUZZY_LAST_RESORT_MIN = 20     # Original minimum
        logger.info("Using aggressive thresholds for single channel matching")    # Process each channel
    for index, chan in enumerate(channels_data):
        normalized_tvg_id = chan.get("tvg_id", "")
        fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"]

        # Send progress update every 5 channels or for the first few
        if send_progress and (index < 5 or index % 5 == 0 or index == total_channels - 1):
            send_epg_matching_progress(
                total_channels,
                len(matched_channels),
                current_channel_name=chan["name"][:50],  # Truncate long names
                stage="matching"
            )
        normalized_tvg_id = chan.get("tvg_id", "")
        fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"]

        # Step 1: Exact TVG ID match
        epg_by_tvg_id = next((epg for epg in epg_data if epg["tvg_id"] == normalized_tvg_id), None)
        if normalized_tvg_id and epg_by_tvg_id:
            chan["epg_data_id"] = epg_by_tvg_id["id"]
            channels_to_update.append(chan)
            matched_channels.append((chan['id'], fallback_name, epg_by_tvg_id["tvg_id"]))
            logger.info(f"Channel {chan['id']} '{fallback_name}' => EPG found by exact tvg_id={epg_by_tvg_id['tvg_id']}")
            continue

        # Step 2: Secondary TVG ID check (legacy compatibility)
        if chan["tvg_id"]:
            epg_match = [epg["id"] for epg in epg_data if epg["tvg_id"] == chan["tvg_id"]]
            if epg_match:
                chan["epg_data_id"] = epg_match[0]
                channels_to_update.append(chan)
                matched_channels.append((chan['id'], fallback_name, chan["tvg_id"]))
                logger.info(f"Channel {chan['id']} '{chan['name']}' => EPG found by secondary tvg_id={chan['tvg_id']}")
                continue

        # Step 2.5: Exact Gracenote ID match
        normalized_gracenote_id = chan.get("gracenote_id", "")
        if normalized_gracenote_id:
            epg_by_gracenote_id = next((epg for epg in epg_data if epg["tvg_id"] == normalized_gracenote_id), None)
            if epg_by_gracenote_id:
                chan["epg_data_id"] = epg_by_gracenote_id["id"]
                channels_to_update.append(chan)
                matched_channels.append((chan['id'], fallback_name, f"gracenote:{epg_by_gracenote_id['tvg_id']}"))
                logger.info(f"Channel {chan['id']} '{fallback_name}' => EPG found by exact gracenote_id={normalized_gracenote_id}")
                continue

        # Step 3: Name-based fuzzy matching
        if not chan["norm_chan"]:
            logger.debug(f"Channel {chan['id']} '{chan['name']}' => empty after normalization, skipping")
            continue

        best_score = 0
        best_epg = None

        # Debug: show what we're matching against
        logger.debug(f"Fuzzy matching '{chan['norm_chan']}' against EPG entries...")

        # Find best fuzzy match
        for row in epg_data:
            if not row.get("norm_name"):
                continue

            base_score = fuzz.ratio(chan["norm_chan"], row["norm_name"])
            bonus = 0

            # Apply region-based bonus/penalty
            if region_code and row.get("tvg_id"):
                combined_text = row["tvg_id"].lower() + " " + row["name"].lower()
                dot_regions = re.findall(r'\.([a-z]{2})', combined_text)

                if dot_regions:
                    if region_code in dot_regions:
                        bonus = 15  # Bigger bonus for matching region
                    else:
                        bonus = -15  # Penalty for different region
                elif region_code in combined_text:
                    bonus = 10

            score = base_score + bonus

            # Debug the best few matches
            if score > 50:  # Only show decent matches
                logger.debug(f"  EPG '{row['name']}' (norm: '{row['norm_name']}') => score: {score} (base: {base_score}, bonus: {bonus})")

            # When scores are equal, prefer higher priority EPG source
            row_priority = row.get('epg_source_priority', 0)
            best_priority = best_epg.get('epg_source_priority', 0) if best_epg else -1

            if score > best_score or (score == best_score and row_priority > best_priority):
                best_score = score
                best_epg = row

        # Log the best score we found
        if best_epg:
            logger.info(f"Channel {chan['id']} '{chan['name']}' => best match: '{best_epg['name']}' (score: {best_score})")
        else:
            logger.debug(f"Channel {chan['id']} '{chan['name']}' => no EPG entries with valid norm_name found")
            continue

        # High confidence match - accept immediately
        if best_score >= FUZZY_HIGH_CONFIDENCE:
            chan["epg_data_id"] = best_epg["id"]
            channels_to_update.append(chan)
            matched_channels.append((chan['id'], chan['name'], best_epg["tvg_id"]))
            logger.info(f"Channel {chan['id']} '{chan['name']}' => matched tvg_id={best_epg['tvg_id']} (score={best_score})")

        # Medium confidence - use ML if available (lazy load models here)
        elif best_score >= FUZZY_MEDIUM_CONFIDENCE and ml_available:
            # Lazy load ML models only when we actually need them
            if st_model is None:
                st_model, util = get_sentence_transformer()

            # Lazy generate embeddings only when we actually need them
            if epg_embeddings is None and st_model and any(row.get("norm_name") for row in epg_data):
                try:
                    logger.info("Generating embeddings for EPG data using ML model (lazy loading)")
                    epg_embeddings = st_model.encode(
                        [row["norm_name"] for row in epg_data if row.get("norm_name")],
                        convert_to_tensor=True
                    )
                except Exception as e:
                    logger.warning(f"Failed to generate embeddings: {e}")
                    epg_embeddings = None

            if epg_embeddings is not None and st_model:
                try:
                    # Generate embedding for this channel
                    chan_embedding = st_model.encode(chan["norm_chan"], convert_to_tensor=True)

                    # Calculate similarity with all EPG embeddings
                    sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0]
                    top_index = int(sim_scores.argmax())
                    top_value = float(sim_scores[top_index])

                    if top_value >= ML_HIGH_CONFIDENCE:
                        # Find the EPG entry that corresponds to this embedding index
                        epg_with_names = [epg for epg in epg_data if epg.get("norm_name")]
                        matched_epg = epg_with_names[top_index]

                        chan["epg_data_id"] = matched_epg["id"]
                        channels_to_update.append(chan)
                        matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"]))
                        logger.info(f"Channel {chan['id']} '{chan['name']}' => matched EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})")
                    else:
                        logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy={best_score}, ML-sim={top_value:.2f} < {ML_HIGH_CONFIDENCE}, trying last resort...")

                        # Last resort: try ML with very low fuzzy threshold
                        if top_value >= ML_LAST_RESORT:  # Dynamic last resort threshold
                            epg_with_names = [epg for epg in epg_data if epg.get("norm_name")]
                            matched_epg = epg_with_names[top_index]

                            chan["epg_data_id"] = matched_epg["id"]
                            channels_to_update.append(chan)
                            matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"]))
                            logger.info(f"Channel {chan['id']} '{chan['name']}' => LAST RESORT match EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})")
                        else:
                            logger.info(f"Channel {chan['id']} '{chan['name']}' => even last resort ML-sim {top_value:.2f} < {ML_LAST_RESORT}, skipping")

                except Exception as e:
                    logger.warning(f"ML matching failed for channel {chan['id']}: {e}")
                    # Fall back to non-ML decision
                    logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy score {best_score} below threshold, skipping")

        # Last resort: Try ML matching even with very low fuzzy scores
        elif best_score >= FUZZY_LAST_RESORT_MIN and ml_available:
            # Lazy load ML models for last resort attempts
            if st_model is None:
                st_model, util = get_sentence_transformer()

            # Lazy generate embeddings for last resort attempts
            if epg_embeddings is None and st_model and any(row.get("norm_name") for row in epg_data):
                try:
                    logger.info("Generating embeddings for EPG data using ML model (last resort lazy loading)")
                    epg_embeddings = st_model.encode(
                        [row["norm_name"] for row in epg_data if row.get("norm_name")],
                        convert_to_tensor=True
                    )
                except Exception as e:
                    logger.warning(f"Failed to generate embeddings for last resort: {e}")
                    epg_embeddings = None

            if epg_embeddings is not None and st_model:
                try:
                    logger.info(f"Channel {chan['id']} '{chan['name']}' => trying ML as last resort (fuzzy={best_score})")
                    # Generate embedding for this channel
                    chan_embedding = st_model.encode(chan["norm_chan"], convert_to_tensor=True)

                    # Calculate similarity with all EPG embeddings
                    sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0]
                    top_index = int(sim_scores.argmax())
                    top_value = float(sim_scores[top_index])

                    if top_value >= ML_LAST_RESORT:  # Dynamic threshold for desperate attempts
                        # Find the EPG entry that corresponds to this embedding index
                        epg_with_names = [epg for epg in epg_data if epg.get("norm_name")]
                        matched_epg = epg_with_names[top_index]

                        chan["epg_data_id"] = matched_epg["id"]
                        channels_to_update.append(chan)
                        matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"]))
                        logger.info(f"Channel {chan['id']} '{chan['name']}' => DESPERATE LAST RESORT match EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})")
                    else:
                        logger.info(f"Channel {chan['id']} '{chan['name']}' => desperate last resort ML-sim {top_value:.2f} < {ML_LAST_RESORT}, giving up")
                except Exception as e:
                    logger.warning(f"Last resort ML matching failed for channel {chan['id']}: {e}")
                    logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {FUZZY_MEDIUM_CONFIDENCE}, giving up")
        else:
            # No ML available or very low fuzzy score
            logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {FUZZY_MEDIUM_CONFIDENCE}, no ML fallback available")

    # Clean up ML models from memory after matching (infrequent operation)
    if _ml_model_cache['sentence_transformer'] is not None:
        logger.info("Cleaning up ML models from memory")
        _ml_model_cache['sentence_transformer'] = None
        gc.collect()

    # Send final progress update
    if send_progress:
        send_epg_matching_progress(
            total_channels,
            len(matched_channels),
            stage="completed"
        )

    return {
        "channels_to_update": channels_to_update,
        "matched_channels": matched_channels
    }

@shared_task
def match_epg_channels():
    """
    Uses integrated EPG matching instead of external script.
    Provides the same functionality with better performance and maintainability.
    """
    try:
        logger.info("Starting integrated EPG matching...")

        # Get region preference
        try:
            region_obj = CoreSettings.objects.get(key="preferred-region")
            region_code = region_obj.value.strip().lower()
        except CoreSettings.DoesNotExist:
            region_code = None

        # Get channels that don't have EPG data assigned
        channels_without_epg = Channel.objects.filter(epg_data__isnull=True)
        logger.info(f"Found {channels_without_epg.count()} channels without EPG data")

        channels_data = []
        for channel in channels_without_epg:
            normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else ""
            normalized_gracenote_id = channel.tvc_guide_stationid.strip().lower() if channel.tvc_guide_stationid else ""
            channels_data.append({
                "id": channel.id,
                "name": channel.name,
                "tvg_id": normalized_tvg_id,
                "original_tvg_id": channel.tvg_id,
                "gracenote_id": normalized_gracenote_id,
                "original_gracenote_id": channel.tvc_guide_stationid,
                "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name,
                "norm_chan": normalize_name(channel.name)  # Always use channel name for fuzzy matching!
            })

        # Get all EPG data from active sources, ordered by source priority (highest first) so we prefer higher priority matches
        epg_data = []
        for epg in EPGData.objects.select_related('epg_source').filter(epg_source__is_active=True):
            normalized_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else ""
            epg_data.append({
                'id': epg.id,
                'tvg_id': normalized_tvg_id,
                'original_tvg_id': epg.tvg_id,
                'name': epg.name,
                'norm_name': normalize_name(epg.name),
                'epg_source_id': epg.epg_source.id if epg.epg_source else None,
                'epg_source_priority': epg.epg_source.priority if epg.epg_source else 0,
            })

        # Sort EPG data by source priority (highest first) so we prefer higher priority matches
        epg_data.sort(key=lambda x: x['epg_source_priority'], reverse=True)

        logger.info(f"Processing {len(channels_data)} channels against {len(epg_data)} EPG entries (from active sources only)")

        # Run EPG matching with progress updates - automatically uses conservative thresholds for bulk operations
        result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True)
        channels_to_update_dicts = result["channels_to_update"]
        matched_channels = result["matched_channels"]

        # Update channels in database
        if channels_to_update_dicts:
            channel_ids = [d["id"] for d in channels_to_update_dicts]
            channels_qs = Channel.objects.filter(id__in=channel_ids)
            channels_list = list(channels_qs)

            # Create mapping from channel_id to epg_data_id
            epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts}

            # Update each channel with matched EPG data
            for channel_obj in channels_list:
                epg_data_id = epg_mapping.get(channel_obj.id)
                if epg_data_id:
                    try:
                        epg_data_obj = EPGData.objects.get(id=epg_data_id)
                        channel_obj.epg_data = epg_data_obj
                    except EPGData.DoesNotExist:
                        logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}")

            # Bulk update all channels
            Channel.objects.bulk_update(channels_list, ["epg_data"])

        total_matched = len(matched_channels)
        if total_matched:
            logger.info(f"Match Summary: {total_matched} channel(s) matched.")
            for (cid, cname, tvg) in matched_channels:
                logger.info(f"  - Channel ID={cid}, Name='{cname}' => tvg_id='{tvg}'")
        else:
            logger.info("No new channels were matched.")

        logger.info("Finished integrated EPG matching.")

        # Send WebSocket update
        channel_layer = get_channel_layer()
        associations = [
            {"channel_id": chan["id"], "epg_data_id": chan["epg_data_id"]}
            for chan in channels_to_update_dicts
        ]

        async_to_sync(channel_layer.group_send)(
            'updates',
            {
                'type': 'update',
                "data": {
                    "success": True,
                    "type": "epg_match",
                    "refresh_channels": True,
                    "matches_count": total_matched,
                    "message": f"EPG matching complete: {total_matched} channel(s) matched",
                    "associations": associations
                }
            }
        )

        return f"Done. Matched {total_matched} channel(s)."

    finally:
        # Clean up ML models from memory after bulk matching
        if _ml_model_cache['sentence_transformer'] is not None:
            logger.info("Cleaning up ML models from memory")
            _ml_model_cache['sentence_transformer'] = None

        # Memory cleanup
        gc.collect()
        from core.utils import cleanup_memory
        cleanup_memory(log_usage=True, force_collection=True)


@shared_task
def match_selected_channels_epg(channel_ids):
    """
    Match EPG data for only the specified selected channels.
    Uses the same integrated EPG matching logic but processes only selected channels.
    """
    try:
        logger.info(f"Starting integrated EPG matching for {len(channel_ids)} selected channels...")

        # Get region preference
        try:
            region_obj = CoreSettings.objects.get(key="preferred-region")
            region_code = region_obj.value.strip().lower()
        except CoreSettings.DoesNotExist:
            region_code = None

        # Get only the specified channels that don't have EPG data assigned
        channels_without_epg = Channel.objects.filter(
            id__in=channel_ids,
            epg_data__isnull=True
        )
        logger.info(f"Found {channels_without_epg.count()} selected channels without EPG data")

        if not channels_without_epg.exists():
            logger.info("No selected channels need EPG matching.")

            # Send WebSocket update
            channel_layer = get_channel_layer()
            async_to_sync(channel_layer.group_send)(
                'updates',
                {
                    'type': 'update',
                    "data": {
                        "success": True,
                        "type": "epg_match",
                        "refresh_channels": True,
                        "matches_count": 0,
                        "message": "No selected channels need EPG matching",
                        "associations": []
                    }
                }
            )
            return "No selected channels needed EPG matching."

        channels_data = []
        for channel in channels_without_epg:
            normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else ""
            normalized_gracenote_id = channel.tvc_guide_stationid.strip().lower() if channel.tvc_guide_stationid else ""
            channels_data.append({
                "id": channel.id,
                "name": channel.name,
                "tvg_id": normalized_tvg_id,
                "original_tvg_id": channel.tvg_id,
                "gracenote_id": normalized_gracenote_id,
                "original_gracenote_id": channel.tvc_guide_stationid,
                "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name,
                "norm_chan": normalize_name(channel.name)
            })

        # Get all EPG data from active sources, ordered by source priority (highest first) so we prefer higher priority matches
        epg_data = []
        for epg in EPGData.objects.select_related('epg_source').filter(epg_source__is_active=True):
            normalized_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else ""
            epg_data.append({
                'id': epg.id,
                'tvg_id': normalized_tvg_id,
                'original_tvg_id': epg.tvg_id,
                'name': epg.name,
                'norm_name': normalize_name(epg.name),
                'epg_source_id': epg.epg_source.id if epg.epg_source else None,
                'epg_source_priority': epg.epg_source.priority if epg.epg_source else 0,
            })

        # Sort EPG data by source priority (highest first) so we prefer higher priority matches
        epg_data.sort(key=lambda x: x['epg_source_priority'], reverse=True)

        logger.info(f"Processing {len(channels_data)} selected channels against {len(epg_data)} EPG entries (from active sources only)")

        # Run EPG matching with progress updates - automatically uses appropriate thresholds
        result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True)
        channels_to_update_dicts = result["channels_to_update"]
        matched_channels = result["matched_channels"]

        # Update channels in database
        if channels_to_update_dicts:
            channel_ids_to_update = [d["id"] for d in channels_to_update_dicts]
            channels_qs = Channel.objects.filter(id__in=channel_ids_to_update)
            channels_list = list(channels_qs)

            # Create mapping from channel_id to epg_data_id
            epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts}

            # Update each channel with matched EPG data
            for channel_obj in channels_list:
                epg_data_id = epg_mapping.get(channel_obj.id)
                if epg_data_id:
                    try:
                        epg_data_obj = EPGData.objects.get(id=epg_data_id)
                        channel_obj.epg_data = epg_data_obj
                    except EPGData.DoesNotExist:
                        logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}")

            # Bulk update all channels
            Channel.objects.bulk_update(channels_list, ["epg_data"])

        total_matched = len(matched_channels)
        if total_matched:
            logger.info(f"Selected Channel Match Summary: {total_matched} channel(s) matched.")
            for (cid, cname, tvg) in matched_channels:
                logger.info(f"  - Channel ID={cid}, Name='{cname}' => tvg_id='{tvg}'")
        else:
            logger.info("No selected channels were matched.")

        logger.info("Finished integrated EPG matching for selected channels.")

        # Send WebSocket update
        channel_layer = get_channel_layer()
        associations = [
            {"channel_id": chan["id"], "epg_data_id": chan["epg_data_id"]}
            for chan in channels_to_update_dicts
        ]

        async_to_sync(channel_layer.group_send)(
            'updates',
            {
                'type': 'update',
                "data": {
                    "success": True,
                    "type": "epg_match",
                    "refresh_channels": True,
                    "matches_count": total_matched,
                    "message": f"EPG matching complete: {total_matched} selected channel(s) matched",
                    "associations": associations
                }
            }
        )

        return f"Done. Matched {total_matched} selected channel(s)."

    finally:
        # Clean up ML models from memory after bulk matching
        if _ml_model_cache['sentence_transformer'] is not None:
            logger.info("Cleaning up ML models from memory")
            _ml_model_cache['sentence_transformer'] = None

        # Memory cleanup
        gc.collect()
        from core.utils import cleanup_memory
        cleanup_memory(log_usage=True, force_collection=True)


@shared_task
def match_single_channel_epg(channel_id):
    """
    Try to match a single channel with EPG data using the integrated matching logic
    that includes both fuzzy and ML-enhanced matching. Returns a dict with match status and message.
    """
    try:
        from apps.channels.models import Channel
        from apps.epg.models import EPGData

        logger.info(f"Starting integrated single channel EPG matching for channel ID {channel_id}")

        # Get the channel
        try:
            channel = Channel.objects.get(id=channel_id)
        except Channel.DoesNotExist:
            return {"matched": False, "message": "Channel not found"}

        # If channel already has EPG data, skip
        if channel.epg_data:
            return {"matched": False, "message": f"Channel '{channel.name}' already has EPG data assigned"}

        # Prepare single channel data for matching (same format as bulk matching)
        normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else ""
        normalized_gracenote_id = channel.tvc_guide_stationid.strip().lower() if channel.tvc_guide_stationid else ""
        channel_data = {
            "id": channel.id,
            "name": channel.name,
            "tvg_id": normalized_tvg_id,
            "original_tvg_id": channel.tvg_id,
            "gracenote_id": normalized_gracenote_id,
            "original_gracenote_id": channel.tvc_guide_stationid,
            "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name,
            "norm_chan": normalize_name(channel.name)  # Always use channel name for fuzzy matching!
        }

        logger.info(f"Channel data prepared: name='{channel.name}', tvg_id='{normalized_tvg_id}', gracenote_id='{normalized_gracenote_id}', norm_chan='{channel_data['norm_chan']}'")

        # Debug: Test what the normalization does to preserve call signs
        test_name = "NBC 11 (KVLY) - Fargo"  # Example for testing
        test_normalized = normalize_name(test_name)
        logger.debug(f"DEBUG normalization example: '{test_name}' → '{test_normalized}' (call sign preserved)")

        # Get all EPG data for matching from active sources - must include norm_name field
        # Ordered by source priority (highest first) so we prefer higher priority matches
        epg_data_list = []
        for epg in EPGData.objects.select_related('epg_source').filter(epg_source__is_active=True, name__isnull=False).exclude(name=''):
            normalized_epg_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else ""
            epg_data_list.append({
                'id': epg.id,
                'tvg_id': normalized_epg_tvg_id,
                'original_tvg_id': epg.tvg_id,
                'name': epg.name,
                'norm_name': normalize_name(epg.name),
                'epg_source_id': epg.epg_source.id if epg.epg_source else None,
                'epg_source_priority': epg.epg_source.priority if epg.epg_source else 0,
            })

        # Sort EPG data by source priority (highest first) so we prefer higher priority matches
        epg_data_list.sort(key=lambda x: x['epg_source_priority'], reverse=True)

        if not epg_data_list:
            return {"matched": False, "message": "No EPG data available for matching (from active sources)"}

        logger.info(f"Matching single channel '{channel.name}' against {len(epg_data_list)} EPG entries")

        # Send progress for single channel matching
        send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="matching")

        # Use the EPG matching function - automatically uses aggressive thresholds for single channel
        result = match_channels_to_epg([channel_data], epg_data_list, send_progress=False)
        channels_to_update = result.get("channels_to_update", [])
        matched_channels = result.get("matched_channels", [])

        if channels_to_update:
            # Find our channel in the results
            channel_match = None
            for update in channels_to_update:
                if update["id"] == channel.id:
                    channel_match = update
                    break

            if channel_match:
                # Apply the match to the channel
                try:
                    epg_data = EPGData.objects.get(id=channel_match['epg_data_id'])
                    channel.epg_data = epg_data
                    channel.save(update_fields=["epg_data"])

                    # Find match details from matched_channels for better reporting
                    match_details = None
                    for match_info in matched_channels:
                        if match_info[0] == channel.id:  # matched_channels format: (channel_id, channel_name, epg_info)
                            match_details = match_info
                            break

                    success_msg = f"Channel '{channel.name}' matched with EPG '{epg_data.name}'"
                    if match_details:
                        success_msg += f" (matched via: {match_details[2]})"

                    logger.info(success_msg)

                    # Send completion progress for single channel
                    send_epg_matching_progress(1, 1, current_channel_name=channel.name, stage="completed")

                    # Clean up ML models from memory after single channel matching
                    if _ml_model_cache['sentence_transformer'] is not None:
                        logger.info("Cleaning up ML models from memory")
                        _ml_model_cache['sentence_transformer'] = None
                        gc.collect()

                    return {
                        "matched": True,
                        "message": success_msg,
                        "epg_name": epg_data.name,
                        "epg_id": epg_data.id
                    }
                except EPGData.DoesNotExist:
                    return {"matched": False, "message": "Matched EPG data not found"}

        # No match found
        # Send completion progress for single channel (failed)
        send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="completed")

        # Clean up ML models from memory after single channel matching
        if _ml_model_cache['sentence_transformer'] is not None:
            logger.info("Cleaning up ML models from memory")
            _ml_model_cache['sentence_transformer'] = None
            gc.collect()

        return {
            "matched": False,
            "message": f"No suitable EPG match found for channel '{channel.name}'"
        }

    except Exception as e:
        logger.error(f"Error in integrated single channel EPG matching: {e}", exc_info=True)

        # Clean up ML models from memory even on error
        if _ml_model_cache['sentence_transformer'] is not None:
            logger.info("Cleaning up ML models from memory after error")
            _ml_model_cache['sentence_transformer'] = None
            gc.collect()

        return {"matched": False, "message": f"Error during matching: {str(e)}"}


def evaluate_series_rules_impl(tvg_id: str | None = None):
    """Synchronous implementation of series rule evaluation; returns details for debugging."""
    result = {"scheduled": 0, "details": []}

    # Serialize all invocations to prevent concurrent evaluations from
    # racing to create duplicate recordings (e.g. multiple EPG sources
    # refreshing simultaneously each firing evaluate_series_rules.delay()).
    # If Redis is unavailable, proceed without lock — the primary and
    # secondary dedup guards still prevent duplicates.
    lock_acquired = False
    try:
        lock_acquired = acquire_task_lock('evaluate_series_rules', 'all')
        if not lock_acquired:
            result["details"].append({"status": "skipped", "reason": "concurrent evaluation in progress"})
            return result
    except (ConnectionError, OSError, AttributeError):
        logger.warning("Could not acquire series rule evaluation lock (Redis unavailable), proceeding without lock")

    try:
        return _evaluate_series_rules_locked(tvg_id, result)
    finally:
        if lock_acquired:
            try:
                release_task_lock('evaluate_series_rules', 'all')
            except (ConnectionError, OSError, AttributeError):
                logger.warning("Could not release series rule evaluation lock")


def _evaluate_series_rules_locked(tvg_id, result):
    """Inner implementation of series rule evaluation, called under lock."""
    from django.utils import timezone
    from apps.channels.models import Recording, Channel
    from apps.epg.models import EPGData, ProgramData

    rules = CoreSettings.get_dvr_series_rules()
    if not isinstance(rules, list) or not rules:
        return result

    # Optionally filter for tvg_id
    if tvg_id:
        rules = [r for r in rules if str(r.get("tvg_id")) == str(tvg_id)]
        if not rules:
            result["details"].append({"tvg_id": tvg_id, "status": "no_rule"})
            return result

    now = timezone.now()
    horizon = now + timedelta(days=7)

    # Preload existing recordings keyed by stable program attributes that
    # survive EPG refreshes (tvg_id + original start/end times stored in
    # custom_properties).  ProgramData.id changes on every EPG refresh so
    # it cannot be used for deduplication.  Only load future recordings
    # to bound the set size — past recordings cannot collide with newly
    # scheduled future programs.
    existing_program_keys = set()
    for cp in Recording.objects.filter(
        end_time__gte=now,
    ).values_list("custom_properties", flat=True):
        try:
            prog_data = (cp or {}).get("program", {})
            tvg_id_val = prog_data.get("tvg_id")
            st = prog_data.get("start_time")
            et = prog_data.get("end_time")
            if tvg_id_val and st and et:
                existing_program_keys.add((str(tvg_id_val), str(st), str(et)))
        except Exception:
            continue

    for rule in rules:
        rv_tvg = str(rule.get("tvg_id") or "").strip()
        mode = (rule.get("mode") or "all").lower()
        series_title = (rule.get("title") or "").strip()
        title_mode = (rule.get("title_mode") or "exact").lower()
        description = (rule.get("description") or "").strip()
        description_mode = (rule.get("description_mode") or "contains").lower()
        try:
            pinned_channel_id = int(rule["channel_id"]) if rule.get("channel_id") not in (None, "") else None
        except (TypeError, ValueError):
            pinned_channel_id = None
        if not rv_tvg:
            result["details"].append({"tvg_id": rv_tvg, "status": "invalid_rule"})
            continue

        epg = EPGData.objects.filter(tvg_id=rv_tvg).first()
        if not epg:
            result["details"].append({"tvg_id": rv_tvg, "status": "no_epg_match"})
            continue

        programs_qs = ProgramData.objects.filter(
                epg=epg,
                end_time__gt=now,
                start_time__lte=horizon,
            )

        from apps.epg.query_utils import parse_text_query

        if series_title:
            if title_mode == "exact":
                programs_qs = programs_qs.filter(title__iexact=series_title)
            else:
                use_regex = title_mode == "regex"
                whole_words = title_mode == "search"
                programs_qs = programs_qs.filter(
                    parse_text_query("title", series_title, use_regex=use_regex, whole_words=whole_words)
                )

        if description:
            use_regex = description_mode == "regex"
            whole_words = description_mode == "search"
            programs_qs = programs_qs.filter(
                parse_text_query("description", description, use_regex=use_regex, whole_words=whole_words)
            )

        programs = list(programs_qs.distinct().order_by("start_time"))

        if pinned_channel_id is not None:
            channel = Channel.objects.filter(id=pinned_channel_id).first()
            if channel is None:
                result["details"].append({"tvg_id": rv_tvg, "status": "pinned_channel_missing", "channel_id": pinned_channel_id})
                continue
        else:
            channel = Channel.objects.filter(epg_data=epg).order_by("channel_number").first()
        if not channel:
            result["details"].append({"tvg_id": rv_tvg, "status": "no_channel_for_epg"})
            continue

        #
        # Many providers list multiple future airings of the same episode
        # (e.g., prime-time and a late-night repeat). Previously we scheduled
        # a recording for each airing which shows up as duplicates in the DVR.
        #
        # To avoid that, we collapse programs to the earliest airing per
        # unique episode using the best identifier available:
        #  - season+episode from ProgramData.custom_properties
        #  - onscreen_episode (e.g., S08E03)
        #  - sub_title (episode name), scoped by tvg_id+series title
        # If none of the above exist, we fall back to keeping each program
        # (usually movies or specials without episode identifiers).
        #
        def _episode_key(p: "ProgramData"):
            try:
                props = p.custom_properties or {}
                season = props.get("season")
                episode = props.get("episode")
                onscreen = props.get("onscreen_episode")
            except Exception:
                season = episode = onscreen = None
            base = f"{p.tvg_id or ''}|{(p.title or '').strip().lower()}"  # series scope
            if season is not None and episode is not None:
                return f"{base}|s{season}e{episode}"
            if onscreen:
                return f"{base}|{str(onscreen).strip().lower()}"
            if p.sub_title:
                return f"{base}|{p.sub_title.strip().lower()}"
            # No reliable episode identity; use the program id to avoid over-merging
            return f"id:{p.id}"

        # Optionally filter to only brand-new episodes before grouping
        if mode == "new":
            filtered = []
            for p in programs:
                try:
                    if (p.custom_properties or {}).get("new"):
                        filtered.append(p)
                except Exception:
                    pass
            programs = filtered

        # Pick the earliest airing for each episode key
        earliest_by_key = {}
        for p in programs:
            k = _episode_key(p)
            cur = earliest_by_key.get(k)
            if cur is None or p.start_time < cur.start_time:
                earliest_by_key[k] = p

        unique_programs = list(earliest_by_key.values())

        created_here = 0
        for prog in unique_programs:
            try:
                # Skip if a recording already exists for this exact airing
                # (keyed by tvg_id + original program times, which are stable
                # across EPG refreshes unlike ProgramData.id).
                prog_key = (str(prog.tvg_id), prog.start_time.isoformat(), prog.end_time.isoformat())
                if prog_key in existing_program_keys:
                    continue
                # Extra guard: DB query using the same stable attributes
                # stored in custom_properties (unadjusted program times,
                # not offset-adjusted Recording.start_time/end_time).
                try:
                    if Recording.objects.filter(
                        custom_properties__program__tvg_id=prog.tvg_id,
                        custom_properties__program__start_time=prog.start_time.isoformat(),
                        custom_properties__program__end_time=prog.end_time.isoformat(),
                    ).exists():
                        continue
                except Exception:
                    continue  # already scheduled/recorded

                # Apply global DVR pre/post offsets (in minutes)
                try:
                    pre_min = int(CoreSettings.get_dvr_pre_offset_minutes())
                except Exception:
                    pre_min = 0
                try:
                    post_min = int(CoreSettings.get_dvr_post_offset_minutes())
                except Exception:
                    post_min = 0

                adj_start = prog.start_time
                adj_end = prog.end_time
                try:
                    if pre_min and pre_min > 0:
                        adj_start = adj_start - timedelta(minutes=pre_min)
                except Exception:
                    pass
                try:
                    if post_min and post_min > 0:
                        adj_end = adj_end + timedelta(minutes=post_min)
                except Exception:
                    pass

                rec = Recording.objects.create(
                    channel=channel,
                    start_time=adj_start,
                    end_time=adj_end,
                    custom_properties={
                        "program": {
                            "id": prog.id,
                            "tvg_id": prog.tvg_id,
                            "title": prog.title,
                            "sub_title": prog.sub_title,
                            "description": prog.description,
                            "start_time": prog.start_time.isoformat(),
                            "end_time": prog.end_time.isoformat(),
                        }
                    },
                )
                existing_program_keys.add(prog_key)
                created_here += 1
                try:
                    prefetch_recording_artwork.apply_async(args=[rec.id], countdown=1)
                except Exception:
                    pass
            except Exception as e:
                result["details"].append({"tvg_id": rv_tvg, "status": "error", "error": str(e)})
                continue
        result["scheduled"] += created_here
        result["details"].append({"tvg_id": rv_tvg, "title": series_title, "status": "ok", "created": created_here})

    # Notify frontend to refresh
    try:
        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)(
            'updates',
            {'type': 'update', 'data': {"success": True, "type": "recordings_refreshed", "scheduled": result["scheduled"]}},
        )
    except Exception:
        pass

    return result


@shared_task
def evaluate_series_rules(tvg_id: str | None = None):
    return evaluate_series_rules_impl(tvg_id)


def reschedule_upcoming_recordings_for_offset_change_impl():
    """Recalculate start/end for all future EPG-based recordings using current DVR offsets.

    Only recordings that have not yet started (start_time > now) and that were
    scheduled from EPG data (custom_properties.program present) are updated.
    """
    from django.utils import timezone
    from django.utils.dateparse import parse_datetime
    from apps.channels.models import Recording

    now = timezone.now()

    try:
        pre_min = int(CoreSettings.get_dvr_pre_offset_minutes())
    except Exception:
        pre_min = 0
    try:
        post_min = int(CoreSettings.get_dvr_post_offset_minutes())
    except Exception:
        post_min = 0

    changed = 0
    scanned = 0

    for rec in Recording.objects.filter(start_time__gt=now).iterator():
        scanned += 1
        try:
            cp = rec.custom_properties or {}
            program = cp.get("program") if isinstance(cp, dict) else None
            if not isinstance(program, dict):
                continue
            base_start = program.get("start_time")
            base_end = program.get("end_time")
            if not base_start or not base_end:
                continue
            start_dt = parse_datetime(str(base_start))
            end_dt = parse_datetime(str(base_end))
            if start_dt is None or end_dt is None:
                continue

            adj_start = start_dt
            adj_end = end_dt
            try:
                if pre_min and pre_min > 0:
                    adj_start = adj_start - timedelta(minutes=pre_min)
            except Exception:
                pass
            try:
                if post_min and post_min > 0:
                    adj_end = adj_end + timedelta(minutes=post_min)
            except Exception:
                pass

            if rec.start_time != adj_start or rec.end_time != adj_end:
                rec.start_time = adj_start
                rec.end_time = adj_end
                rec.save(update_fields=["start_time", "end_time"])
                changed += 1
        except Exception:
            continue

    # Notify frontend to refresh
    try:
        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)(
            'updates',
            {'type': 'update', 'data': {"success": True, "type": "recordings_refreshed", "rescheduled": changed}},
        )
    except Exception:
        pass

    return {"changed": changed, "scanned": scanned, "pre": pre_min, "post": post_min}


@shared_task
def reschedule_upcoming_recordings_for_offset_change():
    return reschedule_upcoming_recordings_for_offset_change_impl()


def _notify_recordings_refresh():
    try:
        from core.utils import send_websocket_update
        send_websocket_update('updates', 'update', {"success": True, "type": "recordings_refreshed"})
    except Exception:
        pass


def purge_recurring_rule_impl(rule_id: int) -> int:
    """Remove all future recordings created by a recurring rule."""
    from django.utils import timezone
    from .models import Recording

    now = timezone.now()
    try:
        removed, _ = Recording.objects.filter(
            start_time__gte=now,
            custom_properties__rule__id=rule_id,
        ).delete()
    except Exception:
        removed = 0
    if removed:
        _notify_recordings_refresh()
    return removed


def sync_recurring_rule_impl(rule_id: int, drop_existing: bool = True, horizon_days: int = 14) -> int:
    """Ensure recordings exist for a recurring rule within the scheduling horizon."""
    from django.utils import timezone
    from .models import RecurringRecordingRule, Recording

    rule = RecurringRecordingRule.objects.filter(pk=rule_id).select_related("channel").first()
    now = timezone.now()
    removed = 0
    if drop_existing:
        removed = purge_recurring_rule_impl(rule_id)

    if not rule or not rule.enabled:
        return 0

    days = rule.cleaned_days()
    if not days:
        return 0

    tz_name = CoreSettings.get_system_time_zone()
    try:
        tz = ZoneInfo(tz_name)
    except Exception:
        logger.warning("Invalid or unsupported time zone '%s'; falling back to Server default", tz_name)
        tz = timezone.get_current_timezone()
    local_today = now.astimezone(tz).date()
    start_limit = rule.start_date or local_today
    end_limit = rule.end_date
    horizon = now + timedelta(days=horizon_days)
    start_window = max(start_limit, local_today)
    if drop_existing and end_limit:
        end_window = end_limit
    else:
        end_window = horizon.astimezone(tz).date()
        if end_limit and end_limit < end_window:
            end_window = end_limit
    if end_window < start_window:
        return 0
    total_created = 0

    for offset in range((end_window - start_window).days + 1):
        target_date = start_window + timedelta(days=offset)
        if target_date.weekday() not in days:
            continue
        if end_limit and target_date > end_limit:
            continue
        try:
            start_dt = timezone.make_aware(datetime.combine(target_date, rule.start_time), tz)
            end_dt = timezone.make_aware(datetime.combine(target_date, rule.end_time), tz)
        except Exception:
            continue
        if end_dt <= start_dt:
            end_dt = end_dt + timedelta(days=1)
        if start_dt <= now:
            continue
        exists = Recording.objects.filter(
            channel=rule.channel,
            start_time=start_dt,
            custom_properties__rule__id=rule.id,
        ).exists()
        if exists:
            continue
        description = rule.name or f"Recurring recording for {rule.channel.name}"
        cp = {
            "rule": {
                "type": "recurring",
                "id": rule.id,
                "days_of_week": days,
                "name": rule.name or "",
            },
            "status": "scheduled",
            "description": description,
            "program": {
                "title": rule.name or rule.channel.name,
                "description": description,
                "start_time": start_dt.isoformat(),
                "end_time": end_dt.isoformat(),
            },
        }
        try:
            Recording.objects.create(
                channel=rule.channel,
                start_time=start_dt,
                end_time=end_dt,
                custom_properties=cp,
            )
            total_created += 1
        except Exception as err:
            logger.warning(f"Failed to create recurring recording for rule {rule.id}: {err}")

    if removed or total_created:
        _notify_recordings_refresh()

    return total_created


@shared_task
def rebuild_recurring_rule(rule_id: int, horizon_days: int = 14):
    return sync_recurring_rule_impl(rule_id, drop_existing=True, horizon_days=horizon_days)


@shared_task
def maintain_recurring_recordings():
    from .models import RecurringRecordingRule

    total = 0
    for rule_id in RecurringRecordingRule.objects.filter(enabled=True).values_list("id", flat=True):
        try:
            total += sync_recurring_rule_impl(rule_id, drop_existing=False)
        except Exception as err:
            logger.warning(f"Recurring rule maintenance failed for {rule_id}: {err}")
    return total


@shared_task
def purge_recurring_rule(rule_id: int):
    return purge_recurring_rule_impl(rule_id)

@shared_task
def _safe_name(s):
    try:
        import re
        s = s or ""
        # Remove forbidden filename characters and normalize spaces
        s = re.sub(r'[\\/:*?"<>|]+', '', s)
        s = s.strip()
        return s
    except Exception:
        return s or ""


def _parse_epg_tv_movie_info(program):
    """Return tuple (is_movie, season, episode, year, sub_title) from EPG ProgramData if available."""
    is_movie = False
    season = None
    episode = None
    year = None
    sub_title = program.get('sub_title') if isinstance(program, dict) else None
    try:
        from apps.epg.models import ProgramData
        prog_id = program.get('id') if isinstance(program, dict) else None
        epg_program = ProgramData.objects.filter(id=prog_id).only('custom_properties').first() if prog_id else None
        if epg_program and epg_program.custom_properties:
            cp = epg_program.custom_properties
            # Determine categories
            cats = [c.lower() for c in (cp.get('categories') or []) if isinstance(c, str)]
            is_movie = 'movie' in cats or 'film' in cats
            season = cp.get('season')
            episode = cp.get('episode')
            onscreen = cp.get('onscreen_episode')
            if (season is None or episode is None) and isinstance(onscreen, str):
                import re as _re
                m = _re.search(r'[sS](\d+)[eE](\d+)', onscreen)
                if m:
                    season = season or int(m.group(1))
                    episode = episode or int(m.group(2))
            d = cp.get('date')
            if d:
                year = str(d)[:4]
    except Exception:
        pass
    return is_movie, season, episode, year, sub_title


def _build_output_paths(channel, program, start_time, end_time, recording_id):
    """
    Build (final_path, hls_dir, final_filename) using DVR templates.

    The HLS working directory is hidden and tagged with the recording id
    (e.g. ``.dvr_71_hls``) so it cannot collide with another recording's
    files and so orphan-cleanup utilities can identify internal scratch
    directories unambiguously.
    """
    from core.models import CoreSettings
    # Root for DVR recordings: fixed to /data/recordings inside the container
    library_root = '/data/recordings'

    is_movie, season, episode, year, sub_title = _parse_epg_tv_movie_info(program)
    show = _safe_name(program.get('title') if isinstance(program, dict) else channel.name)
    title = _safe_name(program.get('title') if isinstance(program, dict) else channel.name)
    sub_title = _safe_name(sub_title)
    season = int(season) if season is not None else 0
    episode = int(episode) if episode is not None else 0
    year = year or str(start_time.year)

    values = {
        'show': show,
        'title': title,
        'sub_title': sub_title,
        'season': season,
        'episode': episode,
        'year': year,
        'channel': _safe_name(channel.name),
        'start': start_time.strftime('%Y%m%d_%H%M%S'),
        'end': end_time.strftime('%Y%m%d_%H%M%S'),
    }

    template = CoreSettings.get_dvr_movie_template() if is_movie else CoreSettings.get_dvr_tv_template()
    # Build relative path from templates with smart fallbacks
    rel_path = None
    if not is_movie and (season == 0 or episode == 0):
        # TV fallback template when S/E are missing
        try:
            tv_fb = CoreSettings.get_dvr_tv_fallback_template()
            rel_path = tv_fb.format(**values)
        except Exception:
            # Older setting support
            try:
                fallback_root = CoreSettings.get_dvr_tv_fallback_dir()
            except Exception:
                fallback_root = "TV_Shows"
            rel_path = f"{fallback_root}/{show}/{values['start']}.mkv"
    if not rel_path:
        try:
            rel_path = template.format(**values)
        except Exception:
            rel_path = None
    # Movie-specific fallback if formatting failed or title missing
    if is_movie and not rel_path:
        try:
            m_fb = CoreSettings.get_dvr_movie_fallback_template()
            rel_path = m_fb.format(**values)
        except Exception:
            rel_path = f"Movies/{values['start']}.mkv"
    # As a last resort for TV
    if not is_movie and not rel_path:
        rel_path = f"TV_Shows/{show}/S{season:02d}E{episode:02d}.mkv"
    # Keep any leading folder like 'Recordings/' from the template so users can
    # structure their library under /data as desired.
    if not rel_path.lower().endswith('.mkv'):
        rel_path = f"{rel_path}.mkv"

    # Normalize path (strip ./)
    if rel_path.startswith('./'):
        rel_path = rel_path[2:]
    final_path = rel_path if rel_path.startswith('/') else os.path.join(library_root, rel_path)
    final_path = os.path.normpath(final_path)

    # Avoid overwriting an existing MKV from a different recording.  The HLS
    # working directory is keyed by recording id, so it cannot collide and is
    # not part of this check.
    base, ext = os.path.splitext(final_path)
    counter = 1
    while True:
        try:
            mkv_occupied = os.stat(final_path).st_size > 0
        except OSError:
            mkv_occupied = False
        if not mkv_occupied:
            break
        counter += 1
        final_path = f"{base}_{counter}{ext}"

    # Ensure directory exists
    os.makedirs(os.path.dirname(final_path), exist_ok=True)

    # HLS working directory: hidden and id-tagged alongside the final MKV.
    hls_dir = os.path.join(os.path.dirname(final_path), f".dvr_{recording_id}_hls")
    return final_path, hls_dir, os.path.basename(final_path)


def get_dvr_stream_base_url():
    """Return the single correct base URL for DVR to reach the TS stream proxy.

    Priority:
    1. DISPATCHARR_INTERNAL_TS_BASE_URL — explicit override, always wins.
    2. Modular mode (DISPATCHARR_ENV=modular) — celery runs in a separate container
       and must reach the web container by its Docker service name on DISPATCHARR_PORT.
       Override the host with DISPATCHARR_WEB_HOST for non-standard compose setups.
    3. AIO / dev / debug — celery shares the container with uwsgi which binds on
       port 5656; use 127.0.0.1 to avoid any nginx layer.
    """
    explicit = os.environ.get('DISPATCHARR_INTERNAL_TS_BASE_URL')
    if explicit:
        return explicit.rstrip('/')

    dispatcharr_env = os.environ.get('DISPATCHARR_ENV', 'aio').lower()

    if dispatcharr_env == 'modular':
        host = os.environ.get('DISPATCHARR_WEB_HOST', 'web')
        port = os.environ.get('DISPATCHARR_PORT', '9191')
        return f'http://{host}:{port}'

    # AIO, dev, debug: celery and uwsgi share the container, reach uwsgi directly
    return 'http://127.0.0.1:5656'


@shared_task
def run_recording(recording_id, channel_id, start_time_str, end_time_str):
    """
    Execute a scheduled recording for the given channel/recording.

    Enhancements:
    - Accepts recording_id so we can persist metadata back to the Recording row
    - Persists basic file info (name/path) to Recording.custom_properties
    - Attempts to capture stream stats from TS proxy (codec, resolution, fps, etc.)
    - Attempts to capture a poster (via program.custom_properties) and store a Logo reference
    """
    from .models import Recording, Logo

    # --- Idempotency guard (prevents duplicate recordings from task redelivery) ---
    # Fail closed: if the DB is unreachable, abort rather than risk a duplicate
    # task overwriting a valid recording.
    try:
        rec_check = Recording.objects.filter(id=recording_id).only("custom_properties").first()
        if not rec_check:
            logger.info(
                f"run_recording called for recording {recording_id} but it no longer exists — skipping."
            )
            return
        status = (rec_check.custom_properties or {}).get("status", "")
        if status in ("recording", "completed", "stopped"):
            logger.warning(
                f"run_recording called for recording {recording_id} but status "
                f"is already '{status}' - skipping duplicate execution."
            )
            return
    except Exception as e:
        logger.error(
            f"Idempotency guard DB check failed for recording {recording_id} "
            f"({type(e).__name__}: {e}) — aborting to prevent potential duplicate."
        )
        return

    # --- Per-recording active lock (prevents concurrent ffmpeg processes
    # writing to the same HLS dir when duplicate recovery dispatches occur). ---
    # The lock is refreshed inside the main loop below; if a worker crashes
    # the TTL expires within ~45s and recovery can re-acquire it.
    _active_lock_key = f"dvr:recording-active:{recording_id}"
    _active_lock_ttl = 45
    _active_lock_redis = None
    try:
        from core.utils import RedisClient
        _active_lock_redis = RedisClient.get_client()
    except Exception:
        _active_lock_redis = None

    if _active_lock_redis is not None:
        try:
            if not _active_lock_redis.set(
                _active_lock_key, "1", ex=_active_lock_ttl, nx=True
            ):
                logger.warning(
                    f"run_recording {recording_id}: another worker holds the "
                    f"active-recording lock, aborting duplicate dispatch."
                )
                return
        except Exception as _le:
            logger.warning(
                f"run_recording {recording_id}: active-lock acquisition error "
                f"({type(_le).__name__}: {_le}), proceeding without lock."
            )
            _active_lock_redis = None

    # --- Clean up the one-off PeriodicTask that dispatched this task ---
    try:
        from apps.channels.signals import revoke_task, _dvr_task_name
        revoke_task(_dvr_task_name(recording_id))
    except Exception as e:
        logger.debug(f"PeriodicTask cleanup failed (non-fatal): {e}")

    channel = Channel.objects.get(id=channel_id)

    start_time = datetime.fromisoformat(start_time_str)
    end_time = datetime.fromisoformat(end_time_str)

    # Build output paths from templates (refined after loading Recording cp below)
    filename = None
    final_path = None
    hls_dir = None

    channel_layer = get_channel_layer()

    async_to_sync(channel_layer.group_send)(
        "updates",
        {
            "type": "update",
            "data": {"success": True, "type": "recording_started", "channel": channel.name}
        },
    )

    logger.info(f"Starting recording for channel {channel.name}")

    # Log system event for recording start
    try:
        from core.utils import log_system_event
        log_system_event(
            'recording_start',
            channel_id=channel.uuid,
            channel_name=channel.name,
            recording_id=recording_id
        )
    except Exception as e:
        logger.error(f"Could not log recording start event: {e}")

    # Try to resolve the Recording row up front
    recording_obj = None
    try:
        recording_obj = Recording.objects.get(id=recording_id)
        # If the stop endpoint already wrote "stopped" before the task started,
        # honor it instead of overwriting with "recording".
        _pre_cp = recording_obj.custom_properties or {}
        if _pre_cp.get("status") == "stopped":
            logger.info(
                f"run_recording {recording_id}: 'stopped' found in DB before stream started "
                f"— task exits without connecting."
            )
            return

        # Prime custom_properties with file info/status
        cp = recording_obj.custom_properties or {}
        cp.update({
            "status": "recording",
            "started_at": str(datetime.now()),
        })
        # Provide a predictable playback URL for the frontend
        cp["file_url"] = f"/api/channels/recordings/{recording_id}/hls/index.m3u8"
        cp["output_file_url"] = cp["file_url"]

        # Determine program info (may include id for deeper details)
        program = cp.get("program") or {}

        # Enrich empty program dicts (manual recordings) from EPG time-slot data.
        if isinstance(program, dict) and not program.get("user_edited") and not program.get("id") and not program.get("title"):
            epg_match = _match_epg_program_by_timeslot(
                channel.epg_data, recording_obj.start_time, recording_obj.end_time,
            )
            if epg_match:
                program.update(epg_match)
                cp["program"] = program

        # Resume into the existing working set if this task is a recovery run.
        # `recover_recordings_on_startup` re-dispatches `run_recording` for an
        # interrupted recording with `_hls_dir` and `file_path` already set in
        # custom_properties.  Re-running `_build_output_paths` would allocate a
        # fresh `Foo_2.mkv` / `.dvr_{id}_hls` pair and orphan the original
        # segments, producing two MKVs at the end.  Reuse the original paths
        # whenever the HLS directory is still on disk.
        existing_hls = cp.get("_hls_dir")
        existing_file = cp.get("file_path")
        if (
            existing_hls
            and existing_file
            and os.path.isdir(existing_hls)
        ):
            final_path = existing_file
            hls_dir = existing_hls
            filename = os.path.basename(final_path)
            try:
                _seg_count = sum(
                    1 for f in os.listdir(hls_dir) if f.endswith(".ts")
                )
            except OSError:
                _seg_count = 0
            logger.info(
                f"run_recording {recording_id}: resuming into existing HLS dir "
                f"{hls_dir} ({_seg_count} segment(s)), final={final_path}"
            )
        else:
            final_path, hls_dir, filename = _build_output_paths(
                channel, program, start_time, end_time, recording_id
            )
        cp["file_name"] = filename
        cp["file_path"] = final_path
        cp["_hls_dir"] = hls_dir

        # Resolve poster art via the shared pipeline (EPG → VOD → TMDB/OMDb →
        # TVMaze/iTunes → direct program fields → Logo table → channel logo).
        poster_logo_id, poster_url = _resolve_poster_for_program(
            channel.name, program, channel_logo_id=channel.logo_id,
        )
        if poster_logo_id:
            cp["poster_logo_id"] = poster_logo_id
        if poster_url and "poster_url" not in cp:
            cp["poster_url"] = poster_url

        # Ensure destination directory exists.  No placeholder MKV is created
        # here on purpose: with the HLS pipeline, the final MKV is materialized
        # by the post-recording concat step, and `/file/` redirects to `/hls/`
        # while no MKV exists, so a 0-byte placeholder serves no functional
        # purpose.  Writing one previously caused an orphan-file bug on server
        # restart in cases where the resume path could not adopt the original
        # final_path, since the second run would generate a fresh placeholder
        # at a new path and leave the original behind.
        try:
            os.makedirs(os.path.dirname(final_path), exist_ok=True)
        except Exception:
            pass

        # Re-read from DB to preserve concurrent changes (e.g., artwork
        # prefetch may have saved poster/rating info while resolving).
        recording_obj.refresh_from_db()
        fresh_cp = recording_obj.custom_properties or {}

        # If the stop endpoint set "stopped" while resolving, honor it.
        if fresh_cp.get("status") == "stopped":
            logger.info(
                f"run_recording {recording_id}: 'stopped' found after metadata "
                f"prep — task exits without streaming."
            )
            return

        # Merge only the keys explicitly set into the fresh copy
        for key in ("status", "started_at", "file_url", "output_file_url",
                     "file_name", "file_path", "_hls_dir",
                     "program", "poster_logo_id", "poster_url"):
            if key in cp:
                fresh_cp[key] = cp[key]
        recording_obj.custom_properties = fresh_cp
        recording_obj.save(update_fields=["custom_properties"])

        # Notify frontends so the tile picks up poster/metadata immediately
        try:
            from core.utils import send_websocket_update
            send_websocket_update('updates', 'update', {
                "success": True,
                "type": "recording_updated",
                "recording_id": recording_id,
            })
        except Exception:
            pass
    except Exception as e:
        logger.debug(f"Unable to prime Recording metadata: {e}")
    interrupted = False
    interrupted_reason = None
    bytes_written = 0

    base_url = get_dvr_stream_base_url()
    logger.info(f"DVR recording {recording_id}: using stream base URL {base_url}")

    # --- DB retry constants ---
    _dvr_db_max_retries = 3
    _dvr_db_retry_interval = 1  # seconds (base for exponential backoff)

    # Redis key used to signal that an HLS client is actively fetching segments.
    # The hls endpoint refreshes this on every .ts request; we check it before
    # deleting the HLS directory so in-flight downloads are not cut short.
    _hls_viewer_key = f"dvr:hls_viewer:{recording_id}"

    # --- HLS recording pipeline ---
    # Launch FFmpeg with the deterministic base URL resolved above.  Wait up to
    # _first_segment_timeout seconds for the first segment to appear before
    # treating the stream as unavailable.
    from django.utils import timezone as _tz

    last_error = None
    ffmpeg_proc = None
    hls_m3u8 = None
    hls_seg_pattern = None
    _stream_confirmed = False

    if hls_dir:
        os.makedirs(hls_dir, exist_ok=True)
        hls_m3u8 = os.path.join(hls_dir, "index.m3u8")
        hls_seg_pattern = os.path.join(hls_dir, "seg_%05d.ts")

    base = base_url

    if not interrupted:
        # Check for stop/delete before starting
        try:
            _pre = Recording.objects.filter(id=recording_id).only("custom_properties").first()
            if _pre is None:
                interrupted = True
                interrupted_reason = "recording_deleted"
            elif (_pre.custom_properties or {}).get("status") == "stopped":
                interrupted = False
                interrupted_reason = "stopped_by_user"
        except Exception:
            pass

    if not interrupted:
        stream_url = f"{base}/proxy/ts/stream/{channel.uuid}"
        logger.info(f"DVR recording {recording_id}: stream URL: {stream_url}")

    if not interrupted:

        # Continue segment numbering from any previous session (server-restart resume)
        existing_segs = sorted(
            f for f in os.listdir(hls_dir) if f.startswith("seg_") and f.endswith(".ts")
        ) if hls_dir else []
        hls_start_number = len(existing_segs)

        ffmpeg_cmd = [
            "ffmpeg", "-y",
            "-reconnect", "1",
            "-reconnect_streamed", "1",
            "-reconnect_delay_max", "5",
            "-user_agent", f"Dispatcharr-DVR/recording-{recording_id}",
            # Regenerate monotonic PTS to handle erratic/discontinuous timestamps
            # from IPTV sources.
            "-fflags", "+genpts",
            "-i", stream_url,
            "-c", "copy",
            # Shift output timestamps so they start from 0, fixing negative PTS
            # values that can prevent segment boundary detection in the HLS muxer.
            "-avoid_negative_ts", "make_zero",
            "-f", "hls",
            "-hls_time", "4",
            "-hls_list_size", "0",
            "-hls_flags", "append_list+omit_endlist+independent_segments",
            "-start_number", str(hls_start_number),
            "-hls_segment_filename", hls_seg_pattern,
            hls_m3u8,
        ]

        logger.info(f"DVR recording {recording_id}: starting FFmpeg — stream URL: {stream_url}")
        logger.info(f"DVR recording {recording_id}: HLS output dir: {hls_dir}")
        logger.debug(f"DVR recording {recording_id}: FFmpeg command: {' '.join(str(a) for a in ffmpeg_cmd)}")
        # Rolling tail of FFmpeg stderr lines for post-mortem diagnostics
        _ffmpeg_stderr_tail = deque(maxlen=200)
        try:
            ffmpeg_proc = subprocess.Popen(
                ffmpeg_cmd,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
            )
        except Exception as _fe:
            last_error = str(_fe)
            logger.warning(f"DVR recording {recording_id}: failed to launch FFmpeg: {_fe}")
            ffmpeg_proc = None

        # Drain FFmpeg stderr in a background thread to prevent the OS pipe
        # buffer (~64 KB on Linux) from filling up, which would block FFmpeg's
        # writes and cause it to silently stop demuxing/segmenting after a few
        # minutes of normal progress output. Lines are logged to the Dispatcharr
        # log and tail-buffered for diagnostics.
        def _drain_ffmpeg_stderr(proc, rec_id, tail):
            # FFmpeg emits progress lines terminated with \r (carriage return)
            # so it can rewrite them in place; only non-progress messages end
            # with \n. Split on either to capture both kinds.
            try:
                buf = bytearray()
                stream = proc.stderr
                while True:
                    byte = stream.read(1)
                    if not byte:
                        break
                    if byte in (b'\r', b'\n'):
                        if buf:
                            line = buf.decode('utf-8', errors='replace').strip()
                            buf.clear()
                            if line:
                                tail.append(line)
                                low = line.lower()
                                if 'error' in low or 'failed' in low or 'invalid' in low:
                                    logger.warning(f"DVR recording {rec_id} ffmpeg: {line}")
                                else:
                                    logger.debug(f"DVR recording {rec_id} ffmpeg: {line}")
                        continue
                    buf.append(byte[0])
                    # Safety net: flush absurdly long lines so a malformed
                    # stream can't grow the buffer without bound.
                    if len(buf) > 4096:
                        line = buf.decode('utf-8', errors='replace').strip()
                        buf.clear()
                        if line:
                            tail.append(line)
                            logger.debug(f"DVR recording {rec_id} ffmpeg: {line}")
                # Flush any trailing content
                if buf:
                    line = buf.decode('utf-8', errors='replace').strip()
                    if line:
                        tail.append(line)
                        logger.debug(f"DVR recording {rec_id} ffmpeg: {line}")
            except Exception as _de:
                logger.debug(f"DVR recording {rec_id}: stderr drain ended: {_de}")

        if ffmpeg_proc is not None and ffmpeg_proc.stderr is not None:
            _stderr_thread = threading.Thread(
                target=_drain_ffmpeg_stderr,
                args=(ffmpeg_proc, recording_id, _ffmpeg_stderr_tail),
                daemon=True,
                name=f"dvr-ffmpeg-stderr-{recording_id}",
            )
            _stderr_thread.start()

        end_timestamp = end_time.timestamp()
        _stop_poll_interval = 2.0
        _last_stop_poll = time.time()
        _ffmpeg_start = time.time()
        _first_segment_timeout = 15.0
        _stall_timeout = 60.0  # seconds without new segments → source stream gone
        _stream_confirmed = False
        _last_seg_count = hls_start_number
        _last_new_seg_time = time.time()
        _last_active_lock_refresh = 0.0
        _active_lock_refresh_interval = 15.0

        while ffmpeg_proc.poll() is None:
            time.sleep(0.5)
            now = time.time()

            # Refresh the per-recording active lock so a concurrent worker
            # cannot acquire it and start a duplicate ffmpeg.
            if (
                _active_lock_redis is not None
                and now - _last_active_lock_refresh >= _active_lock_refresh_interval
            ):
                try:
                    _active_lock_redis.expire(_active_lock_key, _active_lock_ttl)
                except Exception:
                    pass
                _last_active_lock_refresh = now

            segs_now = [
                f for f in os.listdir(hls_dir)
                if f.startswith("seg_") and f.endswith(".ts")
            ] if hls_dir else []

            # Wait for the first segment to confirm data is flowing
            if not _stream_confirmed:
                if segs_now:
                    _stream_confirmed = True
                    _last_seg_count = len(segs_now)
                    _last_new_seg_time = now
                    logger.info(
                        f"DVR recording {recording_id}: first HLS segment written, stream confirmed"
                    )
                elif now - _ffmpeg_start > _first_segment_timeout:
                    logger.warning(
                        f"DVR recording {recording_id}: no HLS segments produced after "
                        f"{_first_segment_timeout}s from {base} — stream unavailable"
                    )
                    ffmpeg_proc.kill()
                    try:
                        ffmpeg_proc.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        pass
                    last_error = f"no_segments_in_{_first_segment_timeout}s_from_{base}"
                    ffmpeg_proc = None
                    break
            else:
                # Stream was confirmed, detect source stalls (e.g. proxy ghost-kills the client)
                if len(segs_now) > _last_seg_count:
                    _last_seg_count = len(segs_now)
                    _last_new_seg_time = now
                else:
                    # Also treat a recently-modified segment file as activity.
                    # With erratic source timestamps, FFmpeg may buffer data inside
                    # a partially-written segment for longer than hls_time, so the
                    # segment count won't increase even though data is flowing.
                    # Only stat the most recent segment (highest filename) to keep
                    # this O(1) per tick instead of O(N) over a long recording.
                    try:
                        if segs_now and hls_dir:
                            _newest = max(segs_now)
                            _newest_mtime = os.path.getmtime(
                                os.path.join(hls_dir, _newest)
                            )
                            if _newest_mtime > _last_new_seg_time:
                                _last_new_seg_time = _newest_mtime
                    except Exception:
                        pass
                if now - _last_new_seg_time > _stall_timeout:
                    logger.warning(
                        f"DVR recording {recording_id}: no new HLS segments for "
                        f"{_stall_timeout:.0f}s — source stream stalled, stopping FFmpeg"
                    )
                    ffmpeg_proc.send_signal(signal.SIGINT)
                    try:
                        ffmpeg_proc.wait(timeout=10)
                    except subprocess.TimeoutExpired:
                        ffmpeg_proc.kill()
                    break

            # Duration check — SIGINT lets FFmpeg write #EXT-X-ENDLIST cleanly
            if now >= end_timestamp:
                logger.info(
                    f"DVR recording {recording_id}: scheduled end time reached, stopping FFmpeg"
                )
                ffmpeg_proc.send_signal(signal.SIGINT)
                try:
                    ffmpeg_proc.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    ffmpeg_proc.kill()
                break

            # Periodic DB poll: stop, delete, end_time extension
            if now - _last_stop_poll >= _stop_poll_interval:
                _last_stop_poll = now
                try:
                    _sc = Recording.objects.filter(
                        id=recording_id
                    ).only("custom_properties", "end_time").first()
                    if _sc is None:
                        logger.info(
                            f"DVR recording {recording_id}: deleted — stopping FFmpeg"
                        )
                        ffmpeg_proc.send_signal(signal.SIGINT)
                        try:
                            ffmpeg_proc.wait(timeout=10)
                        except subprocess.TimeoutExpired:
                            ffmpeg_proc.kill()
                        interrupted = False
                        break
                    if (_sc.custom_properties or {}).get("status") == "stopped":
                        logger.info(
                            f"DVR recording {recording_id}: stop requested — stopping FFmpeg"
                        )
                        ffmpeg_proc.send_signal(signal.SIGINT)
                        try:
                            ffmpeg_proc.wait(timeout=10)
                        except subprocess.TimeoutExpired:
                            ffmpeg_proc.kill()
                        break
                    # Handle end_time extension — just update the deadline, no restart needed
                    try:
                        new_end = _sc.end_time
                        if new_end is not None:
                            if _tz.is_naive(new_end):
                                new_end = _tz.make_aware(new_end)
                            new_ts = new_end.timestamp()
                            if new_ts > end_timestamp:
                                logger.info(
                                    f"DVR recording {recording_id}: end_time extended to {new_end}"
                                )
                                end_timestamp = new_ts
                    except Exception:
                        pass
                except Exception:
                    pass

        # If FFmpeg exited after confirming the stream, log the exit code so we know why it stopped.
        if _stream_confirmed and ffmpeg_proc is not None:
            _exit_code = ffmpeg_proc.poll()
            if _exit_code is not None and _exit_code != 0:
                logger.warning(
                    f"DVR recording {recording_id}: FFmpeg exited unexpectedly (rc={_exit_code}) "
                    f"after stream was confirmed — source stream likely disconnected"
                )

        # If FFmpeg exited without the stream being confirmed, log the tail of
        # captured stderr for diagnosis. Lines were already logged live by the
        # drain thread; this surfaces the recent context in a single message.
        elif not _stream_confirmed and ffmpeg_proc is not None:
            try:
                _exit_code = ffmpeg_proc.poll()
                last_error = f"rc={_exit_code} from {base}"
                _tail_text = "\n".join(_ffmpeg_stderr_tail)
                if _tail_text:
                    logger.warning(
                        f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) "
                        f"for {base} without producing segments.\nFFmpeg stderr tail:\n{_tail_text[-1000:]}"
                    )
                else:
                    logger.warning(
                        f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) "
                        f"for {base} without producing segments (no stderr output)"
                    )
            except Exception:
                pass

    # Ensure FFmpeg has fully exited (covers cases where the loop broke early)
    if ffmpeg_proc is not None and ffmpeg_proc.poll() is None:
        try:
            ffmpeg_proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            ffmpeg_proc.kill()

    # If the loop broke because the Celery worker is shutting down (e.g.
    # docker stop, container update) and the recording window is still open,
    # leave the HLS working directory intact and mark the recording as
    # interrupted.  `recover_recordings_on_startup` will re-dispatch this task
    # on the next boot, and the resume-path logic in the prep block will
    # adopt the existing `_hls_dir` and `file_path` so FFmpeg appends to the
    # same playlist and the eventual concat produces a single MKV.
    #
    # Note: `_DVR_SHUTTING_DOWN` is per-process. The DVR queue currently uses
    # a threads pool worker (one process, all threads see the flag). If the
    # DVR queue is ever switched back to prefork, the parent's flag will not
    # propagate to in-flight child workers.
    if _DVR_SHUTTING_DOWN and time.time() < end_timestamp:
        try:
            _shutdown_rec = Recording.objects.filter(id=recording_id).first()
            if _shutdown_rec:
                _shutdown_cp = _shutdown_rec.custom_properties or {}
                # Preserve _hls_dir / file_path / file_url exactly as they are
                # so the HLS endpoint keeps serving until the worker dies and
                # so recovery can find the working directory on next boot.
                _shutdown_cp["status"] = "interrupted"
                _shutdown_cp["interrupted_reason"] = "server_shutdown"
                _shutdown_cp["ended_at"] = str(datetime.now())
                _shutdown_rec.custom_properties = _shutdown_cp
                _shutdown_rec.save(update_fields=["custom_properties"])
        except Exception as _se:
            logger.warning(
                f"DVR recording {recording_id}: failed to flag interrupted on shutdown: {_se}"
            )
        logger.info(
            f"DVR recording {recording_id}: worker shutting down with "
            f"{int(end_timestamp - time.time())}s of recording window remaining; "
            f"leaving HLS dir intact for resume on next boot, skipping concat."
        )
        return

    if not _stream_confirmed and not interrupted:
        interrupted = True
        interrupted_reason = f"no_stream_data: {last_error or 'ffmpeg_failed'}"

    # Measure bytes from HLS segment files
    try:
        bytes_written = sum(
            os.path.getsize(os.path.join(hls_dir, f))
            for f in os.listdir(hls_dir)
            if f.startswith("seg_") and f.endswith(".ts")
        ) if hls_dir and os.path.isdir(hls_dir) else 0
    except Exception:
        bytes_written = 0
    if bytes_written == 0 and not interrupted:
        _deliberately_stopped = False
        try:
            _rc = Recording.objects.filter(id=recording_id).only("custom_properties").first()
            if _rc and (_rc.custom_properties or {}).get("status") == "stopped":
                _deliberately_stopped = True
        except Exception:
            pass

        if not _deliberately_stopped:
            interrupted = True
            interrupted_reason = f"no_stream_data: {last_error or 'unknown'}"

            # Update DB status immediately so the UI reflects the change on the event below
            try:
                if recording_obj is None:
                    recording_obj = Recording.objects.get(id=recording_id)
                cp_now = recording_obj.custom_properties or {}
                cp_now.update({
                    "status": "interrupted",
                    "ended_at": str(datetime.now()),
                    "file_name": filename or cp_now.get("file_name"),
                    "file_path": final_path or cp_now.get("file_path"),
                    "interrupted_reason": interrupted_reason,
                })
                recording_obj.custom_properties = cp_now
                recording_obj.save(update_fields=["custom_properties"])
            except Exception as e:
                logger.debug(f"Failed to update immediate recording status: {e}")

            async_to_sync(channel_layer.group_send)(
                "updates",
                {
                    "type": "update",
                    "data": {"success": True, "type": "recording_ended", "channel": channel.name}
                },
            )
            # After the loop, the file and response are closed automatically.
            logger.info(f"Finished recording for channel {channel.name}")

    # Log system event for recording end
    try:
        from core.utils import log_system_event
        log_system_event(
            'recording_end',
            channel_id=channel.uuid,
            channel_name=channel.name,
            recording_id=recording_id,
            interrupted=interrupted,
            bytes_written=bytes_written
        )
    except Exception as e:
        logger.error(f"Could not log recording end event: {e}")

    # If the Recording was deleted (cancelled by user), skip post-processing
    recording_cancelled = not Recording.objects.filter(id=recording_id).exists()
    if recording_cancelled:
        logger.info(f"Recording {recording_id} was cancelled — skipping remux and metadata.")
        # Clean up all artifacts for the cancelled recording,
        # including any pre-restart .ts segments from server recovery.
        # Use the in-memory recording_obj since the DB row is already deleted.
        if hls_dir and os.path.isdir(hls_dir):
            try:
                shutil.rmtree(hls_dir)
                logger.info(f"Cleaned up cancelled recording HLS directory: {hls_dir}")
            except Exception as _e:
                logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}")
        if final_path:
            try:
                if os.path.exists(final_path):
                    os.remove(final_path)
                    logger.info(f"Cleaned up cancelled recording MKV placeholder: {final_path}")
            except Exception:
                pass
        return

    # --- Post-processing: concat HLS segments → final MKV ---
    remux_success = False
    hls_m3u8 = os.path.join(hls_dir, "index.m3u8") if hls_dir else None

    def _get_hls_segments(m3u8_path, seg_dir):
        """Return ordered segment paths from an HLS m3u8 playlist."""
        segs = []
        try:
            with open(m3u8_path) as _f:
                for _line in _f:
                    _line = _line.strip()
                    if _line and not _line.startswith('#'):
                        sp = os.path.join(seg_dir, _line) if not os.path.isabs(_line) else _line
                        if os.path.exists(sp):
                            segs.append(sp)
        except Exception as _e:
            logger.warning(f"DVR recording {recording_id}: failed to parse m3u8: {_e}")
        return segs

    segments = (
        _get_hls_segments(hls_m3u8, hls_dir)
        if (hls_m3u8 and os.path.exists(hls_m3u8))
        else []
    )
    if not segments and hls_dir and os.path.isdir(hls_dir):
        # Fallback: sort all segment files by name if m3u8 is missing or empty
        try:
            segments = sorted(
                os.path.join(hls_dir, f)
                for f in os.listdir(hls_dir)
                if f.startswith("seg_") and f.endswith(".ts")
            )
        except Exception:
            segments = []

    if segments:
        concat_list_path = os.path.join(hls_dir, "concat.txt")
        try:
            with open(concat_list_path, "w") as _cl:
                for seg in segments:
                    _escaped = seg.replace("'", "'\\''")
                    _cl.write(f"file '{_escaped}'\n")

            concat_result = subprocess.run(
                [
                    "ffmpeg", "-y",
                    "-f", "concat", "-safe", "0",
                    "-i", concat_list_path,
                    "-c", "copy",
                    final_path,
                ],
                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
            )
            _ok = (
                concat_result.returncode == 0
                and os.path.exists(final_path)
                and os.path.getsize(final_path) > 0
            )
            _fallback_used = False
            # MP4-intermediate fallback: some MPEG-TS streams (parameter set
            # changes mid-stream, weird PMT updates, audio discontinuities)
            # fail to mux directly into Matroska but go through an MP4
            # container cleanly, which can then be remuxed losslessly to
            # MKV.  Try this before declaring the recording lost.
            if not _ok:
                logger.warning(
                    f"DVR recording {recording_id}: direct HLS\u2192MKV concat failed "
                    f"(rc={concat_result.returncode}); attempting MP4-intermediate "
                    f"fallback. stderr: {(concat_result.stderr or '')[:300]}"
                )
                try:
                    if os.path.exists(final_path) and os.path.getsize(final_path) == 0:
                        os.remove(final_path)
                except OSError:
                    pass
                _intermediate_mp4 = os.path.join(
                    hls_dir, f".dvr_{recording_id}_intermediate.mp4"
                )
                try:
                    if os.path.exists(_intermediate_mp4):
                        os.remove(_intermediate_mp4)
                except OSError:
                    pass
                _mp4_concat = subprocess.run(
                    [
                        "ffmpeg", "-y",
                        "-f", "concat", "-safe", "0",
                        "-i", concat_list_path,
                        "-c", "copy",
                        "-bsf:a", "aac_adtstoasc",
                        _intermediate_mp4,
                    ],
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                )
                if (
                    _mp4_concat.returncode == 0
                    and os.path.exists(_intermediate_mp4)
                    and os.path.getsize(_intermediate_mp4) > 0
                ):
                    _mp4_to_mkv = subprocess.run(
                        [
                            "ffmpeg", "-y",
                            "-i", _intermediate_mp4,
                            "-c", "copy",
                            final_path,
                        ],
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                    )
                    if (
                        _mp4_to_mkv.returncode == 0
                        and os.path.exists(final_path)
                        and os.path.getsize(final_path) > 0
                    ):
                        _ok = True
                        _fallback_used = True
                    else:
                        logger.error(
                            f"DVR recording {recording_id}: MP4\u2192MKV remux step "
                            f"failed (rc={_mp4_to_mkv.returncode}). stderr: "
                            f"{(_mp4_to_mkv.stderr or '')[:300]}"
                        )
                else:
                    logger.error(
                        f"DVR recording {recording_id}: HLS\u2192MP4 concat fallback "
                        f"failed (rc={_mp4_concat.returncode}). stderr: "
                        f"{(_mp4_concat.stderr or '')[:300]}"
                    )
                try:
                    if os.path.exists(_intermediate_mp4):
                        os.remove(_intermediate_mp4)
                except OSError:
                    pass

            if _ok:
                remux_success = True
                if _fallback_used:
                    logger.info(
                        f"DVR recording {recording_id}: HLS\u2192MP4\u2192MKV fallback "
                        f"concat succeeded \u2014 {len(segments)} segments \u2192 "
                        f"{os.path.basename(final_path)} "
                        f"({os.path.getsize(final_path):,} bytes)"
                    )
                else:
                    logger.info(
                        f"DVR recording {recording_id}: HLS\u2192MKV concat succeeded \u2014 "
                        f"{len(segments)} segments \u2192 {os.path.basename(final_path)} "
                        f"({os.path.getsize(final_path):,} bytes)"
                    )
                # Update DB so *new* client requests go to /file/ (the final MKV)
                # rather than the soon-to-be-removed /hls/ endpoint.  Important:
                # we do NOT clear _hls_dir here, active viewers still in the
                # 20s heartbeat window need it to keep fetching .ts segments
                # during the viewer-wait grace period below.  _hls_dir is
                # cleared only after the directory is actually removed.
                try:
                    _pre_rmtree_rec = Recording.objects.filter(id=recording_id).first()
                    if _pre_rmtree_rec:
                        _pre_rmtree_cp = _pre_rmtree_rec.custom_properties or {}
                        _pre_rmtree_cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/"
                        _pre_rmtree_cp["output_file_url"] = _pre_rmtree_cp["file_url"]
                        _pre_rmtree_rec.custom_properties = _pre_rmtree_cp
                        _pre_rmtree_rec.save(update_fields=["custom_properties"])
                except Exception as _pre_e:
                    logger.warning(f"DVR recording {recording_id}: pre-rmtree DB update failed: {_pre_e}")
                # Wait for active HLS viewers to finish downloading segments
                # before deleting the directory.  The hls endpoint refreshes
                # _hls_viewer_key on every .ts request with a 20s TTL, so the
                # key expiring naturally means no segment was fetched in the
                # last 20 seconds (client has stopped).  We loop until the key
                # is gone with a 4-hour hard safety cap in case Redis gets stuck.
                _viewer_wait_start = time.time()
                _safety_timeout = 14400  # 4 hours absolute maximum
                try:
                    from core.utils import RedisClient as _RC
                    _rv = _RC.get_client(max_retries=1, retry_interval=0)
                    if _rv and _rv.exists(_hls_viewer_key):
                        logger.info(
                            f"DVR recording {recording_id}: active HLS viewer detected, "
                            f"deferring HLS directory cleanup until client disconnects"
                        )
                        while _rv.exists(_hls_viewer_key):
                            if time.time() - _viewer_wait_start > _safety_timeout:
                                logger.warning(
                                    f"DVR recording {recording_id}: viewer wait safety timeout "
                                    f"({_safety_timeout}s) reached, proceeding with HLS directory cleanup"
                                )
                                break
                            time.sleep(2)
                        logger.info(
                            f"DVR recording {recording_id}: viewer wait complete after "
                            f"{time.time() - _viewer_wait_start:.1f}s"
                        )
                except Exception as _ve:
                    logger.debug(f"DVR recording {recording_id}: viewer wait check failed: {_ve}")
                try:
                    shutil.rmtree(hls_dir)
                    logger.debug(f"Cleaned up HLS directory: {hls_dir}")
                except Exception as _e:
                    logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}")
                # Now that the HLS dir is gone, clear _hls_dir from custom_properties
                # so the hls endpoint will redirect (.m3u8) or 404 (.ts) cleanly.
                try:
                    _post_rmtree_rec = Recording.objects.filter(id=recording_id).first()
                    if _post_rmtree_rec:
                        _post_rmtree_cp = _post_rmtree_rec.custom_properties or {}
                        if "_hls_dir" in _post_rmtree_cp:
                            _post_rmtree_cp.pop("_hls_dir", None)
                            _post_rmtree_rec.custom_properties = _post_rmtree_cp
                            _post_rmtree_rec.save(update_fields=["custom_properties"])
                except Exception as _post_e:
                    logger.warning(f"DVR recording {recording_id}: post-rmtree DB update failed: {_post_e}")
            else:
                logger.error(
                    f"DVR recording {recording_id}: all HLS\u2192MKV concat attempts "
                    f"failed (direct rc={concat_result.returncode}, MP4 fallback also "
                    f"failed). Keeping HLS segments for recovery. stderr: "
                    f"{(concat_result.stderr or '')[:500]}"
                )
        except Exception as _ce:
            logger.error(f"DVR recording {recording_id}: concat exception: {_ce}")
        finally:
            try:
                os.remove(concat_list_path)
            except OSError:
                pass
    else:
        logger.warning(f"DVR recording {recording_id}: no HLS segments found. Nothing to concat")

    # Persist final metadata to Recording (status, ended_at, and stream stats if available)
    try:
        if recording_obj is None:
            recording_obj = Recording.objects.get(id=recording_id)

        # Re-read from DB to get the latest status (stop endpoint may have set it)
        recording_obj.refresh_from_db()
        cp = recording_obj.custom_properties or {}
        cp["ended_at"] = str(datetime.now())

        # Restore file_url to the permanent MKV endpoint now that recording has ended.
        # During recording it pointed to /hls/index.m3u8; clients should use /file/ hereafter.
        cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/"
        cp["output_file_url"] = cp["file_url"]

        # Final status priority: stopped > completed > interrupted.
        # "stopped" is set by the stop endpoint before stream teardown, so
        # refresh_from_db() above guarantees it is visible here.
        db_status_now = cp.get("status", "")
        if db_status_now == "stopped":
            # Deliberate user stop — preserve; do not overwrite with "completed".
            cp.pop("interrupted_reason", None)
        elif not interrupted:
            cp["status"] = "completed"
            cp.pop("interrupted_reason", None)
        else:
            cp["status"] = "interrupted"
            if interrupted_reason:
                cp["interrupted_reason"] = interrupted_reason
        cp["bytes_written"] = bytes_written
        cp["remux_success"] = remux_success

        # Try to get stream stats from TS proxy Redis metadata
        try:
            from core.utils import RedisClient
            from apps.proxy.ts_proxy.redis_keys import RedisKeys
            from apps.proxy.ts_proxy.constants import ChannelMetadataField

            r = RedisClient.get_client()
            if r is not None:
                metadata_key = RedisKeys.channel_metadata(str(channel.uuid))
                md = r.hgetall(metadata_key)
                if md:
                    def _d(bkey, cast=str):
                        v = md.get(bkey)
                        try:
                            if v is None:
                                return None
                            s = v
                            return cast(s) if cast is not str else s
                        except Exception:
                            return None

                    stream_info = {}
                    # Video fields
                    for key, caster in [
                        (ChannelMetadataField.VIDEO_CODEC, str),
                        (ChannelMetadataField.RESOLUTION, str),
                        (ChannelMetadataField.WIDTH, float),
                        (ChannelMetadataField.HEIGHT, float),
                        (ChannelMetadataField.SOURCE_FPS, float),
                        (ChannelMetadataField.PIXEL_FORMAT, str),
                        (ChannelMetadataField.VIDEO_BITRATE, float),
                    ]:
                        val = _d(key, caster)
                        if val is not None:
                            stream_info[key] = val

                    # Audio fields
                    for key, caster in [
                        (ChannelMetadataField.AUDIO_CODEC, str),
                        (ChannelMetadataField.SAMPLE_RATE, float),
                        (ChannelMetadataField.AUDIO_CHANNELS, str),
                        (ChannelMetadataField.AUDIO_BITRATE, float),
                    ]:
                        val = _d(key, caster)
                        if val is not None:
                            stream_info[key] = val

                    if stream_info:
                        cp["stream_info"] = stream_info
        except Exception as e:
            logger.debug(f"Unable to capture stream stats for recording: {e}")

        # Removed: local thumbnail generation. We rely on EPG/VOD/TMDB/OMDb/keyless providers only.

        # Final cancellation guard: destroy() may have deleted the record while
        # remuxing.  If it's gone now, skip saving "interrupted" status and
        # skip the notification — destroy() already sent recording_cancelled.
        if not Recording.objects.filter(id=recording_id).exists():
            logger.info(
                f"Recording {recording_id} was deleted during post-processing — skipping final save."
            )
            return

        def _save_final_metadata():
            recording_obj.custom_properties = cp
            recording_obj.save(update_fields=["custom_properties"])

        _db_retry(
            _save_final_metadata,
            max_retries=_dvr_db_max_retries,
            base_interval=_dvr_db_retry_interval,
            label=f"DVR recording {recording_id}: metadata save",
        )

        # Notify frontends so the UI refreshes immediately (e.g. "Stopped" → "Completed")
        try:
            async_to_sync(channel_layer.group_send)(
                "updates",
                {
                    "type": "update",
                    "data": {"success": True, "type": "recording_ended", "channel": channel.name},
                },
            )
        except Exception:
            pass
    except Exception as e:
        logger.debug(f"Unable to finalize Recording metadata: {e}")

    # Optionally run comskip post-process
    try:
        from core.models import CoreSettings
        if CoreSettings.get_dvr_comskip_enabled():
            comskip_process_recording.delay(recording_id)
    except Exception:
        pass

    # Release the per-recording active lock so a follow-up dispatch (e.g.
    # manual re-record) does not have to wait for TTL expiry.
    if _active_lock_redis is not None:
        try:
            _active_lock_redis.delete(_active_lock_key)
        except Exception:
            pass


@shared_task
def recover_recordings_on_startup():
    """
    On service startup, reschedule or resume recordings to handle server restarts.
    - For recordings whose window includes 'now': mark interrupted and start a new recording for the remainder.
    - For future recordings: ensure a task is scheduled at start_time.
    Uses a Redis lock to ensure only one worker runs this recovery.
    """
    try:
        from django.utils import timezone
        from .models import Recording
        from core.utils import RedisClient
        from .signals import schedule_recording_task, revoke_task

        redis = RedisClient.get_client()
        if redis is None:
            # Fail closed: without Redis we cannot guarantee single-execution
            # across workers, and running unguarded has been shown to start
            # duplicate ffmpeg processes against the same HLS output dir.
            logger.warning(
                "recover_recordings_on_startup: Redis unavailable, "
                "skipping recovery to avoid duplicate recording dispatch."
            )
            return "Redis unavailable"

        lock_key = "dvr:recover_lock"
        # Set lock with 10-minute TTL; must be long enough for Phase 2
        # ffmpeg remux operations on large files.
        if not redis.set(lock_key, "1", ex=600, nx=True):
            return "Recovery already in progress"

        now = timezone.now()

        # Resume in-window recordings.  DB queries and saves use _db_retry
        # to tolerate transient connection errors common during startup.
        active = _db_retry(
            lambda: list(Recording.objects.filter(
                start_time__lte=now, end_time__gt=now
            )),
            label="DVR recovery: fetching active recordings",
        )
        for rec in active:
            try:
                cp = rec.custom_properties or {}
                current_status = cp.get("status", "")

                # Skip recordings that are already in a terminal state.
                # "completed" / "stopped" — user stopped or it finished normally; do NOT
                # overwrite the status and re-schedule (that would cause the
                # Interrupted → In-Progress → Previously-Recorded ghost cycle).
                # NOTE: "recording" is NOT skipped — this function runs on
                # worker_ready, meaning all previous workers are dead.  A
                # recording stuck in "recording" status is from a crashed
                # worker and must be recovered.
                if current_status in ("completed", "stopped"):
                    logger.info(
                        f"recover_recordings_on_startup: skipping recording {rec.id} "
                        f"(status={current_status!r}, already in terminal/active state)."
                    )
                    continue

                # If a live worker still holds the per-recording active lock,
                # skip recovery, that worker is currently writing to the HLS
                # output dir and re-dispatching `run_recording` would spawn a
                # second ffmpeg racing on the same files. The lock auto-expires
                # within ~45s if the worker died, so this only blocks recovery
                # while the recording is genuinely live.
                try:
                    if redis.exists(f"dvr:recording-active:{rec.id}"):
                        logger.info(
                            f"recover_recordings_on_startup: skipping recording {rec.id} "
                            f"(active-recording lock held by live worker)."
                        )
                        continue
                except Exception:
                    pass

                # Mark interrupted due to restart; will flip to 'recording' when task starts
                cp["status"] = "interrupted"
                cp["interrupted_reason"] = "server_restarted"

                # Preserve the pre-restart .ts segment path so run_recording
                # can concatenate it with the resumed segment later.
                hls_dir_path = cp.get("_hls_dir")
                if hls_dir_path and os.path.isdir(hls_dir_path):
                    existing_segs = [f for f in os.listdir(hls_dir_path) if f.endswith(".ts")]
                    logger.info(
                        f"recover_recordings_on_startup: recording {rec.id} — "
                        f"HLS dir has {len(existing_segs)} existing segment(s), will resume"
                    )

                rec.custom_properties = cp
                _db_retry(
                    lambda r=rec: r.save(update_fields=["custom_properties"]),
                    label=f"DVR recovery: recording {rec.id} status update",
                )

                # Revoke the old PeriodicTask so Celery Beat doesn't also
                # fire run_recording for this recording (would be a duplicate).
                old_task_id = rec.task_id
                if old_task_id:
                    try:
                        revoke_task(old_task_id)
                    except Exception:
                        pass

                # Start recording for remaining window.  Use a deterministic
                # task_id so duplicate dispatches (e.g. from a second recovery
                # attempt) are deduplicated by Celery/Redis.
                recovery_task_id = f"dvr-recover-{rec.id}"
                run_recording.apply_async(
                    args=[rec.id, rec.channel_id, str(now), str(rec.end_time)],
                    eta=now,
                    task_id=recovery_task_id,
                )
            except Exception as e:
                logger.warning(f"Failed to resume recording {rec.id}: {e}")

        # Finalize expired recordings that were active when the server crashed
        # but whose end_time has now passed.  Remux the partial .ts and mark
        # as interrupted so the user can watch whatever was captured.
        expired = _db_retry(
            lambda: list(Recording.objects.filter(
                end_time__lte=now,
                custom_properties__status="recording",
            )),
            label="DVR recovery: fetching expired recordings",
        )
        for rec in expired:
            try:
                cp = rec.custom_properties or {}
                hls_dir_path = cp.get("_hls_dir")
                mkv_path = cp.get("file_path")

                if hls_dir_path and os.path.isdir(hls_dir_path) and mkv_path:
                    # Parse m3u8 for ordered segment list; fall back to sorted filenames
                    _m3u8 = os.path.join(hls_dir_path, "index.m3u8")
                    _segs = []
                    if os.path.exists(_m3u8):
                        try:
                            with open(_m3u8) as _f:
                                for _l in _f:
                                    _l = _l.strip()
                                    if _l and not _l.startswith('#'):
                                        _sp = os.path.join(hls_dir_path, _l) if not os.path.isabs(_l) else _l
                                        if os.path.exists(_sp):
                                            _segs.append(_sp)
                        except Exception:
                            pass
                    if not _segs:
                        _segs = sorted(
                            os.path.join(hls_dir_path, f)
                            for f in os.listdir(hls_dir_path)
                            if f.startswith("seg_") and f.endswith(".ts")
                        )

                    if _segs:
                        logger.info(
                            f"recover_recordings_on_startup: recording {rec.id} expired "
                            f"during downtime, concat {len(_segs)} HLS segment(s) \u2192 MKV"
                        )
                        os.makedirs(os.path.dirname(mkv_path), exist_ok=True)
                        _concat_txt = os.path.join(hls_dir_path, "concat.txt")
                        try:
                            with open(_concat_txt, "w") as _cl:
                                for _s in _segs:
                                    _escaped = _s.replace("'", "'\\''")
                                    _cl.write(f"file '{_escaped}'\n")
                            _res = subprocess.run(
                                [
                                    "ffmpeg", "-y",
                                    "-f", "concat", "-safe", "0",
                                    "-i", _concat_txt,
                                    "-c", "copy", mkv_path,
                                ],
                                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                            )
                            if _res.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0:
                                cp["status"] = "interrupted"
                                cp["interrupted_reason"] = "server_restarted_after_end"
                                cp["remux_success"] = True
                                try:
                                    shutil.rmtree(hls_dir_path)
                                except OSError:
                                    pass
                                logger.info(
                                    f"recover_recordings_on_startup: recording {rec.id} HLS\u2192MKV concat succeeded"
                                )
                            else:
                                cp["status"] = "interrupted"
                                cp["interrupted_reason"] = "server_restarted_after_end"
                                cp["remux_success"] = False
                                logger.warning(
                                    f"recover_recordings_on_startup: recording {rec.id} concat failed, keeping HLS dir"
                                )
                        except Exception as _ce:
                            cp["status"] = "interrupted"
                            cp["interrupted_reason"] = "server_restarted_after_end"
                            cp["remux_success"] = False
                            logger.warning(
                                f"recover_recordings_on_startup: recording {rec.id} concat error: {_ce}"
                            )
                        finally:
                            try:
                                os.remove(_concat_txt)
                            except OSError:
                                pass
                    else:
                        cp["status"] = "interrupted"
                        cp["interrupted_reason"] = "server_restarted_after_end"
                        cp["remux_success"] = False
                else:
                    cp["status"] = "interrupted"
                    cp["interrupted_reason"] = "server_restarted_after_end"
                    cp["remux_success"] = False

                rec.custom_properties = cp
                _db_retry(
                    lambda r=rec: r.save(update_fields=["custom_properties"]),
                    label=f"DVR recovery: recording {rec.id} expired status update",
                )
            except Exception as e:
                logger.warning(f"Failed to finalize expired recording {rec.id}: {e}")

        # Ensure future recordings are scheduled.
        # With ClockedSchedule, PeriodicTasks survive restarts in the DB.
        # Only recreate if the PeriodicTask is missing (safety net).
        from django_celery_beat.models import PeriodicTask as _PT
        from apps.channels.signals import _dvr_task_name

        upcoming = _db_retry(
            lambda: list(Recording.objects.filter(
                start_time__gt=now, end_time__gt=now
            )),
            label="DVR recovery: fetching upcoming recordings",
        )

        # Batch-fetch existing PeriodicTask names to avoid N+1 queries
        task_names = {_dvr_task_name(r.id) for r in upcoming}
        existing_tasks = set(_db_retry(
            lambda: list(_PT.objects.filter(name__in=task_names).values_list("name", flat=True)),
            label="DVR recovery: fetching existing periodic tasks",
        )) if task_names else set()

        for rec in upcoming:
            try:
                task_name = _dvr_task_name(rec.id)
                if task_name in existing_tasks:
                    if rec.task_id != task_name:
                        rec.task_id = task_name
                        _db_retry(
                            lambda r=rec: r.save(update_fields=["task_id"]),
                            label=f"DVR recovery: recording {rec.id} task_id update",
                        )
                    continue
                # PeriodicTask missing - recreate it
                task_id = schedule_recording_task(rec)
                if task_id:
                    rec.task_id = task_id
                    _db_retry(
                        lambda r=rec: r.save(update_fields=["task_id"]),
                        label=f"DVR recovery: recording {rec.id} task_id update",
                    )
            except Exception as e:
                logger.warning(f"Failed to schedule recording {rec.id}: {e}")

        # Release the lock early so a subsequent restart can recover
        # immediately.  The 10-minute TTL is only a safety net in case
        # recovery itself crashes before reaching this point.
        if redis:
            redis.delete(lock_key)

        return "Recovery complete"
    except Exception as e:
        logger.error(f"Error during DVR recovery: {e}")
        return f"Error: {e}"

@shared_task
def comskip_process_recording(recording_id: int):
    """Run comskip on the MKV to remove commercials and replace the file in place.
    Safe to call even if comskip is not installed; stores status in custom_properties.comskip.
    """
    import shutil
    from django.db import DatabaseError
    from .models import Recording
    # Helper to broadcast status over websocket
    def _ws(status: str, extra: dict | None = None):
        try:
            from core.utils import send_websocket_update
            payload = {"success": True, "type": "comskip_status", "status": status, "recording_id": recording_id}
            if extra:
                payload.update(extra)
            send_websocket_update('updates', 'update', payload)
        except Exception:
            pass

    try:
        rec = Recording.objects.get(id=recording_id)
    except Recording.DoesNotExist:
        return "not_found"

    cp = rec.custom_properties.copy() if isinstance(rec.custom_properties, dict) else {}

    def _persist_custom_properties():
        """Persist updated custom_properties without raising if the row disappeared."""
        try:
            updated = Recording.objects.filter(pk=recording_id).update(custom_properties=cp)
            if not updated:
                logger.warning(
                    "Recording %s vanished before comskip status could be saved",
                    recording_id,
                )
                return False
        except DatabaseError as db_err:
            logger.warning(
                "Failed to persist comskip status for recording %s: %s",
                recording_id,
                db_err,
            )
            return False
        except Exception as unexpected:
            logger.warning(
                "Unexpected error while saving comskip status for recording %s: %s",
                recording_id,
                unexpected,
            )
            return False
        return True
    file_path = (cp or {}).get("file_path")
    if not file_path or not os.path.exists(file_path):
        return "no_file"

    if isinstance(cp.get("comskip"), dict) and cp["comskip"].get("status") == "completed":
        return "already_processed"

    comskip_bin = shutil.which("comskip")
    if not comskip_bin:
        cp["comskip"] = {"status": "skipped", "reason": "comskip_not_installed"}
        _persist_custom_properties()
        _ws('skipped', {"reason": "comskip_not_installed"})
        return "comskip_missing"

    base, _ = os.path.splitext(file_path)
    edl_path = f"{base}.edl"

    # Notify start
    _ws('started', {"title": (cp.get('program') or {}).get('title') or os.path.basename(file_path)})

    try:
        cmd = [comskip_bin, "--output", os.path.dirname(file_path)]
        # Prefer user-specified INI, fall back to known defaults
        ini_candidates = []
        try:
            custom_ini = CoreSettings.get_dvr_comskip_custom_path()
            if custom_ini:
                ini_candidates.append(custom_ini)
        except Exception as ini_err:
            logger.debug(f"Unable to load custom comskip.ini path: {ini_err}")
        ini_candidates.extend(["/etc/comskip/comskip.ini", "/app/docker/comskip.ini"])
        selected_ini = None
        for ini_path in ini_candidates:
            if ini_path and os.path.exists(ini_path):
                selected_ini = ini_path
                cmd.extend([f"--ini={ini_path}"])
                break
        cmd.append(file_path)
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        # comskip exit codes: 0 = commercials found, 1 = no commercials detected.
        # Negative codes indicate killed by signal; anything else is a real error.
        if result.returncode == 1:
            # No commercials detected — not an error.
            cp["comskip"] = {"status": "completed", "skipped": True}
            if selected_ini:
                cp["comskip"]["ini_path"] = selected_ini
            _persist_custom_properties()
            _ws('skipped', {"reason": "no_commercials_detected"})
            return "no_commercials"
        elif result.returncode != 0:
            stderr_tail = (result.stderr or "").strip().splitlines()
            stderr_tail = stderr_tail[-5:] if stderr_tail else []
            detail = {
                "status": "error",
                "reason": "comskip_failed",
                "returncode": result.returncode,
            }
            if result.returncode < 0:
                try:
                    detail["signal"] = signal.Signals(-result.returncode).name
                except Exception:
                    detail["signal"] = f"signal_{-result.returncode}"
            if stderr_tail:
                detail["stderr"] = "\n".join(stderr_tail)
            if selected_ini:
                detail["ini_path"] = selected_ini
            cp["comskip"] = detail
            _persist_custom_properties()
            _ws('error', {"reason": "comskip_failed", "returncode": result.returncode})
            return "comskip_failed"
    except Exception as e:
        cp["comskip"] = {"status": "error", "reason": f"comskip_failed: {e}"}
        _persist_custom_properties()
        _ws('error', {"reason": str(e)})
        return "comskip_failed"

    if not os.path.exists(edl_path):
        cp["comskip"] = {"status": "error", "reason": "edl_not_found"}
        _persist_custom_properties()
        _ws('error', {"reason": "edl_not_found"})
        return "no_edl"

    # Duration via ffprobe
    def _ffprobe_duration(path):
        try:
            p = subprocess.run([
                "ffprobe", "-v", "error", "-show_entries", "format=duration",
                "-of", "default=noprint_wrappers=1:nokey=1", path
            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True)
            return float(p.stdout.strip())
        except Exception:
            return None

    duration = _ffprobe_duration(file_path)
    if duration is None:
        cp["comskip"] = {"status": "error", "reason": "duration_unknown"}
        _persist_custom_properties()
        _ws('error', {"reason": "duration_unknown"})
        return "no_duration"

    commercials = []
    try:
        with open(edl_path, "r") as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) >= 2:
                    try:
                        s = float(parts[0]); e = float(parts[1])
                        commercials.append((max(0.0, s), min(duration, e)))
                    except Exception:
                        pass
    except Exception:
        pass

    commercials.sort()
    keep = []
    cur = 0.0
    for s, e in commercials:
        if s > cur:
            keep.append((cur, max(cur, s)))
        cur = max(cur, e)
    if cur < duration:
        keep.append((cur, duration))

    if not commercials or sum((e - s) for s, e in commercials) <= 0.5:
        cp["comskip"] = {
            "status": "completed",
            "skipped": True,
            "edl": os.path.basename(edl_path),
        }
        if selected_ini:
            cp["comskip"]["ini_path"] = selected_ini
        _persist_custom_properties()
        _ws('skipped', {"reason": "no_commercials", "commercials": 0})
        return "no_commercials"

    workdir = os.path.dirname(file_path)
    parts = []
    try:
        for idx, (s, e) in enumerate(keep):
            seg = os.path.join(workdir, f"segment_{idx:03d}.mkv")
            dur = max(0.0, e - s)
            if dur <= 0.01:
                continue
            subprocess.run([
                "ffmpeg", "-y", "-ss", f"{s:.3f}", "-i", file_path, "-t", f"{dur:.3f}",
                "-c", "copy", "-avoid_negative_ts", "1", seg
            ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            parts.append(seg)

        if not parts:
            raise RuntimeError("no_parts")

        list_path = os.path.join(workdir, "concat_list.txt")
        with open(list_path, "w") as lf:
            for pth in parts:
                escaped = pth.replace("'", "'\\''")
                lf.write(f"file '{escaped}'\n")

        output_path = os.path.join(workdir, f"{os.path.splitext(os.path.basename(file_path))[0]}.cut.mkv")
        subprocess.run([
            "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_path, "-c", "copy", output_path
        ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        try:
            os.replace(output_path, file_path)
        except Exception:
            shutil.copy(output_path, file_path)

        try:
            os.remove(list_path)
        except Exception:
            pass
        for pth in parts:
            try: os.remove(pth)
            except Exception: pass

        cp["comskip"] = {
            "status": "completed",
            "edl": os.path.basename(edl_path),
            "segments_kept": len(parts),
            "commercials": len(commercials),
        }
        if selected_ini:
            cp["comskip"]["ini_path"] = selected_ini
        _persist_custom_properties()
        _ws('completed', {"commercials": len(commercials), "segments_kept": len(parts)})
        return "ok"
    except Exception as e:
        cp["comskip"] = {"status": "error", "reason": str(e)}
        _persist_custom_properties()
        _ws('error', {"reason": str(e)})
        return f"error:{e}"
def _resolve_poster_for_program(channel_name, program, channel_logo_id=None):
    """Resolve poster URL and/or Logo id for a recording program.

    Callers should enrich the program dict via _match_epg_program_by_timeslot
    before invoking this function so that EPG data is already available.

    Pipeline: EPG images → VOD logo → TMDB/OMDb → TVMaze/iTunes →
    direct program fields → Logo table → Logo creation → channel logo.
    Returns (poster_logo_id, poster_url) where either may be None.
    """
    poster_logo_id = None
    poster_url = None
    epg_props = None

    _title = ((program.get("title") if isinstance(program, dict) else None) or "").strip() or None

    # Guard: if the "title" is really just the channel name (common when EPG
    # has no real program data), don't use it for external API searches —
    # those queries produce false-positive artwork from unrelated shows.
    _title_is_channel_name = False
    if _title and channel_name:
        def _norm_channel(s):
            return s.lower().replace("*", "").replace("-", " ").strip()
        _title_is_channel_name = _norm_channel(_title) == _norm_channel(channel_name)

    # Stage 1: EPG Program images/icon (with URL validation)
    try:
        from apps.epg.models import ProgramData
        prog_id = program.get("id") if isinstance(program, dict) else None
        if prog_id:
            epg_program = ProgramData.objects.filter(id=prog_id).only("custom_properties").first()
            if epg_program and epg_program.custom_properties:
                epg_props = epg_program.custom_properties or {}
                poster_url = _pick_best_image_from_epg_props(epg_props)
                if poster_url and not _validate_url(poster_url):
                    poster_url = None
                if not poster_url:
                    icon = epg_props.get("icon")
                    if isinstance(icon, str) and icon and _validate_url(icon):
                        poster_url = icon
    except Exception:
        pass

    # Stage 2: VOD logo fallback by title
    if not poster_url and not poster_logo_id and _title and not _title_is_channel_name:
        try:
            from apps.vod.models import Movie, Series
            vod_logo = None
            movie = Movie.objects.filter(name__iexact=_title).select_related("logo").first()
            if movie and movie.logo:
                vod_logo = movie.logo
            if not vod_logo:
                series = Series.objects.filter(name__iexact=_title).select_related("logo").first()
                if series and series.logo:
                    vod_logo = series.logo
            if vod_logo:
                poster_logo_id = vod_logo.id
        except Exception:
            pass

    # Stage 3: TMDB/OMDb (keyed APIs)
    if not poster_url and not poster_logo_id and _title and not _title_is_channel_name:
        try:
            tmdb_key = os.environ.get('TMDB_API_KEY')
            omdb_key = os.environ.get('OMDB_API_KEY')
            title = _title
            year = None
            imdb_id = None

            # Derive year and imdb_id from cached EPG data
            if epg_props:
                d = epg_props.get('date')
                if d and len(str(d)) >= 4:
                    year = str(d)[:4]
                imdb_id = epg_props.get('imdb.com_id')

            # TMDB: by IMDb ID
            if not poster_url and tmdb_key and imdb_id:
                try:
                    url = f"https://api.themoviedb.org/3/find/{quote(imdb_id)}?api_key={tmdb_key}&external_source=imdb_id"
                    resp = requests.get(url, timeout=5)
                    if resp.ok:
                        data = resp.json() or {}
                        picks = []
                        for k in ('movie_results', 'tv_results', 'tv_episode_results', 'tv_season_results'):
                            picks.extend(data.get(k) or [])
                        for item in picks:
                            if item.get('poster_path'):
                                poster_url = f"https://image.tmdb.org/t/p/w780{item['poster_path']}"
                                break
                except Exception:
                    pass

            # TMDB: by title (and year if available)
            if not poster_url and tmdb_key and title:
                try:
                    q = quote(title)
                    extra = f"&year={year}" if year else ""
                    url = f"https://api.themoviedb.org/3/search/multi?api_key={tmdb_key}&query={q}{extra}"
                    resp = requests.get(url, timeout=5)
                    if resp.ok:
                        data = resp.json() or {}
                        results = data.get('results') or []
                        results.sort(key=lambda x: float(x.get('popularity') or 0), reverse=True)
                        for item in results:
                            if item.get('poster_path'):
                                poster_url = f"https://image.tmdb.org/t/p/w780{item['poster_path']}"
                                break
                except Exception:
                    pass

            # OMDb fallback
            if not poster_url and omdb_key:
                try:
                    if imdb_id:
                        url = f"https://www.omdbapi.com/?apikey={omdb_key}&i={quote(imdb_id)}"
                    elif title:
                        yy = f"&y={year}" if year else ""
                        url = f"https://www.omdbapi.com/?apikey={omdb_key}&t={quote(title)}{yy}"
                    else:
                        url = None
                    if url:
                        resp = requests.get(url, timeout=5)
                        if resp.ok:
                            data = resp.json() or {}
                            p = data.get('Poster')
                            if p and p != 'N/A':
                                poster_url = p
                except Exception:
                    pass
        except Exception:
            pass

    # Stage 4: Keyless providers (TVMaze & iTunes)
    if not poster_url and not poster_logo_id and _title and not _title_is_channel_name:
        try:
            title = _title
            # TVMaze
            try:
                url = f"https://api.tvmaze.com/singlesearch/shows?q={quote(title)}"
                resp = requests.get(url, timeout=5)
                if resp.ok:
                    data = resp.json() or {}
                    img = (data.get('image') or {})
                    p = img.get('original') or img.get('medium')
                    if p:
                        poster_url = p
            except Exception:
                pass
            # iTunes
            if not poster_url:
                try:
                    for media in ('movie', 'tvShow'):
                        url = f"https://itunes.apple.com/search?term={quote(title)}&media={media}&limit=1"
                        resp = requests.get(url, timeout=5)
                        if resp.ok:
                            data = resp.json() or {}
                            results = data.get('results') or []
                            if results:
                                art = results[0].get('artworkUrl100')
                                if art:
                                    poster_url = art.replace('100x100', '600x600')
                                    break
                except Exception:
                    pass
        except Exception:
            pass

    # Stage 5: Direct fields on program object (with URL validation)
    if not poster_url and not poster_logo_id and isinstance(program, dict):
        for key in ("poster", "cover", "cover_big", "image", "icon"):
            val = program.get(key)
            if isinstance(val, dict):
                candidate = val.get("url")
                if candidate and _validate_url(candidate):
                    poster_url = candidate
                    break
            elif isinstance(val, str) and val and _validate_url(val):
                poster_url = val
                break

    # Stage 6: Search existing Logo entries by program title
    if not poster_logo_id and not poster_url and _title and not _title_is_channel_name:
        try:
            from .models import Logo
            existing = Logo.objects.filter(name__iexact=_title).first()
            if existing:
                poster_logo_id = existing.id
                poster_url = existing.url
        except Exception:
            pass

    # Stage 7: Persist to Logo table if URL available
    if not poster_logo_id and poster_url and len(poster_url) <= 1000:
        try:
            from .models import Logo
            logo, _ = Logo.objects.get_or_create(url=poster_url, defaults={"name": _title or channel_name})
            poster_logo_id = logo.id
        except Exception:
            pass

    # Stage 8: Fall back to channel logo
    if not poster_logo_id and not poster_url and channel_logo_id:
        poster_logo_id = channel_logo_id

    return poster_logo_id, poster_url


@shared_task
def prefetch_recording_artwork(recording_id):
    """Prefetch poster info for a scheduled recording so the UI can show art in Upcoming."""
    try:
        from .models import Recording
        rec = Recording.objects.get(id=recording_id)
        cp = rec.custom_properties or {}

        # Bail out if the recording is already active or finished — run_recording
        # handles poster resolution itself, and saving here can race with status updates.
        current_status = cp.get("status", "")
        if current_status in ("recording", "completed", "stopped", "interrupted"):
            return "skipped: status is " + current_status

        program = cp.get("program") or {}

        # Enrich empty program dicts (manual recordings) from EPG time-slot data.
        # Persists matched title/description for display in the recording card.
        if isinstance(program, dict) and not program.get("user_edited") and not program.get("id") and not program.get("title"):
            epg_match = _match_epg_program_by_timeslot(
                rec.channel.epg_data, rec.start_time, rec.end_time,
            )
            if epg_match:
                program.update(epg_match)
                cp["program"] = program

        poster_logo_id, poster_url = _resolve_poster_for_program(
            rec.channel.name, program, channel_logo_id=rec.channel.logo_id,
        )
        updated = False
        if poster_logo_id and cp.get("poster_logo_id") != poster_logo_id:
            cp["poster_logo_id"] = poster_logo_id
            updated = True
        if poster_url and cp.get("poster_url") != poster_url:
            cp["poster_url"] = poster_url
            updated = True
        # Enrich with rating if available from ProgramData.custom_properties
        try:
            from apps.epg.models import ProgramData
            prog_id = program.get("id") if isinstance(program, dict) else None
            if prog_id:
                epg_program = ProgramData.objects.filter(id=prog_id).only("custom_properties").first()
                if epg_program and isinstance(epg_program.custom_properties, dict):
                    rating_val = epg_program.custom_properties.get("rating")
                    rating_sys = epg_program.custom_properties.get("rating_system")
                    season_val = epg_program.custom_properties.get("season")
                    episode_val = epg_program.custom_properties.get("episode")
                    onscreen = epg_program.custom_properties.get("onscreen_episode")
                    if rating_val and cp.get("rating") != rating_val:
                        cp["rating"] = rating_val
                        updated = True
                    if rating_sys and cp.get("rating_system") != rating_sys:
                        cp["rating_system"] = rating_sys
                        updated = True
                    if season_val is not None and cp.get("season") != season_val:
                        cp["season"] = season_val
                        updated = True
                    if episode_val is not None and cp.get("episode") != episode_val:
                        cp["episode"] = episode_val
                        updated = True
                    if onscreen and cp.get("onscreen_episode") != onscreen:
                        cp["onscreen_episode"] = onscreen
                        updated = True
        except Exception:
            pass

        if updated:
            # Re-read from DB to avoid overwriting status changes made by
            # the stop endpoint or run_recording's final metadata save.
            rec.refresh_from_db()
            fresh_cp = rec.custom_properties or {}
            for key in ("program", "poster_logo_id", "poster_url", "rating",
                        "rating_system", "season", "episode", "onscreen_episode"):
                if key in cp:
                    fresh_cp[key] = cp[key]
            rec.custom_properties = fresh_cp
            rec.save(update_fields=["custom_properties"])
            try:
                from core.utils import send_websocket_update
                send_websocket_update('updates', 'update', {"success": True, "type": "recording_updated", "recording_id": rec.id})
            except Exception:
                pass
        return "ok"
    except Exception as e:
        logger.debug(f"prefetch_recording_artwork failed: {e}")
        return f"error: {e}"


@shared_task(bind=True)
def bulk_create_channels_from_streams(self, stream_ids, channel_profile_ids=None, starting_channel_number=None):
    """
    Asynchronously create channels from a list of stream IDs.
    Provides progress updates via WebSocket.

    Args:
        stream_ids: List of stream IDs to create channels from
        channel_profile_ids: Optional list of channel profile IDs to assign channels to
        starting_channel_number: Optional starting channel number behavior:
            - None: Use provider channel numbers, then auto-assign from 1
            - 0: Start with lowest available number and increment by 1
            - Other number: Use as starting number for auto-assignment
    """
    from apps.channels.models import Stream, Channel, ChannelGroup, ChannelProfile, ChannelProfileMembership, Logo
    from apps.epg.models import EPGData
    from django.db import transaction
    from django.shortcuts import get_object_or_404
    from core.utils import send_websocket_update

    task_id = self.request.id
    total_streams = len(stream_ids)
    created_channels = []
    errors = []

    try:
        # Send initial progress update
        send_websocket_update('updates', 'update', {
            'type': 'bulk_channel_creation_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_streams,
            'status': 'starting',
            'message': f'Starting bulk creation of {total_streams} channels...'
        })

        # Gather current used numbers once
        used_numbers = set(Channel.objects.all().values_list("channel_number", flat=True))

        # Initialize next_number based on starting_channel_number mode
        if starting_channel_number is None:
            # Mode 1: Use provider numbers when available, auto-assign when not
            next_number = 1
        elif starting_channel_number == 0:
            # Mode 2: Start from lowest available number
            next_number = 1
        elif starting_channel_number == -1:
            # Mode 4: Start after the current highest channel number
            highest = Channel.objects.order_by('-channel_number').values_list('channel_number', flat=True).first()
            next_number = (int(highest) + 1) if highest is not None else 1
        else:
            # Mode 3: Start from specified number
            next_number = starting_channel_number

        def get_auto_number():
            nonlocal next_number
            while next_number in used_numbers:
                next_number += 1
            used_numbers.add(next_number)
            return next_number

        logos_to_create = []
        channels_to_create = []
        streams_map = []
        logo_map = []
        profile_map = []

        # Process streams in batches to avoid memory issues
        batch_size = 100
        processed = 0

        for i in range(0, total_streams, batch_size):
            batch_stream_ids = stream_ids[i:i + batch_size]
            # Fetch streams and preserve the order from batch_stream_ids
            batch_streams_dict = {stream.id: stream for stream in Stream.objects.filter(id__in=batch_stream_ids)}
            batch_streams = [batch_streams_dict[stream_id] for stream_id in batch_stream_ids if stream_id in batch_streams_dict]

            # Send progress update
            send_websocket_update('updates', 'update', {
                'type': 'bulk_channel_creation_progress',
                'task_id': task_id,
                'progress': processed,
                'total': total_streams,
                'status': 'processing',
                'message': f'Processing streams {processed + 1}-{min(processed + batch_size, total_streams)} of {total_streams}...'
            })

            for stream in batch_streams:
                try:
                    name = stream.name
                    channel_group = stream.channel_group
                    stream_custom_props = stream.custom_properties or {}

                    # Determine channel number based on starting_channel_number mode
                    channel_number = None

                    if starting_channel_number is None:
                        # Mode 1: Use provider numbers when available (from stream_chno field)
                        if stream.stream_chno is not None:
                            channel_number = stream.stream_chno

                    # For modes 2 and 3 (starting_channel_number == 0 or specific number),
                    # ignore provider numbers and use sequential assignment

                    # Get TVC guide station ID
                    tvc_guide_stationid = None
                    if "tvc-guide-stationid" in stream_custom_props:
                        tvc_guide_stationid = stream_custom_props["tvc-guide-stationid"]

                    # Check if the determined/provider number is available
                    if channel_number is not None and (
                        channel_number in used_numbers
                        or Channel.objects.filter(channel_number=channel_number).exists()
                    ):
                        # Provider number is taken, use auto-assignment
                        channel_number = get_auto_number()
                    elif channel_number is not None:
                        # Provider number is available, use it
                        used_numbers.add(channel_number)
                    else:
                        # No provider number or ignoring provider numbers, use auto-assignment
                        channel_number = get_auto_number()

                    channel_data = {
                        "channel_number": channel_number,
                        "name": name,
                        "tvc_guide_stationid": tvc_guide_stationid,
                        "tvg_id": stream.tvg_id,
                        "is_adult": stream.is_adult,
                    }

                    # Only add channel_group_id if the stream has a channel group
                    if channel_group:
                        channel_data["channel_group_id"] = channel_group.id

                    # Attempt to find existing EPGs with the same tvg-id
                    epgs = EPGData.objects.filter(tvg_id=stream.tvg_id)
                    if epgs:
                        channel_data["epg_data_id"] = epgs.first().id

                    channel = Channel(**channel_data)
                    channels_to_create.append(channel)
                    streams_map.append([stream.id])

                    # Store profile IDs for this channel
                    profile_map.append(channel_profile_ids)

                    # Handle logo - validate URL length to avoid PostgreSQL btree index errors
                    validated_logo_url = validate_logo_url(stream.logo_url) if stream.logo_url else None
                    if validated_logo_url:
                        logos_to_create.append(
                            Logo(
                                url=validated_logo_url,
                                name=stream.name or stream.tvg_id,
                            )
                        )
                        logo_map.append(validated_logo_url)
                    else:
                        logo_map.append(None)

                    processed += 1

                except Exception as e:
                    errors.append({
                        'stream_id': stream.id if 'stream' in locals() else 'unknown',
                        'error': str(e)
                    })
                    processed += 1

        # Create logos first
        if logos_to_create:
            send_websocket_update('updates', 'update', {
                'type': 'bulk_channel_creation_progress',
                'task_id': task_id,
                'progress': processed,
                'total': total_streams,
                'status': 'creating_logos',
                'message': f'Creating {len(logos_to_create)} logos...'
            })
            Logo.objects.bulk_create(logos_to_create, ignore_conflicts=True)

        # Get logo objects for association
        channel_logos = {
            logo.url: logo
            for logo in Logo.objects.filter(
                url__in=[url for url in logo_map if url is not None]
            )
        }

        # Create channels in database
        if channels_to_create:
            send_websocket_update('updates', 'update', {
                'type': 'bulk_channel_creation_progress',
                'task_id': task_id,
                'progress': processed,
                'total': total_streams,
                'status': 'creating_channels',
                'message': f'Creating {len(channels_to_create)} channels in database...'
            })

            with transaction.atomic():
                created_channels = Channel.objects.bulk_create(channels_to_create)

                # Update channels with logos and create stream associations
                update = []
                channel_stream_associations = []
                channel_profile_memberships = []

                for channel, stream_ids, logo_url, profile_ids in zip(
                    created_channels, streams_map, logo_map, profile_map
                ):
                    if logo_url:
                        channel.logo = channel_logos[logo_url]
                        update.append(channel)

                    # Create stream associations
                    for stream_id in stream_ids:
                        from apps.channels.models import ChannelStream
                        channel_stream_associations.append(
                            ChannelStream(channel=channel, stream_id=stream_id, order=0)
                        )

                    # Handle channel profile membership
                    # Semantics:
                    # - None: add to ALL profiles (backward compatible default)
                    # - Empty array []: add to NO profiles
                    # - Sentinel [0] or 0 in array: add to ALL profiles (explicit)
                    # - [1,2,...]: add to specified profile IDs only
                    if profile_ids is None:
                        # Omitted -> add to all profiles (backward compatible)
                        all_profiles = ChannelProfile.objects.all()
                        channel_profile_memberships.extend([
                            ChannelProfileMembership(
                                channel_profile=profile,
                                channel=channel,
                                enabled=True
                            )
                            for profile in all_profiles
                        ])
                    elif isinstance(profile_ids, list) and len(profile_ids) == 0:
                        # Empty array -> add to no profiles
                        pass
                    elif isinstance(profile_ids, list) and 0 in profile_ids:
                        # Sentinel 0 -> add to all profiles (explicit)
                        all_profiles = ChannelProfile.objects.all()
                        channel_profile_memberships.extend([
                            ChannelProfileMembership(
                                channel_profile=profile,
                                channel=channel,
                                enabled=True
                            )
                            for profile in all_profiles
                        ])
                    else:
                        # Specific profile IDs
                        try:
                            specific_profiles = ChannelProfile.objects.filter(id__in=profile_ids)
                            channel_profile_memberships.extend([
                                ChannelProfileMembership(
                                    channel_profile=profile,
                                    channel=channel,
                                    enabled=True
                                )
                                for profile in specific_profiles
                            ])
                        except Exception as e:
                            errors.append({
                                'channel_id': channel.id,
                                'error': f'Failed to add to profiles: {str(e)}'
                            })

                # Bulk update channels with logos
                if update:
                    Channel.objects.bulk_update(update, ["logo"])

                # Bulk create channel-stream associations
                if channel_stream_associations:
                    from apps.channels.models import ChannelStream
                    ChannelStream.objects.bulk_create(channel_stream_associations, ignore_conflicts=True)

                # Bulk create profile memberships
                if channel_profile_memberships:
                    ChannelProfileMembership.objects.bulk_create(channel_profile_memberships, ignore_conflicts=True)

        # Send completion update
        send_websocket_update('updates', 'update', {
            'type': 'bulk_channel_creation_progress',
            'task_id': task_id,
            'progress': total_streams,
            'total': total_streams,
            'status': 'completed',
            'message': f'Successfully created {len(created_channels)} channels',
            'created_count': len(created_channels),
            'error_count': len(errors),
            'errors': errors[:10]  # Send first 10 errors only
        })

        # Send general channel update notification
        send_websocket_update('updates', 'update', {
            'type': 'channels_created',
            'count': len(created_channels)
        })

        return {
            'status': 'completed',
            'created_count': len(created_channels),
            'error_count': len(errors),
            'errors': errors
        }

    except Exception as e:
        logger.error(f"Bulk channel creation failed: {e}")
        send_websocket_update('updates', 'update', {
            'type': 'bulk_channel_creation_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_streams,
            'status': 'failed',
            'message': f'Task failed: {str(e)}',
            'error': str(e)
        })
        raise


@shared_task(bind=True)
def set_channels_names_from_epg(self, channel_ids):
    """
    Celery task to set channel names from EPG data for multiple channels
    """
    from core.utils import send_websocket_update

    task_id = self.request.id
    total_channels = len(channel_ids)
    updated_count = 0
    errors = []

    try:
        logger.info(f"Starting EPG name setting task for {total_channels} channels")

        # Send initial progress
        send_websocket_update('updates', 'update', {
            'type': 'epg_name_setting_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_channels,
            'status': 'running',
            'message': 'Starting EPG name setting...'
        })

        batch_size = 100
        for i in range(0, total_channels, batch_size):
            batch_ids = channel_ids[i:i + batch_size]
            batch_updates = []

            # Get channels and their EPG data
            channels = Channel.objects.filter(id__in=batch_ids).select_related('epg_data')

            for channel in channels:
                try:
                    if channel.epg_data and channel.epg_data.name:
                        if channel.name != channel.epg_data.name:
                            channel.name = channel.epg_data.name
                            batch_updates.append(channel)
                            updated_count += 1
                except Exception as e:
                    errors.append(f"Channel {channel.id}: {str(e)}")
                    logger.error(f"Error processing channel {channel.id}: {e}")

            # Bulk update the batch
            if batch_updates:
                Channel.objects.bulk_update(batch_updates, ['name'])

            # Send progress update
            progress = min(i + batch_size, total_channels)
            send_websocket_update('updates', 'update', {
                'type': 'epg_name_setting_progress',
                'task_id': task_id,
                'progress': progress,
                'total': total_channels,
                'status': 'running',
                'message': f'Updated {updated_count} channel names...',
                'updated_count': updated_count
            })

        # Send completion notification
        send_websocket_update('updates', 'update', {
            'type': 'epg_name_setting_progress',
            'task_id': task_id,
            'progress': total_channels,
            'total': total_channels,
            'status': 'completed',
            'message': f'Successfully updated {updated_count} channel names from EPG data',
            'updated_count': updated_count,
            'error_count': len(errors),
            'errors': errors
        })

        logger.info(f"EPG name setting task completed. Updated {updated_count} channels")
        return {
            'status': 'completed',
            'updated_count': updated_count,
            'error_count': len(errors),
            'errors': errors
        }

    except Exception as e:
        logger.error(f"EPG name setting task failed: {e}")
        send_websocket_update('updates', 'update', {
            'type': 'epg_name_setting_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_channels,
            'status': 'failed',
            'message': f'Task failed: {str(e)}',
            'error': str(e)
        })
        raise


@shared_task(bind=True)
def set_channels_logos_from_epg(self, channel_ids):
    """
    Celery task to set channel logos from EPG data for multiple channels
    Creates logos from EPG icon URLs if they don't exist
    """
    from .models import Logo
    from core.utils import send_websocket_update
    import requests
    from urllib.parse import urlparse

    task_id = self.request.id
    total_channels = len(channel_ids)
    updated_count = 0
    created_logos_count = 0
    errors = []

    try:
        logger.info(f"Starting EPG logo setting task for {total_channels} channels")

        # Send initial progress
        send_websocket_update('updates', 'update', {
            'type': 'epg_logo_setting_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_channels,
            'status': 'running',
            'message': 'Starting EPG logo setting...'
        })

        batch_size = 50  # Smaller batch for logo processing
        for i in range(0, total_channels, batch_size):
            batch_ids = channel_ids[i:i + batch_size]
            batch_updates = []

            # Get channels and their EPG data
            channels = Channel.objects.filter(id__in=batch_ids).select_related('epg_data', 'logo')

            for channel in channels:
                try:
                    if channel.epg_data and channel.epg_data.icon_url:
                        icon_url = channel.epg_data.icon_url.strip()

                        # Try to find existing logo with this URL
                        try:
                            logo = Logo.objects.get(url=icon_url)
                        except Logo.DoesNotExist:
                            # Create new logo from EPG icon URL
                            try:
                                # Generate a name for the logo
                                logo_name = channel.epg_data.name or f"Logo for {channel.epg_data.tvg_id}"

                                # Create the logo record
                                logo = Logo.objects.create(
                                    name=logo_name,
                                    url=icon_url
                                )
                                created_logos_count += 1
                                logger.info(f"Created new logo from EPG: {logo_name} - {icon_url}")

                            except Exception as create_error:
                                errors.append(f"Channel {channel.id}: Failed to create logo from {icon_url}: {str(create_error)}")
                                logger.error(f"Failed to create logo for channel {channel.id}: {create_error}")
                                continue

                        # Update channel logo if different
                        if channel.logo != logo:
                            channel.logo = logo
                            batch_updates.append(channel)
                            updated_count += 1

                except Exception as e:
                    errors.append(f"Channel {channel.id}: {str(e)}")
                    logger.error(f"Error processing channel {channel.id}: {e}")

            # Bulk update the batch
            if batch_updates:
                Channel.objects.bulk_update(batch_updates, ['logo'])

            # Send progress update
            progress = min(i + batch_size, total_channels)
            send_websocket_update('updates', 'update', {
                'type': 'epg_logo_setting_progress',
                'task_id': task_id,
                'progress': progress,
                'total': total_channels,
                'status': 'running',
                'message': f'Updated {updated_count} channel logos, created {created_logos_count} new logos...',
                'updated_count': updated_count,
                'created_logos_count': created_logos_count
            })

        # Send completion notification
        send_websocket_update('updates', 'update', {
            'type': 'epg_logo_setting_progress',
            'task_id': task_id,
            'progress': total_channels,
            'total': total_channels,
            'status': 'completed',
            'message': f'Successfully updated {updated_count} channel logos and created {created_logos_count} new logos from EPG data',
            'updated_count': updated_count,
            'created_logos_count': created_logos_count,
            'error_count': len(errors),
            'errors': errors
        })

        logger.info(f"EPG logo setting task completed. Updated {updated_count} channels, created {created_logos_count} logos")
        return {
            'status': 'completed',
            'updated_count': updated_count,
            'created_logos_count': created_logos_count,
            'error_count': len(errors),
            'errors': errors
        }

    except Exception as e:
        logger.error(f"EPG logo setting task failed: {e}")
        send_websocket_update('updates', 'update', {
            'type': 'epg_logo_setting_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_channels,
            'status': 'failed',
            'message': f'Task failed: {str(e)}',
            'error': str(e)
        })
        raise


@shared_task(bind=True)
def set_channels_tvg_ids_from_epg(self, channel_ids):
    """
    Celery task to set channel TVG-IDs from EPG data for multiple channels
    """
    from core.utils import send_websocket_update

    task_id = self.request.id
    total_channels = len(channel_ids)
    updated_count = 0
    errors = []

    try:
        logger.info(f"Starting EPG TVG-ID setting task for {total_channels} channels")

        # Send initial progress
        send_websocket_update('updates', 'update', {
            'type': 'epg_tvg_id_setting_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_channels,
            'status': 'running',
            'message': 'Starting EPG TVG-ID setting...'
        })

        batch_size = 100
        for i in range(0, total_channels, batch_size):
            batch_ids = channel_ids[i:i + batch_size]
            batch_updates = []

            # Get channels and their EPG data
            channels = Channel.objects.filter(id__in=batch_ids).select_related('epg_data')

            for channel in channels:
                try:
                    if channel.epg_data and channel.epg_data.tvg_id:
                        if channel.tvg_id != channel.epg_data.tvg_id:
                            channel.tvg_id = channel.epg_data.tvg_id
                            batch_updates.append(channel)
                            updated_count += 1
                except Exception as e:
                    errors.append(f"Channel {channel.id}: {str(e)}")
                    logger.error(f"Error processing channel {channel.id}: {e}")

            # Bulk update the batch
            if batch_updates:
                Channel.objects.bulk_update(batch_updates, ['tvg_id'])

            # Send progress update
            progress = min(i + batch_size, total_channels)
            send_websocket_update('updates', 'update', {
                'type': 'epg_tvg_id_setting_progress',
                'task_id': task_id,
                'progress': progress,
                'total': total_channels,
                'status': 'running',
                'message': f'Updated {updated_count} channel TVG-IDs...',
                'updated_count': updated_count
            })

        # Send completion notification
        send_websocket_update('updates', 'update', {
            'type': 'epg_tvg_id_setting_progress',
            'task_id': task_id,
            'progress': total_channels,
            'total': total_channels,
            'status': 'completed',
            'message': f'Successfully updated {updated_count} channel TVG-IDs from EPG data',
            'updated_count': updated_count,
            'error_count': len(errors),
            'errors': errors
        })

        logger.info(f"EPG TVG-ID setting task completed. Updated {updated_count} channels")
        return {
            'status': 'completed',
            'updated_count': updated_count,
            'error_count': len(errors),
            'errors': errors
        }

    except Exception as e:
        logger.error(f"EPG TVG-ID setting task failed: {e}")
        send_websocket_update('updates', 'update', {
            'type': 'epg_tvg_id_setting_progress',
            'task_id': task_id,
            'progress': 0,
            'total': total_channels,
            'status': 'failed',
            'message': f'Task failed: {str(e)}',
            'error': str(e)
        })
        raise