# apps/m3u/tasks.py
import logging
import re
import regex
import requests
import os
import gc
import gzip, zipfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from celery.app.control import Inspect
from celery.result import AsyncResult
from celery import shared_task, current_app, group
from django.conf import settings
from django.core.cache import cache
from django.db import models, transaction
from .models import M3UAccount
from apps.channels.models import Stream, ChannelGroup, ChannelGroupM3UAccount
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.utils import timezone
import time
import json
from core.utils import (
    RedisClient,
    acquire_task_lock,
    release_task_lock,
    TaskLockRenewer,
    natural_sort_key,
    log_system_event,
)
from core.models import CoreSettings, UserAgent
from asgiref.sync import async_to_sync
from core.xtream_codes import Client as XCClient
from core.utils import send_websocket_update
from .utils import normalize_stream_url

logger = logging.getLogger(__name__)

BATCH_SIZE = 1500  # Optimized batch size for threading
m3u_dir = os.path.join(settings.MEDIA_ROOT, "cached_m3u")

_EXTINF_ATTR_RE = re.compile(r'([^\s=]+)\s*=\s*(["\'])(.*?)\2')


def fetch_m3u_lines(account, use_cache=False):
    os.makedirs(m3u_dir, exist_ok=True)
    file_path = os.path.join(m3u_dir, f"{account.id}.m3u")

    """Fetch M3U file lines efficiently."""
    if account.server_url:
        if not use_cache or not os.path.exists(file_path):
            try:
                # Try to get account-specific user agent first
                user_agent_obj = account.get_user_agent()
                user_agent = (
                    user_agent_obj.user_agent
                    if user_agent_obj
                    else "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                )

                logger.debug(
                    f"Using user agent: {user_agent} for M3U account: {account.name}"
                )
                headers = {"User-Agent": user_agent}
                logger.info(f"Fetching from URL {account.server_url}")

                # Set account status to FETCHING before starting download
                account.status = M3UAccount.Status.FETCHING
                account.last_message = "Starting download..."
                account.save(update_fields=["status", "last_message"])

                response = requests.get(
                    account.server_url, headers=headers, stream=True,
                    timeout=(30, 60),  # 30s connect, 60s read between chunks
                )

                # Log the actual response details for debugging
                logger.debug(f"HTTP Response: {response.status_code} from {account.server_url}")
                logger.debug(f"Content-Type: {response.headers.get('content-type', 'Not specified')}")
                logger.debug(f"Content-Length: {response.headers.get('content-length', 'Not specified')}")
                logger.debug(f"Response headers: {dict(response.headers)}")

                # Check if we've been redirected to a different URL
                if hasattr(response, 'url') and response.url != account.server_url:
                    logger.warning(f"Request was redirected from {account.server_url} to {response.url}")

                # Check for ANY non-success status code FIRST (before raise_for_status)
                if response.status_code < 200 or response.status_code >= 300:
                    # For error responses, read the content immediately (not streaming)
                    try:
                        response_content = response.text[:1000]  # Capture up to 1000 characters
                        logger.error(f"Error response content: {response_content!r}")
                    except Exception as e:
                        logger.error(f"Could not read error response content: {e}")
                        response_content = "Could not read error response content"

                    # Provide specific messages for known non-standard codes
                    if response.status_code == 884:
                        error_msg = f"Server returned HTTP 884 (authentication/authorization failure) from URL: {account.server_url}. Server message: {response_content}"
                    elif response.status_code >= 800:
                        error_msg = f"Server returned non-standard HTTP status {response.status_code} from URL: {account.server_url}. Server message: {response_content}"
                    elif response.status_code == 404:
                        error_msg = f"M3U file not found (404) at URL: {account.server_url}. Server message: {response_content}"
                    elif response.status_code == 403:
                        error_msg = f"Access forbidden (403) to M3U file at URL: {account.server_url}. Server message: {response_content}"
                    elif response.status_code == 401:
                        error_msg = f"Authentication required (401) for M3U file at URL: {account.server_url}. Server message: {response_content}"
                    elif response.status_code == 500:
                        error_msg = f"Server error (500) while fetching M3U file from URL: {account.server_url}. Server message: {response_content}"
                    else:
                        error_msg = f"HTTP error ({response.status_code}) while fetching M3U file from URL: {account.server_url}. Server message: {response_content}"

                    logger.error(error_msg)
                    account.status = M3UAccount.Status.ERROR
                    account.last_message = error_msg
                    account.save(update_fields=["status", "last_message"])
                    send_m3u_update(
                        account.id,
                        "downloading",
                        100,
                        status="error",
                        error=error_msg,
                    )
                    return [], False

                # Only call raise_for_status if we have a success code (this should not raise now)
                response.raise_for_status()

                total_size = int(response.headers.get("Content-Length", 0))
                downloaded = 0
                start_time = time.time()
                last_update_time = start_time
                progress = 0
                has_content = False

                # Stream directly to a temp file to avoid holding the entire
                # M3U in memory (large files can be 100MB+, which would use
                # ~3x that in RAM in certain approaches).
                temp_path = file_path + ".tmp"
                try:
                    send_m3u_update(account.id, "downloading", 0)
                    with open(temp_path, "wb") as tmp_file:
                        for chunk in response.iter_content(chunk_size=8192):
                            if chunk:
                                tmp_file.write(chunk)
                                has_content = True

                                downloaded += len(chunk)
                                elapsed_time = time.time() - start_time

                                # Calculate download speed in KB/s
                                speed = downloaded / elapsed_time / 1024  # in KB/s

                                # Calculate progress percentage
                                if total_size and total_size > 0:
                                    progress = (downloaded / total_size) * 100

                                # Time remaining (in seconds)
                                time_remaining = (
                                    (total_size - downloaded) / (speed * 1024)
                                    if speed > 0
                                    else 0
                                )

                                current_time = time.time()
                                if current_time - last_update_time >= 0.5:
                                    last_update_time = current_time
                                    if progress > 0:
                                        # Update the account's last_message with detailed progress info
                                        progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining"
                                        account.last_message = progress_msg
                                        account.save(update_fields=["last_message"])

                                        send_m3u_update(
                                            account.id,
                                            "downloading",
                                            progress,
                                            speed=speed,
                                            elapsed_time=elapsed_time,
                                            time_remaining=time_remaining,
                                            message=progress_msg,
                                        )

                    # Check if we actually received any content
                    logger.info(f"Download completed. Has content: {has_content}, Content length: {downloaded} bytes")
                    if not has_content or downloaded == 0:
                        error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
                        logger.error(error_msg)
                        account.status = M3UAccount.Status.ERROR
                        account.last_message = error_msg
                        account.save(update_fields=["status", "last_message"])
                        send_m3u_update(
                            account.id,
                            "downloading",
                            100,
                            status="error",
                            error=error_msg,
                        )
                        return [], False

                    # Validate the file by reading only the first portion from
                    # disk — no need to load the entire file into memory just
                    # to check the header.
                    VALIDATION_READ_SIZE = 32768  # 32KB covers headers comfortably
                    try:
                        with open(temp_path, "rb") as vf:
                            head_bytes = vf.read(VALIDATION_READ_SIZE)
                        head_str = head_bytes.decode('utf-8', errors='ignore')
                        head_lines = head_str.strip().split('\n')

                        # Count total lines efficiently without loading full file
                        with open(temp_path, "rb") as vf:
                            total_lines = sum(1 for _ in vf)

                        # Log first few lines for debugging (be careful not to log too much)
                        preview_lines = head_lines[:5]
                        logger.info(f"Content preview (first 5 lines): {preview_lines}")
                        logger.info(f"Total lines in content: {total_lines}")

                        # Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
                        is_valid_m3u = False

                        # First, check if this looks like an error response disguised as 200 OK
                        head_lower = head_str.lower()
                        if any(error_indicator in head_lower for error_indicator in [
                            '<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
                            'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
                        ]):
                            logger.warning(f"Content appears to be an error response disguised as HTTP 200: {head_str[:200]!r}")
                            # Continue with M3U validation, but this gives us a clue

                        if head_lines and head_lines[0].strip().upper().startswith('#EXTM3U'):
                            is_valid_m3u = True
                            logger.info("Content validated as M3U: starts with #EXTM3U")
                        elif any(line.strip().startswith('#EXTINF:') for line in head_lines):
                            is_valid_m3u = True
                            logger.info("Content validated as M3U: contains #EXTINF entries")
                        elif any(line.strip().startswith('http') for line in head_lines):
                            # Has HTTP URLs, might be a simple M3U without headers
                            is_valid_m3u = True
                            logger.info("Content validated as M3U: contains HTTP URLs")
                        elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in head_lines):
                            # Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
                            is_valid_m3u = True
                            logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")

                        if not is_valid_m3u:
                            # Log what we actually received for debugging
                            logger.error(f"Invalid M3U content received. First 200 characters: {head_str[:200]!r}")

                            # Try to provide more specific error messages based on content
                            if '<html' in head_lower or '<!doctype html' in head_lower:
                                error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
                            elif 'error' in head_lower or 'not found' in head_lower:
                                error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {head_str[:100]}"
                            elif len(head_str.strip()) == 0:
                                error_msg = f"Server returned completely empty response from URL: {account.server_url}"
                            else:
                                error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U file."
                            logger.error(error_msg)
                            account.status = M3UAccount.Status.ERROR
                            account.last_message = error_msg
                            account.save(update_fields=["status", "last_message"])
                            send_m3u_update(
                                account.id,
                                "downloading",
                                100,
                                status="error",
                                error=error_msg,
                            )
                            return [], False

                    except UnicodeDecodeError:
                        with open(temp_path, "rb") as vf:
                            first_bytes = vf.read(200)
                        logger.error(f"Non-text content received. First 200 bytes: {first_bytes!r}")
                        error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U file."
                        logger.error(error_msg)
                        account.status = M3UAccount.Status.ERROR
                        account.last_message = error_msg
                        account.save(update_fields=["status", "last_message"])
                        send_m3u_update(
                            account.id,
                            "downloading",
                            100,
                            status="error",
                            error=error_msg,
                        )
                        return [], False

                    # Validation passed — promote temp file to final path
                    os.replace(temp_path, file_path)

                    # Final update with 100% progress
                    dl_size = downloaded / 1024 / 1024
                    final_msg = f"Download complete. Size: {dl_size:.2f} MB, Time: {time.time() - start_time:.1f}s"
                    account.last_message = final_msg
                    account.save(update_fields=["last_message"])
                    send_m3u_update(account.id, "downloading", 100, message=final_msg)

                finally:
                    # Clean up temp file on any failure path
                    if os.path.exists(temp_path):
                        try:
                            os.remove(temp_path)
                        except OSError:
                            pass
            except requests.exceptions.HTTPError as e:
                # Handle HTTP errors specifically with more context
                status_code = e.response.status_code if e.response else "unknown"

                # Try to capture the error response content
                response_content = ""
                if e.response:
                    try:
                        response_content = e.response.text[:500]  # Limit to first 500 characters
                        logger.error(f"HTTP error response content: {response_content!r}")
                    except Exception as content_error:
                        logger.error(f"Could not read HTTP error response content: {content_error}")
                        response_content = "Could not read error response content"

                if status_code == 404:
                    error_msg = f"M3U file not found (404) at URL: {account.server_url}. Server message: {response_content}"
                elif status_code == 403:
                    error_msg = f"Access forbidden (403) to M3U file at URL: {account.server_url}. Server message: {response_content}"
                elif status_code == 401:
                    error_msg = f"Authentication required (401) for M3U file at URL: {account.server_url}. Server message: {response_content}"
                elif status_code == 500:
                    error_msg = f"Server error (500) while fetching M3U file from URL: {account.server_url}. Server message: {response_content}"
                else:
                    error_msg = f"HTTP error ({status_code}) while fetching M3U file from URL: {account.server_url}. Server message: {response_content}"

                logger.error(error_msg)
                account.status = M3UAccount.Status.ERROR
                account.last_message = error_msg
                account.save(update_fields=["status", "last_message"])
                send_m3u_update(
                    account.id,
                    "downloading",
                    100,
                    status="error",
                    error=error_msg,
                )
                return [], False
            except requests.exceptions.RequestException as e:
                # Handle other request errors (connection, timeout, etc.)
                if "timeout" in str(e).lower():
                    error_msg = f"Timeout while fetching M3U file from URL: {account.server_url}"
                elif "connection" in str(e).lower():
                    error_msg = f"Connection error while fetching M3U file from URL: {account.server_url}"
                else:
                    error_msg = f"Network error while fetching M3U file from URL: {account.server_url} - {str(e)}"

                logger.error(error_msg)
                account.status = M3UAccount.Status.ERROR
                account.last_message = error_msg
                account.save(update_fields=["status", "last_message"])
                send_m3u_update(
                    account.id,
                    "downloading",
                    100,
                    status="error",
                    error=error_msg,
                )
                return [], False
            except Exception as e:
                # Handle any other unexpected errors
                error_msg = f"Unexpected error while fetching M3U file from URL: {account.server_url} - {str(e)}"
                logger.error(error_msg)
                account.status = M3UAccount.Status.ERROR
                account.last_message = error_msg
                account.save(update_fields=["status", "last_message"])
                send_m3u_update(
                    account.id,
                    "downloading",
                    100,
                    status="error",
                    error=error_msg,
                )
                return [], False

        # Check if the file exists and is not empty (fallback check - should not happen with new validation)
        if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
            error_msg = f"M3U file is unexpectedly missing or empty after validation: {file_path}"
            logger.error(error_msg)
            account.status = M3UAccount.Status.ERROR
            account.last_message = error_msg
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account.id, "downloading", 100, status="error", error=error_msg
            )
            return [], False  # Return empty list and False for success

        try:
            with open(file_path, "r", encoding="utf-8") as f:
                return f.readlines(), True
        except Exception as e:
            error_msg = f"Error reading M3U file: {str(e)}"
            logger.error(error_msg)
            account.status = M3UAccount.Status.ERROR
            account.last_message = error_msg
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account.id, "downloading", 100, status="error", error=error_msg
            )
            return [], False

    elif account.file_path:
        try:
            if account.file_path.endswith(".gz"):
                with gzip.open(account.file_path, "rt", encoding="utf-8") as f:
                    return f.readlines(), True

            elif account.file_path.endswith(".zip"):
                with zipfile.ZipFile(account.file_path, "r") as zip_file:
                    for name in zip_file.namelist():
                        if name.endswith(".m3u"):
                            with zip_file.open(name) as f:
                                return [
                                    line.decode("utf-8") for line in f.readlines()
                                ], True

                    error_msg = (
                        f"No .m3u file found in ZIP archive: {account.file_path}"
                    )
                    logger.warning(error_msg)
                    account.status = M3UAccount.Status.ERROR
                    account.last_message = error_msg
                    account.save(update_fields=["status", "last_message"])
                    send_m3u_update(
                        account.id, "downloading", 100, status="error", error=error_msg
                    )
                    return [], False

            else:
                with open(account.file_path, "r", encoding="utf-8") as f:
                    return f.readlines(), True

        except (IOError, OSError, zipfile.BadZipFile, gzip.BadGzipFile) as e:
            error_msg = f"Error opening file {account.file_path}: {e}"
            logger.error(error_msg)
            account.status = M3UAccount.Status.ERROR
            account.last_message = error_msg
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account.id, "downloading", 100, status="error", error=error_msg
            )
            return [], False

    # Neither server_url nor uploaded_file is available
    error_msg = "No M3U source available (missing URL and file)"
    logger.error(error_msg)
    account.status = M3UAccount.Status.ERROR
    account.last_message = error_msg
    account.save(update_fields=["status", "last_message"])
    send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg)
    return [], False


def get_case_insensitive_attr(attributes, key, default=""):
    """Get attribute value using case-insensitive key lookup."""
    for attr_key, attr_value in attributes.items():
        if attr_key.lower() == key.lower():
            return attr_value
    return default


def parse_is_adult(value):
    try:
        return int(value) == 1
    except (TypeError, ValueError):
        return False


