"""
Video/audio downloading engine based on yt-dlp.

All heavy work (network IO, ffmpeg subprocesses) runs inside worker threads
via asyncio.to_thread, so the Telegram event loop is never blocked. yt-dlp
invokes ffmpeg itself using safe argument lists (no shell), and no user input
is ever passed to a shell.
"""

from __future__ import annotations

import asyncio
import logging
import shutil
import time
import uuid
from pathlib import Path
from typing import Any, Optional

import yt_dlp
from yt_dlp.utils import DownloadError

import config

logger = logging.getLogger(__name__)

# Format selectors per requested quality (as required by the spec).
QUALITY_FORMATS: dict[str, str] = {
    "360": "bestvideo[height<=360]+bestaudio/best[height<=360]",
    "480": "bestvideo[height<=480]+bestaudio/best[height<=480]",
    "720": "bestvideo[height<=720]+bestaudio/best[height<=720]",
    "1080": "bestvideo[height<=1080]+bestaudio/best[height<=1080]",
    "best": "bestvideo+bestaudio/best",
    "mp3": "bestaudio/best",
}

QUALITY_HEIGHTS: tuple[int, ...] = (360, 480, 720, 1080)


class DownloadCancelledError(Exception):
    """Raised inside yt-dlp progress hooks when the user cancels a download."""


class DownloadState:
    """Progress state shared between the yt-dlp worker thread and the
    asyncio progress-updater task (plain attribute writes are atomic enough
    under the GIL for this use case)."""

    def __init__(self) -> None:
        self.phase: str = "downloading"  # downloading | processing | uploading
        self.downloaded: int = 0
        self.total: int = 0
        self.speed: float = 0.0
        self.eta: int = 0
        self.percent: float = 0.0
        self.cancelled: bool = False
        self.finished: bool = False

    def hook(self, d: dict[str, Any]) -> None:
        """yt-dlp progress hook (executed in the worker thread)."""
        if self.cancelled:
            raise DownloadCancelledError("Cancelled by user")
        status = d.get("status")
        if status == "downloading":
            self.phase = "downloading"
            self.downloaded = int(d.get("downloaded_bytes") or 0)
            total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
            self.total = int(total or 0)
            self.speed = float(d.get("speed") or 0.0)
            self.eta = int(d.get("eta") or 0)
            if self.total:
                self.percent = self.downloaded * 100.0 / self.total
        elif status == "finished":
            self.percent = 100.0
            self.phase = "processing"


# --------------------------------------------------------------------------- #
# FFmpeg detection
# --------------------------------------------------------------------------- #

_FFMPEG_PATH: Optional[str] = None
_FFMPEG_CHECKED = False


def ffmpeg_available() -> bool:
    """Return True when an ffmpeg binary is reachable on the system PATH."""
    global _FFMPEG_PATH, _FFMPEG_CHECKED
    if not _FFMPEG_CHECKED:
        _FFMPEG_PATH = shutil.which("ffmpeg")
        _FFMPEG_CHECKED = True
        if _FFMPEG_PATH:
            logger.info("FFmpeg found: %s", _FFMPEG_PATH)
        else:
            logger.warning("FFmpeg was not found on the system PATH")
    return _FFMPEG_PATH is not None


# --------------------------------------------------------------------------- #
# Metadata extraction
# --------------------------------------------------------------------------- #

def _build_extract_opts() -> dict[str, Any]:
    return {
        "quiet": True,
        "no_warnings": True,
        "skip_download": True,
        "noplaylist": True,
        "socket_timeout": 20,
        "retries": 2,
    }


def _extract_info_sync(url: str) -> dict[str, Any]:
    with yt_dlp.YoutubeDL(_build_extract_opts()) as ydl:
        info = ydl.extract_info(url, download=False)
    if info is None:
        raise DownloadError("No metadata returned")
    if "entries" in info:
        entries = [e for e in (info.get("entries") or []) if e]
        if not entries:
            raise DownloadError("Empty playlist result")
        info = entries[0]
    return info


async def extract_info(url: str) -> dict[str, Any]:
    """Extract metadata for `url` inside a worker thread."""
    return await asyncio.to_thread(_extract_info_sync, url)


