"""Audio normalization — convert uploaded media to canonical format before sharding.

Canonical format: mono, 16 kHz, PCM WAV.
This ensures consistent duration measurement, deterministic content hashing,
and uniform input for Whisper across all workers.
"""

from __future__ import annotations

import hashlib
import json
import shutil
import subprocess
from pathlib import Path
from urllib.parse import urljoin, urlparse

import httpx
import structlog

from hive_compute.url_safety import validate_outbound_http_url

logger = structlog.get_logger()
MAX_REMOTE_DOWNLOAD_SIZE = 500 * 1024 * 1024
_REMOTE_DOWNLOAD_USER_AGENT = "HiveCompute/0.1 (+https://hivecompute.fly.dev)"

_SUPPORTED_REMOTE_MEDIA_HOSTS = {
    "youtube.com",
    "www.youtube.com",
    "m.youtube.com",
    "music.youtube.com",
    "youtu.be",
}


def normalize_audio(input_path: str | Path, output_path: str | Path) -> dict:
    """Transcode audio to mono 16kHz PCM WAV. Returns metadata dict.

    Raises RuntimeError if ffmpeg is unavailable or transcoding fails.
    """
    input_path = Path(input_path)
    output_path = Path(output_path)

    if not input_path.exists():
        raise FileNotFoundError(f"Input audio not found: {input_path}")

    if not shutil.which("ffmpeg"):
        raise RuntimeError("ffmpeg is required for audio normalization but was not found on PATH.")

    output_path.parent.mkdir(parents=True, exist_ok=True)
    source_metadata = _probe_source_metadata(input_path)

    try:
        subprocess.run(
            [
                "ffmpeg", "-y",
                "-i", str(input_path),
                "-ar", "16000",
                "-ac", "1",
                "-c:a", "pcm_s16le",
                str(output_path),
            ],
            check=True,
            capture_output=True,
            text=True,
        )
    except subprocess.CalledProcessError as exc:
        stderr = (exc.stderr or "")[:500]
        raise RuntimeError(f"Audio normalization failed (exit {exc.returncode}): {stderr}") from exc

    # Measure duration from the normalized file
    duration = _probe_duration(output_path)
    content_hash = _compute_hash(output_path)
    size_bytes = output_path.stat().st_size

    logger.info(
        "Audio normalized",
        input=str(input_path),
        output=str(output_path),
        duration=duration,
        size_bytes=size_bytes,
    )

    return {
        "duration_seconds": duration,
        "content_hash": content_hash,
        "size_bytes": size_bytes,
        "sample_rate": 16000,
        "channels": 1,
        "channel_count": source_metadata["channel_count"] or 1,
        "format": "pcm_s16le",
        "original_format": source_metadata["original_format"] or _detect_original_format(input_path),
    }


def is_supported_remote_media_source(source_url: str) -> bool:
    """Return whether the URL should be ingested through the coordinator.

    This first version is intentionally narrow: support YouTube-style watch/share
    URLs that are not directly fetchable as audio by workers.
    """
    parsed = urlparse(source_url)
    hostname = (parsed.hostname or "").rstrip(".").lower()
    return hostname in _SUPPORTED_REMOTE_MEDIA_HOSTS