def parse_extinf_line(line: str) -> dict:
    """
    Parse an EXTINF line from an M3U file.
    This function removes the "#EXTINF:" prefix, then extracts all key="value" attributes,
    and treats everything after the last attribute as the display name.

    Returns a dictionary with:
      - 'attributes': a dict of attribute key/value pairs (e.g. tvg-id, tvg-logo, group-title)
      - 'display_name': the text after the attributes (the fallback display name)
      - 'name': the value from tvg-name (if present) or the display name otherwise.
    """
    if not line.startswith("#EXTINF:"):
        return None
    content = line[len("#EXTINF:") :].strip()

    # Single pass: extract all attributes AND track the last attribute position.
    # Keys are normalised to lowercase so downstream code can use plain dict.get()
    attrs = {}
    last_attr_end = 0

    for match in _EXTINF_ATTR_RE.finditer(content):
        attrs[match.group(1).lower()] = match.group(3)
        last_attr_end = match.end()

    # Everything after the last attribute (skipping leading comma and whitespace) is the display name
    if last_attr_end > 0:
        remaining = content[last_attr_end:].strip()
        # Remove leading comma if present
        if remaining.startswith(','):
            remaining = remaining[1:].strip()
        display_name = remaining
    else:
        # No attributes found, try the old comma-split method as fallback
        parts = content.split(',', 1)
        if len(parts) == 2:
            display_name = parts[1].strip()
        else:
            display_name = content.strip()

    # Per the base #EXTINF spec, the comma text is the canonical human-readable title.
    # Fall back to tvc-guide-title, then tvg-name (which some providers use as an EPG key,
    # not a display label), and finally the raw content if everything else is empty.
    name = display_name or attrs.get("tvc-guide-title") or attrs.get("tvg-name") or content.strip()
    return {"attributes": attrs, "display_name": display_name, "name": name}


def iter_m3u_entries(lines):
    """
    Generator that yields fully-assembled M3U stream entries from raw lines.

    Each yielded dict is guaranteed to contain a ``url`` key in addition to the
    fields produced by :func:`parse_extinf_line` (``attributes``, ``display_name``,
    ``name``).  Recognised extended-tag lines that appear *between* an ``#EXTINF``
    and its URL are accumulated into the pending entry so they are available for
    downstream processing:

    - ``#EXTGRP`` — sets ``attributes["group-title"]`` when no ``group-title``
      attribute was present on the ``#EXTINF`` line (explicit attribute wins).
    - ``#EXTVLCOPT`` — stored as a list under the ``vlc_opts`` key.

    Unknown directives (``#KODIPROP``, etc.) and blank lines are
    silently skipped while keeping the pending entry intact.  A second ``#EXTINF``
    before a URL discards the first entry with a warning.  A trailing ``#EXTINF``
    at end-of-file with no URL is also discarded.

    Adding support for a new directive requires only a new ``elif`` branch here;
    no other code needs to change.
    """
    pending = None
    pending_line_num = None
    for line_num, raw_line in enumerate(lines, 1):
        line = raw_line.strip()
        if not line:
            continue

        if line.startswith("#EXTINF"):
            if pending is not None:
                logger.warning(
                    f"Line {pending_line_num}: #EXTINF had no URL (next #EXTINF at line {line_num}); "
                    f"discarding entry: {list(pending['attributes'].items())[:3]}"
                )
            parsed = parse_extinf_line(line)
            if parsed is None:
                logger.warning(f"Line {line_num}: Failed to parse #EXTINF: {line[:200]}")
            pending = parsed  # None if malformed; URL branch guards on `pending is not None`
            pending_line_num = line_num

        elif line.startswith("#EXTGRP:"):
            # Only apply when group-title is absent — explicit attribute wins.
            if pending is not None and "group-title" not in pending["attributes"]:
                pending["attributes"]["group-title"] = line[len("#EXTGRP:"):].strip()
            # else: #EXTGRP outside an entry, or group-title already set — silently skip

        elif line.startswith("#EXTVLCOPT:"):
            if pending is not None:
                pending.setdefault("vlc_opts", []).append(line[len("#EXTVLCOPT:"):])
            # else: #EXTVLCOPT outside an entry — silently skip

        elif pending is not None and line.startswith(("http", "rtsp", "rtp", "udp")):
            pending["url"] = normalize_stream_url(line) if line.startswith("udp") else line
            yield pending
            pending = None
            pending_line_num = None

        # else: unknown directive or bare content — skip, keeping pending intact

    if pending is not None:
        logger.warning(
            f"Line {pending_line_num}: #EXTINF at end of file had no URL; "
            f"discarding entry: {list(pending['attributes'].items())[:3]}"
        )


@shared_task
def refresh_m3u_accounts():
    """Queue background parse for all active M3UAccounts."""
    active_accounts = M3UAccount.objects.filter(is_active=True)
    count = 0
    for account in active_accounts:
        refresh_single_m3u_account.delay(account.id)
        count += 1

    msg = f"Queued M3U refresh for {count} active account(s)."
    logger.info(msg)
    return msg


def check_field_lengths(streams_to_create):
    for stream in streams_to_create:
        for field, value in stream.__dict__.items():
            if isinstance(value, str) and len(value) > 255:
                print(f"{field} --- {value}")

        print("")
        print("")


@shared_task
def process_groups(account, groups, scan_start_time=None):
    """Process groups and update their relationships with the M3U account.

    Args:
        account: M3UAccount instance
        groups: Dict of group names to custom properties
        scan_start_time: Timestamp when the scan started (for consistent last_seen marking)
    """
    # Use scan_start_time if provided, otherwise current time
    # This ensures consistency with stream processing and cleanup logic
    if scan_start_time is None:
        scan_start_time = timezone.now()

    existing_groups = {
        group.name: group
        for group in ChannelGroup.objects.filter(name__in=groups.keys())
    }
    logger.info(f"Currently {len(existing_groups)} existing groups")

    # Check if we should auto-enable new groups based on account settings
    account_custom_props = account.custom_properties or {}
    auto_enable_new_groups_live = account_custom_props.get("auto_enable_new_groups_live", True)

    # Separate existing groups from groups that need to be created
    existing_group_objs = []
    groups_to_create = []

    for group_name, custom_props in groups.items():
        if group_name in existing_groups:
            existing_group_objs.append(existing_groups[group_name])
        else:
            groups_to_create.append(ChannelGroup(name=group_name))

    # Create new groups and fetch them back with IDs
    newly_created_group_objs = []
    if groups_to_create:
        logger.info(f"Creating {len(groups_to_create)} new groups for account {account.id}")
        newly_created_group_objs = list(ChannelGroup.bulk_create_and_fetch(groups_to_create))
        logger.debug(f"Successfully created {len(newly_created_group_objs)} new groups")

    # Combine all groups
    all_group_objs = existing_group_objs + newly_created_group_objs

    # Get existing relationships for this account
    existing_relationships = {
        rel.channel_group.name: rel
        for rel in ChannelGroupM3UAccount.objects.filter(
            m3u_account=account,
            channel_group__name__in=groups.keys()
        ).select_related('channel_group')
    }

    relations_to_create = []
    relations_to_update = []

    for group in all_group_objs:
        custom_props = groups.get(group.name, {})

        if group.name in existing_relationships:
            # Update existing relationship if xc_id has changed (preserve other custom properties)
            existing_rel = existing_relationships[group.name]

            # Get existing custom properties (now JSONB, no need to parse)
            existing_custom_props = existing_rel.custom_properties or {}

            # Get the new xc_id from groups data
            new_xc_id = custom_props.get("xc_id")
            existing_xc_id = existing_custom_props.get("xc_id")

            # Only update if xc_id has changed
            if new_xc_id != existing_xc_id:
                # Merge new xc_id with existing custom properties to preserve user settings
                updated_custom_props = existing_custom_props.copy()
                if new_xc_id is not None:
                    updated_custom_props["xc_id"] = new_xc_id
                elif "xc_id" in updated_custom_props:
                    # Remove xc_id if it's no longer provided (e.g., converting from XC to standard)
                    del updated_custom_props["xc_id"]

                existing_rel.custom_properties = updated_custom_props
                existing_rel.last_seen = scan_start_time
                existing_rel.is_stale = False
                relations_to_update.append(existing_rel)
                logger.debug(f"Updated xc_id for group '{group.name}' from '{existing_xc_id}' to '{new_xc_id}' - account {account.id}")
            else:
                # Update last_seen even if xc_id hasn't changed
                existing_rel.last_seen = scan_start_time
                existing_rel.is_stale = False
                relations_to_update.append(existing_rel)
                logger.debug(f"xc_id unchanged for group '{group.name}' - account {account.id}")
        else:
            # Create new relationship - this group is new to this M3U account
            # Use the auto_enable setting to determine if it should start enabled
            if not auto_enable_new_groups_live:
                logger.info(f"Group '{group.name}' is new to account {account.id} - creating relationship but DISABLED (auto_enable_new_groups_live=False)")

            relations_to_create.append(
                ChannelGroupM3UAccount(
                    channel_group=group,
                    m3u_account=account,
                    custom_properties=custom_props,
                    enabled=auto_enable_new_groups_live,
                    last_seen=scan_start_time,
                    is_stale=False,
                )
            )

    # Bulk create new relationships
    if relations_to_create:
        ChannelGroupM3UAccount.objects.bulk_create(relations_to_create, ignore_conflicts=True)
        logger.debug(f"Created {len(relations_to_create)} new group relationships for account {account.id}")

    # Bulk update existing relationships
    if relations_to_update:
        ChannelGroupM3UAccount.objects.bulk_update(relations_to_update, ['custom_properties', 'last_seen', 'is_stale'])
        logger.info(f"Updated {len(relations_to_update)} existing group relationships for account {account.id}")


def cleanup_stale_group_relationships(account, scan_start_time):
    """
    Remove group relationships that haven't been seen since the stale retention period.
    This follows the same logic as stream cleanup for consistency.
    """
    # Calculate cutoff date for stale group relationships
    stale_cutoff = scan_start_time - timezone.timedelta(days=account.stale_stream_days)
    logger.info(
        f"Removing group relationships not seen since {stale_cutoff} for M3U account {account.id}"
    )

    # Find stale relationships
    stale_relationships = ChannelGroupM3UAccount.objects.filter(
        m3u_account=account,
        last_seen__lt=stale_cutoff
    ).select_related('channel_group')

    relations_to_delete = list(stale_relationships)
    deleted_count = len(relations_to_delete)

    if deleted_count > 0:
        logger.info(
            f"Found {deleted_count} stale group relationships for account {account.id}: "
            f"{[rel.channel_group.name for rel in relations_to_delete]}"
        )

        # Delete the stale relationships
        stale_relationships.delete()

        # Check if any of the deleted relationships left groups with no remaining associations
        orphaned_group_ids = []
        for rel in relations_to_delete:
            group = rel.channel_group

            # Check if this group has any remaining M3U account relationships
            remaining_m3u_relationships = ChannelGroupM3UAccount.objects.filter(
                channel_group=group
            ).exists()

            # Check if this group has any direct channels (not through M3U accounts)
            has_direct_channels = group.related_channels().exists()

            # If no relationships and no direct channels, it's safe to delete
            if not remaining_m3u_relationships and not has_direct_channels:
                orphaned_group_ids.append(group.id)
                logger.debug(f"Group '{group.name}' has no remaining associations and will be deleted")

        # Delete truly orphaned groups
        if orphaned_group_ids:
            deleted_groups = list(ChannelGroup.objects.filter(id__in=orphaned_group_ids).values_list('name', flat=True))
            ChannelGroup.objects.filter(id__in=orphaned_group_ids).delete()
            logger.info(f"Deleted {len(orphaned_group_ids)} orphaned groups that had no remaining associations: {deleted_groups}")
    else:
        logger.debug(f"No stale group relationships found for account {account.id}")

    return deleted_count


def collect_xc_streams(account_id, enabled_groups):
    """Collect all XC streams in a single API call and filter by enabled groups."""
    account = M3UAccount.objects.get(id=account_id)
    all_streams = []

    # Create a mapping from category_id to group info for filtering
    enabled_category_ids = {}
    for group_name, props in enabled_groups.items():
        if "xc_id" in props:
            enabled_category_ids[str(props["xc_id"])] = {
                "name": group_name,
                "props": props
            }

    try:
        with XCClient(
            account.server_url,
            account.username,
            account.password,
            account.get_user_agent(),
        ) as xc_client:

            # Fetch ALL live streams in a single API call (much more efficient)
            logger.info("Fetching ALL live streams from XC provider...")
            all_xc_streams = xc_client.get_all_live_streams()  # Get all streams without category filter

            if not all_xc_streams:
                logger.warning("No live streams returned from XC provider")
                return []

            logger.info(f"Retrieved {len(all_xc_streams)} total live streams from provider")

            # Filter streams based on enabled categories
            filtered_count = 0
            for stream in all_xc_streams:
                # Fall back to a generated name if the provider returns null/empty
                stream_name = stream.get("name") or f"{account.name} - {stream.get('stream_id', 'Unknown')}"
                if not stream.get("name"):
                    logger.warning(
                        f"XC stream has null/empty name; using generated name '{stream_name}' "
                        f"(stream_id={stream.get('stream_id', 'unknown')})"
                    )

                # Get the category_id for this stream
                category_id = str(stream.get("category_id", ""))

                # Only include streams from enabled categories
                if category_id in enabled_category_ids:
                    group_info = enabled_category_ids[category_id]

                    # Convert XC stream to our standard format with all properties preserved
                    stream_data = {
                        "name": stream_name,
                        "url": xc_client.get_stream_url(stream["stream_id"]),
                        "attributes": {
                            "tvg-id": stream.get("epg_channel_id", ""),
                            "tvg-logo": stream.get("stream_icon", ""),
                            "group-title": group_info["name"],
                            # Preserve all XC stream properties as custom attributes
                            "stream_id": str(stream.get("stream_id", "")),
                            "num": stream.get("num"),
                            "category_id": category_id,
                            "stream_type": stream.get("stream_type", ""),
                            "added": stream.get("added", ""),
                            "is_adult": str(stream.get("is_adult", "0")),
                            "custom_sid": stream.get("custom_sid", ""),
                            # Include any other properties that might be present
                            **{k: str(v) for k, v in stream.items() if k not in [
                                "name", "stream_id", "epg_channel_id", "stream_icon",
                                "category_id", "stream_type", "added", "is_adult", "custom_sid", "num"
                            ] and v is not None}
                        }
                    }
                    all_streams.append(stream_data)
                    filtered_count += 1

    except Exception as e:
        logger.error(f"Failed to fetch XC streams: {str(e)}")
        return []

    logger.info(f"Filtered {filtered_count} streams from {len(enabled_category_ids)} enabled categories")
    return all_streams

