from __future__ import annotations

import asyncio
import logging
import time
from collections.abc import Sequence
from pathlib import Path

from app.core.config import Settings
from app.core.exceptions import ProcessingError
from app.core.logger import get_logger

logger = get_logger(__name__)


class FFmpegService:
    """Concurrency-limited, shell-free FFmpeg process runner."""

    def __init__(self, settings: Settings) -> None:
        self.settings = settings
        self._semaphore = asyncio.Semaphore(settings.max_workers)

    async def run(
        self,
        args: Sequence[str | Path],
        *,
        operation: str,
        timeout: float | None = None,
        cancel_event: asyncio.Event | None = None,
    ) -> None:
        command = [self.settings.ffmpeg_binary, "-hide_banner", "-nostdin", "-y"] + [
            str(arg) for arg in args
        ]
        started = time.monotonic()
        # Arguments can contain signed URLs, storage locators, and internal
        # filesystem paths. Keep those out of structured logs while retaining
        # enough bounded context to correlate and diagnose an invocation.
        logger.info(
            "ffmpeg started",
            extra={"operation": operation, "argument_count": len(command) - 1},
        )
        async with self._semaphore:
            try:
                process = await asyncio.create_subprocess_exec(
                    *command,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                stdout_task = asyncio.create_task(_read_limited(process.stdout, 4_000))
                stderr_task = asyncio.create_task(_read_limited(process.stderr, 16_000))
                try:
                    wait_task = asyncio.create_task(process.wait())
                    cancel_task = asyncio.create_task(cancel_event.wait()) if cancel_event else None
                    pending: set[asyncio.Task[object]] = set()
                    waitables: set[asyncio.Task[object]] = {wait_task}
                    if cancel_task is not None:
                        waitables.add(cancel_task)
                    done, pending = await asyncio.wait(
                        waitables,
                        timeout=timeout or self.settings.download_timeout_seconds * 4,
                        return_when=asyncio.FIRST_COMPLETED,
                    )
                    if not done:
                        raise asyncio.TimeoutError
                    if (
                        cancel_task is not None
                        and cancel_task in done
                        and cancel_event
                        and cancel_event.is_set()
                    ):
                        process.kill()
                        await process.wait()
                        await asyncio.gather(stdout_task, stderr_task, return_exceptions=True)
                        raise ProcessingError(
                            "FFmpeg processing was cancelled", details={"cancelled": True}
                        )
                    await wait_task
                except asyncio.TimeoutError:
                    process.kill()
                    await process.wait()
                    await asyncio.gather(stdout_task, stderr_task)
                    raise
                finally:
                    for task in pending if "pending" in locals() else set():
                        task.cancel()
                    if "wait_task" in locals() and not wait_task.done():
                        wait_task.cancel()
                    if (
                        "cancel_task" in locals()
                        and cancel_task is not None
                        and not cancel_task.done()
                    ):
                        cancel_task.cancel()
                stdout, stderr = await asyncio.gather(stdout_task, stderr_task)
            except asyncio.TimeoutError as exc:
                raise ProcessingError("FFmpeg processing timed out") from exc
            except FileNotFoundError as exc:
                raise ProcessingError("FFmpeg is not installed or not available") from exc
        log_data = {
            "operation": operation,
            "argument_count": len(command) - 1,
            "duration": round(time.monotonic() - started, 4),
            "return_code": process.returncode,
            "stdout_bytes": len(stdout),
            "stderr_bytes": len(stderr),
        }
        if process.returncode != 0:
            logger.error("ffmpeg failed", extra=log_data)
            raise ProcessingError(
                "FFmpeg could not process the media",
                details={"operation": operation},
            )
        logger.log(logging.INFO, "ffmpeg completed", extra=log_data)

    async def version(self) -> str:
        """Return the installed FFmpeg version banner."""
        output = await self._capture(["-version"], operation="ffmpeg.version")
        return output.splitlines()[0] if output else "unknown"

    async def codecs(self) -> list[dict[str, str | bool]]:
        """Return structured codec capabilities reported by FFmpeg."""
        output = await self._capture(["-hide_banner", "-codecs"], operation="ffmpeg.codecs")
        codecs: list[dict[str, str | bool]] = []
        for line in output.splitlines():
            if len(line) < 9 or line[0] != " ":
                continue
            flags = line[1:7]
            if (
                flags[0] not in {"D", "."}
                or flags[1] not in {"E", "."}
                or flags[2] not in {"V", "A", "S", "D", "."}
            ):
                continue
            remainder = line[8:].strip()
            if not remainder or " " not in remainder:
                continue
            name, description = remainder.split(maxsplit=1)
            if name == "=":
                continue
            codecs.append(
                {
                    "name": name,
                    "description": description,
                    "decode": flags[0] == "D",
                    "encode": flags[1] == "E",
                    "type": {
                        "V": "video",
                        "A": "audio",
                        "S": "subtitle",
                    }.get(flags[2], "other"),
                }
            )
        return codecs

    async def _capture(self, args: Sequence[str], *, operation: str) -> str:
        command = [self.settings.ffmpeg_binary, *args]
        logger.info(
            "ffmpeg information requested", extra={"operation": operation, "command": command}
        )
        try:
            async with self._semaphore:
                process = await asyncio.create_subprocess_exec(
                    *command,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.STDOUT,
                )
                stdout, _ = await asyncio.wait_for(process.communicate(), timeout=30)
        except FileNotFoundError as exc:
            raise ProcessingError("FFmpeg is not installed or not available") from exc
        except asyncio.TimeoutError as exc:
            if "process" in locals():
                process.kill()
                await process.wait()
            raise ProcessingError("FFmpeg information request timed out") from exc
        if process.returncode != 0:
            raise ProcessingError("FFmpeg information request failed")
        return stdout.decode("utf-8", errors="replace")[-2_000_000:]


async def _read_limited(stream: asyncio.StreamReader | None, limit: int) -> bytes:
    if stream is None:
        return b""
    data = bytearray()
    while chunk := await stream.read(64 * 1024):
        data.extend(chunk)
        if len(data) > limit:
            del data[:-limit]
    return bytes(data)