def download_supported_media(source_url: str, target_dir: str | Path, *, stem: str) -> Path:
    """Download a supported remote media page to a local file.

    The returned path points to the downloaded source container. Callers are
    expected to normalize it afterward and clean it up when finished.
    """
    validate_outbound_http_url(source_url, purpose="remote media source URL")
    if not is_supported_remote_media_source(source_url):
        raise ValueError("Unsupported remote media source")

    try:
        from yt_dlp import YoutubeDL
    except ImportError as exc:
        raise RuntimeError("yt-dlp is required for remote media ingestion but is not installed") from exc

    target_dir = Path(target_dir)
    target_dir.mkdir(parents=True, exist_ok=True)
    output_template = str(target_dir / f"{stem}.%(ext)s")

    logger.info("Downloading remote media", source=source_url[:120], target_dir=str(target_dir))
    with YoutubeDL(
        {
            "format": "bestaudio/best",
            "noplaylist": True,
            "quiet": True,
            "no_warnings": True,
            "restrictfilenames": True,
            "outtmpl": {"default": output_template},
            "overwrites": True,
            "socket_timeout": 30,
        }
    ) as ydl:
        try:
            info = ydl.extract_info(source_url, download=True)
        except Exception as exc:  # pragma: no cover - exact extractor failures vary
            raise RuntimeError(f"Media download failed: {exc}") from exc

        downloaded_path = Path(ydl.prepare_filename(info))

    if not downloaded_path.exists():
        raise RuntimeError("Media download failed: extractor did not produce a local file")

    logger.info("Remote media downloaded", source=source_url[:120], path=str(downloaded_path))
    return downloaded_path


def download_remote_audio(source_url: str, target_dir: str | Path, *, stem: str) -> Path:
    """Download a direct remote audio URL to a local file for coordinator-side ingest."""
    validate_outbound_http_url(source_url, purpose="remote audio source URL")

    target_dir = Path(target_dir)
    target_dir.mkdir(parents=True, exist_ok=True)

    current_url = source_url
    for _ in range(4):
        try:
            with httpx.stream(
                "GET",
                current_url,
                follow_redirects=False,
                timeout=300,
                headers={"User-Agent": _REMOTE_DOWNLOAD_USER_AGENT},
            ) as response:
                if response.status_code in {301, 302, 303, 307, 308}:
                    location = response.headers.get("location")
                    if not location:
                        raise RuntimeError("Remote audio URL redirected without a location header")
                    current_url = urljoin(current_url, location)
                    validate_outbound_http_url(current_url, purpose="remote audio redirect URL")
                    continue

                response.raise_for_status()
                content_length = response.headers.get("content-length")
                if content_length:
                    try:
                        if int(content_length) > MAX_REMOTE_DOWNLOAD_SIZE:
                            raise RuntimeError(
                                f"Remote audio file too large (>{MAX_REMOTE_DOWNLOAD_SIZE // 1024 // 1024}MB)"
                            )
                    except ValueError:
                        pass

                suffix = _guess_download_suffix(
                    url=str(response.url),
                    content_type=response.headers.get("content-type"),
                )
                downloaded_path = target_dir / f"{stem}{suffix}"
                total = 0
                with downloaded_path.open("wb") as handle:
                    for chunk in response.iter_bytes(chunk_size=8192):
                        total += len(chunk)
                        if total > MAX_REMOTE_DOWNLOAD_SIZE:
                            downloaded_path.unlink(missing_ok=True)
                            raise RuntimeError(
                                f"Remote audio file too large (>{MAX_REMOTE_DOWNLOAD_SIZE // 1024 // 1024}MB)"
                            )
                        handle.write(chunk)

                logger.info(
                    "Remote audio downloaded",
                    source=source_url[:120],
                    final_url=str(response.url)[:120],
                    path=str(downloaded_path),
                    size_mb=round(total / 1024 / 1024, 1),
                )
                return downloaded_path
        except httpx.HTTPStatusError as exc:
            raise RuntimeError(f"Remote audio download failed: {exc}") from exc
        except httpx.HTTPError as exc:
            raise RuntimeError(f"Remote audio download failed: {exc}") from exc

    raise RuntimeError("Remote audio URL redirected too many times")