def process_xc_category_direct(account_id, batch, groups, hash_keys):
    from django.db import connections

    # Ensure clean database connections for threading
    connections.close_all()

    account = M3UAccount.objects.get(id=account_id)

    streams_to_create = []
    streams_to_update = []
    stream_hashes = {}

    try:
        with XCClient(
            account.server_url,
            account.username,
            account.password,
            account.get_user_agent(),
        ) as xc_client:
            # Log the batch details to help with debugging
            logger.debug(f"Processing XC batch: {batch}")

            for group_name, props in batch.items():
                # Check if we have a valid xc_id for this group
                if "xc_id" not in props:
                    logger.error(
                        f"Missing xc_id for group {group_name} in batch {batch}"
                    )
                    continue

                # Get actual group ID from the mapping
                group_id = groups.get(group_name)
                if not group_id:
                    logger.error(f"Group {group_name} not found in enabled groups")
                    continue

                try:
                    logger.debug(
                        f"Fetching streams for XC category: {group_name} (ID: {props['xc_id']})"
                    )
                    streams = xc_client.get_live_category_streams(props["xc_id"])

                    if not streams:
                        logger.warning(
                            f"No streams found for XC category {group_name} (ID: {props['xc_id']})"
                        )
                        continue

                    logger.debug(
                        f"Found {len(streams)} streams for category {group_name}"
                    )

                    for stream in streams:
                        name = stream.get("name") or f"{account.name} - {stream.get('stream_id', 'Unknown')}"
                        if not stream.get("name"):
                            logger.warning(
                                f"XC stream has null/empty name in category {group_name}; "
                                f"using generated name '{name}' (stream_id={stream.get('stream_id', 'unknown')})"
                            )
                        raw_stream_id = stream.get("stream_id", "")
                        provider_stream_id = None
                        if raw_stream_id:
                            try:
                                provider_stream_id = int(raw_stream_id)
                            except (ValueError, TypeError):
                                pass
                        url = xc_client.get_stream_url(stream["stream_id"])
                        tvg_id = stream.get("epg_channel_id", "")
                        tvg_logo = stream.get("stream_icon", "")
                        group_title = group_name
                        stream_chno = stream.get("num")
                        # Convert stream_chno to float if valid, otherwise None
                        if stream_chno is not None:
                            try:
                                stream_chno = float(stream_chno)
                            except (ValueError, TypeError):
                                stream_chno = None

                        stream_hash = Stream.generate_hash_key(
                            name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title,
                            account_type='XC', stream_id=provider_stream_id
                        )
                        stream_props = {
                            "name": name,
                            "url": url,
                            "logo_url": tvg_logo,
                            "tvg_id": tvg_id,
                            "m3u_account": account,
                            "channel_group_id": int(group_id),
                            "stream_hash": stream_hash,
                            "custom_properties": stream,
                            "is_adult": parse_is_adult(stream.get("is_adult", 0)),
                            "is_stale": False,
                            "stream_id": provider_stream_id,
                            "stream_chno": stream_chno,
                        }

                        if stream_hash not in stream_hashes:
                            stream_hashes[stream_hash] = stream_props
                except Exception as e:
                    logger.error(
                        f"Error processing XC category {group_name} (ID: {props['xc_id']}): {str(e)}"
                    )
                    continue

        # Process all found streams
        existing_streams = {
            s.stream_hash: s
            for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()).select_related('m3u_account').only(
                'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno'
            )
        }

        for stream_hash, stream_props in stream_hashes.items():
            if stream_hash in existing_streams:
                obj = existing_streams[stream_hash]
                # Optimized field comparison for XC streams
                changed = (
                    obj.name != stream_props["name"] or
                    obj.url != stream_props["url"] or
                    obj.logo_url != stream_props["logo_url"] or
                    obj.tvg_id != stream_props["tvg_id"] or
                    obj.custom_properties != stream_props["custom_properties"] or
                    obj.is_adult != stream_props["is_adult"] or
                    obj.stream_id != stream_props["stream_id"] or
                    obj.stream_chno != stream_props["stream_chno"]
                )

                if changed:
                    for key, value in stream_props.items():
                        setattr(obj, key, value)
                    obj.last_seen = timezone.now()
                    obj.updated_at = timezone.now()  # Update timestamp only for changed streams
                    obj.is_stale = False
                    streams_to_update.append(obj)
                else:
                    # Always update last_seen, even if nothing else changed
                    obj.last_seen = timezone.now()
                    obj.is_stale = False
                    # Don't update updated_at for unchanged streams
                    streams_to_update.append(obj)

                # Remove from existing_streams since we've processed it
                del existing_streams[stream_hash]
            else:
                stream_props["last_seen"] = timezone.now()
                stream_props["updated_at"] = (
                    timezone.now()
                )  # Set initial updated_at for new streams
                stream_props["is_stale"] = False
                streams_to_create.append(Stream(**stream_props))

        try:
            with transaction.atomic():
                if streams_to_create:
                    Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True)

                if streams_to_update:
                    # Simplified bulk update for better performance
                    Stream.objects.bulk_update(
                        streams_to_update,
                        ['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno'],
                        batch_size=150  # Smaller batch size for XC processing
                    )

                # Update last_seen for any remaining existing streams that weren't processed
                if len(existing_streams.keys()) > 0:
                    Stream.objects.bulk_update(existing_streams.values(), ["last_seen"])
        except Exception as e:
            logger.error(f"Bulk operation failed for XC streams: {str(e)}")

        retval = f"Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."

    except Exception as e:
        logger.error(f"XC category processing error: {str(e)}")
        retval = f"Error processing XC batch: {str(e)}"
    finally:
        # Clean up database connections for threading
        connections.close_all()

    # Aggressive garbage collection
    del streams_to_create, streams_to_update, stream_hashes, existing_streams
    gc.collect()

    return retval


def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
    """Processes a batch of M3U streams using bulk operations with thread-safe DB connections."""
    from django.db import connections

    # Ensure clean database connections for threading
    connections.close_all()

    account = M3UAccount.objects.get(id=account_id)

    compiled_filters = [
        (
            re.compile(
                f.regex_pattern,
                (
                    re.IGNORECASE
                    if (f.custom_properties or {}).get(
                        "case_sensitive", True
                    )
                    == False
                    else 0
                ),
            ),
            f,
        )
        for f in account.filters.order_by("order")
    ]

    streams_to_create = []
    streams_to_update = []
    stream_hashes = {}

    name_max_length = Stream._meta.get_field('name').max_length

    logger.debug(f"Processing batch of {len(batch)} for M3U account {account_id}")
    if compiled_filters:
        logger.debug(f"Using compiled filters: {[f[1].regex_pattern for f in compiled_filters]}")
    for stream_info in batch:
        try:
            name, url = stream_info["name"], stream_info["url"]

            # Validate URL length - maximum of 4096 characters
            if url and len(url) > 4096:
                logger.warning(f"Skipping stream '{name}': URL too long ({len(url)} characters, max 4096)")
                continue

            # Truncate name if it exceeds the model field limit
            if name and len(name) > name_max_length:
                logger.warning(f"Stream name too long ({len(name)} > {name_max_length}), truncating: {name[:80]}...")
                name = name[:name_max_length]

            tvg_id, tvg_logo = get_case_insensitive_attr(
                stream_info["attributes"], "tvg-id", ""
            ), get_case_insensitive_attr(stream_info["attributes"], "tvg-logo", "")
            group_title = get_case_insensitive_attr(
                stream_info["attributes"], "group-title", "Default Group"
            )
            logger.debug(f"Processing stream: {name} - {url} in group {group_title}")
            include = True
            for pattern, filter in compiled_filters:
                logger.trace(f"Checking filter pattern {pattern}")
                target = name
                if filter.filter_type == "url":
                    target = url
                elif filter.filter_type == "group":
                    target = group_title

                if pattern.search(target or ""):
                    logger.debug(
                        f"Stream {name} - {url} matches filter pattern {filter.regex_pattern}"
                    )
                    include = not filter.exclude
                    break

            if not include:
                logger.debug(f"Stream excluded by filter, skipping.")
                continue

            # Filter out disabled groups for this account
            if group_title not in groups:
                logger.debug(
                    f"Skipping stream in disabled or excluded group: {group_title}"
                )
                continue

            # Determine provider-specific fields first
            provider_stream_id = None
            channel_num = None
            account_type_for_hash = None

            if account.account_type == M3UAccount.Types.XC:
                account_type_for_hash = 'XC'
                raw_stream_id = stream_info["attributes"].get("stream_id", "")
                if raw_stream_id:
                    try:
                        provider_stream_id = int(raw_stream_id)
                    except (ValueError, TypeError):
                        pass
                raw_num = stream_info["attributes"].get("num")
                if raw_num is not None:
                    try:
                        channel_num = float(raw_num)
                    except (ValueError, TypeError):
                        pass
            else:
                # For standard M3U accounts, check for tvg-chno or channel-number
                tvg_chno = get_case_insensitive_attr(stream_info["attributes"], "tvg-chno", None)
                if tvg_chno is None:
                    tvg_chno = get_case_insensitive_attr(stream_info["attributes"], "channel-number", None)
                if tvg_chno is not None:
                    try:
                        channel_num = float(tvg_chno)
                    except (ValueError, TypeError):
                        pass

            # Generate hash once with all parameters
            stream_hash = Stream.generate_hash_key(
                name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title,
                account_type=account_type_for_hash, stream_id=provider_stream_id
            )

            stream_props = {
                "name": name,
                "url": url,
                "logo_url": tvg_logo,
                "tvg_id": tvg_id,
                "m3u_account": account,
                "channel_group_id": int(groups.get(group_title)),
                "stream_hash": stream_hash,
                "custom_properties": {**stream_info["attributes"], "vlc_opts": stream_info["vlc_opts"]} if "vlc_opts" in stream_info else stream_info["attributes"],
                "is_adult": parse_is_adult(stream_info["attributes"].get("is_adult", 0)),
                "is_stale": False,
                "stream_id": provider_stream_id,
                "stream_chno": channel_num,
            }

            if stream_hash not in stream_hashes:
                stream_hashes[stream_hash] = stream_props
        except Exception as e:
            logger.error(f"Failed to process stream {name}: {e}")
            logger.error(json.dumps(stream_info))

    existing_streams = {
        s.stream_hash: s
        for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()).select_related('m3u_account').only(
            'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno'
        )
    }

    for stream_hash, stream_props in stream_hashes.items():
        if stream_hash in existing_streams:
            obj = existing_streams[stream_hash]
            # Optimized field comparison
            changed = (
                obj.name != stream_props["name"] or
                obj.url != stream_props["url"] or
                obj.logo_url != stream_props["logo_url"] or
                obj.tvg_id != stream_props["tvg_id"] or
                obj.custom_properties != stream_props["custom_properties"] or
                obj.is_adult != stream_props["is_adult"] or
                obj.stream_id != stream_props["stream_id"] or
                obj.stream_chno != stream_props["stream_chno"]
            )

            # Always update last_seen
            obj.last_seen = timezone.now()

            if changed:
                # Only update fields that changed and set updated_at
                obj.name = stream_props["name"]
                obj.url = stream_props["url"]
                obj.logo_url = stream_props["logo_url"]
                obj.tvg_id = stream_props["tvg_id"]
                obj.custom_properties = stream_props["custom_properties"]
                obj.is_adult = stream_props["is_adult"]
                obj.stream_id = stream_props["stream_id"]
                obj.stream_chno = stream_props["stream_chno"]
                obj.updated_at = timezone.now()

            # Always mark as not stale since we saw it in this refresh
            obj.is_stale = False

            streams_to_update.append(obj)
        else:
            # New stream
            stream_props["last_seen"] = timezone.now()
            stream_props["updated_at"] = timezone.now()
            stream_props["is_stale"] = False
            streams_to_create.append(Stream(**stream_props))

    try:
        with transaction.atomic():
            if streams_to_create:
                Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True)

            if streams_to_update:
                # Update all streams in a single bulk operation
                Stream.objects.bulk_update(
                    streams_to_update,
                    ['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno'],
                    batch_size=200
                )
    except Exception as e:
        logger.error(f"Bulk operation failed: {str(e)}")

    retval = f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."

    # Clean up database connections for threading
    connections.close_all()

    # Free batch data structures (reference-counted deallocation)
    del streams_to_create, streams_to_update, stream_hashes, existing_streams

    return retval


def cleanup_streams(account_id, scan_start_time=timezone.now):
    account = M3UAccount.objects.get(id=account_id, is_active=True)
    existing_groups = ChannelGroup.objects.filter(
        m3u_accounts__m3u_account=account,
        m3u_accounts__enabled=True,
    ).values_list("id", flat=True)
    logger.info(
        f"Found {len(existing_groups)} active groups for M3U account {account_id}"
    )

    # Calculate cutoff date for stale streams
    stale_cutoff = scan_start_time - timezone.timedelta(days=account.stale_stream_days)
    logger.info(
        f"Removing streams not seen since {stale_cutoff} for M3U account {account_id}"
    )

    # Delete streams that are not in active groups
    streams_to_delete = Stream.objects.filter(m3u_account=account).exclude(
        channel_group__in=existing_groups
    )

    # Also delete streams that haven't been seen for longer than stale_stream_days
    stale_streams = Stream.objects.filter(
        m3u_account=account, last_seen__lt=stale_cutoff
    )

    deleted_count = streams_to_delete.count()
    stale_count = stale_streams.count()

    streams_to_delete.delete()
    stale_streams.delete()

    total_deleted = deleted_count + stale_count
    logger.info(
        f"Cleanup for M3U account {account_id} complete: {deleted_count} streams removed due to group filter, {stale_count} removed as stale"
    )

    # Return the total count of deleted streams
    return total_deleted