def get_available_qualities(info: dict[str, Any]) -> list[str]:
    """Map the formats reported by yt-dlp onto the qualities the bot offers.

    A quality bucket is offered only when a video format whose height falls
    into that bucket actually exists, so the UI never promises a quality the
    source cannot provide. When heights are unknown but video exists, the
    generic 'best' option is offered instead.
    """
    heights: set[int] = set()
    has_video = False
    for fmt in info.get("formats") or []:
        if fmt.get("vcodec") in (None, "none"):
            continue
        has_video = True
        height = fmt.get("height")
        if height:
            heights.add(int(height))
    if info.get("height"):
        heights.add(int(info["height"]))
        has_video = True

    qualities: list[str] = []
    for quality in QUALITY_HEIGHTS:
        low = quality - 60
        if any(low <= h <= quality for h in heights):
            qualities.append(str(quality))
    if not qualities and has_video:
        qualities.append("best")
    return qualities


def estimate_size(info: dict[str, Any]) -> Optional[int]:
    """Rough total size estimate (best video + best audio) in bytes."""
    approx = info.get("filesize") or info.get("filesize_approx")
    if approx:
        return int(approx)
    best_video = 0
    best_audio = 0
    for fmt in info.get("formats") or []:
        size = fmt.get("filesize") or fmt.get("filesize_approx")
        if not size:
            continue
        size = int(size)
        is_video = fmt.get("vcodec") not in (None, "none")
        is_audio = fmt.get("acodec") not in (None, "none")
        if is_video:
            best_video = max(best_video, size)
        elif is_audio:
            best_audio = max(best_audio, size)
    if best_video or best_audio:
        return best_video + best_audio
    return None

def friendly_error(exc: Exception) -> str:
    """Map a yt-dlp error to a safe user-facing Persian message.

    Internal details (URLs, stack traces, host names) are never exposed.
    """
    text = str(exc).lower()

    if "unsupported url" in text:
        return "❌ این لینک توسط موتور دانلود پشتیبانی نمی‌شود."

    if "requested format" in text or "format not available" in text:
        return "❌ این کیفیت در دسترس نیست."

    if (
        "sign in" in text
        or "login" in text
        or "cookies" in text
        or "confirm you" in text
        or "rate-limit" in text
        or "rate limit" in text
    ):
        return "❌ اینستاگرام نیاز به ورود به حساب یا محدودیت درخواست دارد. لطفاً بعداً دوباره تلاش کنید."

    if "private" in text:
        return "❌ این ویدیو خصوصی است و قابل دانلود نیست."

    if "geo" in text and "restrict" in text or "not available in your country" in text:
        return "❌ این ویدیو برای منطقه شما در دسترس نیست."

    if "removed" in text or "unavailable" in text or "not available" in text:
        return "❌ این ویدیو در دسترس نیست یا حذف شده است."

    if "404" in text or "not found" in text:
        return "❌ ویدیو پیدا نشد."

    if (
        "timed out" in text
        or "timeout" in text
        or "network" in text
        or "connection" in text
    ):
        return "❌ خطای شبکه رخ داد. لطفاً کمی بعد دوباره تلاش کنید."

    if "ffmpeg" in text:
        return "❌ پردازش فایل ناموفق بود (FFmpeg). لطفاً بعداً تلاش کنید."

    return "❌ دریافت اطلاعات ویدیو موفق نشد."

# --------------------------------------------------------------------------- #
# Downloading
# --------------------------------------------------------------------------- #

_semaphore: Optional[asyncio.Semaphore] = None


def get_semaphore() -> asyncio.Semaphore:
    """Lazily created concurrency limiter (MAX_CONCURRENT_DOWNLOADS)."""
    global _semaphore
    if _semaphore is None:
        _semaphore = asyncio.Semaphore(config.MAX_CONCURRENT_DOWNLOADS)
    return _semaphore


