"""
Index-time enrichment for the Videos group app: technical probe (ffprobe),
filename parsing (title_parse), and thumbnail generation (ffmpeg) for a
newly-added video IndexEntry.

Runs through its own small bounded worker pool — separate from the streaming
transcode pool (docs/mediacenter.md §5.2, mirroring webrtc_server.py's
`_transcode_semaphore`) — so indexing a large library never blocks on this,
and enrichment never competes with an active viewer for CPU. The scan itself
already put the entry in the index with hash/size/type only; this fills in
the rest asynchronously and hands the result back via a callback.
"""

import asyncio
import logging
from pathlib import Path
from typing import Awaitable, Callable

import blake3

from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer import title_parse
from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS
from meshbay_node.media_cache import MediaCache
from meshbay_node.media_probe import probe_video

log = logging.getLogger(__name__)

DEFAULT_MAX_CONCURRENT = 2
PROBE_TIMEOUT_SECS = 30
THUMB_TIMEOUT_SECS = 30
THUMB_WIDTH = 320
# "Show/SeasonFolder/episode.mkv" is the expected shape, with a little slack
# for an extra wrapper folder — not an attempt to find the exact group root.
MAX_ANCESTOR_DEPTH = 4
# Bounds the "borrow a title from a sibling episode filename" scan (§3.4) so
# a folder with thousands of files costs a fixed, small amount of work.
MAX_SIBLINGS_CHECKED = 20


def _season_from_ancestors(file_path: Path) -> int | None:
    folder = file_path.parent
    for _ in range(MAX_ANCESTOR_DEPTH):
        if folder is None or folder == folder.parent:
            break
        season = title_parse.season_from_folder_name(folder.name)
        if season is not None:
            return season
        folder = folder.parent
    return None


def _title_from_siblings(file_path: Path) -> str | None:
    """
    §3.4: an episode filename with no show name in it borrows the title from
    a representative sibling in the same folder, never from the folder name
    alone (an acronym-named show folder is a real, observed case).
    """
    try:
        names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file())
    except OSError:
        return None
    checked = 0
    for name in names:
        if name == file_path.name:
            continue
        if Path(name).suffix.lower() not in MEDIA_EXTENSIONS["video"]:
            continue
        checked += 1
        if checked > MAX_SIBLINGS_CHECKED:
            break
        parsed = title_parse.parse_episode_filename(name)
        if parsed.display_title:
            return parsed.display_title
    return None


async def _make_thumbnail(file_path: Path, duration: float | None) -> bytes | None:
    """One ffmpeg frame grab at ~10% of duration (or 5s if unknown), scaled down."""
    seek = max(0.0, (duration or 50.0) * 0.1)
    proc = await asyncio.create_subprocess_exec(
        "ffmpeg", "-v", "error", "-ss", str(seek), "-i", str(file_path),
        "-frames:v", "1", "-vf", f"scale={THUMB_WIDTH}:-1",
        "-f", "image2", "-c:v", "mjpeg", "pipe:1",
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, _ = await asyncio.wait_for(proc.communicate(), THUMB_TIMEOUT_SECS)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        return None
    return stdout or None


class Enricher:
    """Owns the node's bounded index-time enrichment pool."""

    def __init__(self, media_cache: MediaCache, max_concurrent: int = DEFAULT_MAX_CONCURRENT):
        self._media_cache = media_cache
        self._sem = asyncio.Semaphore(max_concurrent)
        self._tasks: set[asyncio.Task] = set()

    def spawn(
        self, entry: IndexEntry, file_path: Path,
        on_done: Callable[[str, dict], Awaitable[None]],
    ) -> asyncio.Task:
        """
        Fire-and-forget one file's enrichment. `on_done(file_id, fields)` is
        awaited with the index fields to merge in once ready — never blocks
        the caller (a scan or watchdog event). The reference this method
        returns is what keeps the task alive; callers should hold it the
        same way `WebRTCPeerSession._spawn` holds streaming tasks.
        """
        task = asyncio.ensure_future(self._run(entry, file_path, on_done))
        self._tasks.add(task)

        def _cleanup(t: asyncio.Task) -> None:
            self._tasks.discard(t)
            if not t.cancelled() and t.exception():
                log.error("Enrichment failed for %s: %s", entry.id[:12], t.exception(),
                          exc_info=t.exception())
        task.add_done_callback(_cleanup)
        return task

    async def _run(
        self, entry: IndexEntry, file_path: Path,
        on_done: Callable[[str, dict], Awaitable[None]],
    ) -> None:
        async with self._sem:
            fields: dict = {}
            duration: float | None = None
            try:
                _codec, duration, _has_audio, width, height, _raw = await asyncio.wait_for(
                    probe_video(str(file_path)), timeout=PROBE_TIMEOUT_SECS)
                fields["duration"] = int(duration) if duration else None
                fields["width"] = width
                fields["height"] = height
            except Exception as e:
                log.warning("Probe failed for %s: %s", file_path, e)

            ep = title_parse.parse_episode_filename(entry.name)
            if ep.episode is not None:
                title = ep.display_title or await asyncio.to_thread(
                    _title_from_siblings, file_path)
                season = ep.season
                if season is None:
                    season = await asyncio.to_thread(_season_from_ancestors, file_path)
                fields["display_title"] = title or title_parse.naive_title(entry.name)
                fields["season"] = season
                fields["episode"] = ep.episode
            else:
                mv = title_parse.parse_movie_filename(entry.name)
                fields["display_title"] = mv.display_title or mv.naive_title

            try:
                thumb = await _make_thumbnail(file_path, duration)
            except Exception as e:
                log.warning("Thumbnail generation failed for %s: %s", file_path, e)
                thumb = None
            if thumb:
                thumb_hash = blake3.blake3(thumb).hexdigest()
                await self._media_cache.put_thumb(thumb_hash, entry.id, thumb)
                fields["thumb_hash"] = thumb_hash

            await on_done(entry.id, fields)