@shared_task
def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_start_time=None):
    """Refresh M3U groups for an account.

    Args:
        account_id: ID of the M3U account
        use_cache: Whether to use cached M3U file
        full_refresh: Whether this is part of a full refresh
        scan_start_time: Timestamp when the scan started (for consistent last_seen marking)
    """
    if not acquire_task_lock("refresh_m3u_account_groups", account_id):
        return f"Task already running for account_id={account_id}.", None

    lock_renewer = TaskLockRenewer("refresh_m3u_account_groups", account_id)
    lock_renewer.start()

    try:
        account = M3UAccount.objects.get(id=account_id, is_active=True)
    except M3UAccount.DoesNotExist:
        lock_renewer.stop()
        release_task_lock("refresh_m3u_account_groups", account_id)
        return f"M3UAccount with ID={account_id} not found or inactive.", None

    extinf_data = []
    groups = {"Default Group": {}}

    if account.account_type == M3UAccount.Types.XC:
        # Log detailed information about the account
        logger.info(
            f"Processing XC account {account_id} with URL: {account.server_url}"
        )
        logger.debug(
            f"Username: {account.username}, Has password: {'Yes' if account.password else 'No'}"
        )

        # Validate required fields
        if not account.server_url:
            error_msg = "Missing server URL for Xtream Codes account"
            logger.error(error_msg)
            account.status = M3UAccount.Status.ERROR
            account.last_message = error_msg
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account_id, "processing_groups", 100, status="error", error=error_msg
            )
            lock_renewer.stop()
            release_task_lock("refresh_m3u_account_groups", account_id)
            return error_msg, None

        if not account.username or not account.password:
            error_msg = "Missing username or password for Xtream Codes account"
            logger.error(error_msg)
            account.status = M3UAccount.Status.ERROR
            account.last_message = error_msg
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account_id, "processing_groups", 100, status="error", error=error_msg
            )
            lock_renewer.stop()
            release_task_lock("refresh_m3u_account_groups", account_id)
            return error_msg, None

        try:
            # Ensure server URL is properly formatted
            server_url = account.server_url.rstrip("/")
            if not (
                server_url.startswith("http://") or server_url.startswith("https://")
            ):
                server_url = f"http://{server_url}"

            # User agent handling - completely rewritten
            try:
                # Debug the user agent issue
                logger.debug(f"Getting user agent for account {account.id}")

                # Use a hardcoded user agent string to avoid any issues with object structure
                user_agent_string = (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                )

                try:
                    # Try to get the user agent directly from the database
                    if account.user_agent_id:
                        ua_obj = UserAgent.objects.get(id=account.user_agent_id)
                        if (
                            ua_obj
                            and hasattr(ua_obj, "user_agent")
                            and ua_obj.user_agent
                        ):
                            user_agent_string = ua_obj.user_agent
                            logger.debug(
                                f"Using user agent from account: {user_agent_string}"
                            )
                    else:
                        # Get default user agent from CoreSettings
                        default_ua_id = CoreSettings.get_default_user_agent_id()
                        logger.debug(
                            f"Default user agent ID from settings: {default_ua_id}"
                        )
                        if default_ua_id:
                            ua_obj = UserAgent.objects.get(id=default_ua_id)
                            if (
                                ua_obj
                                and hasattr(ua_obj, "user_agent")
                                and ua_obj.user_agent
                            ):
                                user_agent_string = ua_obj.user_agent
                                logger.debug(
                                    f"Using default user agent: {user_agent_string}"
                                )
                except Exception as e:
                    logger.warning(
                        f"Error getting user agent, using fallback: {str(e)}"
                    )

                logger.debug(f"Final user agent string: {user_agent_string}")
            except Exception as e:
                user_agent_string = (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                )
                logger.warning(
                    f"Exception in user agent handling, using fallback: {str(e)}"
                )

            logger.info(
                f"Creating XCClient with URL: {account.server_url}, Username: {account.username}, User-Agent: {user_agent_string}"
            )

            # Create XCClient with explicit error handling
            try:
                with XCClient(
                    account.server_url, account.username, account.password, user_agent_string
                ) as xc_client:
                    logger.info(f"XCClient instance created successfully")

                    # Queue async profile refresh task to run in background
                    # This prevents any delay in the main refresh process
                    try:
                        logger.info(f"Queueing background profile refresh for account {account.name}")
                        refresh_account_profiles.delay(account.id)
                    except Exception as e:
                        logger.warning(f"Failed to queue profile refresh task: {str(e)}")
                        # Don't fail the main refresh if profile refresh can't be queued

                    # Get categories with detailed error handling
                    try:
                        logger.info(f"Getting live categories from XC server")
                        xc_categories = xc_client.get_live_categories()
                        logger.info(
                            f"Found {len(xc_categories)} categories: {xc_categories}"
                        )

                        # Validate response
                        if not isinstance(xc_categories, list):
                            error_msg = (
                                f"Unexpected response from XC server: {xc_categories}"
                            )
                            logger.error(error_msg)
                            account.status = M3UAccount.Status.ERROR
                            account.last_message = error_msg
                            account.save(update_fields=["status", "last_message"])
                            send_m3u_update(
                                account_id,
                                "processing_groups",
                                100,
                                status="error",
                                error=error_msg,
                            )
                            lock_renewer.stop()
                            release_task_lock("refresh_m3u_account_groups", account_id)
                            return error_msg, None

                        if len(xc_categories) == 0:
                            logger.warning("No categories found in XC server response")

                        for category in xc_categories:
                            cat_name = category.get("category_name", "Unknown Category")
                            cat_id = category.get("category_id", "0")
                            logger.info(f"Adding category: {cat_name} (ID: {cat_id})")
                            groups[cat_name] = {
                                "xc_id": cat_id,
                            }
                    except Exception as e:
                        # Determine if this is an authentication error or category retrieval error
                        error_str = str(e).lower()
                        # Check for authentication-related keywords or HTTP status codes commonly used for auth failures
                        is_auth_error = any(keyword in error_str for keyword in [
                            'auth', 'credential', 'login', 'unauthorized', 'forbidden',
                            '401', '403', '512', '513'  # HTTP status codes: 401 Unauthorized, 403 Forbidden, 512-513 (non-standard auth failure)
                        ])

                        if is_auth_error:
                            error_msg = f"Failed to authenticate with XC server: {str(e)}"
                        else:
                            error_msg = f"Failed to get categories from XC server: {str(e)}"

                        logger.error(error_msg)
                        account.status = M3UAccount.Status.ERROR
                        account.last_message = error_msg
                        account.save(update_fields=["status", "last_message"])
                        send_m3u_update(
                            account_id,
                            "processing_groups",
                            100,
                            status="error",
                            error=error_msg,
                        )
                        lock_renewer.stop()
                        release_task_lock("refresh_m3u_account_groups", account_id)
                        return error_msg, None

            except Exception as e:
                error_msg = f"Failed to create XC Client: {str(e)}"
                logger.error(error_msg)
                account.status = M3UAccount.Status.ERROR
                account.last_message = error_msg
                account.save(update_fields=["status", "last_message"])
                send_m3u_update(
                    account_id,
                    "processing_groups",
                    100,
                    status="error",
                    error=error_msg,
                )
                lock_renewer.stop()
                release_task_lock("refresh_m3u_account_groups", account_id)
                return error_msg, None
        except Exception as e:
            error_msg = f"Unexpected error occurred in XC Client: {str(e)}"
            logger.error(error_msg)
            account.status = M3UAccount.Status.ERROR
            account.last_message = error_msg
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account_id, "processing_groups", 100, status="error", error=error_msg
            )
            lock_renewer.stop()
            release_task_lock("refresh_m3u_account_groups", account_id)
            return error_msg, None
    else:
        lines, success = fetch_m3u_lines(account, use_cache)
        if not success:
            # If fetch failed, don't continue processing
            lock_renewer.stop()
            release_task_lock("refresh_m3u_account_groups", account_id)
            return f"Failed to fetch M3U data for account_id={account_id}.", None

        # Log basic file structure for debugging
        logger.debug(f"Processing {len(lines)} lines from M3U file")

        valid_stream_count = 0

        for entry in iter_m3u_entries(lines):
            valid_stream_count += 1
            group_title_attr = get_case_insensitive_attr(entry["attributes"], "group-title", "")
            if group_title_attr and group_title_attr not in groups:
                logger.debug(f"Found new group for M3U account {account_id}: '{group_title_attr}'")
                groups[group_title_attr] = {}
            extinf_data.append(entry)

            if valid_stream_count % 1000 == 0:
                logger.debug(f"Processed {valid_stream_count} valid streams so far for M3U account: {account_id}")

        logger.info(f"M3U parsing complete - Valid streams: {valid_stream_count}")

        # Log group statistics
        logger.info(
            f"Found {len(groups)} groups in M3U file: {', '.join(list(groups.keys())[:20])}"
            + ("..." if len(groups) > 20 else "")
        )

        # Cache processed data
        cache_path = os.path.join(m3u_dir, f"{account_id}.json")
        with open(cache_path, "w", encoding="utf-8") as f:
            json.dump(
                {
                    "extinf_data": extinf_data,
                    "groups": groups,
                },
                f,
            )
            logger.debug(f"Cached parsed M3U data to {cache_path}")

    send_m3u_update(account_id, "processing_groups", 0)

    process_groups(account, groups, scan_start_time)

    lock_renewer.stop()
    release_task_lock("refresh_m3u_account_groups", account_id)

    if not full_refresh:
        # Use update() instead of save() to avoid triggering signals
        M3UAccount.objects.filter(id=account_id).update(
            status=M3UAccount.Status.PENDING_SETUP,
            last_message="M3U groups loaded. Please select groups or refresh M3U to complete setup.",
        )
        send_m3u_update(
            account_id,
            "processing_groups",
            100,
            status="pending_setup",
            message="M3U groups loaded. Please select groups or refresh M3U to complete setup.",
        )

    return extinf_data, groups


def delete_m3u_refresh_task_by_id(account_id):
    """
    Delete the periodic task associated with an M3U account ID.
    Can be called directly or from the post_delete signal.
    Returns True if a task was found and deleted, False otherwise.
    """
    try:
        task = None
        task_name = f"m3u_account-refresh-{account_id}"

        # Look for task by name
        try:
            from django_celery_beat.models import PeriodicTask, IntervalSchedule

            task = PeriodicTask.objects.get(name=task_name)
            logger.debug(f"Found task by name: {task.id} for M3UAccount {account_id}")
        except PeriodicTask.DoesNotExist:
            logger.warning(f"No PeriodicTask found with name {task_name}")
            return False

        # Now delete the task and its interval
        if task:
            # Store interval info before deleting the task
            interval_id = None
            if hasattr(task, "interval") and task.interval:
                interval_id = task.interval.id

                # Count how many TOTAL tasks use this interval (including this one)
                tasks_with_same_interval = PeriodicTask.objects.filter(
                    interval_id=interval_id
                ).count()
                logger.debug(
                    f"Interval {interval_id} is used by {tasks_with_same_interval} tasks total"
                )

            # Delete the task first
            task_id = task.id
            task.delete()
            logger.debug(f"Successfully deleted periodic task {task_id}")

            # Now check if we should delete the interval
            # We only delete if it was the ONLY task using this interval
            if interval_id and tasks_with_same_interval == 1:
                try:
                    interval = IntervalSchedule.objects.get(id=interval_id)
                    logger.debug(
                        f"Deleting interval schedule {interval_id} (not shared with other tasks)"
                    )
                    interval.delete()
                    logger.debug(f"Successfully deleted interval {interval_id}")
                except IntervalSchedule.DoesNotExist:
                    logger.warning(f"Interval {interval_id} no longer exists")
            elif interval_id:
                logger.debug(
                    f"Not deleting interval {interval_id} as it's shared with {tasks_with_same_interval-1} other tasks"
                )

            return True
        return False
    except Exception as e:
        logger.error(
            f"Error deleting periodic task for M3UAccount {account_id}: {str(e)}",
            exc_info=True,
        )
        return False


def _next_available_number(used_numbers, start, end=None):
    """
    Return the smallest integer >= start that is not present in `used_numbers`.

    When `end` is provided (inclusive upper bound for range-constrained groups),
    returns None if the search exceeds that bound instead of running
    indefinitely. The search is O(cluster size) per call against the set;
    maintaining the cursor monotonically across calls keeps the
    "next_available" numbering mode from becoming O(N^2) on large groups.
    """
    n = start
    while n in used_numbers:
        n += 1
        if end is not None and n > end:
            return None
    if end is not None and n > end:
        return None
    return n


def _pick_target_number(
    mode,
    stream,
    used_numbers,
    fixed_cursor,
    fallback_start,
    end_number=None,
    range_start=None,
):
    """
    Return the channel number a given stream should claim under the group's
    numbering mode, or None if the configured range is exhausted.

    Shared by the existing-channel renumber pass and the new-channel create
    pass so both honor identical mode semantics: provider-supplied number
    when available and free, otherwise fall back; or always next-available;
    or fixed-cursor sequential.

    `range_start`, when provided, is the inclusive lower bound for the
    group's configured numbering range. Provider-supplied numbers below
    this bound fall back to the next-available picker so freshly-created
    channels never land outside the configured range.
    """
    if mode == "provider":
        chno = stream.stream_chno
        if (
            chno is not None
            and chno not in used_numbers
            and (range_start is None or chno >= range_start)
            and (end_number is None or chno <= end_number)
        ):
            return chno
        # No usable provider number: walk from fallback_start, bumped up
        # to range_start when set so the fallback never lands below the
        # configured range.
        effective_start = (
            max(fallback_start, range_start)
            if range_start is not None
            else fallback_start
        )
        return _next_available_number(used_numbers, effective_start, end=end_number)
    if mode == "next_available":
        return _next_available_number(used_numbers, 1, end=end_number)
    return _next_available_number(used_numbers, fixed_cursor, end=end_number)


def _custom_properties_as_dict(value):
    """
    Normalize a JSONField-backed custom_properties value into a dict.

    Historical data has rows where the field holds a JSON-encoded string
    instead of a dict. Django's JSONField serializes whatever it gets, so
    `.get()` on one of those rows raises AttributeError and aborts the
    entire sync. Treat string values as JSON to parse, and fall back to an
    empty dict for anything that isn't a dict after parsing.
    """
    import json

    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except (ValueError, TypeError):
            logger.warning(
                "custom_properties stored as non-JSON string; ignoring: %r",
                value[:100],
            )
            return {}
        return parsed if isinstance(parsed, dict) else {}
    return {}


def _classify_sync_failure(exc):
    """
    Map an exception raised during per-stream sync to a coarse typed
    reason used by the completion notification's grouped failure list.
    Keeps the bucket count small so the modal stays readable; the
    underlying exception text is preserved verbatim in ``error``.
    """
    from django.db import IntegrityError

    if isinstance(exc, IntegrityError):
        return "INTEGRITY_ERROR"
    return "OTHER"


