"""Media bot — Matrix companion for the homelab media stack.

Wraps Seerr (search/request), Emby (library/playback), Sonarr/Radarr (queue/calendar),
NZBGet + qBittorrent (active downloads).

Each Matrix sender is mapped to per-service user IDs via plugin config; unmapped
senders are rejected. All replies are prefixed with the sender's MXID so notifications
in shared rooms stay readable (matches the Books Bot convention).
"""

from __future__ import annotations

import asyncio
from datetime import date, datetime, timedelta, timezone
from typing import Any, Optional, Type

import aiohttp
from aiohttp.web import Request, Response, json_response

from maubot import MessageEvent, Plugin
from maubot.handlers import command, event, web
from mautrix.types import (
    EventType,
    Format,
    ImageInfo,
    MediaMessageEventContent,
    MessageType,
    ReactionEvent,
    RoomID,
    TextMessageEventContent,
)
from mautrix.util.async_db import UpgradeTable
from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper

from .clients.arr import ArrError, LidarrClient, RadarrClient, SonarrClient
from .clients.downloads import DownloadError, NzbgetClient, QbtClient
from .clients.emby import EmbyClient, EmbyError
from .clients.seerr import SeerrClient, SeerrError
from .db import upgrade_table


def _is_us_dst(dt: datetime) -> bool:
    """Approximate US DST: second Sunday of March → first Sunday of November.
    Used because the maubot container ships without tzdata for ZoneInfo.
    """
    y = dt.year
    march1 = datetime(y, 3, 1)
    second_sunday = march1 + timedelta(days=((6 - march1.weekday()) % 7) + 7)
    nov1 = datetime(y, 11, 1)
    first_sunday = nov1 + timedelta(days=(6 - nov1.weekday()) % 7)
    naive = dt.replace(tzinfo=None) if dt.tzinfo else dt
    return second_sunday <= naive < first_sunday


def _local_now() -> datetime:
    """Approximate Indianapolis local time (Eastern, with DST). Naive."""
    utc = datetime.now(timezone.utc)
    offset = -4 if _is_us_dst(utc) else -5
    return (utc + timedelta(hours=offset)).replace(tzinfo=None)


# --- Seerr availability codes ----------------------------------------------
# https://api-docs.overseerr.dev — `Status` enum
SEERR_AVAIL = {
    1: "unknown",
    2: "pending",
    3: "processing",
    4: "partial",
    5: "available",
}


def _human_bytes(n: float) -> str:
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if abs(n) < 1024:
            return f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} PB"


def _human_eta(seconds: int | float | None) -> str:
    if not seconds or seconds < 0 or seconds > 8640000:
        return "?"
    s = int(seconds)
    h, rem = divmod(s, 3600)
    m, s = divmod(rem, 60)
    if h:
        return f"{h}h{m}m"
    if m:
        return f"{m}m{s}s"
    return f"{s}s"


class Config(BaseProxyConfig):
    def do_update(self, helper: ConfigUpdateHelper) -> None:
        helper.copy("http_timeout")
        helper.copy("default_results")
        helper.copy("posters_enabled")
        helper.copy("admin_users")
        helper.copy("notifications_room")
        helper.copy("seerr_webhook_secret")
        helper.copy("sonarr_webhook_secret")
        helper.copy("digest_enabled")
        helper.copy("digest_hour")
        helper.copy("ntfy_url")
        helper.copy("emby_cleaner_topic")
        helper.copy("seerr")
        helper.copy("sonarr")
        helper.copy("radarr")
        helper.copy("lidarr")
        helper.copy("emby")
        helper.copy("nzbget")
        helper.copy("qbittorrent")
        helper.copy("user_map")


# Reactions counted as approve/decline. Skin-tone variants pass via startswith.
APPROVE_KEYS = ("👍", "✅", "🟢")
DECLINE_KEYS = ("👎", "❌", "🟥")


def _matches(key: str, candidates: tuple[str, ...]) -> bool:
    return any(key.startswith(c) for c in candidates)


