"""YouTube-offline: hent videoer med yt-dlp, servér dem via Emby, og ryd op når de er SET.

Bevidst forskel fra mediaclean.py: dér tæller selv delvis afspilning som "set", fordi
formålet dér er at frede alt der er rørt. Her er det modsat — en video slettes KUN når
den er eksplicit **markeret som set** (Emby `UserData.Played`). Halvt sete videoer og
videoer der er lagt fra sig midtvejs bliver liggende. Favoritter slettes aldrig.

Sikkerhedsnettet er lige så vigtigt som reglen: oprydningen rører **kun filer portalen
selv har hentet**, registreret med fuld sti i yt_state.json. Alt andet i YouTube-
biblioteket — fx det der lå i /slow/shared/yt i forvejen — er urørligt.
"""
import asyncio
import fcntl
import json
import os
import re
import shutil
import time
import uuid
from xml.sax.saxutils import escape

import httpx

import config

TICKS = 10_000_000  # Emby: 1 sek = 10^7 ticks


# ---------------- state + settings ----------------
def _load(path, default):
    try:
        return json.loads(path.read_text("utf-8"))
    except Exception:
        return default


def _save(path, data):
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), "utf-8")


def get_settings() -> dict:
    s = dict(config.YT_DEFAULTS)
    s.update(_load(config.YT_SETTINGS_PATH, {}) or {})
    return s


def save_settings(patch: dict) -> dict:
    s = get_settings()
    s.update(patch)
    _save(config.YT_SETTINGS_PATH, s)
    return s


def get_state() -> dict:
    st = _load(config.YT_STATE_PATH, {}) or {}
    st.setdefault("jobs", {})      # id -> {url,status,title,path,size,added,finished,error}
    st.setdefault("leaving", {})   # sti -> {title,size,first_seen,delete_after}
    st.setdefault("log", [])       # seneste hændelser
    st.setdefault("subs", {})      # id -> abonnement (playliste/kanal), se add_sub()
    return st


def save_state(st: dict):
    _save(config.YT_STATE_PATH, st)


def _log(st: dict, text: str):
    st["log"].insert(0, {"t": time.time(), "text": text})
    del st["log"][60:]


def configured() -> bool:
    return bool(config.EMBY_URL and config.EMBY_API_KEY
                and shutil.which(config.YT_DLP_BIN)
                and os.path.isdir(config.YT_DOWNLOAD_DIR))


def _human(n: int) -> str:
    n = float(n or 0)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if n < 1024 or unit == "TB":
            return f"{n:.1f} {unit}" if unit != "B" else f"{int(n)} B"
        n /= 1024


# ---------------- download ----------------
def add_job(url: str, playlist: bool = False, sub_id: str = "",
            video_id: str = "") -> dict:
    """Læg en URL i køen. Selve hentningen sker i process_queue().

    `sub_id` binder jobbet til et abonnement. Det er dét felt kanal-cyklussen
    tæller på — manuelle hentninger (sub_id="") og playliste-abonnementer tæller
    aldrig med i en kanals X, præcis som aftalt.
    """
    st = get_state()
    jid = uuid.uuid4().hex[:12]
    job = {"id": jid, "url": url.strip(), "playlist": bool(playlist), "status": "queued",
           "title": "", "path": "", "size": 0, "added": time.time(),
           "finished": 0, "error": "", "sub_id": sub_id, "video_id": video_id,
           "videos": []}
    st["jobs"][jid] = job
    _log(st, f"Lagt i kø: {url.strip()[:80]}")
    save_state(st)
    return job


def _set_job(jid: str, **kw):
    st = get_state()
    job = st["jobs"].get(jid)
    if not job:
        return
    job.update(kw)
    save_state(st)


def _build_cmd(job: dict, settings: dict, with_subs: bool, klient: str = "") -> list:
    # yt-dlp henter til en skjult staging-mappe under video-id'et; organiser()
    # flytter bagefter videoen ind som Kanal/Season ÅÅÅÅ/... med NFO og billede.
    out = os.path.join(staging_dir(), "%(id)s.%(ext)s")
    cmd = [config.YT_DLP_BIN,
           "-f", settings.get("quality") or config.YT_DEFAULTS["quality"],
           "--merge-output-format", "mp4",
           "--no-progress", "--newline", "--no-warnings",
           "--concurrent-fragments", "4",
           "--retries", "5", "--sleep-requests", "1",
           "--print", "after_move:%(id)s|%(upload_date)s|%(filepath)s",
           # Titel, kanal, beskrivelse og upload-dato til NFO'en, og YouTubes
           # miniature som afsnitsbillede i Emby.
           "--write-info-json", "--write-thumbnail", "--convert-thumbnails", "jpg",
           "-o", out]

    if settings.get("metadata", True):
        # Titel/kanal/dato ind i filen + miniature som cover, så Emby har noget at
        # vise uden at skulle gætte ud fra filnavnet.
        cmd += ["--embed-metadata", "--embed-thumbnail", "--embed-chapters"]

    if with_subs:
        # "hvis de er der": rigtige undertekster først, ellers YouTubes automatiske.
        # Både indlejret (rejser med filen når Emby-appen henter den ned) og som
        # .srt-sidecar. Sidecars ryddes med i slettetrinnet.
        cmd += ["--sub-langs", settings.get("sub_langs") or config.YT_DEFAULTS["sub_langs"],
                "--write-subs", "--write-auto-subs", "--convert-subs", "srt", "--embed-subs"]

    if settings.get("sponsorblock", True):
        # Klip sponsorater/selvpromovering ud, og behold resten som kapitler man
        # selv kan springe i. Kræver ffmpeg. Videoer uden indsendte segmenter
        # påvirkes ikke (yt-dlp henter dem fra SponsorBlock-API'et pr. video).
        cmd += ["--sponsorblock-remove", "default", "--sponsorblock-mark", "all"]

    if klient:
        # Valgfrit: bed YouTube om videoen som en anden klient (se FORSOEG).
        cmd += ["--extractor-args", f"youtube:player_client={klient}"]

    if not job.get("playlist"):
        cmd.append("--no-playlist")
    else:
        # En enkelt privat/slettet video må ikke vælte resten af playlisten.
        cmd.append("--ignore-errors")
    cmd.append(job["url"])
    return cmd