@shared_task
def sync_auto_channels(account_id, scan_start_time=None):
    """
    Automatically create/update/delete channels to match streams in groups with auto_channel_sync enabled.
    Preserves existing channel UUIDs to maintain M3U link integrity.
    Called after M3U refresh completes successfully.
    """
    from apps.channels.models import (
        Channel,
        ChannelGroup,
        ChannelGroupM3UAccount,
        Stream,
        ChannelStream,
    )
    from apps.epg.models import EPGData
    from django.utils import timezone

    try:
        account = M3UAccount.objects.get(id=account_id)
        logger.info(f"Starting auto channel sync for M3U account {account.name}")

        # Always use scan_start_time as the cutoff for last_seen
        if scan_start_time is not None:
            if isinstance(scan_start_time, str):
                scan_start_time = timezone.datetime.fromisoformat(scan_start_time)
        else:
            scan_start_time = timezone.now()

        # Get groups with auto sync enabled for this account
        auto_sync_groups = ChannelGroupM3UAccount.objects.filter(
            m3u_account=account, enabled=True, auto_channel_sync=True
        ).select_related("channel_group")

        channels_created = 0
        channels_updated = 0
        channels_deleted = 0
        channels_failed = 0
        # Per-failure context for the completion notification. Each entry
        # carries a typed ``reason`` so the modal can group counts by
        # cause; the cap keeps the WebSocket payload bounded but is sized
        # generously to cover realistic multi-provider failure sets.
        # Full per-stream detail still goes to ``logger.warning`` for
        # power-user diagnostics regardless of the cap.
        failed_stream_details = []
        FAILURE_LOG_LIMIT = 1000

        # Group range reservations (start+end) are advisory and NOT seeded
        # here: two groups with overlapping ranges must cooperate, so only
        # actually-occupied numbers constrain assignment.
        # Hidden auto-created channels stay in the seed because the renumber
        # loop iterates current provider streams (which excludes hidden
        # ones); excluding them here would let sync reclaim their numbers.
        used_numbers = set(
            Channel.objects.exclude(
                auto_created=True,
                auto_created_by=account,
                hidden_from_output=False,
            ).values_list("channel_number", flat=True)
        )
        # Override pins are global reservations: effective_channel_number
        # uses the override, so the picker must treat those numbers as
        # taken or sync can produce duplicate effective channel numbers.
        from apps.channels.models import ChannelOverride

        used_numbers.update(
            ChannelOverride.objects.filter(
                channel_number__isnull=False
            ).values_list("channel_number", flat=True)
        )
        used_numbers.discard(None)

        for group_relation in auto_sync_groups:
            channel_group = group_relation.channel_group
            start_number = group_relation.auto_sync_channel_start or 1.0
            # Optional upper bound; _next_available_number returns None when
            # exhausted, which the per-stream loop converts to a failure.
            end_number = group_relation.auto_sync_channel_end

            # Get force_dummy_epg, group_override, and regex patterns from group custom_properties
            group_custom_props = {}
            force_dummy_epg = False  # Backward compatibility: legacy option to disable EPG
            override_group_id = None
            name_regex_pattern = None
            name_replace_pattern = None
            name_match_regex = None
            name_match_exclude_regex = None
            channel_profile_ids = None
            channel_sort_order = None
            channel_sort_reverse = False
            stream_profile_id = None
            custom_logo_id = None
            custom_epg_id = None  # New option: select specific EPG source (takes priority over force_dummy_epg)
            channel_numbering_mode = "fixed"  # Default mode
            channel_numbering_fallback = 1  # Default fallback for provider mode
            group_custom_props = _custom_properties_as_dict(
                group_relation.custom_properties
            )
            if group_custom_props:
                force_dummy_epg = group_custom_props.get("force_dummy_epg", False)
                override_group_id = group_custom_props.get("group_override")
                name_regex_pattern = group_custom_props.get("name_regex_pattern")
                name_replace_pattern = group_custom_props.get(
                    "name_replace_pattern"
                )
                name_match_regex = group_custom_props.get("name_match_regex")
                name_match_exclude_regex = group_custom_props.get(
                    "name_match_exclude_regex"
                )
                channel_profile_ids = group_custom_props.get("channel_profile_ids")
                custom_epg_id = group_custom_props.get("custom_epg_id")
                channel_sort_order = group_custom_props.get("channel_sort_order")
                channel_sort_reverse = group_custom_props.get(
                    "channel_sort_reverse", False
                )
                stream_profile_id = group_custom_props.get("stream_profile_id")
                custom_logo_id = group_custom_props.get("custom_logo_id")
                channel_numbering_mode = group_custom_props.get("channel_numbering_mode", "fixed")
                channel_numbering_fallback = group_custom_props.get("channel_numbering_fallback", 1)

            # Determine which group to use for created channels
            target_group = channel_group
            if override_group_id:
                try:
                    target_group = ChannelGroup.objects.get(id=override_group_id)
                    logger.info(
                        f"Using override group '{target_group.name}' instead of '{channel_group.name}' for auto-created channels"
                    )
                except ChannelGroup.DoesNotExist:
                    logger.warning(
                        f"Override group with ID {override_group_id} not found, using original group '{channel_group.name}'"
                    )

            logger.info(
                f"Processing auto sync for group: {channel_group.name} (mode: {channel_numbering_mode}, start: {start_number})"
            )

            # Get all current streams in this group for this M3U account, filter out stale streams
            current_streams = Stream.objects.filter(
                m3u_account=account,
                channel_group=channel_group,
                last_seen__gte=scan_start_time,
            )

            # Pre-compile with Python re so an invalid pattern fails here
            # rather than as a Postgres exception inside the per-stream loop.
            if name_match_regex:
                try:
                    re.compile(name_match_regex)
                    current_streams = current_streams.filter(
                        name__iregex=name_match_regex
                    )
                except re.error as e:
                    logger.warning(
                        f"Invalid name_match_regex '{name_match_regex}' for group '{channel_group.name}': {e}. Skipping name filter."
                    )

            # Exclude regex runs after the include filter so the two
            # compose: include narrows, exclude removes from what's left.
            if name_match_exclude_regex:
                try:
                    re.compile(name_match_exclude_regex)
                    current_streams = current_streams.exclude(
                        name__iregex=name_match_exclude_regex
                    )
                except re.error as e:
                    logger.warning(
                        f"Invalid name_match_exclude_regex '{name_match_exclude_regex}' for group '{channel_group.name}': {e}. Skipping exclude filter."
                    )

            # --- APPLY CHANNEL SORT ORDER ---
            streams_is_list = False  # Track if we converted to list
            if channel_sort_order and channel_sort_order != "":
                if channel_sort_order == "name":
                    # Use natural sorting for names to handle numbers correctly
                    current_streams = list(current_streams)
                    current_streams.sort(
                        key=lambda stream: natural_sort_key(stream.name),
                        reverse=channel_sort_reverse,
                    )
                    streams_is_list = True
                elif channel_sort_order == "tvg_id":
                    order_prefix = "-" if channel_sort_reverse else ""
                    current_streams = current_streams.order_by(f"{order_prefix}tvg_id")
                elif channel_sort_order == "updated_at":
                    order_prefix = "-" if channel_sort_reverse else ""
                    current_streams = current_streams.order_by(
                        f"{order_prefix}updated_at"
                    )
                else:
                    logger.warning(
                        f"Unknown channel_sort_order '{channel_sort_order}' for group '{channel_group.name}'. Using provider order."
                    )
                    order_prefix = "-" if channel_sort_reverse else ""
                    current_streams = current_streams.order_by(f"{order_prefix}id")
            else:
                # Provider order (default) - can still be reversed
                order_prefix = "-" if channel_sort_reverse else ""
                current_streams = current_streams.order_by(f"{order_prefix}id")

            # Scoped to this group so the loop below runs in O(group size).
            # Multi-stream channels are deduped by channel_id so every
            # stream_id maps to the same in-memory Channel instance and
            # post-loop bulk_update writes the merged state.
            existing_channel_map = {}
            existing_channels_by_id = {}
            existing_channel_streams = (
                ChannelStream.objects.filter(
                    channel__auto_created=True,
                    channel__auto_created_by=account,
                    stream__m3u_account=account,
                    stream__channel_group=channel_group,
                )
                .select_related("channel")
            )
            for cs in existing_channel_streams:
                if cs.stream_id and cs.channel_id:
                    canonical = existing_channels_by_id.setdefault(
                        cs.channel_id, cs.channel
                    )
                    existing_channel_map[cs.stream_id] = canonical

            # Track which streams we've processed
            processed_stream_ids = set()

            # Check if we have streams - handle both QuerySet and list cases
            has_streams = (
                len(current_streams) > 0
                if streams_is_list
                else current_streams.exists()
            )

            # Bulk pre-fetch collapses N+1 Logo/EPGData lookups into a
            # pair of in_bulk() calls.
            from apps.channels.models import Logo
            from apps.epg.models import EPGSource

            # Resolve the group's custom EPG source once.
            custom_epg_source = None
            custom_dummy_epg_data = None
            if custom_epg_id:
                try:
                    custom_epg_source = EPGSource.objects.get(id=custom_epg_id)
                    if custom_epg_source.source_type == "dummy":
                        custom_dummy_epg_data = (
                            EPGData.objects.filter(
                                epg_source=custom_epg_source
                            ).first()
                        )
                        if not custom_dummy_epg_data:
                            logger.warning(
                                f"No EPGData found for dummy EPG source "
                                f"{custom_epg_source.name} (ID: {custom_epg_id})"
                            )
                except EPGSource.DoesNotExist:
                    logger.warning(
                        f"Custom EPG source with ID {custom_epg_id} not found "
                        f"for group '{channel_group.name}', falling back to "
                        f"auto-match"
                    )

            # Resolve the group's custom logo once.
            custom_logo = None
            if custom_logo_id:
                try:
                    custom_logo = Logo.objects.get(id=custom_logo_id)
                except Logo.DoesNotExist:
                    logger.warning(
                        f"Custom logo with ID {custom_logo_id} not found for "
                        f"group '{channel_group.name}', falling back to stream "
                        f"logos"
                    )

            logo_cache_by_url = {}
            epg_cache_by_tvg_id = {}
            if has_streams:
                # Collect unique URLs / tvg_ids in one DB call each.
                stream_iter = (
                    current_streams
                    if streams_is_list
                    else list(current_streams.values("logo_url", "tvg_id"))
                )
                unique_logo_urls = {
                    s.get("logo_url") if isinstance(s, dict) else getattr(s, "logo_url", None)
                    for s in stream_iter
                }
                unique_logo_urls.discard(None)
                unique_logo_urls.discard("")
                if unique_logo_urls:
                    logo_cache_by_url = {
                        lg.url: lg
                        for lg in Logo.objects.filter(url__in=unique_logo_urls)
                    }

                unique_tvg_ids = {
                    s.get("tvg_id") if isinstance(s, dict) else getattr(s, "tvg_id", None)
                    for s in stream_iter
                }
                unique_tvg_ids.discard(None)
                unique_tvg_ids.discard("")
                # Skip the EPG cache when force_dummy_epg with no
                # custom source override; the resolver always returns None.
                want_epg_cache = unique_tvg_ids and (
                    not force_dummy_epg or custom_epg_id
                )
                if want_epg_cache:
                    # Scope to the group's pinned source so foreign-source
                    # tvg_id matches do not leak in.
                    epg_q = EPGData.objects.filter(tvg_id__in=unique_tvg_ids)
                    if (
                        custom_epg_source is not None
                        and custom_epg_source.source_type != "dummy"
                    ):
                        epg_q = epg_q.filter(epg_source=custom_epg_source)
                    epg_cache_by_tvg_id = {d.tvg_id: d for d in epg_q}

            def _resolve_logo_for_stream(stream):
                """Return a Logo for stream.logo_url, creating it once if needed."""
                url = getattr(stream, "logo_url", None)
                if not url:
                    return None
                cached = logo_cache_by_url.get(url)
                if cached is not None:
                    return cached
                created, _ = Logo.objects.get_or_create(
                    url=url,
                    defaults={"name": stream.name or stream.tvg_id or "Unknown"},
                )
                logo_cache_by_url[url] = created
                return created

            def _resolve_epg_for_stream(stream):
                """Return the EPGData row that should be assigned to this
                stream's channel. Encodes all four group-level EPG modes:

                  1. custom dummy source:           single shared EPGData
                  2. custom non-dummy source:       cache lookup, scoped to
                                                    that source
                  3. force_dummy_epg with no custom: None (clear EPG)
                  4. default auto-match:            cache lookup, any source

                The cache (epg_cache_by_tvg_id) is built once above with the
                correct scope so the per-stream lookup is a dict access.
                """
                if custom_epg_source is not None:
                    if custom_epg_source.source_type == "dummy":
                        return custom_dummy_epg_data
                    tvg_id = getattr(stream, "tvg_id", None)
                    if not tvg_id:
                        return None
                    return epg_cache_by_tvg_id.get(tvg_id)
                if force_dummy_epg:
                    return None
                tvg_id = getattr(stream, "tvg_id", None)
                if not tvg_id:
                    return None
                return epg_cache_by_tvg_id.get(tvg_id)

            if not has_streams:
                logger.debug(f"No streams found in group {channel_group.name}")
                # No streams left in the group: drop the visible auto
                # channels. Hidden channels are preserved so the hide
                # flag survives temporary provider drops (event/PPV).
                channels_to_delete = [
                    ch
                    for ch in existing_channel_map.values()
                    if not ch.hidden_from_output
                ]
                if channels_to_delete:
                    deleted_count = len(channels_to_delete)
                    Channel.objects.filter(
                        id__in=[ch.id for ch in channels_to_delete]
                    ).delete()
                    channels_deleted += deleted_count
                    logger.debug(
                        f"Deleted {deleted_count} auto channels (no streams remaining)"
                    )
                continue

            # Prepare profiles to assign to new channels
            from apps.channels.models import ChannelProfile, ChannelProfileMembership

            if (
                channel_profile_ids
                and isinstance(channel_profile_ids, list)
                and len(channel_profile_ids) > 0
            ):
                # Convert all to int (in case they're strings)
                try:
                    profile_ids = [int(pid) for pid in channel_profile_ids]
                except Exception:
                    profile_ids = []
                profiles_to_assign = list(
                    ChannelProfile.objects.filter(id__in=profile_ids)
                )
            else:
                profiles_to_assign = list(ChannelProfile.objects.all())

            # Get stream profile to assign if specified
            from core.models import StreamProfile
            stream_profile_to_assign = None
            if stream_profile_id:
                try:
                    stream_profile_to_assign = StreamProfile.objects.get(id=int(stream_profile_id))
                    logger.info(
                        f"Will assign stream profile '{stream_profile_to_assign.name}' to auto-synced streams in group '{channel_group.name}'"
                    )
                except (StreamProfile.DoesNotExist, ValueError, TypeError):
                    logger.warning(
                        f"Stream profile with ID {stream_profile_id} not found for group '{channel_group.name}', streams will use default profile"
                    )
                    stream_profile_to_assign = None

            current_channel_number = start_number

            # Renumber existing channels to match sort order. Compact
            # mode skips this; the end-of-iteration pack is the source
            # of truth and would overwrite the renumber.
            compact_mode = bool(group_custom_props.get("compact_numbering"))
            channels_to_renumber = []
            temp_channel_number = start_number

            for stream in current_streams if not compact_mode else []:
                if stream.id in existing_channel_map:
                    channel = existing_channel_map[stream.id]

                    target_number = _pick_target_number(
                        channel_numbering_mode,
                        stream,
                        used_numbers,
                        temp_channel_number,
                        channel_numbering_fallback,
                        end_number=end_number,
                        range_start=start_number,
                    )

                    # Range exhausted: leave the channel at its existing
                    # number. The renumber pass is sort-optimization only;
                    # no failure record needed.
                    if target_number is None:
                        # Preserve the channel's current number in used_numbers
                        if channel.channel_number is not None:
                            used_numbers.add(channel.channel_number)
                        continue

                    # Add this number to used_numbers so we don't reuse it in this batch
                    used_numbers.add(target_number)

                    if channel.channel_number != target_number:
                        channel.channel_number = target_number
                        channels_to_renumber.append(channel)
                        logger.debug(
                            f"Will renumber channel '{channel.name}' to {target_number}"
                        )

                    # Only increment temp_channel_number in fixed mode
                    if channel_numbering_mode == "fixed":
                        temp_channel_number += 1.0
                        if temp_channel_number % 1 != 0:  # Has decimal
                            temp_channel_number = int(temp_channel_number) + 1.0

            # Bulk update channel numbers if any need renumbering
            if channels_to_renumber:
                Channel.objects.bulk_update(
                    channels_to_renumber, ["channel_number"], batch_size=500
                )
                logger.info(
                    f"Renumbered {len(channels_to_renumber)} channels to maintain sort order"
                )

            # When the group's configured range is narrower than its existing
            # channels span, any non-hidden auto-created channel whose number
            # falls outside [start, end] gets deleted. The new-channel
            # creation loop below picks up the freed stream and re-creates
            # the channel at a slot inside the new range, so the net user
            # outcome is a renumber, not a failure. Counted in
            # channels_deleted; the replacement counts in channels_created.
            # Hidden channels are preserved. Runs BEFORE new-channel creation
            # so slots freed by the deletions are available to incoming
            # streams.
            if end_number is not None:
                overflow_delete_ids = []
                for stream_id, ch in list(existing_channel_map.items()):
                    if ch.hidden_from_output:
                        continue
                    num = ch.channel_number
                    if num is None:
                        continue
                    if num < start_number or num > end_number:
                        overflow_delete_ids.append(ch.id)
                        existing_channel_map.pop(stream_id, None)
                        used_numbers.discard(num)
                if overflow_delete_ids:
                    deleted = Channel.objects.filter(
                        id__in=overflow_delete_ids
                    ).delete()
                    removed_count = (
                        deleted[1].get("dispatcharr_channels.Channel", 0)
                        if isinstance(deleted, tuple) and len(deleted) > 1
                        else len(overflow_delete_ids)
                    )
                    channels_deleted += removed_count
                    logger.info(
                        f"Deleted {removed_count} channels outside the "
                        f"range {int(start_number)}-{int(end_number)} for "
                        f"group '{channel_group.name}'"
                    )

            # Reset channel number counter for processing new channels
            current_channel_number = start_number

            # Per-channel changes are buffered and bulk_update'd once after the
            # loop. update_fields is set explicitly so post_save signals only
            # fire for receivers whose tracked field actually changed.
            existing_dirty_channels = []
            existing_dirty_ids = set()
            existing_dirty_field_set = set()
            # Subset of channels whose epg_data actually changed in this
            # pass. Used by the dispatcher below to fire the EPG parse
            # task only for those, not for every channel in
            # existing_dirty_channels.
            epg_dirty_channel_ids = set()

            # New channels are buffered and bulk_create'd after the loop.
            # bulk_create skips post_save, so the EPG parse task is dispatched
            # once per unique epg_data_id below rather than per channel.
            # Pairs are (Channel(), Stream) so the post-loop step can attach
            # ChannelStream rows using the IDs Postgres returns.
            new_channels_pending = []

            for stream in current_streams:
                processed_stream_ids.add(stream.id)
                try:
                    # Parse custom properties for additional info
                    stream_custom_props = stream.custom_properties or {}
                    tvc_guide_stationid = stream_custom_props.get("tvc-guide-stationid")

                    # --- REGEX FIND/REPLACE LOGIC ---
                    original_name = stream.name
                    new_name = original_name
                    if name_regex_pattern is not None:
                        # If replace is None, treat as empty string (remove match)
                        replace = (
                            name_replace_pattern
                            if name_replace_pattern is not None
                            else ""
                        )
                        try:
                            # Convert $1, $2, etc. to \1, \2, etc. for consistency with M3U profiles
                            safe_replace_pattern = re.sub(r'\$(\d+)', r'\\\1', replace)
                            new_name = re.sub(
                                name_regex_pattern, safe_replace_pattern, original_name
                            )
                        except re.error as e:
                            logger.warning(
                                f"Regex error for group '{channel_group.name}': {e}. Using original name."
                            )
                            new_name = original_name

                    # Check if we already have a channel for this stream
                    existing_channel = existing_channel_map.get(stream.id)

                    if existing_channel:
                        # Track only the fields that actually changed, so the
                        # eventual UPDATE writes one column per change instead
                        # of every column on every channel. The dirty list is
                        # accumulated and bulk_update'd after the loop -
                        # which avoids issuing an UPDATE per channel and
                        # avoids firing the EPG post_save signal on saves
                        # that didn't touch epg_data.
                        dirty_fields = []

                        if existing_channel.name != new_name:
                            existing_channel.name = new_name
                            dirty_fields.append("name")

                        if existing_channel.tvg_id != stream.tvg_id:
                            existing_channel.tvg_id = stream.tvg_id
                            dirty_fields.append("tvg_id")

                        if existing_channel.tvc_guide_stationid != tvc_guide_stationid:
                            existing_channel.tvc_guide_stationid = tvc_guide_stationid
                            dirty_fields.append("tvc_guide_stationid")

                        # The group override may direct sync to a different
                        # ChannelGroup than the one currently on the row.
                        if existing_channel.channel_group_id != target_group.id:
                            existing_channel.channel_group = target_group
                            dirty_fields.append("channel_group")

                        # Logo: custom group setting wins; otherwise stream logo
                        current_logo = (
                            custom_logo
                            if custom_logo_id and custom_logo is not None
                            else _resolve_logo_for_stream(stream)
                        )
                        current_logo_id = current_logo.id if current_logo else None
                        if existing_channel.logo_id != current_logo_id:
                            existing_channel.logo = current_logo
                            dirty_fields.append("logo")

                        # EPG: handled centrally by _resolve_epg_for_stream
                        current_epg_data = _resolve_epg_for_stream(stream)
                        current_epg_id = (
                            current_epg_data.id if current_epg_data else None
                        )
                        if existing_channel.epg_data_id != current_epg_id:
                            existing_channel.epg_data = current_epg_data
                            dirty_fields.append("epg_data")
                            if current_epg_id is not None:
                                epg_dirty_channel_ids.add(existing_channel.id)

                        # Stream profile: only set if group has one configured
                        if (
                            stream_profile_to_assign is not None
                            and existing_channel.stream_profile_id
                            != stream_profile_to_assign.id
                        ):
                            existing_channel.stream_profile = stream_profile_to_assign
                            dirty_fields.append("stream_profile")

                        if dirty_fields:
                            # Multi-stream channels appear once per stream;
                            # dedupe by id so bulk_update does not double-fire
                            # and channels_updated does not double-count.
                            if existing_channel.id not in existing_dirty_ids:
                                existing_dirty_channels.append(existing_channel)
                                existing_dirty_ids.add(existing_channel.id)
                                channels_updated += 1
                            existing_dirty_field_set.update(dirty_fields)

                    else:
                        # Range exhaustion is surfaced to the user via the
                        # completion notification, not swallowed.
                        target_number = _pick_target_number(
                            channel_numbering_mode,
                            stream,
                            used_numbers,
                            current_channel_number,
                            channel_numbering_fallback,
                            end_number=end_number,
                            range_start=start_number,
                        )

                        if target_number is None:
                            channels_failed += 1
                            if len(failed_stream_details) < FAILURE_LOG_LIMIT:
                                failed_stream_details.append({
                                    "stream_name": stream.name,
                                    "stream_id": stream.id,
                                    "group": channel_group.name,
                                    "reason": "RANGE_EXHAUSTED",
                                    "error": (
                                        f"Channel number range "
                                        f"{int(start_number)}-{int(end_number)} is full"
                                    ),
                                })
                            processed_stream_ids.add(stream.id)
                            continue

                        # Add this number to used_numbers
                        used_numbers.add(target_number)

                        # Resolve every FK BEFORE the create call so the
                        # initial INSERT carries the complete row.
                        new_logo = (
                            custom_logo
                            if custom_logo_id and custom_logo is not None
                            else _resolve_logo_for_stream(stream)
                        )
                        new_epg_data = _resolve_epg_for_stream(stream)

                        new_channels_pending.append(
                            (
                                Channel(
                                    channel_number=target_number,
                                    name=new_name,
                                    tvg_id=stream.tvg_id,
                                    tvc_guide_stationid=tvc_guide_stationid,
                                    channel_group=target_group,
                                    user_level=0,
                                    auto_created=True,
                                    auto_created_by=account,
                                    logo=new_logo,
                                    epg_data=new_epg_data,
                                    stream_profile=stream_profile_to_assign,
                                ),
                                stream,
                            )
                        )

                    # Increment channel number for next iteration (only in fixed mode)
                    if channel_numbering_mode == "fixed":
                        current_channel_number += 1.0
                        if current_channel_number % 1 != 0:  # Has decimal
                            current_channel_number = int(current_channel_number) + 1.0

                except Exception as e:
                    logger.error(
                        f"Error processing auto channel for stream {stream.name}: {str(e)}"
                    )
                    channels_failed += 1
                    if len(failed_stream_details) < FAILURE_LOG_LIMIT:
                        failed_stream_details.append({
                            "stream_name": stream.name,
                            "stream_id": stream.id,
                            "group": channel_group.name,
                            "reason": _classify_sync_failure(e),
                            "error": str(e),
                        })
                    continue

            # Bulk-create channels, then dependent rows using the IDs
            # Postgres returns. bulk_create skips post_save, so the EPG
            # parse task is dispatched explicitly per-epg_data_id below
            # to avoid flooding Celery at scale.
            if new_channels_pending:
                channel_objs = [pair[0] for pair in new_channels_pending]
                streams_for_new = [pair[1] for pair in new_channels_pending]
                Channel.objects.bulk_create(channel_objs, batch_size=500)

                ChannelStream.objects.bulk_create(
                    [
                        ChannelStream(
                            channel_id=channel_objs[i].id,
                            stream_id=streams_for_new[i].id,
                            order=0,
                        )
                        for i in range(len(channel_objs))
                    ],
                    batch_size=500,
                )

                if profiles_to_assign:
                    ChannelProfileMembership.objects.bulk_create(
                        [
                            ChannelProfileMembership(
                                channel_id=ch.id,
                                channel_profile_id=profile.id,
                                enabled=True,
                            )
                            for ch in channel_objs
                            for profile in profiles_to_assign
                        ],
                        ignore_conflicts=True,
                        batch_size=500,
                    )

                channels_created += len(channel_objs)

                # One EPG parse task per unique EPGData replaces the
                # per-channel post_save dispatch bypassed by bulk_create.
                from apps.epg.tasks import parse_programs_for_tvg_id

                unique_epg_ids = {
                    ch.epg_data_id for ch in channel_objs if ch.epg_data_id
                }
                for epg_id in unique_epg_ids:
                    parse_programs_for_tvg_id.delay(epg_id)

                logger.debug(
                    f"Bulk created {len(channel_objs)} channels in group "
                    f"'{channel_group.name}'; dispatched "
                    f"{len(unique_epg_ids)} unique EPG parse task(s)"
                )

            # bulk_update writes only the columns named in `fields` and
            # bypasses post_save, so the EPG refresh signal cannot fire here.
            # Dispatch one parse task per unique EPGData id when epg_data was
            # in the dirty set, mirroring the new-channel path above.
            if existing_dirty_channels:
                Channel.objects.bulk_update(
                    existing_dirty_channels,
                    fields=list(existing_dirty_field_set),
                    batch_size=500,
                )
                if epg_dirty_channel_ids:
                    from apps.epg.tasks import parse_programs_for_tvg_id

                    # Dispatch only for channels whose epg_data_id changed.
                    # Other dirty channels would queue redundant parses.
                    unique_epg_ids = {
                        ch.epg_data_id
                        for ch in existing_dirty_channels
                        if ch.id in epg_dirty_channel_ids and ch.epg_data_id
                    }
                    for epg_id in unique_epg_ids:
                        parse_programs_for_tvg_id.delay(epg_id)
                logger.debug(
                    f"Bulk updated {len(existing_dirty_channels)} existing "
                    f"channels (fields: {sorted(existing_dirty_field_set)})"
                )

            # Reconcile ChannelProfileMembership in two writes: one
            # bulk_update for enable-flips, one bulk_create for missing
            # rows. Avoids a per-channel save loop.
            existing_channel_ids = [
                c.id for c in existing_channel_map.values()
            ]
            target_profile_ids = {p.id for p in profiles_to_assign}
            if existing_channel_ids:
                membership_rows = list(
                    ChannelProfileMembership.objects.filter(
                        channel_id__in=existing_channel_ids
                    ).only("id", "channel_id", "channel_profile_id", "enabled")
                )
                memberships_by_channel = {}
                for m in membership_rows:
                    memberships_by_channel.setdefault(m.channel_id, []).append(m)

                rows_to_flip = []
                rows_to_create = []
                for ch_id in existing_channel_ids:
                    rows = memberships_by_channel.get(ch_id, [])
                    have_for_target = set()
                    for m in rows:
                        if m.channel_profile_id in target_profile_ids:
                            have_for_target.add(m.channel_profile_id)
                            if not m.enabled:
                                m.enabled = True
                                rows_to_flip.append(m)
                        else:
                            if m.enabled:
                                m.enabled = False
                                rows_to_flip.append(m)
                    missing = target_profile_ids - have_for_target
                    for pid in missing:
                        rows_to_create.append(
                            ChannelProfileMembership(
                                channel_id=ch_id,
                                channel_profile_id=pid,
                                enabled=True,
                            )
                        )

                if rows_to_flip:
                    ChannelProfileMembership.objects.bulk_update(
                        rows_to_flip, ["enabled"], batch_size=500
                    )
                if rows_to_create:
                    ChannelProfileMembership.objects.bulk_create(
                        rows_to_create, ignore_conflicts=True, batch_size=500
                    )
                if rows_to_flip or rows_to_create:
                    logger.debug(
                        f"Reconciled memberships for "
                        f"{len(existing_channel_ids)} channels "
                        f"({len(rows_to_flip)} flipped, "
                        f"{len(rows_to_create)} created)"
                    )

            # Delete channels whose streams have all disappeared.
            # Hidden channels are preserved so event/PPV holds across
            # provider drops.
            channel_streams_in_group = {}
            for stream_id, channel in existing_channel_map.items():
                channel_streams_in_group.setdefault(channel.id, []).append(
                    (stream_id, channel)
                )
            channels_to_delete = []
            for ch_id, pairs in channel_streams_in_group.items():
                channel = pairs[0][1]
                if channel.hidden_from_output:
                    continue
                stream_ids = {sid for sid, _ in pairs}
                if not (stream_ids & processed_stream_ids):
                    channels_to_delete.append(channel)

            if channels_to_delete:
                deleted_count = len(channels_to_delete)
                Channel.objects.filter(
                    id__in=[ch.id for ch in channels_to_delete]
                ).delete()
                channels_deleted += deleted_count
                logger.debug(
                    f"Deleted {deleted_count} auto channels for removed streams"
                )

            # Compact-mode pack: hidden channels release their number and
            # visible channels pack contiguously into [start, end]. Runs
            # after create/update/delete so the channel set is stable.
            if compact_mode:
                from apps.channels.compact_numbering import repack_group

                pack_result = repack_group(group_relation)
                if pack_result["failed"]:
                    channels_failed += pack_result["failed"]
                    if (
                        len(failed_stream_details) < FAILURE_LOG_LIMIT
                    ):
                        failed_stream_details.append(
                            {
                                "stream_name": None,
                                "stream_id": None,
                                "group": channel_group.name,
                                "reason": "RANGE_EXHAUSTED",
                                "error": (
                                    f"Compact pack: {pack_result['failed']} "
                                    f"visible channel(s) could not fit in range "
                                    f"{int(start_number)}"
                                    + (
                                        f"-{int(end_number)}"
                                        if end_number
                                        else "+"
                                    )
                                ),
                            }
                        )
                logger.debug(
                    f"Compact pack for group '{channel_group.name}': "
                    f"{pack_result['assigned']} assigned, "
                    f"{pack_result['released']} released, "
                    f"{pack_result['failed']} failed"
                )

        # Cleanup mode read from account.custom_properties.orphan_channel_cleanup:
        # "always" (default; key absent) removes every orphan auto channel;
        # "preserve_customized" keeps those with a ChannelOverride row;
        # "never" disables cleanup. Hidden channels are preserved across all
        # modes so event/PPV channels that come and go are not silently lost.
        cleanup_mode = (account.custom_properties or {}).get(
            "orphan_channel_cleanup", "always"
        )
        if cleanup_mode != "never":
            orphaned_channels = Channel.objects.filter(
                auto_created=True,
                auto_created_by=account,
                hidden_from_output=False,
            ).exclude(
                id__in=ChannelStream.objects.filter(
                    stream__m3u_account=account,
                    stream__isnull=False,
                ).values_list("channel_id", flat=True)
            )
            if cleanup_mode == "preserve_customized":
                orphaned_channels = orphaned_channels.filter(override__isnull=True)

            _, per_model = orphaned_channels.delete()
            deleted_channels = per_model.get("dispatcharr_channels.Channel", 0)
            if deleted_channels:
                channels_deleted += deleted_channels
                logger.info(
                    f"Deleted {deleted_channels} orphaned auto channels with no valid streams (mode={cleanup_mode})"
                )

        logger.info(
            f"Auto channel sync complete for account {account.name}: "
            f"{channels_created} created, {channels_updated} updated, "
            f"{channels_deleted} deleted, {channels_failed} failed"
        )
        return {
            "status": "ok",
            "channels_created": channels_created,
            "channels_updated": channels_updated,
            "channels_deleted": channels_deleted,
            "channels_failed": channels_failed,
            "failed_stream_details": failed_stream_details,
        }

    except Exception as e:
        logger.error(f"Error in auto channel sync for account {account_id}: {str(e)}")
        return {
            "status": "error",
            "error": str(e),
            "channels_created": 0,
            "channels_updated": 0,
            "channels_deleted": 0,
            "channels_failed": 0,
            "failed_stream_details": [],
        }