class MediaBot(Plugin):
    config: Config
    session: aiohttp.ClientSession

    seerr: SeerrClient
    sonarr: SonarrClient
    radarr: RadarrClient
    lidarr: LidarrClient
    emby: EmbyClient
    nzbget: NzbgetClient
    qbt: QbtClient

    # Per-(room, sender) cache of last search/trending results — used by
    # `!media request <N>` to skip re-typing the query. In-memory only; a bot
    # restart clears it (acceptable, results are cheap to rebuild).
    _search_cache: dict[tuple[str, str], list[dict]]

    # Map of bot-message event_id → Seerr request_id, for reaction-based
    # approve/decline. Capped at 200; oldest entries are evicted on overflow.
    _pending_requests: dict[str, int]

    # Background task running the daily digest scheduler.
    _digest_task: Optional[asyncio.Task]

    @classmethod
    def get_db_upgrade_table(cls) -> UpgradeTable:
        return upgrade_table

    async def start(self) -> None:
        self.config.load_and_update()
        timeout = aiohttp.ClientTimeout(total=self.config["http_timeout"])
        self.session = aiohttp.ClientSession(timeout=timeout)

        s, so, r, l, e, n, q = (
            self.config["seerr"], self.config["sonarr"], self.config["radarr"],
            self.config["lidarr"], self.config["emby"],
            self.config["nzbget"], self.config["qbittorrent"],
        )
        self.seerr = SeerrClient(self.session, s["url"], s["api_key"])
        self.sonarr = SonarrClient(self.session, so["url"], so["api_key"])
        self.radarr = RadarrClient(self.session, r["url"], r["api_key"])
        self.lidarr = LidarrClient(self.session, l["url"], l["api_key"])
        self.emby = EmbyClient(self.session, e["url"], e["api_key"])
        self.nzbget = NzbgetClient(self.session, n["url"], n["username"], n["password"])
        self.qbt = QbtClient(self.session, q["url"], q["username"], q["password"])

        # Seerr is optional. When disabled, search/request fall back to direct
        # Sonarr/Radarr lookup+add, and the seerr-only commands return a hint.
        self.seerr_enabled = self._service_configured(s)

        self._search_cache = {}
        self._pending_requests = {}
        await self._ensure_db_schema()
        self._digest_task = None
        if self.config["digest_enabled"] and (self.config["notifications_room"] or "").strip():
            self._digest_task = asyncio.create_task(self._digest_loop())
        self.log.info("Media bot started — users=%d", len(self.config["user_map"] or {}))

    async def stop(self) -> None:
        if self._digest_task and not self._digest_task.done():
            self._digest_task.cancel()
        if hasattr(self, "session"):
            await self.session.close()

    @classmethod
    def get_config_class(cls) -> Type[BaseProxyConfig]:
        return Config

    # ---------- helpers ----------

    @staticmethod
    def _service_configured(block: dict | None) -> bool:
        """A service block is 'configured' if it has a non-empty url and a
        non-placeholder api_key. Used to gate optional integrations."""
        if not block:
            return False
        url = (block.get("url") or "").strip()
        key = (block.get("api_key") or "").strip()
        return bool(url) and key not in ("", "CHANGEME")

    def _resolve_user(self, sender: str) -> Optional[dict]:
        return (self.config["user_map"] or {}).get(sender)

    def _media_label(self) -> str:
        """Display label for the configured Emby/Jellyfin server."""
        kind = ((self.config["emby"] or {}).get("type") or "emby").lower()
        return "Jellyfin" if kind == "jellyfin" else "Emby"

    async def _say(self, evt: MessageEvent, message: str) -> None:
        await evt.respond(f"{evt.sender}\n\n{message}")

    async def _reject_unmapped(self, evt: MessageEvent) -> bool:
        if not self._resolve_user(evt.sender):
            self.log.warning("Unauthorized sender: %s", evt.sender)
            await self._say(evt, "Sorry, your Matrix user isn't authorized to use the media bot.")
            return True
        return False

    def _user_pref(self, sender: str, key: str, default: Any) -> Any:
        cfg = self._resolve_user(sender) or {}
        val = cfg.get(key)
        return val if val is not None else default

    def _result_count(self, sender: str) -> int:
        return int(self._user_pref(sender, "result_count", self.config["default_results"]))

    def _stash_search(self, evt: MessageEvent, results: list[dict]) -> None:
        """Cache search/trending results so the user can `!media request <N>`."""
        self._search_cache[(evt.room_id, evt.sender)] = list(results)

    def _track_request(self, event_id: str, request_id: int) -> None:
        self._pending_requests[event_id] = request_id
        if len(self._pending_requests) > 200:
            oldest = next(iter(self._pending_requests))
            del self._pending_requests[oldest]

    async def _post_poster(self, room_id: str, image_url: str, caption: str) -> bool:
        """Download a poster URL, upload to Matrix, send as m.image. Returns True on success."""
        if not self.config["posters_enabled"]:
            return False
        try:
            async with self.session.get(image_url) as r:
                if r.status >= 400:
                    self.log.info("poster fetch %s → %s", image_url, r.status)
                    return False
                data = await r.read()
                mime = r.headers.get("Content-Type", "image/jpeg").split(";")[0].strip()
            mxc = await self.client.upload_media(data, mime_type=mime)
            content = MediaMessageEventContent(
                msgtype=MessageType.IMAGE,
                body=caption,
                url=mxc,
                info=ImageInfo(mimetype=mime, size=len(data)),
            )
            await self.client.send_message(RoomID(room_id), content)
            return True
        except Exception as ex:
            self.log.warning("poster upload failed: %s", ex)
            return False

    # ---------- top-level command ----------

    @command.new("media", aliases=["m"], help="Media stack bot — try `!media help`",
                 require_subcommand=True)
    async def media(self) -> None:
        pass

    @media.subcommand("help", help="Show available commands")
    async def cmd_help(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        msg = (
            "**Media bot commands**\n\n"
            "*Search & request*\n"
            "- `!media search <query>` — top results (numbered)\n"
            "- `!media request <query> [--tv|--movie]` — request top hit\n"
            "- `!media request <N>` — pick item N from your last search/trending\n"
            + ("- `!media requests` — your pending/processing requests\n"
               "- `!media trending` — what's trending (numbered)\n\n"
               if self.seerr_enabled else
               "_(Seerr not configured — search/request go directly to Sonarr+Radarr; "
               "no approval workflow.)_\n\n")
            +
            f"*Library ({self._media_label()})*\n"
            "- `!media nowplaying` — active sessions\n"
            "- `!media recent [movies|tv]` — recently added\n"
            "- `!media watched` — what you recently finished\n"
            "- `!media find <query>` — search the existing library\n"
            "- `!media resume` — your continue-watching list\n"
            "- `!media random [movie|tv]` — random unwatched pick\n\n"
            "*Sonarr / Radarr*\n"
            "- `!media queue` — combined queue\n"
            "- `!media upcoming` — Sonarr calendar (next 7 days)\n"
            "- `!media missing` — Sonarr wanted/missing\n"
            f"- `!media health` — Sonarr/Radarr/Lidarr/Seerr/{self._media_label()}/NZBGet/qBt status\n\n"
            "*Lidarr (music)*\n"
            "- `!media music <q>` — search MusicBrainz (numbered)\n"
            "- `!media music add <q|N>` — add the artist to Lidarr\n\n"
            "*Downloads (NZBGet + qBt)*\n"
            "- `!media activity` — current downloads\n"
            "- `!media completed` — finished in the last 24h\n"
            "- `!media speed` — aggregate down/up speed\n"
            "- `!media pause` / `!media unpause` — global pause/resume\n\n"
            "*Subscriptions*\n"
            "- `!media subscribe <show>` — ping me when new episodes import\n"
            "- `!media unsubscribe <show>` — stop pings\n"
            "- `!media subscriptions` — list yours\n"
            "- `!media digest` — fire today's digest now (admin/test)\n\n"
            "_Admins can react 👍/👎 on a request to approve/decline._"
        )
        await self._say(evt, msg)

    # ---------- Seerr ----------

    @media.subcommand("search", help="Search Seerr for a movie or show")
    @command.argument("query", pass_raw=True, required=True)
    async def cmd_search(self, evt: MessageEvent, query: str) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        if self.seerr_enabled:
            try:
                results = await self.seerr.search(query)
            except SeerrError as ex:
                await self._say(evt, f"Search failed: {ex}")
                return
            results = [r for r in results if r.get("mediaType") in ("movie", "tv")]
            for r in results:
                r["_source"] = "seerr"
        else:
            results = await self._arr_search(query)
        if not results:
            await self._say(evt, f"No results for **{query}**.")
            return
        n = self._result_count(evt.sender)
        shown = results[:n]
        self._stash_search(evt, shown)
        lines = [f"**Top {len(shown)} for '{query}':**"]
        for i, r in enumerate(shown, 1):
            lines.append(f"{i}. " + self._fmt_seerr(r, leading_dash=False))
        lines.append("")
        lines.append("_Pick one with `!media request <N>`._")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("request",
                      help="Request a movie/show. Pass `<query>` or a number from a recent search.")
    @command.argument("query", pass_raw=True, required=True)
    async def cmd_request(self, evt: MessageEvent, query: str) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        user_cfg = self._resolve_user(evt.sender) or {}
        seerr_uid = user_cfg.get("seerr_user_id")
        if self.seerr_enabled and not seerr_uid:
            await self._say(evt, "Your Matrix user has no `seerr_user_id` mapped in `user_map`.")
            return

        # Numbered selection: !media request 2 → 2nd item from last search/trending
        stripped = query.strip()
        if stripped.isdigit():
            cached = self._search_cache.get((evt.room_id, evt.sender)) or []
            idx = int(stripped) - 1
            if not cached:
                await self._say(evt, "No recent search results to pick from. Run `!media search <query>` first.")
                return
            if idx < 0 or idx >= len(cached):
                await self._say(evt, f"Pick a number 1-{len(cached)}.")
                return
            await self._do_request(evt, seerr_uid, cached[idx])
            return

        # Free-text path: optional --tv/--movie flag, else fall back to user default
        forced_type: Optional[str] = None
        cleaned: list[str] = []
        for w in stripped.split():
            if w == "--tv":
                forced_type = "tv"
            elif w == "--movie":
                forced_type = "movie"
            else:
                cleaned.append(w)
        if forced_type is None:
            forced_type = user_cfg.get("default_media_type")
        q = " ".join(cleaned).strip()
        if not q:
            await self._say(evt, "Need a search query, e.g. `!media request dune --movie`.")
            return
        if self.seerr_enabled:
            try:
                results = await self.seerr.search(q)
            except SeerrError as ex:
                await self._say(evt, f"Search failed: {ex}")
                return
            candidates = [r for r in results if r.get("mediaType") in ("movie", "tv")]
            for r in candidates:
                r["_source"] = "seerr"
        else:
            candidates = await self._arr_search(q, media_type=forced_type)
        if forced_type:
            candidates = [r for r in candidates if r.get("mediaType") == forced_type]
        if not candidates:
            await self._say(evt, f"No matching results for **{q}**.")
            return
        await self._do_request(evt, seerr_uid, candidates[0])

    async def _do_request(self, evt: MessageEvent, seerr_uid: Optional[int], item: dict) -> None:
        source = item.get("_source") or ("seerr" if self.seerr_enabled else "arr")
        if source == "seerr":
            await self._do_request_seerr(evt, seerr_uid, item)
        else:
            await self._do_request_arr(evt, item)

    async def _do_request_seerr(self, evt: MessageEvent, seerr_uid: Optional[int], item: dict) -> None:
        media_type = item.get("mediaType")
        tmdb_id = item.get("id")
        title = item.get("title") or item.get("name") or "?"
        if not (media_type and tmdb_id):
            await self._say(evt, f"**{title}** is missing a media type or TMDB id — can't request.")
            return
        try:
            req = await self.seerr.request(media_type, tmdb_id, seerr_uid)
        except SeerrError as ex:
            await self._say(evt, f"Request failed: {ex}")
            return
        status_id = (req or {}).get("status")
        status = {1: "pending approval", 2: "approved", 3: "declined"}.get(status_id, str(status_id))
        request_id = (req or {}).get("id")

        # Optional poster preview
        poster = SeerrClient.poster_url(item)
        if poster:
            await self._post_poster(evt.room_id, poster, f"{title} ({media_type})")

        msg = f"Requested **{title}** ({media_type}) — status: *{status}*."
        if status_id == 1 and request_id:
            msg += "\n\n_Admins: react 👍 to approve or 👎 to decline._"
        sent_id = await evt.respond(f"{evt.sender}\n\n{msg}")
        if request_id and status_id == 1 and sent_id:
            self._track_request(sent_id, request_id)

    async def _do_request_arr(self, evt: MessageEvent, item: dict) -> None:
        """Add a movie/series directly to Radarr/Sonarr (no Seerr in the loop)."""
        media_type = item.get("mediaType")
        title = item.get("title") or item.get("name") or "?"
        raw = item.get("_raw") or {}
        if media_type == "movie":
            ok, msg = await self._add_radarr(raw)
        elif media_type == "tv":
            ok, msg = await self._add_sonarr(raw)
        else:
            await self._say(evt, f"**{title}** is neither movie nor tv — can't add.")
            return

        # Poster preview if Radarr/Sonarr returned one in the lookup payload
        poster = next(
            (img.get("remoteUrl") or img.get("url")
             for img in (raw.get("images") or [])
             if img.get("coverType") == "poster"),
            None,
        )
        if poster:
            await self._post_poster(evt.room_id, poster, f"{title} ({media_type})")

        prefix = "Added" if ok else "Failed"
        await self._say(evt, f"{prefix} **{title}** ({media_type}) — {msg}")

    async def _add_sonarr(self, series: dict) -> tuple[bool, str]:
        cfg = self.config["sonarr"] or {}
        try:
            qpid = cfg.get("quality_profile_id")
            root = cfg.get("root_folder_path")
            if not qpid:
                qps = await self.sonarr.quality_profiles()
                qpid = qps[0]["id"] if qps else None
            if not root:
                rfs = await self.sonarr.root_folders()
                root = rfs[0]["path"] if rfs else None
            lpid = cfg.get("language_profile_id")
            if lpid is None:
                lps = await self.sonarr.language_profiles()
                lpid = lps[0]["id"] if lps else None
        except ArrError as ex:
            return False, f"Sonarr defaults lookup failed: {ex}"
        if not (qpid and root):
            return False, "Sonarr is missing a quality profile or root folder."

        payload = dict(series)
        payload.update({
            "qualityProfileId": qpid,
            "rootFolderPath": root,
            "monitored": True,
            "seasonFolder": True,
            "addOptions": {
                "monitor": cfg.get("monitor", "all"),
                "searchForMissingEpisodes": bool(cfg.get("search_on_add", True)),
                "searchForCutoffUnmetEpisodes": False,
            },
        })
        if lpid:  # Sonarr v3 only
            payload["languageProfileId"] = lpid
        try:
            await self.sonarr.add_series(payload)
        except ArrError as ex:
            return False, f"Sonarr add failed: {ex}"
        searched = "search kicked off" if cfg.get("search_on_add", True) else "no search triggered"
        return True, f"Sonarr profile {qpid}, root `{root}`, {searched}."

    async def _add_radarr(self, movie: dict) -> tuple[bool, str]:
        cfg = self.config["radarr"] or {}
        try:
            qpid = cfg.get("quality_profile_id")
            root = cfg.get("root_folder_path")
            if not qpid:
                qps = await self.radarr.quality_profiles()
                qpid = qps[0]["id"] if qps else None
            if not root:
                rfs = await self.radarr.root_folders()
                root = rfs[0]["path"] if rfs else None
        except ArrError as ex:
            return False, f"Radarr defaults lookup failed: {ex}"
        if not (qpid and root):
            return False, "Radarr is missing a quality profile or root folder."

        payload = dict(movie)
        payload.update({
            "qualityProfileId": qpid,
            "rootFolderPath": root,
            "monitored": True,
            "minimumAvailability": cfg.get("minimum_availability", "released"),
            "addOptions": {
                "monitor": cfg.get("monitor", "movieOnly"),
                "searchForMovie": bool(cfg.get("search_on_add", True)),
            },
        })
        try:
            await self.radarr.add_movie(payload)
        except ArrError as ex:
            return False, f"Radarr add failed: {ex}"
        searched = "search kicked off" if cfg.get("search_on_add", True) else "no search triggered"
        return True, f"Radarr profile {qpid}, root `{root}`, {searched}."

    async def _arr_search(self, query: str, media_type: Optional[str] = None) -> list[dict]:
        """Direct Sonarr+Radarr lookup, normalized to the Seerr search-item shape
        so the rest of the bot (formatting, caching, request flow) doesn't care
        which source produced the result."""
        tasks = []
        if media_type != "movie":
            tasks.append(("tv", self.sonarr.lookup(query)))
        if media_type != "tv":
            tasks.append(("movie", self.radarr.lookup(query)))
        results: list[dict] = []
        gathered = await asyncio.gather(
            *(t[1] for t in tasks), return_exceptions=True,
        )
        for (mt, _), data in zip(tasks, gathered):
            if isinstance(data, Exception):
                self.log.warning("arr lookup failed for %s: %s", mt, data)
                continue
            for item in data:
                results.append(self._normalize_arr_item(item, mt))
        return results

    @staticmethod
    def _normalize_arr_item(item: dict, media_type: str) -> dict:
        """Reshape a Sonarr/Radarr lookup result so it slots into the same code
        paths as a Seerr search result."""
        year = item.get("year")
        date_field = "releaseDate" if media_type == "movie" else "firstAirDate"
        return {
            "title": item.get("title") or item.get("name") or "?",
            "mediaType": media_type,
            date_field: f"{year}-01-01" if year else "",
            "id": item.get("tmdbId") if media_type == "movie" else item.get("tvdbId"),
            "tmdbId": item.get("tmdbId"),
            "tvdbId": item.get("tvdbId"),
            "_source": "radarr" if media_type == "movie" else "sonarr",
            "_raw": item,
        }

    @media.subcommand("requests", help="Show your pending/processing Seerr requests")
    async def cmd_requests(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        if not self.seerr_enabled:
            await self._say(evt, "Seerr isn't configured — `!media requests` is unavailable.")
            return
        user_cfg = self._resolve_user(evt.sender)
        seerr_uid = (user_cfg or {}).get("seerr_user_id")
        if not seerr_uid:
            await self._say(evt, "Your Matrix user has no `seerr_user_id` mapped in `user_map`.")
            return
        try:
            reqs = await self.seerr.user_requests(seerr_uid, take=10)
        except SeerrError as ex:
            await self._say(evt, f"Lookup failed: {ex}")
            return
        if not reqs:
            await self._say(evt, "No pending or processing requests.")
            return
        lines = [f"**Your requests ({len(reqs)}):**"]
        for r in reqs:
            mi = r.get("media") or {}
            t = mi.get("title") or mi.get("name") or f"tmdb:{mi.get('tmdbId')}"
            mt = mi.get("mediaType") or "?"
            avail = SEERR_AVAIL.get(mi.get("status", 0), "?")
            lines.append(f"- *{t}* ({mt}) — {avail}")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("trending", help="Show what's trending in Seerr")
    async def cmd_trending(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        if not self.seerr_enabled:
            await self._say(evt, "Seerr isn't configured — `!media trending` is unavailable.")
            return
        try:
            results = await self.seerr.trending()
        except SeerrError as ex:
            await self._say(evt, f"Trending fetch failed: {ex}")
            return
        n = self._result_count(evt.sender)
        results = [r for r in results if r.get("mediaType") in ("movie", "tv")][:n]
        if not results:
            await self._say(evt, "Nothing trending right now.")
            return
        for r in results:
            r["_source"] = "seerr"
        self._stash_search(evt, results)
        lines = [f"**Trending ({len(results)}):**"]
        for i, r in enumerate(results, 1):
            lines.append(f"{i}. " + self._fmt_seerr(r, leading_dash=False))
        lines.append("")
        lines.append("_Pick one with `!media request <N>`._")
        await self._say(evt, "\n".join(lines))

    def _fmt_seerr(self, r: dict, leading_dash: bool = True) -> str:
        title = r.get("title") or r.get("name") or "?"
        mt = r.get("mediaType") or "?"
        date_s = r.get("releaseDate") or r.get("firstAirDate") or ""
        year = date_s[:4] if date_s else ""
        info = (r.get("mediaInfo") or {}).get("status")
        avail = SEERR_AVAIL.get(info, None) if info else None
        bits = [f"*{title}*"]
        if year:
            bits.append(f"({year})")
        bits.append(f"— {mt}")
        if avail:
            bits.append(f"[{avail}]")
        prefix = "- " if leading_dash else ""
        return prefix + " ".join(bits)

    # ---------- Emby ----------

    @media.subcommand("nowplaying", aliases=["np"], help="Active Emby sessions")
    async def cmd_nowplaying(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        try:
            sessions = await self.emby.sessions()
        except EmbyError as ex:
            await self._say(evt, f"{self._media_label()} fetch failed: {ex}")
            return
        active = [s for s in sessions if s.get("NowPlayingItem")]
        if not active:
            await self._say(evt, f"Nothing playing on {self._media_label()} right now.")
            return
        lines = [f"**Now playing on {self._media_label()} ({len(active)}):**"]
        for s in active:
            item = s.get("NowPlayingItem") or {}
            user = s.get("UserName") or "?"
            client = s.get("Client") or ""
            t = self._fmt_emby_item(item)
            ps = s.get("PlayState") or {}
            pos = ps.get("PositionTicks") or 0
            run = item.get("RunTimeTicks") or 0
            pct = (pos / run * 100) if run else 0
            paused = " (paused)" if ps.get("IsPaused") else ""
            lines.append(f"- **{user}** — {t} · {pct:.0f}%{paused} [{client}]")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("recent", help="Recently added — `!media recent [movies|tv]`")
    @command.argument("kind", required=False)
    async def cmd_recent(self, evt: MessageEvent, kind: str = "") -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        # Use the sender's emby_user_id if set, else fall back to the first mapped user's id
        user_cfg = self._resolve_user(evt.sender) or {}
        emby_uid = user_cfg.get("emby_user_id")
        if not emby_uid or emby_uid == "TBD":
            for v in (self.config["user_map"] or {}).values():
                if v and v.get("emby_user_id") and v["emby_user_id"] != "TBD":
                    emby_uid = v["emby_user_id"]
                    break
        if not emby_uid:
            await self._say(evt, "No usable Emby user_id in config.")
            return
        item_types = None
        if kind.lower() in ("movie", "movies"):
            item_types = "Movie"
        elif kind.lower() in ("tv", "series", "shows"):
            item_types = "Series,Episode"
        try:
            items = await self.emby.recently_added(emby_uid, limit=10, item_types=item_types)
        except EmbyError as ex:
            await self._say(evt, f"{self._media_label()} fetch failed: {ex}")
            return
        if not items:
            await self._say(evt, "Nothing recently added.")
            return
        label = f" ({kind})" if kind else ""
        lines = [f"**Recently added{label} ({len(items)}):**"]
        for it in items:
            lines.append("- " + self._fmt_emby_item(it))
        await self._say(evt, "\n".join(lines))

    @media.subcommand("watched", help="Your recently watched items on Emby")
    async def cmd_watched(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        user_cfg = self._resolve_user(evt.sender) or {}
        emby_uid = user_cfg.get("emby_user_id")
        if not emby_uid or emby_uid == "TBD":
            await self._say(evt, "Your Matrix user has no `emby_user_id` mapped in `user_map`.")
            return
        try:
            items = await self.emby.user_played(emby_uid, limit=10)
        except EmbyError as ex:
            await self._say(evt, f"{self._media_label()} fetch failed: {ex}")
            return
        if not items:
            await self._say(evt, "No recently watched items.")
            return
        lines = [f"**Recently watched ({len(items)}):**"]
        for it in items:
            lines.append("- " + self._fmt_emby_item(it))
        await self._say(evt, "\n".join(lines))

    @media.subcommand("find", help="Search the existing Emby library")
    @command.argument("query", pass_raw=True, required=True)
    async def cmd_find(self, evt: MessageEvent, query: str) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        emby_uid = self._any_emby_uid(evt.sender)
        if not emby_uid:
            await self._say(evt, "No usable Emby user_id in config.")
            return
        try:
            items = await self.emby.find(emby_uid, query, limit=10)
        except EmbyError as ex:
            await self._say(evt, f"{self._media_label()} fetch failed: {ex}")
            return
        if not items:
            await self._say(evt, f"Nothing in the library matching **{query}**.")
            return
        lines = [f"**Found in library ({len(items)}):**"]
        for it in items:
            lines.append("- " + self._fmt_emby_item(it))
        await self._say(evt, "\n".join(lines))

    @media.subcommand("resume", help="Your continue-watching list")
    async def cmd_resume(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        user_cfg = self._resolve_user(evt.sender) or {}
        emby_uid = user_cfg.get("emby_user_id")
        if not emby_uid or emby_uid == "TBD":
            await self._say(evt, "Your Matrix user has no `emby_user_id` mapped in `user_map`.")
            return
        try:
            items = await self.emby.resume(emby_uid, limit=10)
        except EmbyError as ex:
            await self._say(evt, f"{self._media_label()} fetch failed: {ex}")
            return
        if not items:
            await self._say(evt, "Nothing to resume — you've finished everything you started.")
            return
        lines = [f"**Continue watching ({len(items)}):**"]
        for it in items:
            pct = ""
            ud = it.get("UserData") or {}
            played_pct = ud.get("PlayedPercentage")
            if played_pct:
                pct = f" · {played_pct:.0f}%"
            lines.append("- " + self._fmt_emby_item(it) + pct)
        await self._say(evt, "\n".join(lines))

    @media.subcommand("random", help="Random unwatched pick — `!media random [movie|tv]`")
    @command.argument("kind", required=False)
    async def cmd_random(self, evt: MessageEvent, kind: str = "") -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        user_cfg = self._resolve_user(evt.sender) or {}
        emby_uid = user_cfg.get("emby_user_id")
        if not emby_uid or emby_uid == "TBD":
            emby_uid = self._any_emby_uid(evt.sender)
        if not emby_uid:
            await self._say(evt, "No usable Emby user_id in config.")
            return
        item_type = "Movie"
        if kind.lower() in ("tv", "series", "show", "shows"):
            item_type = "Series"
        try:
            item = await self.emby.random_unplayed(emby_uid, item_type=item_type)
        except EmbyError as ex:
            await self._say(evt, f"{self._media_label()} fetch failed: {ex}")
            return
        if not item:
            await self._say(evt, f"No unwatched {item_type.lower()}s found — go you.")
            return
        line = self._fmt_emby_item(item)
        overview = (item.get("Overview") or "").strip()
        if len(overview) > 240:
            overview = overview[:240].rsplit(" ", 1)[0] + "…"
        poster = self.emby.poster_url(item)
        if poster:
            await self._post_poster(evt.room_id, poster, item.get("Name") or "poster")
        msg = f"**Random pick:** {line}"
        if overview:
            msg += f"\n\n{overview}"
        await self._say(evt, msg)

    def _any_emby_uid(self, sender: str) -> Optional[str]:
        """Sender's emby_user_id, falling back to any other mapped user's id."""
        cfg = self._resolve_user(sender) or {}
        uid = cfg.get("emby_user_id")
        if uid and uid != "TBD":
            return uid
        for v in (self.config["user_map"] or {}).values():
            if v and v.get("emby_user_id") and v["emby_user_id"] != "TBD":
                return v["emby_user_id"]
        return None

    def _fmt_emby_item(self, it: dict) -> str:
        t = it.get("Type")
        name = it.get("Name") or "?"
        year = it.get("ProductionYear")
        if t == "Episode":
            series = it.get("SeriesName") or "?"
            sn = it.get("ParentIndexNumber")
            ep = it.get("IndexNumber")
            tag = f"S{sn:02d}E{ep:02d}" if sn and ep else ""
            return f"*{series}* {tag} — {name}".strip()
        if year:
            return f"*{name}* ({year})"
        return f"*{name}*"

    # ---------- Sonarr / Radarr ----------

    @media.subcommand("queue", help="Combined Sonarr + Radarr queue")
    async def cmd_queue(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        try:
            sonarr_q, radarr_q = await asyncio.gather(
                self.sonarr.queue(), self.radarr.queue(),
                return_exceptions=True,
            )
        except Exception as ex:
            await self._say(evt, f"Queue fetch failed: {ex}")
            return
        lines: list[str] = []
        if isinstance(sonarr_q, Exception):
            lines.append(f"**Sonarr:** unreachable ({sonarr_q})")
        else:
            lines.append(f"**Sonarr queue ({len(sonarr_q)}):**")
            for q in sonarr_q[:10]:
                series = (q.get("series") or {}).get("title") or "?"
                ep = q.get("episode") or {}
                tag = ""
                if ep.get("seasonNumber") is not None and ep.get("episodeNumber") is not None:
                    tag = f" S{ep['seasonNumber']:02d}E{ep['episodeNumber']:02d}"
                status = q.get("status") or "?"
                pct = self._arr_pct(q)
                lines.append(f"- *{series}*{tag} — {status} {pct}")
        lines.append("")
        if isinstance(radarr_q, Exception):
            lines.append(f"**Radarr:** unreachable ({radarr_q})")
        else:
            lines.append(f"**Radarr queue ({len(radarr_q)}):**")
            for q in radarr_q[:10]:
                title = (q.get("movie") or {}).get("title") or q.get("title") or "?"
                status = q.get("status") or "?"
                pct = self._arr_pct(q)
                lines.append(f"- *{title}* — {status} {pct}")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("upcoming", help="Sonarr calendar — episodes airing in next 7 days")
    async def cmd_upcoming(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        today = date.today()
        end = today + timedelta(days=7)
        try:
            cal = await self.sonarr.calendar(today.isoformat(), end.isoformat())
        except ArrError as ex:
            await self._say(evt, f"Calendar fetch failed: {ex}")
            return
        if not cal:
            await self._say(evt, "Nothing airing in the next 7 days.")
            return
        lines = [f"**Upcoming ({len(cal)}):**"]
        for ep in cal[:15]:
            series = (ep.get("series") or {}).get("title") or "?"
            sn = ep.get("seasonNumber")
            en = ep.get("episodeNumber")
            tag = f"S{sn:02d}E{en:02d}" if sn is not None and en is not None else ""
            t = ep.get("title") or ""
            air = (ep.get("airDateUtc") or ep.get("airDate") or "")[:10]
            lines.append(f"- {air}: *{series}* {tag} — {t}")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("missing", help="Sonarr wanted/missing list")
    async def cmd_missing(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        try:
            items = await self.sonarr.missing(page_size=10)
        except ArrError as ex:
            await self._say(evt, f"Missing fetch failed: {ex}")
            return
        if not items:
            await self._say(evt, "No missing episodes — Sonarr is happy.")
            return
        lines = [f"**Wanted/Missing ({len(items)}):**"]
        for ep in items:
            series = (ep.get("series") or {}).get("title") or "?"
            sn = ep.get("seasonNumber")
            en = ep.get("episodeNumber")
            tag = f"S{sn:02d}E{en:02d}" if sn is not None and en is not None else ""
            t = ep.get("title") or ""
            air = (ep.get("airDateUtc") or ep.get("airDate") or "")[:10]
            lines.append(f"- *{series}* {tag} — {t} (aired {air})")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("health", help="Status of all backing services")
    async def cmd_health(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        (sonarr_h, radarr_h, lidarr_h,
         seerr_s, emby_s, nzb_s, qbt_s) = await asyncio.gather(
            self.sonarr.health(),
            self.radarr.health(),
            self.lidarr.health(),
            self.seerr.status(),
            self.emby.system_info(),
            self.nzbget.status(),
            self.qbt.transfer_info(),
            return_exceptions=True,
        )

        blocks: list[list[str]] = []

        def arr_block(label: str, data) -> list[str]:
            if isinstance(data, Exception):
                return [f"**{label}:** ❌ unreachable ({data})"]
            if not data:
                return [f"**{label}:** ✅ healthy"]
            out = [f"**{label}:** ⚠️ {len(data)} warning(s)"]
            for h in data:
                src = h.get("source") or "?"
                msg = h.get("message") or "?"
                out.append(f"  - [{src}] {msg}")
            return out

        blocks.append(arr_block("Sonarr", sonarr_h))
        blocks.append(arr_block("Radarr", radarr_h))
        blocks.append(arr_block("Lidarr", lidarr_h))

        if isinstance(seerr_s, Exception):
            blocks.append([f"**Seerr:** ❌ unreachable ({seerr_s})"])
        else:
            ver = seerr_s.get("version") or "?"
            update = seerr_s.get("commitsBehind")
            tag = f" (update available, {update} behind)" if update else ""
            blocks.append([f"**Seerr:** ✅ v{ver}{tag}"])

        media_label = self._media_label()
        if isinstance(emby_s, Exception):
            blocks.append([f"**{media_label}:** ❌ unreachable ({emby_s})"])
        else:
            ver = emby_s.get("Version") or "?"
            name = emby_s.get("ServerName") or media_label
            blocks.append([f"**{media_label}:** ✅ {name} v{ver}"])

        if isinstance(nzb_s, Exception):
            blocks.append([f"**NZBGet:** ❌ unreachable ({nzb_s})"])
        else:
            paused = nzb_s.get("DownloadPaused") or nzb_s.get("ServerStandBy")
            ver = nzb_s.get("Version") or "?"
            blocks.append([
                f"**NZBGet:** ⏸️ paused (v{ver})" if paused else f"**NZBGet:** ✅ v{ver}"
            ])

        if isinstance(qbt_s, Exception):
            blocks.append([f"**qBittorrent:** ❌ unreachable ({qbt_s})"])
        else:
            conn = qbt_s.get("connection_status") or "?"
            # 'connected' = healthy; 'firewalled' = reachable but inbound blocked;
            # 'disconnected' = no peers reachable (often VPN issue).
            icon = {"connected": "✅", "firewalled": "⚠️"}.get(conn, "❌")
            blocks.append([f"**qBittorrent:** {icon} {conn}"])

        await self._say(evt, "\n\n".join("\n".join(b) for b in blocks))

    def _arr_pct(self, q: dict) -> str:
        size = q.get("size") or 0
        left = q.get("sizeleft") or 0
        if size <= 0:
            return ""
        pct = (1 - left / size) * 100
        return f"{pct:.0f}%"

    # ---------- Downloads ----------

    @media.subcommand("activity", aliases=["dl"], help="NZBGet + qBittorrent active downloads")
    async def cmd_activity(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        nzb_res, qbt_res = await asyncio.gather(
            self.nzbget.listgroups(),
            self.qbt.downloading(),
            return_exceptions=True,
        )
        lines: list[str] = []
        # NZBGet
        if isinstance(nzb_res, Exception):
            lines.append(f"**NZBGet:** unreachable ({nzb_res})")
        else:
            active = [g for g in nzb_res if (g.get("RemainingSizeMB") or 0) > 0]
            lines.append(f"**NZBGet ({len(active)} active):**" if active else "**NZBGet:** idle")
            for g in active[:8]:
                name = g.get("NZBNicename") or g.get("NZBName") or "?"
                rate_mbps = (g.get("DownloadRate") or 0) / 1024 / 1024  # NZBGet reports B/s
                size_mb = g.get("FileSizeMB") or 0
                left_mb = g.get("RemainingSizeMB") or 0
                pct = (1 - left_mb / size_mb) * 100 if size_mb else 0
                lines.append(f"- *{name}* — {pct:.0f}% · {rate_mbps:.1f} MB/s")
        lines.append("")
        # qBittorrent
        if isinstance(qbt_res, Exception):
            lines.append(f"**qBittorrent:** unreachable ({qbt_res})")
        else:
            lines.append(f"**qBittorrent ({len(qbt_res)} active):**" if qbt_res else "**qBittorrent:** idle")
            for t in qbt_res[:8]:
                name = t.get("name") or "?"
                pct = (t.get("progress") or 0) * 100
                speed = _human_bytes(t.get("dlspeed") or 0) + "/s"
                eta = _human_eta(t.get("eta"))
                lines.append(f"- *{name}* — {pct:.0f}% · {speed} · ETA {eta}")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("speed", help="Aggregate down/up speeds across NZBGet + qBt")
    async def cmd_speed(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        nzb_status, qbt_xfer = await asyncio.gather(
            self.nzbget.status(), self.qbt.transfer_info(),
            return_exceptions=True,
        )
        lines: list[str] = []
        if isinstance(nzb_status, Exception):
            lines.append(f"**NZBGet:** unreachable ({nzb_status})")
        else:
            dl = nzb_status.get("DownloadRate") or 0  # bytes/s
            paused = nzb_status.get("DownloadPaused") or nzb_status.get("ServerPaused")
            tag = " (paused)" if paused else ""
            lines.append(f"**NZBGet:** ↓ {_human_bytes(dl)}/s{tag}")
        if isinstance(qbt_xfer, Exception):
            lines.append(f"**qBittorrent:** unreachable ({qbt_xfer})")
        else:
            dl = qbt_xfer.get("dl_info_speed") or 0
            ul = qbt_xfer.get("up_info_speed") or 0
            state = qbt_xfer.get("connection_status") or "?"
            lines.append(f"**qBittorrent:** ↓ {_human_bytes(dl)}/s · ↑ {_human_bytes(ul)}/s [{state}]")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("completed", help="Downloads finished in the last 24h")
    async def cmd_completed(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        cutoff = datetime.now(timezone.utc).timestamp() - 86400
        nzb_hist, qbt_all = await asyncio.gather(
            self.nzbget.history(), self.qbt.all_torrents(),
            return_exceptions=True,
        )
        lines: list[str] = []
        # NZBGet history items have HistoryTime (epoch) and FileSizeMB
        if isinstance(nzb_hist, Exception):
            lines.append(f"**NZBGet:** unreachable ({nzb_hist})")
        else:
            recent = [h for h in nzb_hist if (h.get("HistoryTime") or 0) >= cutoff]
            lines.append(f"**NZBGet — completed last 24h ({len(recent)}):**")
            for h in recent[:10]:
                name = h.get("Name") or h.get("NZBName") or "?"
                size_mb = h.get("FileSizeMB") or 0
                status = h.get("Status") or "?"
                lines.append(f"- *{name}* — {size_mb / 1024:.1f} GB [{status}]")
        lines.append("")
        # qBt: completion_on (epoch); finished torrents have it set
        if isinstance(qbt_all, Exception):
            lines.append(f"**qBittorrent:** unreachable ({qbt_all})")
        else:
            recent = [t for t in qbt_all
                      if (t.get("completion_on") or 0) >= cutoff
                      and (t.get("completion_on") or 0) > 0]
            lines.append(f"**qBittorrent — completed last 24h ({len(recent)}):**")
            for t in recent[:10]:
                name = t.get("name") or "?"
                size = t.get("size") or 0
                lines.append(f"- *{name}* — {_human_bytes(size)}")
        await self._say(evt, "\n".join(lines))

    @media.subcommand("pause", help="Pause all downloads (NZBGet + qBittorrent)")
    async def cmd_pause(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        results = await asyncio.gather(
            self.nzbget.pause(), self.qbt.pause_all(),
            return_exceptions=True,
        )
        nzb_ok = not isinstance(results[0], Exception)
        qbt_ok = not isinstance(results[1], Exception)
        msg = (
            f"NZBGet: {'paused ⏸' if nzb_ok else f'failed ({results[0]})'}\n"
            f"qBittorrent: {'paused ⏸' if qbt_ok else f'failed ({results[1]})'}"
        )
        await self._say(evt, msg)

    @media.subcommand("unpause", help="Resume all downloads (NZBGet + qBittorrent)")
    async def cmd_unpause(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        results = await asyncio.gather(
            self.nzbget.unpause(), self.qbt.resume_all(),
            return_exceptions=True,
        )
        nzb_ok = not isinstance(results[0], Exception)
        qbt_ok = not isinstance(results[1], Exception)
        msg = (
            f"NZBGet: {'resumed ▶' if nzb_ok else f'failed ({results[0]})'}\n"
            f"qBittorrent: {'resumed ▶' if qbt_ok else f'failed ({results[1]})'}"
        )
        await self._say(evt, msg)

    # ---------- Reactions: admin approve/decline ----------

    @event.on(EventType.REACTION)
    async def on_reaction(self, evt: ReactionEvent) -> None:
        relates = getattr(evt.content, "relates_to", None)
        if not relates or not relates.event_id or not relates.key:
            return
        request_id = self._pending_requests.get(relates.event_id)
        if not request_id:
            return  # not one of our request messages

        admins = self.config["admin_users"] or []
        if evt.sender not in admins:
            self.log.info("Reaction from non-admin %s ignored", evt.sender)
            return

        key = relates.key
        if _matches(key, APPROVE_KEYS):
            label, fn = "approved ✅", self.seerr.approve
        elif _matches(key, DECLINE_KEYS):
            label, fn = "declined ❌", self.seerr.decline
        else:
            return  # unrelated reaction

        try:
            await fn(request_id)
        except SeerrError as ex:
            await self.client.send_text(evt.room_id, text=f"Request {request_id} action failed: {ex}")
            return
        self._pending_requests.pop(relates.event_id, None)
        await self.client.send_text(
            evt.room_id, text=f"Request {request_id} {label} (by {evt.sender})"
        )

    # ---------- Webhook: Seerr → Matrix notifications ----------

    @web.post("/seerr-webhook")
    async def seerr_webhook(self, req: Request) -> Response:
        secret = self.config["seerr_webhook_secret"] or ""
        if secret:
            # Case-insensitive scheme per RFC 7235 — Seerr's UI lowercases "Bearer"
            auth = (req.headers.get("Authorization") or "").strip()
            scheme, _, token = auth.partition(" ")
            if scheme.lower() != "bearer" or token != secret:
                self.log.warning("Seerr webhook bad auth from %s", req.remote)
                return Response(status=401, text="unauthorized")

        try:
            payload = await req.json()
        except Exception:
            return Response(status=400, text="invalid json")

        room = (self.config["notifications_room"] or "").strip()
        if not room:
            return json_response({"ok": False, "error": "notifications_room not configured"})

        text, html = self._format_seerr_event(payload)
        try:
            content = TextMessageEventContent(
                msgtype=MessageType.NOTICE,
                body=text,
                format=Format.HTML,
                formatted_body=html,
            )
            await self.client.send_message(RoomID(room), content)
        except Exception as ex:
            self.log.exception("Failed to post Seerr webhook to %s", room)
            return json_response({"ok": False, "error": str(ex)}, status=500)

        poster = payload.get("image")
        if poster and self.config["posters_enabled"]:
            await self._post_poster(room, poster, payload.get("subject") or "poster")
        return json_response({"ok": True})

    def _format_seerr_event(self, p: dict) -> tuple[str, str]:
        nt = (p.get("notification_type") or "").upper()
        subject = p.get("subject") or "?"
        message = p.get("message") or ""
        media = p.get("media") or {}
        request = p.get("request") or {}
        media_type = media.get("media_type") or "?"
        requester = request.get("requestedBy_username") or ""

        emoji = {
            "MEDIA_PENDING": "📥",
            "MEDIA_APPROVED": "✅",
            "MEDIA_AUTO_APPROVED": "✅",
            "MEDIA_AVAILABLE": "🎬",
            "MEDIA_DECLINED": "❌",
            "MEDIA_FAILED": "⚠️",
            "ISSUE_CREATED": "🐛",
            "ISSUE_COMMENT": "💬",
            "ISSUE_RESOLVED": "🛠",
            "TEST_NOTIFICATION": "🔧",
        }.get(nt, "📣")

        label = nt.replace("_", " ").title() or "Notification"
        text_parts = [f"{emoji} {label} — {subject}"]
        if media_type and media_type != "?":
            text_parts.append(f"({media_type})")
        if requester:
            text_parts.append(f"— requested by {requester}")
        text = " ".join(text_parts)
        if message:
            text += f"\n\n{message}"

        html = (
            f"{emoji} <strong>{label}</strong> — <em>{subject}</em>"
            + (f" ({media_type})" if media_type != '?' else "")
            + (f" — requested by <strong>{requester}</strong>" if requester else "")
            + (f"<br/><br/>{message}" if message else "")
        )
        return text, html

    # ---------- Lidarr music ----------

    # Cache of last music lookup per (room, sender) so `!media music add <N>` works.
    _music_cache: dict[tuple[str, str], list[dict]] = {}

    @media.subcommand("music",
                      help="Search MusicBrainz via Lidarr. `add <q>` or `add <N>` to add.")
    @command.argument("query", pass_raw=True, required=True)
    async def cmd_music(self, evt: MessageEvent, query: str) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        parts = query.strip().split(None, 1)
        if parts and parts[0].lower() == "add":
            arg = parts[1] if len(parts) > 1 else ""
            await self._music_add(evt, arg)
            return
        await self._music_search(evt, query.strip())

    async def _music_search(self, evt: MessageEvent, term: str) -> None:
        if not term:
            await self._say(evt, "Need a search query, e.g. `!media music radiohead`.")
            return
        try:
            results = await self.lidarr.lookup(term)
        except ArrError as ex:
            await self._say(evt, f"Lidarr lookup failed: {ex}")
            return
        if not results:
            await self._say(evt, f"No artists found for **{term}**.")
            return
        n = self._result_count(evt.sender)
        shown = results[:n]
        self._music_cache[(evt.room_id, evt.sender)] = list(shown)
        if len(self._music_cache) > 200:
            del self._music_cache[next(iter(self._music_cache))]
        lines = [f"**Top {len(shown)} for '{term}':**"]
        for i, a in enumerate(shown, 1):
            name = a.get("artistName") or "?"
            disambig = a.get("disambiguation") or ""
            albums = a.get("statistics", {}).get("albumCount") or len(a.get("remoteAlbums") or [])
            extra = []
            if disambig:
                extra.append(disambig)
            if albums:
                extra.append(f"{albums} albums")
            tail = f" ({', '.join(extra)})" if extra else ""
            lines.append(f"{i}. *{name}*{tail}")
        lines.append("")
        lines.append("_Add with `!media music add <N>`._")
        await self._say(evt, "\n".join(lines))

    async def _music_add(self, evt: MessageEvent, arg: str) -> None:
        arg = arg.strip()
        if not arg:
            await self._say(evt, "`!media music add <query>` or `!media music add <N>`.")
            return
        # Numbered selection
        if arg.isdigit():
            cached = self._music_cache.get((evt.room_id, evt.sender)) or []
            idx = int(arg) - 1
            if not cached:
                await self._say(evt, "No recent music search to pick from. Run `!media music <query>` first.")
                return
            if idx < 0 or idx >= len(cached):
                await self._say(evt, f"Pick a number 1-{len(cached)}.")
                return
            artist = cached[idx]
        else:
            try:
                results = await self.lidarr.lookup(arg)
            except ArrError as ex:
                await self._say(evt, f"Lidarr lookup failed: {ex}")
                return
            if not results:
                await self._say(evt, f"No artists found for **{arg}**.")
                return
            artist = results[0]

        # Resolve profile/folder defaults if not set in config
        cfg = self.config["lidarr"] or {}
        try:
            qpid = cfg.get("quality_profile_id")
            mpid = cfg.get("metadata_profile_id")
            root = cfg.get("root_folder_path")
            if not qpid:
                qps = await self.lidarr.quality_profiles()
                qpid = qps[0]["id"] if qps else None
            if not mpid:
                mps = await self.lidarr.metadata_profiles()
                mpid = mps[0]["id"] if mps else None
            if not root:
                rfs = await self.lidarr.root_folders()
                root = rfs[0]["path"] if rfs else None
        except ArrError as ex:
            await self._say(evt, f"Lidarr defaults lookup failed: {ex}")
            return
        if not (qpid and mpid and root):
            await self._say(evt, "Lidarr is missing a quality/metadata profile or root folder — set up in Lidarr first.")
            return

        payload = {
            "artistName": artist.get("artistName"),
            "foreignArtistId": artist.get("foreignArtistId"),
            "qualityProfileId": qpid,
            "metadataProfileId": mpid,
            "rootFolderPath": root,
            "monitored": True,
            "monitorNewItems": "all",
            "addOptions": {
                "monitor": cfg.get("monitor", "all"),
                "searchForMissingAlbums": bool(cfg.get("search_on_add", True)),
            },
        }
        # Lidarr's add endpoint also expects images + a few other passthroughs from lookup
        for k in ("images", "links", "genres", "tags", "ratings", "overview",
                  "disambiguation", "artistType", "status", "remotePoster"):
            if k in artist:
                payload[k] = artist[k]
        try:
            added = await self.lidarr.add_artist(payload)
        except ArrError as ex:
            await self._say(evt, f"Add failed: {ex}")
            return
        name = added.get("artistName") or payload["artistName"]
        await self._say(
            evt,
            f"Added **{name}** to Lidarr. Quality: profile {qpid}, root: `{root}`. "
            f"{'Search kicked off.' if cfg.get('search_on_add', True) else 'No search triggered.'}"
        )

    # ---------- Subscriptions ----------

    @media.subcommand("subscribe", aliases=["sub"],
                      help="Subscribe to a Sonarr series — pings you when new episodes land")
    @command.argument("query", pass_raw=True, required=True)
    async def cmd_subscribe(self, evt: MessageEvent, query: str) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        try:
            series = await self.sonarr.list_series()
        except ArrError as ex:
            await self._say(evt, f"Sonarr lookup failed: {ex}")
            return
        q = query.lower().strip()
        matches = [s for s in series if q in (s.get("title") or "").lower()]
        if not matches:
            await self._say(evt, f"No Sonarr series matches **{query}**. Add it via Sonarr first.")
            return
        if len(matches) > 1:
            titles = ", ".join(f"*{s['title']}*" for s in matches[:5])
            await self._say(evt, f"Multiple matches: {titles}. Be more specific.")
            return
        s = matches[0]
        try:
            await self.database.execute(
                "INSERT OR REPLACE INTO subscriptions (mxid, sonarr_series_id, title) "
                "VALUES ($1, $2, $3)",
                evt.sender, s["id"], s["title"],
            )
        except Exception as ex:
            await self._say(evt, f"Subscribe failed: {ex}")
            return
        await self._say(evt, f"Subscribed to **{s['title']}**. You'll be pinged on new episode imports.")

    @media.subcommand("unsubscribe", aliases=["unsub"], help="Unsubscribe from a series")
    @command.argument("query", pass_raw=True, required=True)
    async def cmd_unsubscribe(self, evt: MessageEvent, query: str) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        rows = await self.database.fetch(
            "SELECT sonarr_series_id, title FROM subscriptions WHERE mxid = $1",
            evt.sender,
        )
        q = query.lower().strip()
        matches = [r for r in rows if q in (r["title"] or "").lower()]
        if not matches:
            await self._say(evt, f"You aren't subscribed to anything matching **{query}**.")
            return
        if len(matches) > 1:
            titles = ", ".join(f"*{r['title']}*" for r in matches[:5])
            await self._say(evt, f"Multiple matches: {titles}. Be more specific.")
            return
        m = matches[0]
        await self.database.execute(
            "DELETE FROM subscriptions WHERE mxid = $1 AND sonarr_series_id = $2",
            evt.sender, m["sonarr_series_id"],
        )
        await self._say(evt, f"Unsubscribed from **{m['title']}**.")

    @media.subcommand("subscriptions", aliases=["subs"], help="List your subscriptions")
    async def cmd_subscriptions(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        rows = await self.database.fetch(
            "SELECT title, added_at FROM subscriptions WHERE mxid = $1 ORDER BY title",
            evt.sender,
        )
        if not rows:
            await self._say(evt, "No subscriptions yet — `!media subscribe <show>` to add one.")
            return
        lines = [f"**Your subscriptions ({len(rows)}):**"]
        for r in rows:
            lines.append(f"- *{r['title']}*")
        await self._say(evt, "\n".join(lines))

    # ---------- Sonarr webhook ----------

    @web.post("/sonarr-webhook")
    async def sonarr_webhook(self, req: Request) -> Response:
        secret = self.config["sonarr_webhook_secret"] or ""
        if secret:
            auth = (req.headers.get("Authorization") or "").strip()
            scheme, _, token = auth.partition(" ")
            if scheme.lower() != "bearer" or token != secret:
                self.log.warning("Sonarr webhook bad auth from %s", req.remote)
                return Response(status=401, text="unauthorized")
        try:
            payload = await req.json()
        except Exception:
            return Response(status=400, text="invalid json")

        event_type = (payload.get("eventType") or "").lower()
        # Fire on Download (import) and Upgrade events; ignore Grab + Test
        if event_type == "test":
            return json_response({"ok": True, "test": True})
        if event_type not in ("download",):
            return json_response({"ok": True, "ignored": event_type})

        series = payload.get("series") or {}
        series_id = series.get("id")
        title = series.get("title") or "?"
        eps = payload.get("episodes") or []
        if not series_id:
            return json_response({"ok": False, "error": "no series.id"})

        subs = await self.database.fetch(
            "SELECT mxid FROM subscriptions WHERE sonarr_series_id = $1",
            series_id,
        )
        if not subs:
            return json_response({"ok": True, "subscribers": 0})

        room = (self.config["notifications_room"] or "").strip()
        if not room:
            return json_response({"ok": False, "error": "notifications_room not configured"})

        ep_tags = []
        for ep in eps:
            sn = ep.get("seasonNumber")
            en = ep.get("episodeNumber")
            if sn is not None and en is not None:
                ep_tags.append(f"S{sn:02d}E{en:02d}")
        ep_str = " ".join(ep_tags) if ep_tags else ""
        upgrade = " (upgrade)" if payload.get("isUpgrade") else ""

        mention_html = ", ".join(
            f'<a href="https://matrix.to/#/{r["mxid"]}">{r["mxid"]}</a>' for r in subs
        )
        mention_plain = ", ".join(r["mxid"] for r in subs)
        text = f"📺 New: {title} {ep_str}{upgrade} — {mention_plain}"
        html = f"📺 New: <strong>{title}</strong> {ep_str}{upgrade} — {mention_html}"
        try:
            content = TextMessageEventContent(
                msgtype=MessageType.NOTICE,
                body=text,
                format=Format.HTML,
                formatted_body=html,
            )
            await self.client.send_message(RoomID(room), content)
        except Exception as ex:
            self.log.exception("Sonarr webhook send failed")
            return json_response({"ok": False, "error": str(ex)}, status=500)
        return json_response({"ok": True, "subscribers": len(subs)})

    # ---------- Daily digest ----------

    @media.subcommand("digest", help="Manually fire today's digest")
    async def cmd_digest(self, evt: MessageEvent) -> None:
        await evt.mark_read()
        if await self._reject_unmapped(evt):
            return
        room = (self.config["notifications_room"] or "").strip()
        if not room:
            await self._say(evt, "Set `notifications_room` in config first.")
            return
        try:
            await self._send_digest(room)
            await self._say(evt, "Digest sent ✅")
        except Exception as ex:
            self.log.exception("digest failed")
            await self._say(evt, f"Digest failed: {ex}")

    async def _digest_loop(self) -> None:
        """Sleep until digest_hour each day, then send. Survives any error
        except CancelledError so a single failure doesn't kill the schedule."""
        try:
            while True:
                try:
                    hour = int(self.config["digest_hour"] or 8)
                    now = _local_now()
                    target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
                    if target <= now:
                        target += timedelta(days=1)
                    wait_s = (target - now).total_seconds()
                    self.log.info("Digest scheduled in %.0fs (next: %s)", wait_s, target.isoformat())
                    await asyncio.sleep(wait_s)

                    today = date.today().isoformat()
                    try:
                        row = await self.database.fetchrow(
                            "SELECT last_run_date FROM digest_state WHERE id = 1"
                        )
                    except Exception:
                        self.log.exception("digest_state read failed; firing anyway")
                        row = None
                    if row and row["last_run_date"] == today:
                        continue

                    room = (self.config["notifications_room"] or "").strip()
                    if not room:
                        continue
                    await self._send_digest(room)
                    try:
                        await self.database.execute(
                            "UPDATE digest_state SET last_run_date = $1 WHERE id = 1", today
                        )
                    except Exception:
                        self.log.exception("digest_state write failed (idempotency lost)")
                except asyncio.CancelledError:
                    raise
                except Exception:
                    self.log.exception("digest loop iteration failed; sleeping 1h then retrying")
                    await asyncio.sleep(3600)
        except asyncio.CancelledError:
            self.log.info("digest loop cancelled")
            raise

    async def _ensure_db_schema(self) -> None:
        """Defensive: confirm subscriptions + digest_state exist on startup.
        If maubot's UpgradeTable migration didn't run for some reason (it
        silently failed to create tables once — see v0.4 history), this
        creates them with IF NOT EXISTS so the bot self-heals."""
        if not getattr(self, "database", None):
            self.log.warning("self.database is None — !media subscribe/digest won't persist")
            return
        try:
            await self.database.execute(
                """
                CREATE TABLE IF NOT EXISTS subscriptions (
                    mxid             TEXT    NOT NULL,
                    sonarr_series_id INTEGER NOT NULL,
                    title            TEXT    NOT NULL,
                    added_at         TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (mxid, sonarr_series_id)
                )
                """
            )
            await self.database.execute(
                """
                CREATE TABLE IF NOT EXISTS digest_state (
                    id            INTEGER PRIMARY KEY CHECK (id = 1),
                    last_run_date TEXT
                )
                """
            )
            await self.database.execute(
                "INSERT INTO digest_state (id, last_run_date) VALUES (1, NULL) "
                "ON CONFLICT (id) DO NOTHING"
            )
            self.log.info("DB schema verified (subscriptions + digest_state)")
        except Exception:
            self.log.exception("DB schema bootstrap failed")

    async def _send_digest(self, room: str) -> None:
        """Build and post the daily digest to `room`.

        Skips the Completed section entirely when every finished release is
        already represented in Recently Added; skips Queued when both arrs are
        empty; splits long bodies (e.g. Sunday cleanup recap) across numbered
        messages so Matrix clients don't truncate.
        """
        cutoff = datetime.now(timezone.utc).timestamp() - 86400
        emby_uid = self._any_emby_uid("")
        added: list[dict] = []
        if emby_uid:
            try:
                added = await self.emby.recently_added(emby_uid, limit=20)
            except EmbyError:
                added = []
        nzb_hist, qbt_all, sonarr_q, radarr_q = await asyncio.gather(
            self.nzbget.history(), self.qbt.all_torrents(),
            self.sonarr.queue(), self.radarr.queue(),
            return_exceptions=True,
        )

        nzb_recent = []
        if not isinstance(nzb_hist, Exception):
            nzb_recent = [h for h in nzb_hist if (h.get("HistoryTime") or 0) >= cutoff]
        qbt_recent = []
        if not isinstance(qbt_all, Exception):
            qbt_recent = [t for t in qbt_all
                          if (t.get("completion_on") or 0) >= cutoff
                          and (t.get("completion_on") or 0) > 0]

        s_q = sonarr_q if not isinstance(sonarr_q, Exception) else []
        r_q = radarr_q if not isinstance(radarr_q, Exception) else []

        added_titles = {self._title_slug(self._emby_match_title(it)) for it in added}
        added_titles.discard("")

        def _is_unique(release_name: str) -> bool:
            slug = self._title_slug(release_name)
            if not slug:
                return True
            return not any(t and t in slug for t in added_titles)

        nzb_unique = [h for h in nzb_recent
                      if _is_unique(h.get("Name") or h.get("NZBName") or "")]
        qbt_unique = [t for t in qbt_recent if _is_unique(t.get("name") or "")]

        local = _local_now()
        is_sunday = local.weekday() == 6  # Mon=0..Sun=6
        today_str = local.strftime("%A, %b %d")
        lines = [f"**📰 Daily digest — {today_str}**", ""]
        lines.append(f"**🆕 Recently added ({len(added)}):**")
        for it in added[:8]:
            lines.append("- " + self._fmt_emby_item(it))
        if not added:
            lines.append("- (nothing in the last sweep)")

        if nzb_unique or qbt_unique:
            lines.append("")
            lines.append(
                f"**✅ Also completed (not yet in {self._media_label()}): "
                f"{len(nzb_unique) + len(qbt_unique)}** "
                f"(NZBGet: {len(nzb_unique)} · qBt: {len(qbt_unique)})"
            )
            for h in nzb_unique[:5]:
                name = h.get("Name") or h.get("NZBName") or "?"
                size_mb = h.get("FileSizeMB") or 0
                lines.append(f"- *{name}* — {size_mb / 1024:.1f} GB")
            for t in qbt_unique[:5]:
                lines.append(f"- *{t.get('name','?')}* — {_human_bytes(t.get('size') or 0)}")

        if s_q or r_q:
            lines.append("")
            lines.append(f"**📥 Queued: {len(s_q)} TV · {len(r_q)} movies**")

        if is_sunday:
            cleaner_lines = await self._fetch_emby_cleaner_recap()
            if cleaner_lines:
                lines.append("")
                lines.append("**🧹 Weekly cleanup (emby-cleaner):**")
                lines.extend(cleaner_lines)

        chunks = self._chunk_lines(lines)
        total = len(chunks)
        for idx, chunk in enumerate(chunks, 1):
            body = chunk if total == 1 else f"**({idx}/{total})**\n\n{chunk}"
            await self.client.send_message(
                RoomID(room),
                TextMessageEventContent(msgtype=MessageType.NOTICE, body=body),
            )

    @staticmethod
    def _emby_match_title(it: dict) -> str:
        """Pick the title that should match a release name — series name for
        episodes, item name otherwise."""
        if it.get("Type") == "Episode":
            return it.get("SeriesName") or it.get("Name") or ""
        return it.get("Name") or ""

    @staticmethod
    def _title_slug(s: str) -> str:
        """Normalize a title or release name to lowercase alphanumerics so
        'The.Bear.S03E10.1080p...' and 'The Bear' compare equal-ish."""
        return "".join(c for c in s.lower() if c.isalnum())

    @staticmethod
    def _chunk_lines(lines: list[str], max_len: int = 3000) -> list[str]:
        """Pack lines into chunks <= max_len chars on line boundaries."""
        body = "\n".join(lines)
        if len(body) <= max_len:
            return [body]
        chunks: list[str] = []
        cur: list[str] = []
        cur_len = 0
        for line in lines:
            extra = len(line) + (1 if cur else 0)
            if cur and cur_len + extra > max_len:
                chunks.append("\n".join(cur))
                cur = [line]
                cur_len = len(line)
            else:
                cur.append(line)
                cur_len += extra
        if cur:
            chunks.append("\n".join(cur))
        return chunks

    async def _fetch_emby_cleaner_recap(self) -> list[str]:
        """Pull the last 12h of messages from the emby-cleaner ntfy topic."""
        ntfy = (self.config["ntfy_url"] or "").rstrip("/")
        topic = self.config["emby_cleaner_topic"] or "emby-cleaner"
        if not ntfy:
            return []
        url = f"{ntfy}/{topic}/json?poll=1&since=12h"
        try:
            async with self.session.get(url) as r:
                if r.status >= 400:
                    self.log.info("ntfy fetch %s → %s", url, r.status)
                    return []
                text = await r.text()
        except Exception as ex:
            self.log.warning("ntfy fetch failed: %s", ex)
            return []
        out: list[str] = []
        for raw in text.splitlines():
            raw = raw.strip()
            if not raw:
                continue
            try:
                msg = __import__("json").loads(raw)
            except Exception:
                continue
            title = msg.get("title") or ""
            body = (msg.get("message") or "").strip()
            if title:
                out.append(f"- *{title}*")
            for line in body.splitlines()[:6]:
                line = line.strip()
                if line:
                    out.append(f"  {line}")
        return out[:25]