def _build_ydl_opts(quality: str, token: str, hook) -> dict[str, Any]:
    """Build yt-dlp options for a download.

    The output template uses a byte-truncated title (Unicode-safe) plus the
    unique session token, which yields unique, sanitized temporary filenames.
    """
    outtmpl = str(config.DOWNLOAD_PATH / ("%(title).80B_" + token + ".%(ext)s"))
    opts: dict[str, Any] = {
        "format": QUALITY_FORMATS[quality],
        "outtmpl": outtmpl,
        "quiet": True,
        "no_warnings": True,
        "noplaylist": True,
        "socket_timeout": 30,
        "retries": 3,
        "fragment_retries": 3,
        "concurrent_fragment_downloads": 4,
        "overwrites": True,
        "windowsfilenames": True,  # also safe on Linux
        "noprogress": True,
        "progress_hooks": [hook],
    }
    if quality == "mp3":
        opts["postprocessors"] = [
            {"key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "192"},
            {"key": "FFmpegMetadata"},
        ]
    else:
        opts["merge_output_format"] = "mp4"
    return opts


def _download_sync(url: str, opts: dict[str, Any]) -> dict[str, Any]:
    with yt_dlp.YoutubeDL(opts) as ydl:
        info = ydl.extract_info(url, download=True)
    if info is None:
        raise DownloadError("Download returned no data")
    if "entries" in info:
        entries = [e for e in (info.get("entries") or []) if e]
        if not entries:
            raise DownloadError("Empty playlist result")
        info = entries[0]
    return info


def _find_result_file(info: dict[str, Any], token: str) -> Optional[Path]:
    """Locate the final file on disk (post-merge / post-convert path)."""
    for item in info.get("requested_downloads") or []:
        path = item.get("filepath") or item.get("filename")
        if path and Path(path).exists():
            return Path(path)
    # Fallback: newest file carrying the unique session token.
    try:
        candidates = sorted(
            config.DOWNLOAD_PATH.glob(f"*{token}*"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )
    except OSError:
        return None
    return candidates[0] if candidates else None


async def download_media(
    url: str,
    quality: str,
    state: DownloadState,
    token: Optional[str] = None,
) -> tuple[Path, dict[str, Any]]:
    """Download `url` in the requested quality.

    Concurrency is limited with an asyncio.Semaphore. Raises
    DownloadCancelledError when the user cancels; yt-dlp DownloadError and
    other exceptions propagate to the caller for friendly mapping.
    """
    if quality not in QUALITY_FORMATS:
        raise ValueError(f"Unknown quality: {quality}")
    if quality == "mp3" and not ffmpeg_available():
        raise RuntimeError("FFmpeg is required for MP3 conversion but was not found")

    token = token or uuid.uuid4().hex[:10]
    async with get_semaphore():
        opts = _build_ydl_opts(quality, token, state.hook)
        try:
            info = await asyncio.to_thread(_download_sync, url, opts)
        except DownloadError as exc:
            if state.cancelled:
                raise DownloadCancelledError("Cancelled by user") from exc
            raise
        except asyncio.CancelledError:
            state.cancelled = True
            raise
    path = await asyncio.to_thread(_find_result_file, info, token)
    if path is None:
        raise RuntimeError("Downloaded file could not be located on disk")
    return path, info


# --------------------------------------------------------------------------- #
# File cleanup
# --------------------------------------------------------------------------- #

def _delete_file_sync(path: Path) -> None:
    try:
        if path.exists():
            path.unlink()
    except OSError as exc:
        logger.warning("Could not delete %s: %s", path, exc)


async def delete_file(path: Path) -> None:
    """Delete a downloaded temporary file (best effort)."""
    await asyncio.to_thread(_delete_file_sync, path)


def _cleanup_files_sync(max_age_seconds: float) -> int:
    removed = 0
    now = time.time()
    if not config.DOWNLOAD_PATH.exists():
        return 0
    for item in config.DOWNLOAD_PATH.iterdir():
        try:
            if item.is_file() and now - item.stat().st_mtime > max_age_seconds:
                item.unlink()
                removed += 1
        except OSError:
            continue
    return removed


async def cleanup_old_files(max_age_seconds: float = 2 * 3600) -> int:
    """Delete files left inside the downloads folder older than the TTL."""
    return await asyncio.to_thread(_cleanup_files_sync, max_age_seconds)