def get_transformed_credentials(account, profile=None):
    """
    Get transformed credentials for XtreamCodes API calls.

    Args:
        account: M3UAccount instance
        profile: M3UAccountProfile instance (optional, if not provided will use primary profile)

    Returns:
        tuple: (transformed_url, transformed_username, transformed_password)
    """
    import re
    import urllib.parse

    # If no profile is provided, find the primary active profile
    if profile is None:
        try:
            from apps.m3u.models import M3UAccountProfile
            profile = M3UAccountProfile.objects.filter(
                m3u_account=account,
                is_active=True
            ).first()
            if profile:
                logger.debug(f"Using primary profile '{profile.name}' for URL transformation")
            else:
                logger.debug(f"No active profiles found for account {account.name}, using base credentials")
        except Exception as e:
            logger.warning(f"Could not get primary profile for account {account.name}: {e}")
            profile = None

    base_url = account.server_url
    base_username = account.username
    base_password = account.password    # Build a complete URL with credentials (similar to how IPTV URLs are structured)
    # Format: http://server.com:port/live/username/password/1234.ts
    if base_url and base_username and base_password:
        # Remove trailing slash from server URL if present
        clean_server_url = base_url.rstrip('/')

        # Build the complete URL with embedded credentials
        complete_url = f"{clean_server_url}/live/{base_username}/{base_password}/1234.ts"
        logger.debug(f"Built complete URL: {complete_url}")

        # Apply profile-specific transformations if profile is provided
        if profile and profile.search_pattern and profile.replace_pattern:
            try:
                # Handle backreferences: convert JS-style $<name> -> \g<name>, $1 -> \1
                # regex module accepts JS-style (?<name>...) named groups natively
                safe_replace_pattern = regex.sub(r'\$<([^>]+)>', r'\\g<\1>', profile.replace_pattern)
                safe_replace_pattern = regex.sub(r'\$(\d+)', r'\\\1', safe_replace_pattern)

                # Apply transformation to the complete URL
                transformed_complete_url = regex.sub(profile.search_pattern, safe_replace_pattern, complete_url)
                logger.info(f"Transformed complete URL: {complete_url} -> {transformed_complete_url}")

                # Extract components from the transformed URL
                # Pattern: http://server.com:port/live/username/password/1234.ts
                parsed_url = urllib.parse.urlparse(transformed_complete_url)
                path_parts = [part for part in parsed_url.path.split('/') if part]

                if len(path_parts) >= 4 and path_parts[-1] == '1234.ts':
                    # Extract username and password from the known structure:
                    # .../{live}/{username}/{password}/1234.ts
                    # Using negative indices so sub-paths in the server URL don't shift extraction
                    transformed_username = path_parts[-3]
                    transformed_password = path_parts[-2]

                    # Rebuild server URL without the username/password path
                    transformed_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
                    if parsed_url.port:
                        transformed_url = f"{parsed_url.scheme}://{parsed_url.hostname}:{parsed_url.port}"

                    logger.debug(f"Extracted transformed credentials:")
                    logger.debug(f"  Server URL: {transformed_url}")
                    logger.debug(f"  Username: {transformed_username}")
                    logger.debug(f"  Password: {transformed_password}")

                    return transformed_url, transformed_username, transformed_password
                else:
                    logger.warning(f"Could not extract credentials from transformed URL: {transformed_complete_url}")
                    return base_url, base_username, base_password

            except Exception as e:
                logger.error(f"Error transforming URL for profile {profile.name if profile else 'unknown'}: {e}")
                return base_url, base_username, base_password
        else:
            # No profile or no transformation patterns
            return base_url, base_username, base_password
    else:
        logger.warning(f"Missing credentials for account {account.name}")
        return base_url, base_username, base_password