def probe_audio_duration(source: str | Path) -> float | None:
    """Best-effort duration probe for local paths or remote URLs.

    Returns None when ffprobe is unavailable or the source cannot be probed.
    Remote URLs are restricted to http/https to prevent SSRF via ffprobe's
    support for file://, ftp://, smb://, etc.
    """
    source_text = str(source)

    looks_like_url = "://" in source_text or source_text.startswith("data:")
    if looks_like_url:
        try:
            validate_outbound_http_url(source_text, purpose="audio probe URL")
        except ValueError:
            logger.warning("probe_audio_duration: blocked unsafe url", source=source_text[:80])
            return None

    if shutil.which("ffprobe"):
        try:
            result = subprocess.run(
                [
                    "ffprobe",
                    "-v", "quiet",
                    "-analyzeduration", "5000000",
                    "-probesize", "5000000",
                    "-show_entries", "format=duration",
                    "-of", "csv=p=0",
                    source_text,
                ],
                check=False,
                capture_output=True,
                text=True,
                timeout=20,
            )
        except (subprocess.SubprocessError, OSError):
            return None

        try:
            duration = float(result.stdout.strip())
        except (ValueError, AttributeError):
            duration = None

        if duration and duration > 0:
            return duration

    local_path = Path(source_text)
    if local_path.exists():
        return _probe_duration(local_path)
    return None


def _probe_duration(path: Path) -> float:
    """Get audio duration in seconds using ffprobe, falling back to wave module."""
    if shutil.which("ffprobe"):
        result = subprocess.run(
            [
                "ffprobe",
                "-v", "quiet",
                "-show_entries", "format=duration",
                "-of", "csv=p=0",
                str(path),
            ],
            capture_output=True,
            text=True,
        )
        try:
            return float(result.stdout.strip())
        except (ValueError, AttributeError):
            pass

    # Fallback: parse the WAV container directly (accurate for PCM WAV)
    import wave
    try:
        with wave.open(str(path), "rb") as wf:
            return wf.getnframes() / wf.getframerate()
    except Exception:
        # Last resort: raw size estimate
        return max(0, path.stat().st_size - 44) / 32000.0


def _compute_hash(path: Path) -> str:
    """SHA-256 hash of the normalized audio file."""
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()


def _detect_original_format(path: Path) -> str:
    suffix = path.suffix.lower().lstrip(".")
    return suffix or "unknown"


def _probe_source_metadata(path: Path) -> dict[str, int | str | None]:
    if shutil.which("ffprobe"):
        try:
            result = subprocess.run(
                [
                    "ffprobe",
                    "-v", "quiet",
                    "-print_format", "json",
                    "-show_format",
                    "-show_streams",
                    str(path),
                ],
                check=False,
                capture_output=True,
                text=True,
            )
            payload = json.loads(result.stdout or "{}")
        except (subprocess.SubprocessError, OSError, json.JSONDecodeError):
            payload = {}
    else:
        payload = {}

    if not isinstance(payload, dict):
        payload = {}

    streams = payload.get("streams") or []
    audio_stream = next((stream for stream in streams if stream.get("codec_type") == "audio"), {})
    format_name = (payload.get("format") or {}).get("format_name")

    original_format = None
    if isinstance(format_name, str) and format_name.strip():
        original_format = format_name.split(",")[0].strip().lower()

    channel_count = audio_stream.get("channels")
    if isinstance(channel_count, int) and channel_count > 0:
        resolved_channels = channel_count
    else:
        resolved_channels = None

    return {
        "channel_count": resolved_channels,
        "original_format": original_format,
    }


def _guess_download_suffix(*, url: str, content_type: str | None = None) -> str:
    path_suffix = Path(urlparse(url).path).suffix.lower()
    if path_suffix and len(path_suffix) <= 8:
        return path_suffix

    normalized_content_type = (content_type or "").split(";")[0].strip().lower()
    known_by_type = {
        "audio/mpeg": ".mp3",
        "audio/mp3": ".mp3",
        "audio/wav": ".wav",
        "audio/x-wav": ".wav",
        "audio/wave": ".wav",
        "audio/flac": ".flac",
        "audio/x-flac": ".flac",
        "audio/ogg": ".ogg",
        "audio/webm": ".webm",
        "audio/mp4": ".m4a",
        "audio/x-m4a": ".m4a",
        "video/mp4": ".mp4",
        "video/webm": ".webm",
    }
    return known_by_type.get(normalized_content_type, ".audio")
