weircon-front/mcp_server.py

"""Portalens egen MCP-server (FastMCP) — read-only indblik i portal/homelab-data
plus en ntfy-notify-handling. Monteres in-process i main.py (uvicorn-lifespan).

Tools genbruger de eksisterende portal-moduler. Alt pakkes i try/except, så en
downstream-fejl giver en pæn fejl frem for at vælte MCP-kaldet.
"""
import asyncio

from fastmcp import FastMCP

import authentik as ak
import config
import mediaclean as mc
import services as svc
import store
import versions
import ytdl as yt

mcp = FastMCP("weircon-portal")


def _safe(v):
    return v


@mcp.tool
async def service_versions() -> dict:
    """Versions/opdaterings-overblik for de overvågede tjenester (current vs latest)."""
    try:
        return {"services": await versions.all_versions()}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def matomo_summary() -> dict:
    """Kort Matomo-trafikoverblik for weircon.dk."""
    try:
        return await svc.matomo_summary()
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def egress_status() -> dict:
    """Random-proxy egress-status: tunneler, rollover, inflight."""
    try:
        return await svc.rp_status()
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def portal_apps() -> dict:
    """Katalog over Authentik-apps (navn, gruppe, host) som portalen kender."""
    try:
        return {"apps": await ak.fetch_all_apps_admin()}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
def media_cleanup_status() -> dict:
    """Status for medie-oprydningen: indstillinger, state (på-vej-ud/behold) og om den er konfigureret."""
    try:
        return {"configured": mc.configured(), "settings": mc.get_settings(),
                "state": mc.get_state()}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
def ip_labels() -> dict:
    """IP/præfiks → label & land som vist i statistikken."""
    try:
        return {"labels": store.load_json(config.IP_LABELS_LOCAL, {})}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def notify(message: str, title: str = "Claude") -> dict:
    """Send en push-notifikation til dig via ntfy."""
    try:
        await mc.ntfy(title, message)
        return {"sent": True}
    except Exception as e:
        return {"sent": False, "error": str(e)[:200]}


# ---------------- YouTube offline ----------------
@mcp.tool
def yt_status() -> dict:
    """Status for YouTube-offline: hentede videoer (set/favorit/hvornår de slettes),
    download-kø, overvågede playlister/kanaler og de aktuelle indstillinger.

    Videoer slettes 7 dage efter de er markeret som set i Emby. Favoritter slettes aldrig.
    """
    try:
        st = yt.get_state()
        owned = yt.owned_paths()
        leaving = st["leaving"]
        videos = [{"titel": t, "sti": p, "stoerrelse": yt._human(yt.file_size(p)),
                   "slettes_efter": (leaving.get(p) or {}).get("delete_after")}
                  for p, t in owned.items()]
        jobs = sorted(st["jobs"].values(), key=lambda j: j["added"], reverse=True)
        return {
            "konfigureret": yt.configured(),
            "indstillinger": yt.get_settings(),
            "videoer": videos,
            "paa_vej_ud": len(leaving),
            "koe": [{"url": j["url"], "status": j["status"], "titel": j.get("title"),
                     "fejl": j.get("error")} for j in jobs[:15]
                    if j["status"] in ("queued", "running", "error")],
            "abonnementer": [
                {"id": sb["id"], "navn": sb["title"], "type": sb["kind"],
                 "cycle": sb.get("cycle") or None, "aktiv": sb.get("enabled", True),
                 "paa_disken": len(yt.sub_files(st, sb["id"]))}
                for sb in st["subs"].values()],
        }
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def yt_download(url: str, playlist: bool = False) -> dict:
    """Hent en YouTube-video ned til Emby-biblioteket, klar til offline-brug.

    Returnerer med det samme; selve hentningen kører i baggrunden (kan tage minutter
    for en lang video). Brug yt_status til at følge den. Sæt playlist=True for at
    hente hele playlisten bag URL'en i stedet for kun den enkelte video.
    """
    try:
        url = (url or "").strip()
        if not url.startswith(("http://", "https://")):
            return {"error": "Ikke en URL"}
        job = yt.add_job(url, playlist=bool(playlist))
        asyncio.create_task(yt.process_queue())
        return {"lagt_i_koe": True, "job": job["id"], "url": job["url"],
                "bemaerk": "hentes i baggrunden — se yt_status"}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def yt_watch(url: str, kind: str = "", cycle: int = 0) -> dict:
    """Overvåg en playliste eller kanal og hent nye videoer automatisk.

    kind = "playlist" eller "channel" (tom = gæt ud fra URL'en).
    For en kanal SKAL cycle sættes: cycle=X holder de X nyeste videoer liggende,
    og en ny video skubber den ældste ud. Favoritter tæller ikke med i X og slettes
    aldrig. En playliste henter kun videoer der lægges i den fremover — ikke dem
    der allerede ligger der.
    """
    try:
        url = (url or "").strip()
        if not url.startswith(("http://", "https://")):
            return {"error": "Ikke en URL"}
        kind = kind if kind in ("playlist", "channel") else yt.sub_kind_from_url(url)
        if kind == "channel" and int(cycle or 0) < 1:
            return {"error": "En kanal kraever cycle >= 1 (antal nyeste videoer der holdes)"}
        sub = await yt.add_sub(url, kind=kind, cycle=int(cycle or 0) if kind == "channel" else 0)
        asyncio.create_task(_yt_sync_all())
        return {"overvaages": True, "id": sub["id"], "navn": sub["title"],
                "type": sub["kind"], "cycle": sub["cycle"] or None,
                "bemaerk": "tjekker og henter i baggrunden"}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
def yt_unwatch(sub_id: str) -> dict:
    """Stop med at overvåge en playliste/kanal. Allerede hentede videoer bliver
    liggende og følger herefter kun den almindelige set-regel. Id'et fås fra yt_status."""
    try:
        return {"fjernet": yt.del_sub(sub_id)}
    except Exception as e:
        return {"error": str(e)[:200]}


@mcp.tool
async def yt_check_subs() -> dict:
    """Tjek alle overvågede playlister/kanaler nu i stedet for at vente på natkørslen."""
    try:
        res = await yt.sync_subs()
        asyncio.create_task(_yt_sync_all())
        return {"tjekket": res["checked"], "nye_i_koe": res["queued"],
                "fejl": res["errors"]}
    except Exception as e:
        return {"error": str(e)[:200]}


async def _yt_sync_all():
    """Baggrundsarbejde: hent kø og haandhaev kanal-cyklus. Fejl slugges bevidst —
    den natlige timer koerer det samme igen."""
    try:
        await yt.process_queue()
        await yt.enforce_cycles()
    except Exception:
        pass