@shared_task
def refresh_account_profiles(account_id):
    """Refresh account information for all active profiles of an XC account.

    This task runs asynchronously in the background after account refresh completes.
    It includes rate limiting delays between profile authentications to prevent provider bans.
    """
    from django.conf import settings
    import time

    try:
        account = M3UAccount.objects.get(id=account_id, is_active=True)

        if account.account_type != M3UAccount.Types.XC:
            logger.debug(f"Account {account_id} is not XC type, skipping profile refresh")
            return f"Account {account_id} is not an XtreamCodes account"

        from apps.m3u.models import M3UAccountProfile

        profiles = M3UAccountProfile.objects.filter(
            m3u_account=account,
            is_active=True
        )

        if not profiles.exists():
            logger.info(f"No active profiles found for account {account.name}")
            return f"No active profiles for account {account_id}"

        # Get user agent for this account
        try:
            user_agent_string = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            if account.user_agent_id:
                from core.models import UserAgent
                ua_obj = UserAgent.objects.get(id=account.user_agent_id)
                if ua_obj and hasattr(ua_obj, "user_agent") and ua_obj.user_agent:
                    user_agent_string = ua_obj.user_agent
        except Exception as e:
            logger.warning(f"Error getting user agent, using fallback: {str(e)}")
        logger.debug(f"Using user agent for profile refresh: {user_agent_string}")
        # Get rate limiting delay from settings
        profile_delay = getattr(settings, 'XC_PROFILE_REFRESH_DELAY', 2.5)

        profiles_updated = 0
        profiles_failed = 0

        logger.info(f"Starting background refresh for {profiles.count()} profiles of account {account.name}")

        for idx, profile in enumerate(profiles):
            try:
                # Add delay between profiles to prevent rate limiting (except for first profile)
                if idx > 0:
                    logger.info(f"Waiting {profile_delay}s before refreshing next profile to avoid rate limiting")
                    time.sleep(profile_delay)

                # Get transformed credentials for this specific profile
                profile_url, profile_username, profile_password = get_transformed_credentials(account, profile)

                # Create a separate XC client for this profile's credentials
                with XCClient(
                    profile_url,
                    profile_username,
                    profile_password,
                    user_agent_string
                ) as profile_client:
                    # Authenticate with this profile's credentials
                    if profile_client.authenticate():
                        # Get account information specific to this profile's credentials
                        profile_account_info = profile_client.get_account_info()

                        # Merge with existing custom_properties if they exist
                        existing_props = profile.custom_properties or {}
                        existing_props.update(profile_account_info)
                        profile.custom_properties = existing_props
                        profile.save(update_fields=['custom_properties'])

                        profiles_updated += 1
                        logger.info(f"Updated account information for profile '{profile.name}' ({profiles_updated}/{profiles.count()})")
                    else:
                        profiles_failed += 1
                        logger.warning(f"Failed to authenticate profile '{profile.name}' with transformed credentials")

            except Exception as profile_error:
                profiles_failed += 1
                logger.error(f"Failed to update account information for profile '{profile.name}': {str(profile_error)}")
                # Continue with other profiles even if one fails

        result_msg = f"Profile refresh complete for account {account.name}: {profiles_updated} updated, {profiles_failed} failed"
        logger.info(result_msg)
        return result_msg

    except M3UAccount.DoesNotExist:
        error_msg = f"Account {account_id} not found"
        logger.error(error_msg)
        return error_msg
    except Exception as e:
        error_msg = f"Error refreshing profiles for account {account_id}: {str(e)}"
        logger.error(error_msg)
        return error_msg


@shared_task
def refresh_account_info(profile_id):
    """Refresh only the account information for a specific M3U profile."""
    if not acquire_task_lock("refresh_account_info", profile_id):
        return f"Account info refresh task already running for profile_id={profile_id}."

    try:
        from apps.m3u.models import M3UAccountProfile
        import re

        profile = M3UAccountProfile.objects.get(id=profile_id)
        account = profile.m3u_account

        if account.account_type != M3UAccount.Types.XC:
            release_task_lock("refresh_account_info", profile_id)
            return f"Profile {profile_id} belongs to account {account.id} which is not an XtreamCodes account."

        # Get transformed credentials using the helper function
        transformed_url, transformed_username, transformed_password = get_transformed_credentials(account, profile)

        # Initialize XtreamCodes client with extracted/transformed credentials
        client = XCClient(
            transformed_url,
            transformed_username,
            transformed_password,
            account.get_user_agent(),
        )        # Authenticate and get account info
        auth_result = client.authenticate()
        if not auth_result:
            error_msg = f"Authentication failed for profile {profile.name} ({profile_id})"
            logger.error(error_msg)

            # Send error notification to frontend via websocket
            send_websocket_update(
                "updates",
                "update",
                {
                    "type": "account_info_refresh_error",
                    "profile_id": profile_id,
                    "profile_name": profile.name,
                    "error": "Authentication failed with the provided credentials",
                    "message": f"Failed to authenticate profile '{profile.name}'. Please check the credentials."
                }
            )

            release_task_lock("refresh_account_info", profile_id)
            return error_msg

        # Get account information
        account_info = client.get_account_info()

        # Update only this specific profile with the new account info
        if not profile.custom_properties:
            profile.custom_properties = {}
        profile.custom_properties.update(account_info)
        profile.save()

        # Send success notification to frontend via websocket
        send_websocket_update(
            "updates",
            "update",
            {
                "type": "account_info_refresh_success",
                "profile_id": profile_id,
                "profile_name": profile.name,
                "message": f"Account information successfully refreshed for profile '{profile.name}'"
            }
        )

        release_task_lock("refresh_account_info", profile_id)
        return f"Account info refresh completed for profile {profile_id} ({profile.name})."

    except M3UAccountProfile.DoesNotExist:
        error_msg = f"Profile {profile_id} not found"
        logger.error(error_msg)

        send_websocket_update(
            "updates",
            "update",
            {
                "type": "account_refresh_error",
                "profile_id": profile_id,
                "error": "Profile not found",
                "message": f"Profile {profile_id} not found"
            }
        )

        release_task_lock("refresh_account_info", profile_id)
        return error_msg
    except Exception as e:
        error_msg = f"Error refreshing account info for profile {profile_id}: {str(e)}"
        logger.error(error_msg)

        send_websocket_update(
            "updates",
            "update",
            {
                "type": "account_refresh_error",
                "profile_id": profile_id,
                "error": str(e),
                "message": f"Failed to refresh account info: {str(e)}"
            }
        )

        release_task_lock("refresh_account_info", profile_id)
        return error_msg
@shared_task(time_limit=3600, soft_time_limit=3500)
def refresh_single_m3u_account(account_id):
    """Splits M3U processing into chunks and dispatches them as parallel tasks."""
    if not acquire_task_lock("refresh_single_m3u_account", account_id):
        return f"Task already running for account_id={account_id}."

    # Keep the lock alive while this long-running task is working.
    # Without renewal, the 300s lock TTL can expire during large
    # downloads/parses, allowing duplicate tasks to start.
    lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id)
    lock_renewer.start()

    try:
        return _refresh_single_m3u_account_impl(account_id)
    finally:
        # Guaranteed cleanup on all exit paths (success, exception, early return)
        lock_renewer.stop()
        release_task_lock("refresh_single_m3u_account", account_id)