async def _attempt(job: dict, settings: dict, with_subs: bool, klient: str = "") -> dict:
    dest = config.YT_DOWNLOAD_DIR
    proc = await asyncio.create_subprocess_exec(
        *_build_cmd(job, settings, with_subs, klient),
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    stdout, stderr = await proc.communicate()
    if proc.returncode is not None and proc.returncode < 0:
        # Dræbt udefra, typisk fordi portalen genstartede midt i hentningen.
        return {"ok": False, "error": f"afbrudt (signal {-proc.returncode})"}

    # --print after_move giver én linje pr. hentet fil: id|upload_date|sti.
    # Video-id og upload-dato gemmes, fordi cyklussen skal kunne sortere efter
    # hvor ny videoen ER, ikke hvornår vi tilfældigvis hentede den.
    videos = []
    for ln in stdout.decode("utf-8", "replace").splitlines():
        parts = ln.split("|", 2)
        if len(parts) == 3 and parts[2].startswith(dest):
            videos.append({"id": parts[0], "upload_date": parts[1].strip("NA "),
                           "path": parts[2]})
    err = (stderr.decode("utf-8", "replace").strip().splitlines()
           or [f"ukendt fejl (kode {proc.returncode})"])[-1][:300]
    if proc.returncode != 0 and not (job.get("playlist") and videos):
        return {"ok": False, "error": err}
    if not videos:
        return {"ok": False, "error": "yt-dlp meldte ingen fil — allerede hentet?"}

    # Ind i Kanal/Season ÅÅÅÅ/ med NFO og billede, så Emby viser det som en serie.
    titler = []
    for v in videos:
        o = await organiser(v["path"])
        v["path"] = o["path"]
        v["upload_date"] = o["upload_date"] or v["upload_date"]
        titler.append(o["title"])
    paths = [v["path"] for v in videos]
    size = sum(file_size(p) for p in paths if os.path.exists(p))
    title = titler[0] if len(titler) == 1 else f"{titler[0]} (+{len(titler) - 1} fra playlisten)"
    noter = [] if with_subs else ["uden undertekster"]
    if proc.returncode != 0:
        noter.append(f"nogle videoer i playlisten blev sprunget over ({err[:120]})")
    return {"ok": True, "title": title, "path": paths[0], "paths": paths,
            "videos": videos, "size": size, "note": "; ".join(noter)}


# ---------------- Emby-struktur: kanal = serie ----------------
# Hver kanal bliver en serie i et Emby "Serier"-bibliotek:
#   <Kanal>/tvshow.nfo, poster.jpg (kanalens profilbillede), fanart.jpg (banner)
#   <Kanal>/Season ÅÅÅÅ/<Kanal> - sÅÅÅÅeMMDDnn - <Titel> [<id>].mp4
#     + .nfo (titel, beskrivelse, upload-dato) + -thumb.jpg + .<sprog>.srt
# Afsnitsnummeret er upload-datoen (MMDD) + et løbenummer, så Emby sorterer
# efter hvornår videoen kom på YouTube, ikke hvornår vi hentede den.
STAGING_NAVN = ".staging"
KANAL_FILER = ("tvshow.nfo", "poster.jpg", "fanart.jpg", "banner.jpg")
_UGYLDIG = re.compile(r'[\\/:*?"<>|\x00-\x1f]')


def staging_dir() -> str:
    d = os.path.join(config.YT_DOWNLOAD_DIR, STAGING_NAVN)
    os.makedirs(d, exist_ok=True)
    return d


def _rent(navn: str, maks: int = 120) -> str:
    s = re.sub(r"\s+", " ", _UGYLDIG.sub("", navn or "")).strip().rstrip(". ")
    return s[:maks].rstrip(". ") or "Ukendt"


def _dato(d: str) -> str:
    return f"{d[:4]}-{d[4:6]}-{d[6:8]}" if d and len(d) == 8 and d.isdigit() else ""


def _skriv_tekst(sti: str, tekst: str) -> None:
    tmp = sti + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(tekst)
    os.replace(tmp, sti)


def _nfo(rod: str, felter: list) -> str:
    linjer = ['<?xml version="1.0" encoding="utf-8" standalone="yes"?>', f"<{rod}>"]
    for tag, v, attr in felter:
        if v not in (None, ""):
            linjer.append(f"  <{tag}{attr}>{escape(str(v))}</{tag}>")
    linjer += ["  <lockdata>true</lockdata>", f"</{rod}>", ""]
    return "\n".join(linjer)


async def _hent_billede(url: str, sti: str) -> bool:
    try:
        async with httpx.AsyncClient(timeout=30, follow_redirects=True) as c:
            r = await c.get(url)
            if r.status_code == 200 and r.content:
                with open(sti + ".tmp", "wb") as f:
                    f.write(r.content)
                os.replace(sti + ".tmp", sti)
                return True
    except Exception:                                             # noqa: BLE001
        pass
    return False


async def _kanal_info(kanal_dir: str, kanal: str, info: dict) -> None:
    """tvshow.nfo + plakat/baggrund for kanalen (kun når de mangler)."""
    nfo = os.path.join(kanal_dir, "tvshow.nfo")
    if os.path.exists(nfo) and os.path.exists(os.path.join(kanal_dir, "poster.jpg")):
        return
    url = info.get("channel_url") or (
        f"https://www.youtube.com/channel/{info['channel_id']}" if info.get("channel_id") else "")
    meta = {}
    if url:
        try:
            proc = await asyncio.create_subprocess_exec(
                config.YT_DLP_BIN, "--dump-single-json", "--flat-playlist", "--playlist-items", "0",
                "--no-warnings", url, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
            out, _ = await asyncio.wait_for(proc.communicate(), timeout=120)
            meta = json.loads(out.decode("utf-8", "replace") or "{}")
        except Exception:                                         # noqa: BLE001
            meta = {}
    if not os.path.exists(nfo):
        _skriv_tekst(nfo, _nfo("tvshow", [
            ("title", kanal, ""), ("plot", meta.get("description") or "", ""),
            ("studio", "YouTube", ""),
            ("uniqueid", info.get("channel_id") or "", ' type="youtube" default="true"')]))
    thumbs = meta.get("thumbnails") or []
    def find(id_, pred):
        for t in thumbs:
            if t.get("id") == id_ and t.get("url"):
                return t["url"]
        kandidater = [t for t in thumbs if t.get("url") and pred(t)]
        kandidater.sort(key=lambda t: (t.get("width") or 0), reverse=True)
        return kandidater[0]["url"] if kandidater else ""
    avatar = find("avatar_uncropped", lambda t: t.get("width") and t.get("width") == t.get("height"))
    banner = find("banner_uncropped", lambda t: (t.get("width") or 0) > 2 * (t.get("height") or 1))
    if avatar and not os.path.exists(os.path.join(kanal_dir, "poster.jpg")):
        await _hent_billede(avatar, os.path.join(kanal_dir, "poster.jpg"))
    if banner and not os.path.exists(os.path.join(kanal_dir, "fanart.jpg")):
        await _hent_billede(banner, os.path.join(kanal_dir, "fanart.jpg"))


async def organiser(staging_path: str, info: dict | None = None) -> dict:
    """Flyt en hentet video fra staging ind i kanalens serie, med NFO og billede.
    Returnerer {path, title, upload_date}."""
    sd = os.path.dirname(staging_path)
    vid = os.path.basename(staging_path).split(".")[0]
    if info is None:
        try:
            with open(os.path.join(sd, f"{vid}.info.json"), encoding="utf-8") as f:
                info = json.load(f)
        except (OSError, ValueError):
            info = {}
    kanal = _rent(info.get("channel") or info.get("uploader") or "Ukendt kanal", 80)
    dato = info.get("upload_date") or info.get("release_date") or time.strftime("%Y%m%d")
    aar, md = dato[:4], dato[4:8]
    kanal_dir = os.path.join(config.YT_DOWNLOAD_DIR, kanal)
    saeson_dir = os.path.join(kanal_dir, f"Season {aar}")
    os.makedirs(saeson_dir, exist_ok=True)
    await _kanal_info(kanal_dir, kanal, info)

    # Samme video igen (fx nyt forsøg): genbrug dens afsnitsnummer.
    ep, brugt = 0, set()
    for f in os.listdir(saeson_dir):
        m = re.search(rf" - s{aar}e(\d{{6}}) - ", f)
        if not m:
            continue
        if f"[{vid}]" in f:
            ep = int(m.group(1))
        brugt.add(int(m.group(1)))
    if not ep:
        base = int(md) * 100 if md.isdigit() else 0
        ep = next(base + n for n in range(1, 100) if base + n not in brugt)
    titel = info.get("title") or vid
    stem = f"{kanal} - s{aar}e{ep:06d} - {_rent(titel)} [{vid}]"
    final = os.path.join(saeson_dir, stem + os.path.splitext(staging_path)[1])
    os.replace(staging_path, final)
    for f in os.listdir(sd):
        if not f.startswith(vid + "."):
            continue
        src, rest = os.path.join(sd, f), f[len(vid) + 1:]
        if rest == "info.json":
            os.remove(src)
        elif rest in ("jpg", "jpeg", "png", "webp"):
            os.replace(src, os.path.join(saeson_dir, f"{stem}-thumb.{'jpg' if rest == 'jpeg' else rest}"))
        elif not rest.endswith(".part"):
            os.replace(src, os.path.join(saeson_dir, f"{stem}.{rest}"))
    varighed = info.get("duration")
    _skriv_tekst(os.path.join(saeson_dir, stem + ".nfo"), _nfo("episodedetails", [
        ("title", titel, ""), ("showtitle", kanal, ""), ("season", aar, ""), ("episode", ep, ""),
        ("plot", info.get("description") or "", ""), ("aired", _dato(dato), ""),
        ("premiered", _dato(dato), ""), ("year", aar, ""),
        ("runtime", int(varighed // 60) or "" if varighed else "", ""), ("studio", "YouTube", ""),
        ("uniqueid", vid, ' type="youtube" default="true"')]))
    return {"path": final, "title": titel, "upload_date": dato}


def _ryd_tomme_mapper(saeson_dir: str) -> None:
    """Efter en sletning: fjern tom sæsonmappe, og hele kanalen når den ikke har
    flere videoer. Rører kun mapper under download-mappen og kun kendte filer."""
    rod = os.path.realpath(config.YT_DOWNLOAD_DIR)
    s = os.path.realpath(saeson_dir)
    if s == rod or not s.startswith(rod + os.sep):
        return
    try:
        kanal = os.path.dirname(s) if os.path.basename(s).startswith("Season ") else s
        if s != kanal and not os.listdir(s):
            os.rmdir(s)
        if kanal == rod or not kanal.startswith(rod + os.sep) or os.path.basename(kanal) == STAGING_NAVN:
            return
        for _, _, fs in os.walk(kanal):
            if any(f.lower().endswith((".mp4", ".mkv", ".webm")) for f in fs):
                return
        for f in KANAL_FILER:
            if os.path.exists(os.path.join(kanal, f)):
                os.remove(os.path.join(kanal, f))
        for d in os.listdir(kanal):
            dd = os.path.join(kanal, d)
            if d.startswith("Season ") and os.path.isdir(dd) and not os.listdir(dd):
                os.rmdir(dd)
        if not os.listdir(kanal):
            os.rmdir(kanal)
    except OSError:
        pass


def _slet_video(path: str) -> int:
    """Slet videoen med alle sidecars og ryd tomme mapper. Returnerer bytes."""
    size = 0
    for side in _sidecars(path):
        size += os.path.getsize(side)
        os.remove(side)
    if os.path.exists(path):
        size += os.path.getsize(path)
        os.remove(path)
    _ryd_tomme_mapper(os.path.dirname(path))
    return size


# Fejl der ofte går over af sig selv: YouTube afviser en enkelt forespørgsel
# (403/429), eller forbindelsen falder. yt-dlp gentager ikke selv en 403.
_MIDLERTIDIG = ("403", "forbidden", "429", "too many requests", "timed out",
                "connection reset", "remote end closed", "temporary failure", "afbrudt")
# Forsøg: (pause før, YouTube-klient). Samme klient med stigende pause: testet
# 2026-10-06 gav kun standardklienten 1080p (android_vr faldt til 360p, tv,
# web_safari, mweb og tv_simply gav slet ingen formater), og 403'erne kom af
# mange samtidige hentninger, som kø-låsen nu forhindrer. Klientfeltet bliver
# stående, hvis en anden klient senere viser sig bedre.
FORSOEG = [(0, ""), (30, ""), (120, "")]


def midlertidig(err: str) -> bool:
    e = (err or "").lower()
    return any(m in e for m in _MIDLERTIDIG)


async def _run_ytdlp(job: dict, settings: dict) -> dict:
    """Kør yt-dlp for ét job. Returnerer {ok, title, path, size, error}.

    Undertekster er et gode, ikke et krav: YouTube rate-limiter dem (HTTP 429)
    uafhængigt af selve videoen, og uden dette fald-tilbage ville en 429 på en
    undertekstfil vælte hele hentningen. Vi prøver derfor én gang mere uden dem.

    Midlertidige fejl (403/429/timeout) prøves igen op til to gange med
    stigende pause (FORSOEG).
    """
    want_subs = bool(settings.get("subtitles", True))
    res = {"ok": False, "error": "ingen forsøg"}
    for nr, (pause, klient) in enumerate(FORSOEG, start=1):
        if pause:
            await asyncio.sleep(pause)
        res = await _attempt(job, settings, with_subs=want_subs, klient=klient)
        if not res["ok"] and want_subs and "subtitle" in res["error"].lower():
            res = await _attempt(job, settings, with_subs=False, klient=klient)
            if res["ok"]:
                res["note"] = "undertekster sprunget over (YouTube afviste dem lige nu)"
        if res["ok"]:
            if nr > 1:
                res["note"] = "; ".join(x for x in (res.get("note"), f"lykkedes i forsøg {nr}") if x)
            return res
        if not midlertidig(res["error"]):
            return res
    res["error"] = f"{res['error']} (opgivet efter {len(FORSOEG)} forsøg)"
    return res


def _sidecars(path: str) -> list:
    """Filer der hører til videoen (undertekster, miniature) og skal slettes med den.

    yt-dlp navngiver dem efter samme stamme: "Titel [id].da.srt", "... .en.srt",
    "... .jpg". Uden dette trin ville en slettet video efterlade sine .srt-filer,
    og Emby ville blive ved med at vise et tomt punkt i biblioteket.
    """
    stem = path.rsplit(".", 1)[0]
    folder = os.path.dirname(path) or "."
    base = os.path.basename(stem)
    out = []
    try:
        for f in os.listdir(folder):
            full = os.path.join(folder, f)
            if full != path and (f.startswith(base + ".") or f.startswith(base + "-thumb.")) \
                    and os.path.isfile(full):
                out.append(full)
    except OSError:
        pass
    return out


def _koe_laas():
    """Én kørsel af køen ad gangen, også på tværs af processer (web + timer).
    Mange samtidige hentninger fra samme IP er netop det, YouTube svarer 403 på."""
    f = open(os.path.join(config.DATA, "yt_koe.lock"), "w")
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        f.close()
        return None
    return f


def retry_job(jid: str) -> bool:
    """Læg et fejlet job i kø igen (knappen "Prøv igen")."""
    st = get_state()
    j = st["jobs"].get(jid)
    if not j or j["status"] != "error":
        return False
    j.update(status="queued", error="", finished=0)
    _log(st, f"Prøver igen: {(j.get('title') or j['url'])[:80]}")
    save_state(st)
    return True


async def process_queue() -> dict:
    """Hent alt der står i kø, ét job ad gangen. Kaldes både fra web-ruten og
    fra timeren; kører en anden allerede køen, gør denne ingenting."""
    laas = _koe_laas()
    if laas is None:
        return {"done": 0, "failed": 0, "busy": True}
    try:
        return await _process_queue()
    finally:
        laas.close()


async def _process_queue() -> dict:
    settings = get_settings()
    done, failed = 0, 0
    # Vi har låsen, så et job der står som "running" blev afbrudt (fx af en
    # genstart af portalen midt i hentningen). Læg det i kø igen.
    st = get_state()
    haengt = [j for j in st["jobs"].values() if j["status"] == "running"]
    for j in haengt:
        j["status"] = "queued"
        _log(st, f"Genoptager afbrudt hentning: {(j.get('title') or j['url'])[:80]}")
    if haengt:
        save_state(st)
    while True:
        st = get_state()
        queued = [j for j in st["jobs"].values() if j["status"] == "queued"]
        if not queued:
            break
        job = sorted(queued, key=lambda j: j["added"])[0]
        _set_job(job["id"], status="running")
        try:
            res = await _run_ytdlp(job, settings)
        except Exception as e:                                    # noqa: BLE001
            res = {"ok": False, "error": str(e)[:300]}
        st = get_state()
        j = st["jobs"].get(job["id"], {})
        if res.get("ok"):
            j.update(status="done", title=res["title"], path=res["path"],
                     size=res["size"], finished=time.time(), error="")
            # Playlister giver flere filer; registrér dem alle så oprydningen
            # kender dem (ellers ville kun den første være "vores").
            j["paths"] = res.get("paths") or [res["path"]]
            j["videos"] = res.get("videos") or []
            note = f" — {res['note']}" if res.get("note") else ""
            _log(st, f"Hentet: {res['title']} ({_human(res['size'])}){note}")
            j["note"] = res.get("note") or ""
            done += 1
        else:
            j.update(status="error", error=res.get("error") or "", finished=time.time())
            _log(st, f"FEJL: {job['url'][:60]} — {j['error'][:120]}")
            failed += 1
        save_state(st)
    if done:
        await refresh_library()
    return {"done": done, "failed": failed}


def file_size(path: str) -> int:
    """Størrelse af videoen inkl. dens undertekst-/miniature-sidecars."""
    try:
        return os.path.getsize(path) + sum(os.path.getsize(x) for x in _sidecars(path))
    except OSError:
        return 0


def owned_paths() -> dict:
    """Sti -> jobtitel for alt portalen selv har hentet (og som stadig ligger der)."""
    out = {}
    for j in get_state()["jobs"].values():
        if j["status"] != "done":
            continue
        for p in (j.get("paths") or ([j["path"]] if j.get("path") else [])):
            if p:
                out[os.path.realpath(p)] = j.get("title") or os.path.basename(p)
    return out


# ---------------- abonnementer (playliste / kanal) ----------------
def sub_kind_from_url(url: str) -> str:
    """Gæt om en URL er en playliste eller en kanal."""
    u = url.lower()
    if "list=" in u or "/playlist" in u:
        return "playlist"
    return "channel"


def _remote_url(sub: dict) -> str:
    """Kanal-URL'er skal pege på videofanen, ellers lister yt-dlp fanerne selv."""
    url = sub["url"]
    if sub["kind"] != "channel":
        return url
    tail = url.rstrip("/").rsplit("/", 1)[-1].lower()
    if tail in ("videos", "shorts", "streams", "live"):
        return url
    return url.rstrip("/") + "/videos"


async def _ytdlp_lines(args: list) -> list:
    proc = await asyncio.create_subprocess_exec(
        config.YT_DLP_BIN, *args,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        err = (stderr.decode("utf-8", "replace").strip().splitlines() or ["ukendt fejl"])[-1]
        raise RuntimeError(err[:200])
    return [ln for ln in stdout.decode("utf-8", "replace").splitlines() if ln.strip()]


async def list_remote(sub: dict, limit: int = 0) -> list:
    """Videoer i playlisten/kanalen, nyeste først. --flat-playlist henter kun
    listen, ikke videoerne, så et tjek er billigt."""
    args = ["--flat-playlist", "--no-warnings", "--print", "%(id)s|%(title)s"]
    if limit:
        args += ["--playlist-end", str(int(limit))]
    args.append(_remote_url(sub))
    out = []
    for ln in await _ytdlp_lines(args):
        vid, _, title = ln.partition("|")
        if vid and vid != "NA":
            out.append({"id": vid, "title": title or vid})
    return out


async def add_sub(url: str, kind: str = "", cycle: int = 0) -> dict:
    """Opret et abonnement. Titlen hentes fra YouTube så listen er læsbar."""
    url = url.strip()
    kind = kind or sub_kind_from_url(url)
    sid = uuid.uuid4().hex[:12]
    sub = {"id": sid, "url": url, "kind": kind, "cycle": int(cycle or 0),
           "title": url, "enabled": True, "added": time.time(), "last_check": 0,
           "primed": False, "seen": {}}
    try:
        lines = await _ytdlp_lines(["--flat-playlist", "--playlist-end", "1",
                                    "--no-warnings", "--print",
                                    "%(playlist_title)s", _remote_url(sub)])
        if lines and lines[0] not in ("NA", ""):
            sub["title"] = lines[0]
    except Exception:
        pass
    st = get_state()
    st["subs"][sid] = sub
    _log(st, f"Abonnement tilføjet: {sub['title']}"
             + (f" (cycle {sub['cycle']})" if sub["cycle"] else ""))
    save_state(st)
    return sub


def del_sub(sid: str) -> bool:
    """Fjern abonnementet. Allerede hentede videoer bliver liggende — de er dine,
    og de følger herefter kun den almindelige set-regel."""
    st = get_state()
    sub = st["subs"].pop(sid, None)
    if sub:
        for j in st["jobs"].values():
            if j.get("sub_id") == sid:
                j["sub_id"] = ""
        _log(st, f"Abonnement fjernet: {sub['title']}")
        save_state(st)
    return bool(sub)


def set_sub(sid: str, **kw) -> dict:
    st = get_state()
    sub = st["subs"].get(sid)
    if sub:
        sub.update(kw)
        save_state(st)
    return sub or {}


def sub_files(st: dict, sid: str) -> list:
    """Filer der stadig ligger på disken og stammer fra dette abonnement."""
    out = []
    for j in st["jobs"].values():
        if j.get("sub_id") != sid or j["status"] != "done":
            continue
        for v in (j.get("videos") or []):
            path = os.path.realpath(v.get("path") or "")
            if path and os.path.exists(path):
                out.append({"path": path, "upload_date": v.get("upload_date") or "",
                            "title": j.get("title") or os.path.basename(path),
                            "job": j["id"], "finished": j.get("finished") or 0})
    return out


async def sync_subs() -> dict:
    """Tjek alle aktive abonnementer og læg nye videoer i køen.

    Asymmetrien mellem de to typer er med vilje:
    · **playliste** — første tjek markerer alt eksisterende som kendt uden at hente.
      En playliste kan indeholde hundredvis af videoer, og du bad om *nye*.
    · **kanal** — første tjek henter de X nyeste med det samme; det er hele pointen
      med "cycle X". Resten af kanalens bagkatalog markeres som kendt, så den
      aldrig begynder at hente gammelt indhold.

    `seen` husker hvert video-id vi nogensinde har hentet — også efter videoen er
    slettet igen. Uden det ville kanalovervågningen genhente en video, som
    set-reglen lige har ryddet væk, i en uendelig løkke.
    """
    res = {"checked": 0, "queued": 0, "errors": []}
    for sid in list(get_state()["subs"]):
        st = get_state()
        sub = st["subs"].get(sid)
        if not sub or not sub.get("enabled", True):
            continue
        cycle = int(sub.get("cycle") or 0)
        try:
            limit = (cycle + 10) if (sub["kind"] == "channel" and cycle) else 0
            remote = await list_remote(sub, limit)
        except Exception as e:                                    # noqa: BLE001
            res["errors"].append(f"{sub['title']}: {e}")
            continue

        st = get_state()
        sub = st["subs"][sid]
        sub["last_check"] = time.time()
        seen = sub.setdefault("seen", {})
        first = not sub.get("primed")
        unseen = [v for v in remote if v["id"] not in seen]

        if sub["kind"] == "channel" and cycle:
            # Hent højst én cyklus ad gangen: er der pludselig 30 nye videoer,
            # er det spild at hente alle 30 og straks smide de 27 ud igen.
            take = unseen[:cycle]
        elif first:
            take = []                      # playliste: prim uden at hente
        else:
            take = unseen

        for v in take:
            seen[v["id"]] = time.time()
        if first:
            # Alt andet der findes lige nu regnes som bagkatalog.
            for v in unseen:
                seen.setdefault(v["id"], time.time())
            sub["primed"] = True
        save_state(st)

        for v in reversed(take):           # ældste først, så rækkefølgen er naturlig
            add_job(f"https://www.youtube.com/watch?v={v['id']}",
                    sub_id=sid, video_id=v["id"])
            res["queued"] += 1
        res["checked"] += 1
    return res


async def enforce_cycles() -> dict:
    """Hold hver kanals cyklus på X videoer.

    Tæller KUN videoer hentet af selve kanal-abonnementet. Manuelle hentninger,
    playliste-abonnementer og favoritter tæller ikke med — og favoritter slettes
    aldrig, så en favoritering frigør en plads i cyklussen til næste video.
    """
    res = {"evicted": [], "freed": 0, "errors": []}
    st = get_state()
    chans = [s for s in st["subs"].values()
             if s["kind"] == "channel" and int(s.get("cycle") or 0) > 0]
    if not chans:
        return res

    inv = await emby_yt_items()
    if not inv["ok"]:
        # Uden favorit-data kunne vi komme til at slette noget fredet. Rør intet.
        res["errors"].append(f"Emby svarede ikke ({inv['error']}) — cyklus står stille")
        return res
    favs = {it["path"] for it in inv["items"] if it["favorite"]}

    st = get_state()
    for sub in chans:
        files = [f for f in sub_files(st, sub["id"]) if f["path"] not in favs]
        # nyeste først: upload-dato hvis vi har den, ellers hentetidspunkt
        files.sort(key=lambda f: (f["upload_date"] or "", f["finished"]), reverse=True)
        for f in files[int(sub["cycle"]):]:
            try:
                size = _slet_video(f["path"])
                st["leaving"].pop(f["path"], None)
                job = st["jobs"].get(f["job"])
                if job:
                    job["status"] = "deleted"
                res["evicted"].append(f["title"])
                res["freed"] += size
                _log(st, f"Ude af cyklus ({sub['title']}): {f['title']} ({_human(size)})")
            except Exception as ex:                               # noqa: BLE001
                res["errors"].append(f"{f['title']}: {ex}")
    save_state(st)
    if res["evicted"]:
        await refresh_library()
    return res


def delete_video(path: str) -> dict:
    """Slet én video manuelt (knappen på /admin/yt), med samme sikkerhedsnet som
    oprydningen: KUN filer portalen selv har hentet, og sidecars med.

    Video-id'et bliver stående i abonnementets `seen`, så en overvåget kanal ikke
    henter den igen. Er det jobbets sidste fil, markeres jobbet som slettet.
    """
    real = os.path.realpath(path or "")
    if real not in owned_paths():
        return {"ok": False, "error": "Den video har portalen ikke selv hentet, så den røres ikke"}
    st = get_state()
    try:
        size = _slet_video(real)
    except OSError as e:
        return {"ok": False, "error": f"Kunne ikke slette: {e}"}
    st["leaving"].pop(real, None)
    titel = os.path.basename(real)
    for j in st["jobs"].values():
        paths = [p for p in (j.get("paths") or ([j["path"]] if j.get("path") else [])) if p]
        if real not in {os.path.realpath(p) for p in paths}:
            continue
        titel = j.get("title") or titel
        rest = [p for p in paths if os.path.realpath(p) != real]
        if j.get("paths") is not None:
            j["paths"] = rest
        if j.get("videos"):
            j["videos"] = [v for v in j["videos"] if os.path.realpath(v.get("path") or "") != real]
        if not rest:
            j["status"] = "deleted"
    _log(st, f"Slettet manuelt: {titel} ({_human(size)})")
    save_state(st)
    return {"ok": True, "title": titel, "size": size}


# ---------------- Emby ----------------
async def _emby(client, path, **params):
    params["api_key"] = config.EMBY_API_KEY
    r = await client.get(f"{config.EMBY_URL}/emby{path}", params=params)
    r.raise_for_status()
    return r.json()


async def _library_id(client, name: str):
    try:
        folders = await _emby(client, "/Library/VirtualFolders")
    except Exception:
        return None
    for f in folders:
        if (f.get("Name") or "").lower() == name.lower():
            return f.get("ItemId")
    return None


def _marked_played(ud: dict) -> bool:
    """SET = eksplicit markeret som set. Ikke 50%, ikke 'påbegyndt'.

    Derfor kun UserData.Played — bevidst uden PlaybackPositionTicks/PlayedPercentage,
    som mediaclean.py netop tæller med i sin modsatte hensigt.
    """
    return bool((ud or {}).get("Played"))


async def emby_yt_items() -> dict:
    """Alle videoer i YouTube-biblioteket med set/favorit aggregeret over ALLE brugere."""
    settings = get_settings()
    out = {"ok": False, "items": [], "error": ""}
    try:
        async with httpx.AsyncClient(timeout=45) as c:
            lib = await _library_id(c, settings.get("library_name") or "YouTube")
            if not lib:
                out["error"] = (f"Fandt ikke Emby-biblioteket "
                                f"'{settings.get('library_name')}'")
                return out
            users = [u["Id"] for u in await _emby(c, "/Users")]
            merged = {}
            for uid in users:
                data = await _emby(c, f"/Users/{uid}/Items", ParentId=lib,
                                   Recursive="true", Fields="Path,UserData")
                for it in data.get("Items", []):
                    path = it.get("Path") or ""
                    if not path:
                        continue
                    key = os.path.realpath(path)
                    m = merged.setdefault(key, {
                        "id": it["Id"], "name": it.get("Name") or os.path.basename(path),
                        "path": key, "played": False, "favorite": False})
                    ud = it.get("UserData") or {}
                    if _marked_played(ud):
                        m["played"] = True
                    if ud.get("IsFavorite"):
                        m["favorite"] = True
            out["items"] = list(merged.values())
            out["ok"] = True
    except Exception as e:                                        # noqa: BLE001
        out["error"] = str(e)[:200]
    return out


async def refresh_library():
    try:
        async with httpx.AsyncClient(timeout=30) as c:
            await c.post(f"{config.EMBY_URL}/emby/Library/Refresh",
                         params={"api_key": config.EMBY_API_KEY})
    except Exception:
        pass


async def ntfy(title, message, click=None, tags=None):
    if not config.NTFY_URL:
        return
    headers = {"Title": title}
    if tags:
        headers["Tags"] = tags
    if click:
        headers["Click"] = click
        headers["Actions"] = f"view, Åbn admin, {click}"
    if config.NTFY_TOKEN:
        headers["Authorization"] = f"Bearer {config.NTFY_TOKEN}"
    try:
        async with httpx.AsyncClient(timeout=10) as c:
            await c.post(f"{config.NTFY_URL}/{config.NTFY_MEDIA_TOPIC}",
                         content=message.encode("utf-8"), headers=headers)
    except Exception:
        pass


# ---------------- oprydning ----------------
async def run_cycle(dry_run: bool = False) -> dict:
    """Markér sete videoer til sletning efter karens, og slet de udløbne.

    Rører kun stier portalen selv har hentet. Favorit = fredet, og en video der
    favoritteres i karensperioden fjernes igen fra listen.
    """
    settings = get_settings()
    res = {"seen": 0, "new": [], "spared": [], "deleted": [], "freed": 0,
           "enabled": bool(settings.get("enabled")) and not dry_run, "errors": [], "error": ""}
    inv = await emby_yt_items()
    if not inv["ok"]:
        res["error"] = inv["error"]
        return res

    owned = owned_paths()
    st = get_state()
    now = time.time()
    grace = float(settings.get("grace_days") or 7) * 86400

    for it in inv["items"]:
        path = it["path"]
        if path not in owned:
            continue                      # ikke vores fil — rør den aldrig
        res["seen"] += 1
        entry = st["leaving"].get(path)
        if it["favorite"] or not it["played"]:
            # Favorit eller ikke markeret som set → skal ikke slettes.
            if entry:
                st["leaving"].pop(path, None)
                res["spared"].append(it["name"])
                _log(st, f"Fredet igen: {it['name']}"
                         + (" (favorit)" if it["favorite"] else " (ikke længere markeret set)"))
            continue
        if not entry:
            size = os.path.getsize(path) if os.path.exists(path) else 0
            st["leaving"][path] = {"title": it["name"], "size": size,
                                   "first_seen": now, "delete_after": now + grace}
            res["new"].append({"title": it["name"], "size": size})

    # slet udløbne
    for path, e in list(st["leaving"].items()):
        if now < e["delete_after"]:
            continue
        if not res["enabled"]:
            continue
        try:
            if os.path.exists(path):
                res["freed"] += _slet_video(path)
            st["leaving"].pop(path, None)
            # marker jobbet som slettet så owned_paths ikke tæller den igen
            for j in st["jobs"].values():
                paths = j.get("paths") or ([j.get("path")] if j.get("path") else [])
                if os.path.realpath(path) in [os.path.realpath(p) for p in paths if p]:
                    j["status"] = "deleted"
            res["deleted"].append(e["title"])
            _log(st, f"Slettet efter karens: {e['title']} ({_human(e.get('size') or 0)})")
        except Exception as ex:                                   # noqa: BLE001
            res["errors"].append(f"{e['title']}: {ex}")

    save_state(st)

    if res["deleted"]:
        await refresh_library()
        await ntfy(f"Slettede {len(res['deleted'])} sete videoer",
                   "Frigjorde " + _human(res["freed"]) + ":\n"
                   + "\n".join("· " + t for t in res["deleted"][:15]),
                   click="https://weircon.dk/admin/yt", tags="wastebasket")
    elif res["new"]:
        await ntfy(f"{len(res['new'])} sete videoer på vej ud",
                   f"Slettes om {int(settings.get('grace_days') or 7)} dage, "
                   "medmindre du favoritter dem:\n"
                   + "\n".join(f"· {i['title']}" for i in res["new"][:15]),
                   click="https://weircon.dk/admin/yt", tags="hourglass")
    return res