def _refresh_single_m3u_account_impl(account_id):
    """Implementation of M3U account refresh with guaranteed memory cleanup."""
    # Record start time
    refresh_start_timestamp = timezone.now()  # For the cleanup function
    start_time = time.time()  # For tracking elapsed time as float
    streams_created = 0
    streams_updated = 0
    streams_deleted = 0

    try:
        account = M3UAccount.objects.get(id=account_id, is_active=True)
        if not account.is_active:
            logger.debug(f"Account {account_id} is not active, skipping.")
            return

        # Set status to fetching
        account.status = M3UAccount.Status.FETCHING
        account.save(update_fields=['status'])

        filters = list(account.filters.all())

        # Check if VOD is enabled for this account
        vod_enabled = False
        if account.custom_properties:
            custom_props = account.custom_properties or {}
            vod_enabled = custom_props.get('enable_vod', False)

    except M3UAccount.DoesNotExist:
        # The M3U account doesn't exist, so delete the periodic task if it exists
        logger.warning(
            f"M3U account with ID {account_id} not found, but task was triggered. Cleaning up orphaned task."
        )

        # Call the helper function to delete the task
        if delete_m3u_refresh_task_by_id(account_id):
            logger.info(
                f"Successfully cleaned up orphaned task for M3U account {account_id}"
            )
        else:
            logger.debug(f"No orphaned task found for M3U account {account_id}")

        return f"M3UAccount with ID={account_id} not found or inactive, task cleaned up"

    # Fetch M3U lines and handle potential issues
    extinf_data = []
    groups = None

    cache_path = os.path.join(m3u_dir, f"{account_id}.json")
    if os.path.exists(cache_path):
        try:
            with open(cache_path, "r") as file:
                data = json.load(file)

            extinf_data = data["extinf_data"]
            groups = data["groups"]
            del data  # Free top-level dict; extinf_data/groups retain their references
        except json.JSONDecodeError as e:
            # Handle corrupted JSON file
            logger.error(
                f"Error parsing cached M3U data for account {account_id}: {str(e)}"
            )

            # Backup the corrupted file for potential analysis
            backup_path = f"{cache_path}.corrupted"
            try:
                os.rename(cache_path, backup_path)
                logger.info(f"Renamed corrupted cache file to {backup_path}")
            except OSError as rename_err:
                logger.warning(
                    f"Failed to rename corrupted cache file: {str(rename_err)}"
                )

            # Reset the data to empty structures
            extinf_data = []
            groups = None
        except Exception as e:
            logger.error(f"Unexpected error reading cached M3U data: {str(e)}")
            extinf_data = []
            groups = None

    if not extinf_data:
        try:
            logger.info(f"Calling refresh_m3u_groups for account {account_id}")
            result = refresh_m3u_groups(account_id, full_refresh=True, scan_start_time=refresh_start_timestamp)
            logger.trace(f"refresh_m3u_groups result: {result}")

            # Check for completely empty result or missing groups
            if not result or result[1] is None:
                logger.error(
                    f"Failed to refresh M3U groups for account {account_id}: {result}"
                )
                return "Failed to update m3u account - download failed or other error"

            extinf_data, groups = result

            # XC accounts can have empty extinf_data but valid groups
            try:
                account = M3UAccount.objects.get(id=account_id)
                is_xc_account = account.account_type == M3UAccount.Types.XC
            except M3UAccount.DoesNotExist:
                is_xc_account = False

            # For XC accounts, empty extinf_data is normal at this stage
            if not extinf_data and not is_xc_account:
                logger.error(f"No streams found for non-XC account {account_id}")
                account.status = M3UAccount.Status.ERROR
                account.last_message = "No streams found in M3U source"
                account.save(update_fields=["status", "last_message"])
                send_m3u_update(
                    account_id, "parsing", 100, status="error", error="No streams found"
                )
        except Exception as e:
            logger.error(f"Exception in refresh_m3u_groups: {str(e)}", exc_info=True)
            account.status = M3UAccount.Status.ERROR
            account.last_message = f"Error refreshing M3U groups: {str(e)}"
            account.save(update_fields=["status", "last_message"])
            send_m3u_update(
                account_id,
                "parsing",
                100,
                status="error",
                error=f"Error refreshing M3U groups: {str(e)}",
            )
            return "Failed to update m3u account"

    # Only proceed with parsing if we actually have data and no errors were encountered
    # Get account type to handle XC accounts differently
    try:
        is_xc_account = account.account_type == M3UAccount.Types.XC
    except Exception:
        is_xc_account = False

    # Modified validation logic for different account types
    if (not groups) or (not is_xc_account and not extinf_data):
        logger.error(f"No data to process for account {account_id}")
        account.status = M3UAccount.Status.ERROR
        account.last_message = "No data available for processing"
        account.save(update_fields=["status", "last_message"])
        send_m3u_update(
            account_id,
            "parsing",
            100,
            status="error",
            error="No data available for processing",
        )
        return "Failed to update m3u account, no data available"

    hash_keys = CoreSettings.get_m3u_hash_key().split(",")

    existing_groups = {
        group.name: group.id
        for group in ChannelGroup.objects.filter(
            m3u_accounts__m3u_account=account,  # Filter by the M3UAccount
            m3u_accounts__enabled=True,  # Filter by the enabled flag in the join table
        )
    }

    try:
        # Set status to parsing
        account.status = M3UAccount.Status.PARSING
        account.save(update_fields=["status"])

        # Commit any pending transactions before threading
        from django.db import transaction
        transaction.commit()

        # Initialize stream counters
        streams_created = 0
        streams_updated = 0

        if account.account_type == M3UAccount.Types.STADNARD:
            logger.debug(
                f"Processing Standard account ({account_id}) with groups: {existing_groups}"
            )
            # Break into batches and process with threading - use global batch size
            batches = [
                extinf_data[i : i + BATCH_SIZE]
                for i in range(0, len(extinf_data), BATCH_SIZE)
            ]

            logger.info(f"Processing {len(extinf_data)} streams in {len(batches)} thread batches")

            # Use 2 threads for optimal database connection handling
            max_workers = min(2, len(batches))
            logger.debug(f"Using {max_workers} threads for processing")

            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                # Submit batch processing tasks using direct functions (now thread-safe)
                future_to_batch = {
                    executor.submit(process_m3u_batch_direct, account_id, batch, existing_groups, hash_keys): i
                    for i, batch in enumerate(batches)
                }

                completed_batches = 0
                total_batches = len(batches)

                # Process completed batches as they finish
                for future in as_completed(future_to_batch):
                    batch_idx = future_to_batch[future]
                    try:
                        result = future.result()
                        completed_batches += 1

                        # Extract stream counts from result
                        if isinstance(result, str):
                            try:
                                created_match = re.search(r"(\d+) created", result)
                                updated_match = re.search(r"(\d+) updated", result)
                                if created_match and updated_match:
                                    created_count = int(created_match.group(1))
                                    updated_count = int(updated_match.group(1))
                                    streams_created += created_count
                                    streams_updated += updated_count
                            except (AttributeError, ValueError):
                                pass

                        # Send progress update
                        progress = int((completed_batches / total_batches) * 100)
                        current_elapsed = time.time() - start_time

                        if progress > 0:
                            estimated_total = (current_elapsed / progress) * 100
                            time_remaining = max(0, estimated_total - current_elapsed)
                        else:
                            time_remaining = 0

                        send_m3u_update(
                            account_id,
                            "parsing",
                            progress,
                            elapsed_time=current_elapsed,
                            time_remaining=time_remaining,
                            streams_processed=streams_created + streams_updated,
                        )

                        logger.debug(f"Thread batch {completed_batches}/{total_batches} completed")

                    except Exception as e:
                        logger.error(f"Error in thread batch {batch_idx}: {str(e)}")
                        completed_batches += 1  # Still count it to avoid hanging

            logger.info(f"Thread-based processing completed for account {account_id}")
        else:
            # For XC accounts, get the groups with their custom properties containing xc_id
            logger.debug(f"Processing XC account with groups: {existing_groups}")

            # Get the ChannelGroupM3UAccount entries with their custom_properties
            channel_group_relationships = ChannelGroupM3UAccount.objects.filter(
                m3u_account=account, enabled=True
            ).select_related("channel_group")

            filtered_groups = {}
            for rel in channel_group_relationships:
                group_name = rel.channel_group.name
                group_id = rel.channel_group.id

                # Load the custom properties with the xc_id
                custom_props = rel.custom_properties or {}
                if "xc_id" in custom_props:
                    filtered_groups[group_name] = {
                        "xc_id": custom_props["xc_id"],
                        "channel_group_id": group_id,
                    }
                    logger.debug(
                        f"Added group {group_name} with xc_id {custom_props['xc_id']}"
                    )
                else:
                    logger.warning(
                        f"No xc_id found in custom properties for group {group_name}"
                    )

            logger.info(
                f"Filtered {len(filtered_groups)} groups for processing: {filtered_groups}"
            )

            # Collect all XC streams in a single API call and filter by enabled categories
            logger.info("Fetching all XC streams from provider and filtering by enabled categories...")
            all_xc_streams = collect_xc_streams(account_id, filtered_groups)

            if not all_xc_streams:
                logger.warning("No streams collected from XC groups")
            else:
                # Now batch by stream count (like standard M3U processing)
                batches = [
                    all_xc_streams[i : i + BATCH_SIZE]
                    for i in range(0, len(all_xc_streams), BATCH_SIZE)
                ]

                logger.info(f"Processing {len(all_xc_streams)} XC streams in {len(batches)} batches")

                # Free the original list; batches hold independent sliced copies
                del all_xc_streams

                # Use threading for XC stream processing - now with consistent batch sizes
                max_workers = min(4, len(batches))
                logger.debug(f"Using {max_workers} threads for XC stream processing")

                with ThreadPoolExecutor(max_workers=max_workers) as executor:
                    # Submit stream batch processing tasks (reuse standard M3U processing)
                    future_to_batch = {
                        executor.submit(process_m3u_batch_direct, account_id, batch, existing_groups, hash_keys): i
                        for i, batch in enumerate(batches)
                    }

                    completed_batches = 0
                    total_batches = len(batches)

                    # Process completed batches as they finish
                    for future in as_completed(future_to_batch):
                        batch_idx = future_to_batch[future]
                        try:
                            result = future.result()
                            completed_batches += 1

                            # Extract stream counts from result
                            if isinstance(result, str):
                                try:
                                    created_match = re.search(r"(\d+) created", result)
                                    updated_match = re.search(r"(\d+) updated", result)
                                    if created_match and updated_match:
                                        created_count = int(created_match.group(1))
                                        updated_count = int(updated_match.group(1))
                                        streams_created += created_count
                                        streams_updated += updated_count
                                except (AttributeError, ValueError):
                                    pass

                            # Send progress update
                            progress = int((completed_batches / total_batches) * 100)
                            current_elapsed = time.time() - start_time

                            if progress > 0:
                                estimated_total = (current_elapsed / progress) * 100
                                time_remaining = max(0, estimated_total - current_elapsed)
                            else:
                                time_remaining = 0

                            send_m3u_update(
                                account_id,
                                "parsing",
                                progress,
                                elapsed_time=current_elapsed,
                                time_remaining=time_remaining,
                                streams_processed=streams_created + streams_updated,
                            )

                            logger.debug(f"XC thread batch {completed_batches}/{total_batches} completed")

                        except Exception as e:
                            logger.error(f"Error in XC thread batch {batch_idx}: {str(e)}")
                            completed_batches += 1  # Still count it to avoid hanging

                logger.info(f"XC thread-based processing completed for account {account_id}")

        # Ensure all database transactions are committed before cleanup
        logger.info(
            f"All thread processing completed, ensuring DB transactions are committed before cleanup"
        )
        # Force a simple DB query to ensure connection sync
        Stream.objects.filter(
            id=-1
        ).exists()  # This will never find anything but ensures DB sync

        # Mark streams that weren't seen in this refresh as stale (pending deletion)
        stale_stream_count = Stream.objects.filter(
            m3u_account=account,
            last_seen__lt=refresh_start_timestamp
        ).update(is_stale=True)
        logger.info(f"Marked {stale_stream_count} streams as stale for account {account_id}")

        # Mark group relationships that weren't seen in this refresh as stale (pending deletion)
        stale_group_count = ChannelGroupM3UAccount.objects.filter(
            m3u_account=account,
            last_seen__lt=refresh_start_timestamp
        ).update(is_stale=True)
        logger.info(f"Marked {stale_group_count} group relationships as stale for account {account_id}")

        # Now run cleanup
        streams_deleted = cleanup_streams(account_id, refresh_start_timestamp)

        # Cleanup stale group relationships (follows same retention policy as streams)
        cleanup_stale_group_relationships(account, refresh_start_timestamp)

        # Run auto channel sync after successful refresh
        auto_sync_message = ""
        auto_sync_result = {}
        try:
            auto_sync_result = sync_auto_channels(
                account_id, scan_start_time=str(refresh_start_timestamp)
            ) or {}
            logger.info(
                f"Auto channel sync result for account {account_id}: {auto_sync_result}"
            )
            if auto_sync_result.get("status") == "ok":
                created = auto_sync_result.get("channels_created", 0)
                updated = auto_sync_result.get("channels_updated", 0)
                deleted = auto_sync_result.get("channels_deleted", 0)
                failed = auto_sync_result.get("channels_failed", 0)
                if created or updated or deleted or failed:
                    parts = []
                    if created:
                        parts.append(f"{created} channel(s) created")
                    if updated:
                        parts.append(f"{updated} updated")
                    if deleted:
                        parts.append(f"{deleted} deleted")
                    if failed:
                        parts.append(f"{failed} failed")
                    auto_sync_message = f" Auto-sync: {', '.join(parts)}."
            elif auto_sync_result.get("status") == "error":
                auto_sync_message = (
                    f" Auto-sync error: {auto_sync_result.get('error', 'unknown')}."
                )
        except Exception as e:
            logger.error(
                f"Error running auto channel sync for account {account_id}: {str(e)}"
            )

        # Calculate elapsed time
        elapsed_time = time.time() - start_time

        # Calculate total streams processed
        streams_processed = streams_created + streams_updated

        # Set status to success and update timestamp BEFORE sending the final update
        account.status = M3UAccount.Status.SUCCESS
        account.last_message = (
            f"Processing completed in {elapsed_time:.1f} seconds. "
            f"Streams: {streams_created} created, {streams_updated} updated, {streams_deleted} removed. "
            f"Total processed: {streams_processed}.{auto_sync_message}"
        )
        account.updated_at = timezone.now()
        account.save(update_fields=["status", "last_message", "updated_at"])

        # Log system event for M3U refresh
        log_system_event(
            event_type='m3u_refresh',
            account_name=account.name,
            elapsed_time=round(elapsed_time, 2),
            streams_created=streams_created,
            streams_updated=streams_updated,
            streams_deleted=streams_deleted,
            total_processed=streams_processed,
        )

        # Send final update with complete metrics and explicitly include success status
        send_m3u_update(
            account_id,
            "parsing",
            100,
            status="success",  # Explicitly set status to success
            elapsed_time=elapsed_time,
            time_remaining=0,
            streams_processed=streams_processed,
            streams_created=streams_created,
            streams_updated=streams_updated,
            streams_deleted=streams_deleted,
            # Structured auto-sync counts so the frontend can render a
            # warning card when anything failed, without parsing the
            # free-text last_message.
            channels_created=auto_sync_result.get("channels_created", 0),
            channels_updated=auto_sync_result.get("channels_updated", 0),
            channels_deleted=auto_sync_result.get("channels_deleted", 0),
            channels_failed=auto_sync_result.get("channels_failed", 0),
            failed_stream_details=auto_sync_result.get("failed_stream_details", []),
            message=account.last_message,
        )

        # Trigger VOD refresh if enabled and account is XtreamCodes type
        if vod_enabled and account.account_type == M3UAccount.Types.XC:
            logger.info(f"VOD is enabled for account {account_id}, triggering VOD refresh")
            try:
                from apps.vod.tasks import refresh_vod_content
                refresh_vod_content.delay(account_id)
                logger.info(f"VOD refresh task queued for account {account_id}")
            except Exception as e:
                logger.error(f"Failed to queue VOD refresh for account {account_id}: {str(e)}")

    except Exception as e:
        logger.error(f"Error processing M3U for account {account_id}: {str(e)}")
        try:
            account.status = M3UAccount.Status.ERROR
            account.last_message = f"Error processing M3U: {str(e)}"
            account.save(update_fields=["status", "last_message"])
        except Exception:
            logger.debug(f"Failed to update account {account_id} status during error handling")
        raise  # Re-raise the exception for Celery to handle
    finally:
        # Free large data structures regardless of success or failure
        if 'existing_groups' in locals():
            del existing_groups
        if 'extinf_data' in locals():
            del extinf_data
        if 'groups' in locals():
            del groups
        if 'batches' in locals():
            del batches
        if 'all_xc_streams' in locals():
            del all_xc_streams
        if 'data' in locals():
            del data
        if 'filtered_groups' in locals():
            del filtered_groups
        if 'channel_group_relationships' in locals():
            del channel_group_relationships

        # Remove cache file after processing (success or failure)
        cache_path = os.path.join(m3u_dir, f"{account_id}.json")
        try:
            os.remove(cache_path)
        except OSError:
            pass

    return f"Dispatched jobs complete."


def send_m3u_update(account_id, action, progress, **kwargs):
    # Start with the base data dictionary
    data = {
        "progress": progress,
        "type": "m3u_refresh",
        "account": account_id,
        "action": action,
    }

    # Only fetch the account when we actually need to fill in missing fields.
    # Many callers in tight loops already pass status/message; skip the DB hit then.
    if "status" not in kwargs or "message" not in kwargs:
        try:
            account = M3UAccount.objects.only("status", "last_message").get(id=account_id)
            if "status" not in kwargs:
                data["status"] = account.status
            if "message" not in kwargs and account.last_message:
                data["message"] = account.last_message
        except Exception:
            pass

    # Add the additional key-value pairs from kwargs
    data.update(kwargs)
    send_websocket_update("updates", "update", data, collect_garbage=False)

    # Explicitly clear data reference to help garbage collection
    data = None


def evaluate_profile_expiration_notification(profile):
    """
    Evaluate a single M3UAccountProfile's expiration date and create, update,
    or delete the corresponding SystemNotification accordingly.

    Returns the notification key that should remain active (warning or expired),
    or None if the profile is not expiring soon and any stale notifications were removed.
    This return value is used by the bulk task to track active keys for stale cleanup.
    """
    from core.models import SystemNotification
    from core.utils import send_websocket_notification, send_notification_dismissed

    exp = profile.exp_date
    if not exp:
        return None

    now = timezone.now()
    warning_threshold = now + timezone.timedelta(days=7)
    warning_key = f"xc-exp-warning-{profile.id}"
    expired_key = f"xc-exp-expired-{profile.id}"

    if exp <= now:
        # Already expired — delete warning, create/update expired notification
        deleted_warning = list(
            SystemNotification.objects.filter(notification_key=warning_key)
            .values_list("notification_key", flat=True)
        )
        SystemNotification.objects.filter(notification_key=warning_key).delete()
        for key in deleted_warning:
            send_notification_dismissed(key)

        notification, created = SystemNotification.objects.update_or_create(
            notification_key=expired_key,
            defaults={
                "notification_type": SystemNotification.NotificationType.WARNING,
                "priority": SystemNotification.Priority.HIGH,
                "title": f"Account Expired: {profile.name}",
                "message": (
                    f'Profile "{profile.name}" on M3U account '
                    f'"{profile.m3u_account.name}" has expired '
                    f"(expired {exp.strftime('%Y-%m-%d %H:%M UTC')})."
                ),
                "action_data": {
                    "profile_id": profile.id,
                    "account_id": profile.m3u_account.id,
                    "account_name": profile.m3u_account.name,
                    "profile_name": profile.name,
                    "exp_date": exp.isoformat(),
                },
                "is_active": True,
                "admin_only": True,
            },
        )
        send_websocket_notification(notification)
        return expired_key

    elif exp <= warning_threshold:
        # Expiring within 7 days — delete expired notification, create/update warning
        deleted_expired = list(
            SystemNotification.objects.filter(notification_key=expired_key)
            .values_list("notification_key", flat=True)
        )
        SystemNotification.objects.filter(notification_key=expired_key).delete()
        for key in deleted_expired:
            send_notification_dismissed(key)

        days_left = (exp - now).days
        if days_left == 0:
            expires_in_str = "today"
        elif days_left == 1:
            expires_in_str = "in 1 day"
        else:
            expires_in_str = f"in {days_left} days"

        notification, created = SystemNotification.objects.update_or_create(
            notification_key=warning_key,
            defaults={
                "notification_type": SystemNotification.NotificationType.WARNING,
                "priority": SystemNotification.Priority.NORMAL,
                "title": f"Account Expiring: {profile.name}",
                "message": (
                    f'Profile "{profile.name}" on M3U account '
                    f'"{profile.m3u_account.name}" expires {expires_in_str} '
                    f"(expires {exp.strftime('%Y-%m-%d %H:%M UTC')})."
                ),
                "action_data": {
                    "profile_id": profile.id,
                    "account_id": profile.m3u_account.id,
                    "account_name": profile.m3u_account.name,
                    "profile_name": profile.name,
                    "exp_date": exp.isoformat(),
                },
                "is_active": True,
                "admin_only": True,
            },
        )
        send_websocket_notification(notification)
        return warning_key

    else:
        # Not expiring soon — delete any stale notifications
        deleted_keys = list(
            SystemNotification.objects.filter(
                notification_key__in=[warning_key, expired_key]
            ).values_list("notification_key", flat=True)
        )
        SystemNotification.objects.filter(
            notification_key__in=[warning_key, expired_key]
        ).delete()
        for key in deleted_keys:
            send_notification_dismissed(key)
        return None


@shared_task
def check_account_expirations():
    """
    Daily task: check all account profiles for upcoming expirations.
    Creates/updates SystemNotifications for profiles expiring within 7 days.
    Uses separate notification keys for warning vs expired so users can
    dismiss the 7-day warning and still receive the expired notification.
    """
    from apps.m3u.models import M3UAccountProfile
    from core.models import SystemNotification
    from core.utils import send_notification_dismissed

    # Find all active profiles with an exp_date that is set
    expiring_profiles = (
        M3UAccountProfile.objects.filter(
            m3u_account__is_active=True,
            is_active=True,
            exp_date__isnull=False,
        )
        .select_related("m3u_account")
    )

    active_notification_keys = set()

    for profile in expiring_profiles:
        active_key = evaluate_profile_expiration_notification(profile)
        if active_key:
            active_notification_keys.add(active_key)

    # Delete stale notifications for profiles whose expiration was extended
    stale = SystemNotification.objects.filter(
        is_active=True,
    ).filter(
        models.Q(notification_key__startswith="xc-exp-warning-") |
        models.Q(notification_key__startswith="xc-exp-expired-")
    ).exclude(notification_key__in=active_notification_keys)
    stale_keys = list(stale.values_list("notification_key", flat=True))
    stale.delete()
    for key in stale_keys:
        send_notification_dismissed(key)

    logger.info(
        f"Account expiration check complete: {len(active_notification_keys)} active notifications"
    )
