rippy/docker/worker/tasks.py

"""Rip- und Transcode-Tasks — bewusst GETRENNT (23.07.2026, Basis für Etappe 12/20).

Warum zwei Tasks: Rippen braucht das Laufwerk (läuft immer lokal), Kompression
braucht nur CPU/GPU + Zugriff auf die Rohdatei. Als eigener Celery-Task auf der
Queue "transcode" kann die Kompression damit auch ein Remote-Worker mit GPU
übernehmen (optionales Add-on) — und fehlgeschlagene Kompressionen lassen sich
neu anstoßen, ohne die Disc neu zu rippen (POST /jobs/{id}/retry-transcode).

Zwei Stufen (Commander-Entscheid 23.07.): MakeMKV rippt verlustfrei (einziger
Weg durch AACS — HandBrake kann verschlüsselte Discs nicht lesen), HandBrake
komprimiert danach auf Arbeitsgröße. Die Rohdatei liegt nur temporär in
/app/temp und wird nach Erfolg gelöscht (Setting keepOriginal behält sie).
"""

import errno
import glob
import json
import os
import posixpath
import shutil
import time

import requests

import db
import makemkv_daten
import medien
import notify
from celery_app import celery_app

# fcntl gibt es nur unter Linux — der NATIVE Windows-Transcode-Worker
# (deploy/worker-windows) lädt dieses Modul auch, bedient aber nur die
# transcode-Queue. Rippen ohne detection ist unten hart verriegelt.
try:
    from detection import detect_disc_type, disc_size_bytes
except ImportError:  # Windows: kein fcntl
    detect_disc_type = None
    disc_size_bytes = None

from ripping import (
    RIP_OUTPUT_DIR,
    RipAbbruch,
    komprimieren_fuer,
    lies_datei_dauer,
    lies_titel_info,
    preset_fuer,
    rip_cd,
    rip_video,
    run_handbrake,
    wirf_disc_aus,
)

API_URL = os.getenv("API_URL", "http://api:8000")


def _transcode_queue(node: str):
    """Ziel-Queue für die Kompression (siehe celery_client.transcode_queue):
    gewählter Worker via worker_direct, wenn online — sonst geteilte Queue."""
    if not node:
        return "transcode"
    try:
        from celery.utils import worker_direct
        antworten = celery_app.control.ping(timeout=1.0) or []
        online = {k for antwort in antworten for k in antwort.keys()}
        if node in online:
            return worker_direct(node)
    except Exception:
        pass
    return "transcode"


def pfad_lokal(pfad: str, mapping: str = None) -> str:
    """Übersetzt Rippy-Container-Pfade für native Worker (pure Funktion).

    Der native Windows-Worker sieht /app/temp und /app/media nicht — er
    mountet Rippys Freigabe als Netzlaufwerk und setzt RIPPY_PATH_MAP,
    z. B. "/app/media=Z:\\media;/app/temp=Y:\\temp". Ohne Mapping (Docker-
    Worker: identische Mounts) kommt der Pfad unverändert zurück.
    """
    mapping = mapping if mapping is not None else os.getenv("RIPPY_PATH_MAP", "")
    if not pfad or not mapping:
        return pfad
    for paar in mapping.split(";"):
        if "=" not in paar:
            continue
        quelle, ziel = paar.split("=", 1)
        if quelle and pfad.startswith(quelle):
            rest = pfad[len(quelle):]
            if "\\" in ziel:
                rest = rest.replace("/", "\\")
            return ziel + rest
    return pfad

RAW_DIR = os.getenv("RAW_DIR", "/app/temp/raw")
MEDIA_ROOT = "/app/media"

# Wie oft während einer Kompression nachgesehen wird, ob der Nutzer abgebrochen
# hat. Eine DB-Abfrage alle paar Sekunden ist nichts gegen einen Encode, der
# Stunden läuft — und „Abbrechen" fühlt sich damit sofort an.
ABBRUCH_INTERVALL_SEKUNDEN = 5


def unter_wurzel(pfad: str, wurzel: str) -> bool:
    """Liegt `pfad` wirklich unterhalb von `wurzel` (oder IST es die Wurzel)?

    Pure Funktion, testbar. Ein nacktes `startswith()` genügt nicht:
    „/app/media-boese/x" beginnt mit „/app/media", liegt aber ausserhalb
    (Befund 25.07.2026 bei der Durchsicht). Immer „/" als Trenner — das sind
    Container-Pfade, auch wenn ein nativer Windows-Worker das Modul lädt.

    Gleichlautend in docker/api/main.py; es gibt kein geteiltes Paket zwischen
    den Containern.
    """
    if not pfad or not wurzel:
        return False
    sauber = wurzel.rstrip("/") or "/"
    return pfad == sauber or pfad.startswith(sauber + "/")


def _zielbasis(target_dir, disc_type: str) -> str:
    """Ablagebasis: vom Nutzer gewähltes Ziel (validiert) oder Standard.

    posixpath statt os.path — aus demselben Grund wie in _arbeitsverzeichnis:
    das sind IMMER Container-Pfade. os.path.normpath macht unter Windows
    Backslashes daraus, und dann greift die MEDIA_ROOT-Prüfung nicht mehr, das
    gewählte Ziel fiele still auf den Standard zurück. Aufgefallen 25.07.2026,
    als der Test dafür erstmals unter Windows lief. Live war es nie: aufgerufen
    wird nur aus rip_disc, und das ist auf Windows-Workern verriegelt.
    """
    if target_dir:
        normalisiert = posixpath.normpath(target_dir)
        if unter_wurzel(normalisiert, MEDIA_ROOT):
            return normalisiert
    return posixpath.join(RIP_OUTPUT_DIR, disc_type)


def _arbeitsverzeichnis(einstellungen: dict, job_wahl: str = "") -> str:
    """Basis für Roh-Rips. Reihenfolge: Wahl DIESES Rips → UI-Setting
    `workDir` → Container-Default /app/temp/raw.

    Hintergrund (Commander 24.07.): Die VM-Platte (150 GB) reicht für BD-50,
    aber eine 4K-UHD (bis 100 GB roh + Kompression daneben) sprengt sie —
    das Arbeitsverzeichnis muss deshalb auf ein großes Ziel umlegbar sein.

    Pro Rip wählbar seit 25.07.2026 (Commander-Wunsch): beim „Rippen starten"
    entscheidet man je Disc, wo die Rohdaten landen. Das Setting bleibt der
    Standard — und ist damit der Wert, der bei Vollautomatik-Rips greift, bei
    denen niemand gefragt wird.
    """
    # posixpath statt os.path: Das sind IMMER Container-Pfade (/app/media/...),
    # auch wenn ein nativer Windows-Worker dieses Modul lädt — der übersetzt
    # sie erst später mit pfad_lokal(). os.path.normpath macht unter Windows
    # Backslashes daraus, und dann greift die MEDIA_ROOT-Prüfung nicht mehr.
    for kandidat in (job_wahl, einstellungen.get("workDir")):
        wert = (kandidat or "").strip()
        if wert:
            normalisiert = posixpath.normpath(wert)
            if unter_wurzel(normalisiert, MEDIA_ROOT):
                return normalisiert
    return RAW_DIR


def _frei_bytes(pfad: str) -> int:
    """Freier Platz am Pfad (nächster existierender Elternordner zählt)."""
    kandidat = pfad
    while kandidat and not os.path.exists(kandidat):
        kandidat = os.path.dirname(kandidat)
    try:
        return shutil.disk_usage(kandidat or "/").free
    except OSError:
        return -1


def _platz_pruefen(pfad: str, benoetigt: int, zweck: str) -> str:
    """Gibt eine Klartext-Fehlermeldung zurück, wenn der Platz nicht reicht —
    sonst leeren String. VOR dem Rip geprüft: eine volle Platte nach 40 GB
    ist der teuerste Fehlschlag, den es gibt."""
    frei = _frei_bytes(pfad)
    if frei < 0 or benoetigt <= 0:
        return ""  # unbekannt → nicht raten, lieber rippen lassen
    if frei >= benoetigt:
        return ""
    return (
        f"Zu wenig Platz für {zweck} in {pfad}: "
        f"{frei / 1024**3:.1f} GB frei, benötigt ~{benoetigt / 1024**3:.1f} GB. "
        "Abhilfe: anderes Ziel wählen oder das Arbeitsverzeichnis unter "
        "Einstellungen → Verarbeitung auf eine große Freigabe legen."
    )


def _makemkv_key_anwenden(einstellungen: dict) -> None:
    """MakeMKV-Beta-Key aus den UI-Einstellungen aktivieren (schlägt Env).

    Damit ist der Monats-Key ohne Rebuild/Neustart aktualisierbar
    (Einstellungen → System). Format wie entrypoint.sh: settings.conf.

    Ergänzend statt überschreibend (Befund 25.07.2026): das Datenverzeichnis
    ist jetzt persistent, und hier stand vorher ein open(..., "w") — das warf
    vor JEDEM Rip alles andere aus der settings.conf, z. B. app_UpdateEnable
    aus dem entrypoint. Beide Schreiber müssen gleich arbeiten, sonst kommt
    der Fehler beim nächsten Rip still zurück.
    """
    key = (einstellungen.get("makemkvAppKey") or "").strip()
    if not key:
        return
    pfad = os.path.join(makemkv_daten.DATEN_DIR, "settings.conf")
    try:
        os.makedirs(makemkv_daten.DATEN_DIR, exist_ok=True)
        try:
            with open(pfad, encoding="utf-8", errors="replace") as f:
                alt = f.read()
        except OSError:
            alt = ""
        with open(pfad, "w", encoding="utf-8", newline="\n") as f:
            f.write(makemkv_daten.settings_conf_zusammenfuehren(alt, key))
    except OSError as e:
        db.add_log("warning", "worker", f"MakeMKV-Key konnte nicht gesetzt werden: {e}")


def _ordner_groesse(pfad: str) -> int:
    """Belegter Platz eines Ordners in Bytes (0, wenn nicht lesbar)."""
    summe = 0
    for wurzel, _, dateien in os.walk(pfad):
        for name in dateien:
            try:
                summe += os.path.getsize(os.path.join(wurzel, name))
            except OSError:
                pass
    return summe


def _original_aufheben(job_id: str, raw_dir: str, final_dir: str) -> None:
    """Roh-Rip zusätzlich aufheben — darf den Job NIEMALS scheitern lassen.

    Befund 25.07.2026 (Akira-UHD, echter Schaden): Hier stand ein nacktes
    shutil.move(). Zwischen Arbeitsverzeichnis (/app/temp) und Ziel
    (/app/media) scheitert os.rename mit EXDEV, shutil.move fällt auf Kopieren
    zurück — eine 74-GB-Vollkopie auf dieselbe Platte, bis sie mit ENOSPC voll
    war. Ergebnis: Platte 100 % voll, Worker-Container startete nicht mehr, und
    der Job galt als FEHLGESCHLAGEN, obwohl die komprimierte Datei längst
    fertig war. Der Nutzer sah nur „nichts da".

    ## Warum hier NICHT vorhergesagt, sondern versucht wird

    Die erste Fassung dieses Schutzes verglich `os.stat(...).st_dev` und
    schloss aus gleichen Werten auf „reines Umhängen, kein Platz nötig". Am
    25.07.2026 im Worker-Container nachgemessen — beides zugleich wahr:

        st_dev /app/temp  = 2050
        st_dev /app/media = 2050      → also identisch
        os.rename(...)    → EXDEV, „Invalid cross-device link"

    Der Kernel vergleicht bei rename() den **Mount**, nicht das Gerät. /app/temp
    (Docker-Volume) und /app/media (Bind-Mount) sind zwei Mounts DERSELBEN
    ext4-Partition. Die st_dev-Prüfung war deshalb wirkungslos: sie sah
    „gleiches Dateisystem", übersprang die Platzprüfung, und shutil.move kopierte
    doch. Der Schutz hätte genau den Schaden zugelassen, gegen den er gebaut war.

    Also: erst rename VERSUCHEN. Klappt es, ist es umgehängt und fertig.
    Kommt EXDEV, steht fest, dass kopiert werden müsste — und erst dann wird
    der Platz geprüft. Das ist keine Vermutung mehr, sondern die Antwort des
    Kernels.
    """
    ziel_original = os.path.join(final_dir, "original")
    try:
        # Der billige Weg zuerst — und er ist gleichzeitig der einzige
        # verlässliche Test, ob überhaupt umgehängt werden kann.
        try:
            os.rename(raw_dir, ziel_original)
            db.add_log(
                "info", "worker",
                f"Job {job_id}: Original behalten unter {ziel_original} (umgehängt, "
                "kein zusätzlicher Platz nötig)",
            )
            return
        except OSError as e:
            if e.errno != errno.EXDEV:
                raise  # etwas anderes ist schiefgelaufen → unten ehrlich melden

        # Ab hier ist eine echte Kopie unvermeidlich. Jetzt lohnt die Platzfrage.
        benoetigt = _ordner_groesse(raw_dir)
        frei = _frei_bytes(final_dir)
        if frei < benoetigt * 1.05:
            db.add_log(
                "warning", "worker",
                f"Job {job_id}: Original NICHT aufgehoben — Ziel liegt auf einem "
                f"anderen Mount, es müsste kopiert werden. Dafür wären "
                f"{benoetigt / 1024**3:.1f} GB nötig, frei sind nur "
                f"{frei / 1024**3:.1f} GB. Die Roh-Datei bleibt unter "
                f"{raw_dir} liegen. Abhilfe: Arbeitsverzeichnis "
                "(Einstellungen → Verarbeitung) auf dieselbe Freigabe legen "
                "wie das Ziel — dann wird nur umgehängt statt kopiert."
            )
            return
        shutil.move(raw_dir, ziel_original)
        db.add_log("info", "worker", f"Job {job_id}: Original behalten unter {ziel_original}")
    except OSError as e:
        # Halb geschriebene Kopie wegräumen, sonst belegt sie für immer Platz.
        shutil.rmtree(ziel_original, ignore_errors=True)
        db.add_log(
            "warning", "worker",
            f"Job {job_id}: Original konnte nicht aufgehoben werden ({e}). "
            f"Die komprimierte Datei ist fertig; die Roh-Datei bleibt unter {raw_dir}."
        )


def _abbruch_angefordert(job_id: str) -> bool:
    """Kooperativer Abbruch: hat der Nutzer über die API abgebrochen?"""
    return db.get_job_status(job_id) == "canceling"


def _serien_episoden_zuordnen(ausgabe: str, serie: str, staffel, meta: dict) -> list:
    """Episoden-Zuordnung per Laufzeitabgleich (ARM-Wunde #395, besser gelöst).

    Laufzeiten der Staffel kommen von der API (TMDB); die der Dateien vom
    HandBrake-Scan. Nur bei EINDEUTIGEM Treffer wird umbenannt. Wirft nie.
    """
    if meta.get("source") != "tmdb" or meta.get("type") != "tv" or not meta.get("id"):
        return ["Episoden-Zuordnung übersprungen (nur mit TMDB-Serien-Treffer möglich)"]
    try:
        antwort = requests.get(
            f"{API_URL}/metadata/tv/{meta['id']}/season/{int(staffel)}", timeout=30
        )
        if antwort.status_code != 200:
            return [f"Episoden-Laufzeiten nicht abrufbar (HTTP {antwort.status_code})"]
        episoden = [
            (e["episode"], (e.get("runtime") or 0) * 60)
            for e in antwort.json().get("episodes", [])
        ]
    except (requests.RequestException, ValueError) as e:
        return [f"Episoden-Laufzeiten nicht abrufbar: {e}"]
    if not episoden:
        return ["Keine Episoden-Laufzeiten bei TMDB hinterlegt — Dateinamen bleiben"]

    dateien = sorted(
        os.path.join(ausgabe, f) for f in os.listdir(ausgabe) if f.endswith(".mkv")
    )
    if not dateien:
        return []
    dauern = [lies_datei_dauer(p) for p in dateien]
    if 0 in dauern:
        return ["Datei-Laufzeiten nicht lesbar — Dateinamen bleiben"]

    zuordnung = medien.matche_episoden(dauern, episoden)
    if not zuordnung:
        return [
            "Keine EINDEUTIGE Episoden-Zuordnung über die Laufzeiten — "
            "Dateinamen bleiben (lieber ehrlich als falsch benannt)"
        ]
    return medien.episoden_umbenennen(ausgabe, serie, staffel, zuordnung)


def _benachrichtigen(job_id: str, betreff: str, text: str, level: str) -> None:
    """Webhook-Meldung bei Job-Ende — best effort, nie job-entscheidend."""
    einstellungen = db.get_settings()
    url = (einstellungen.get("notificationWebhook") or "").strip()
    if not url:
        return
    try:
        notify.sende(url, betreff, text, level)
    except Exception as e:
        db.add_log("warning", "notify", f"Job {job_id}: Benachrichtigung fehlgeschlagen: {e}")


def _job_abschliessen(job_id: str, ergebnis: dict) -> None:
    """Schreibt den Endzustand eines Jobs (completed/failed) nach Postgres,
    bereitet bei Erfolg fürs Medien-System auf (NFO/Poster) und meldet das
    Ende per Webhook (falls konfiguriert)."""
    job = db.get_job(job_id) or {}
    titel = job.get("title") or job_id[:8]

    if ergebnis.get("status") == "cancelled":
        db.update_job(
            job_id, status="failed", error="Abgebrochen durch Nutzer",
            finished_at=db.utcnow(),
        )
        db.add_log("warning", "worker", f"Job {job_id}: abgebrochen — Rohdaten bleiben erhalten")
        _benachrichtigen(job_id, f'⚠️ Rippy: „{titel}" abgebrochen',
                         "Der Job wurde auf Wunsch abgebrochen — Rohdaten bleiben erhalten.",
                         "warning")
        return
    if ergebnis.get("status") == "success":
        ausgabe = ergebnis.get("output_dir")
        if ausgabe and (job.get("disc_type") in ("dvd", "bluray", "uhd")):
            einstellungen = db.get_settings()
            try:
                meta = json.loads(job.get("meta") or "{}")
            except ValueError:
                meta = {}
            serie = meta.get("series")
            staffel = meta.get("season")

            # Serien-Rips: Episoden per Laufzeit zuordnen und umbenennen
            # (Show S01E02.mkv) — nur bei EINDEUTIGER Zuordnung, sonst
            # bleiben die MakeMKV-Namen (ehrlich geloggt).
            if serie and staffel:
                for meldung in _serien_episoden_zuordnen(ausgabe, serie, staffel, meta):
                    db.add_log("info", "worker", f"Job {job_id}: {meldung}")

            # Media-Server-Aufbereitung (Jellyfin/Emby/Kodi: NFO + Poster);
            # bei Serien wandern tvshow.nfo/poster in den Serien-Ordner.
            for meldung in medien.aufbereiten(
                ausgabe, einstellungen.get("mediaServer") or "none",
                serie or job.get("title") or "", meta.get("year"), meta,
                serien_root=os.path.dirname(ausgabe) if serie and staffel else None,
            ):
                db.add_log("info", "worker", f"Job {job_id}: {meldung}")

            # Bibliotheks-Refresh (Jellyfin/Emby): der Server scannt sofort
            meldung = medien.bibliothek_refresh(
                einstellungen.get("mediaServer") or "none",
                (einstellungen.get("jellyfinUrl") or "").strip(),
                (einstellungen.get("jellyfinApiKey") or "").strip(),
            )
            if meldung:
                db.add_log("info", "worker", f"Job {job_id}: {meldung}")
        db.update_job(
            job_id,
            status="completed",
            progress=100,
            output_path=ausgabe,
            finished_at=db.utcnow(),
        )
        db.add_log("success", "worker", f"Job {job_id}: abgeschlossen → {ausgabe}")
        _benachrichtigen(job_id, f'✅ Rippy: „{titel}" ist fertig',
                         f"Abgelegt unter: {ausgabe}", "success")
    else:
        fehler = ergebnis.get("error", "unbekannter Fehler")
        db.update_job(
            job_id,
            status="failed",
            error=fehler,
            finished_at=db.utcnow(),
        )
        db.add_log("error", "worker", f"Job {job_id}: {fehler}")
        _benachrichtigen(job_id, f'❌ Rippy: „{titel}" fehlgeschlagen', fehler, "error")


@celery_app.task(bind=True, name="worker.tasks.rip_disc")
def rip_disc(self, device_path: str, job_id: str, target_dir: str = None):
    """Stufe 1: Rippt die Disc; bei Video folgt die Kompression als eigener Task.

    target_dir (optional): vom Nutzer gewähltes Ablageziel unter /app/media —
    dort eingehängte Shares (NFS/SMB) sind damit direkt wählbar.
    """
    db.init_db()

    if detect_disc_type is None:
        ergebnis = {
            "status": "error",
            "error": "Dieser Worker kann nur komprimieren (nativer "
                     "Transcode-Worker ohne Laufwerks-Zugriff) — gerippt "
                     "wird auf der Rippy-Hauptmaschine.",
        }
        _job_abschliessen(job_id, ergebnis)
        return ergebnis

    if _abbruch_angefordert(job_id):
        _job_abschliessen(job_id, {"status": "cancelled"})
        return {"status": "cancelled"}

    disc_type = detect_disc_type(device_path)

    if disc_type in ("no_disc", "unknown"):
        fehler = (
            "Keine Disc im Laufwerk" if disc_type == "no_disc"
            else "Disc-Typ nicht erkennbar"
        )
        db.update_job(job_id, status="failed", error=fehler, finished_at=db.utcnow())
        db.add_log("error", "worker", f"Job {job_id}: {fehler} ({device_path})")
        return {"status": "error", "error": fehler, "disc_type": disc_type}

    db.update_job(job_id, status="running", disc_type=disc_type)
    db.add_log("info", "worker", f"Job {job_id}: {disc_type}-Rip gestartet ({device_path})")

    letzter = [-1]

    def fortschritt(progress: int, message: str = ""):
        # MakeMKV liefert viele PRGV-Zeilen pro Sekunde — DB nur bei Änderung.
        if progress == letzter[0]:
            return
        letzter[0] = progress
        if _abbruch_angefordert(job_id):
            raise RipAbbruch()
        self.update_state(
            state="PROGRESS",
            meta={"progress": progress, "status": "ripping", "message": message},
        )
        db.update_job(job_id, progress=progress)

    # MakeMKV-Meldungen ins Log (Befund 25.07.2026): Bis dahin überlebte NUR
    # die letzte Zeile ("Failed to open disc"), und die sagt nichts. Dass
    # MakeMKV bei der UHD-Disc nicht einmal versucht, einen Schlüssel zu
    # holen, war deshalb nur per Hand-Lauf im Container zu sehen.
    # Gedrosselt, weil das UI global nur die letzten 200 Zeilen zeigt: jede
    # Meldung höchstens einmal, insgesamt höchstens MAX_MELDUNGEN je Rip.
    # Code 1003 ist MakeMKVs eigenes DEBUG-Rauschen (am 25.07. beobachtet).
    MAX_MELDUNGEN = 40
    gesehen = set()

    def melde_makemkv(code: int, text: str):
        if code == 1003 or len(gesehen) >= MAX_MELDUNGEN or text in gesehen:
            return
        gesehen.add(text)
        db.add_log("info", "makemkv", f"Job {job_id}: {text[:300]}")

    einstellungen = db.get_settings()
    ist_video = disc_type in ("dvd", "bluray", "uhd")
    # Je Disc-Typ abwählbar (siehe komprimieren_fuer): 4K verlustfrei behalten,
    # DVDs trotzdem schrumpfen — vorher gab es nur alles oder nichts.
    transcode_an = ist_video and komprimieren_fuer(disc_type, einstellungen)

    if ist_video:
        # UI-Key schlägt Env-Key — Monats-Key ohne Rebuild aktualisierbar
        _makemkv_key_anwenden(einstellungen)

    # Sprechender Zielordner „<Titel> (Jahr)" statt Job-UUID — Jellyfin & Co.
    # erkennen den Film am Ordnernamen. UUID bleibt Fallback ohne Titel.
    # Serien-Rips landen stattdessen in <Serie>/Season NN (Staffel-Flow).
    job = db.get_job(job_id) or {}
    try:
        meta = json.loads(job.get("meta") or "{}")
    except ValueError:
        meta = {}
    if meta.get("series") and meta.get("season"):
        final_dir = medien.serien_ordner(
            _zielbasis(target_dir, disc_type), meta["series"], meta["season"]
        )
    else:
        final_dir = medien.zielordner(
            _zielbasis(target_dir, disc_type),
            job.get("title") or "", meta.get("year"), job_id,
        )

    # Titel-Wahl: explizite Auswahl aus der Track-Tabelle schlägt alles;
    # sonst „Nur Hauptfilm" (pro Rip wählbar, Fallback: Setting);
    # bei Serien-Discs ohne Auswahl zählen ALLE Episoden-Titel.
    titel_liste = [int(t) for t in (meta.get("titles") or []) if str(t).isdigit()]
    nur_hauptfilm = meta.get("main_feature_only")
    if nur_hauptfilm is None:
        nur_hauptfilm = bool(einstellungen.get("mainFeatureOnly", False))
    if meta.get("series") or titel_liste:
        nur_hauptfilm = False
    # Geplantes Ziel sofort sichtbar machen (UI-Detail + retry-transcode)
    db.update_job(job_id, output_path=final_dir)

    # meta["work_dir"] = die Wahl aus dem Rip-Dialog; leer = Setting/Default.
    raw_dir = os.path.join(_arbeitsverzeichnis(einstellungen, meta.get("work_dir")), job_id)

    # Platz-Check VOR dem Rip: Disc-Größe ist per ioctl bekannt — eine volle
    # Platte nach 40 GB wäre der teuerste Fehlschlag (4K-UHD: bis 100 GB roh).
    try:
        disc_bytes = disc_size_bytes(device_path) if disc_size_bytes else 0
    except OSError:
        disc_bytes = 0
    marge = 1024**3  # 1 GB Sicherheitsabstand
    if ist_video:
        rip_ziel = raw_dir if transcode_an else final_dir
        fehler = _platz_pruefen(rip_ziel, disc_bytes + marge, "den Roh-Rip")
        if not fehler and transcode_an and einstellungen.get("keepOriginal", False):
            fehler = _platz_pruefen(final_dir, disc_bytes + marge, "das Original (keepOriginal)")
        if fehler:
            ergebnis = {"status": "error", "error": fehler}
            _job_abschliessen(job_id, ergebnis)
            return ergebnis

    if disc_type == "cd":
        ergebnis = rip_cd(
            device_path, job_id, progress_cb=fortschritt, output_dir=final_dir,
            auswerfen=bool(einstellungen.get("autoEject", True)),
        )
    elif transcode_an:
        # Roh-Rip ins Arbeitsverzeichnis (wird nach erfolgreicher Kompression gelöscht)
        ergebnis = rip_video(
            device_path, job_id, disc_type,
            progress_cb=fortschritt,
            output_dir=raw_dir,
            nur_hauptfilm=nur_hauptfilm,
            titel_liste=titel_liste,
            log_cb=melde_makemkv,
        )
    else:
        ergebnis = rip_video(
            device_path, job_id, disc_type,
            progress_cb=fortschritt, output_dir=final_dir,
            nur_hauptfilm=nur_hauptfilm,
            titel_liste=titel_liste,
            log_cb=melde_makemkv,
        )

    # MakeMKV-Fehler in Klartext übersetzen — "Failed to open disc" allein
    # hilft niemandem.
    if ergebnis.get("status") == "error":
        fehler_text = ergebnis.get("error") or ""
        if "volume key is unknown" in fehler_text:
            # Befund 25.07.2026, auf BEIDEN Maschinen gemessen: Laufwerk und
            # MakeMKV sind in Ordnung. makemkvcon unter Linux ruft die
            # Disc-Schlüssel schlicht nie ab — die Windows-Version tut es
            # (Meldung 3338). Hier stand vorher erst "Disc zu neu" und danach
            # "der Schlüssel-Kanal ist tot"; beides war falsch und hat in die
            # Irre geschickt. Herleitung im Kopf von makemkv_daten.py.
            speicher = makemkv_daten.schluesselspeicher_status()
            anzahl = speicher.get("schluessel", 0)
            ergebnis["error"] += (
                " — Klartext: Laufwerk und Rippy sind in Ordnung, MakeMKV "
                "liest die Disc. Es fehlt nur der Schlüssel dieser Pressung. "
                "Der Grund: makemkvcon holt Schlüssel unter Linux nie selbst "
                "nach — die Windows-Version schon. "
                + (
                    "Dieser Worker kennt aktuell GAR KEINEN Disc-Schlüssel. "
                    if not anzahl
                    else f"Dieser Worker kennt {anzahl} Disc-Schlüssel, "
                    "diese Pressung ist nicht dabei. "
                )
                + "Abhilfe: MakeMKV auf einem Windows-PC installieren, die "
                "Disc dort einmal öffnen, dann die Datei _private_data.tar "
                "aus dem MakeMKV-Datenverzeichnis unter Einstellungen → "
                "System hochladen. Wirkt ab dem nächsten Rip. Klappt auch "
                "das nicht, kennt MakeMKV die Pressung selbst nicht — dann "
                "hilft nur eine KEYDB.cfg (ebenfalls dort hochladbar) oder "
                "das Einreichen des AACS-Dumps im MakeMKV-Forum, Bereich "
                "'Ultra HD Blu-ray'. Der Dump steht unter Einstellungen → "
                "System zum Download bereit."
            )
        elif disc_type == "uhd" and "Failed to open disc" in fehler_text:
            ergebnis["error"] += (
                " — 4K-UHD erkannt: Das Laufwerk kann UHD-Discs vermutlich "
                "nicht entschlüsseln. Dafür ist eine LibreDrive-Firmware nötig "
                "(MakeMKV-Forum 'Ultimate UHD Drives Flashing Guide'). "
                "Normale BD/DVD gehen weiterhin."
            )

    # Automatischer Auswurf. Die Disc ist nach dem Rip nicht mehr nötig — die
    # Kompression arbeitet auf der Datei, nicht am Laufwerk.
    #
    # Befund 25.07.2026: Die Einstellung („Disc nach erfolgreichem Ripping
    # automatisch auswerfen", Standard ein) wurde von NIEMANDEM gelesen. Bei
    # DVD/Blu-ray warf Rippy deshalb nie aus, bei Audio-CD dagegen immer, weil
    # abcde `-x` fest verdrahtet bekam. Jetzt entscheidet die Einstellung beides.
    if ergebnis.get("status") == "success" and ist_video and einstellungen.get("autoEject", True):
        if wirf_disc_aus(device_path):
            db.add_log("info", "worker", f"Job {job_id}: Disc ausgeworfen")
        else:
            db.add_log(
                "warning", "worker",
                f"Job {job_id}: Disc konnte nicht ausgeworfen werden ({device_path}) — "
                "der Rip ist davon unberührt.",
            )

    if ergebnis.get("status") == "success" and transcode_an:
        # Kompression als eigener Task — an den im Rip-Dialog GEWÄHLTEN Worker
        # (worker_direct), sonst an die geteilte transcode-Queue (irgendein
        # freier Worker, inkl. Remote-GPU).
        ziel_queue = _transcode_queue(meta.get("transcode_node"))
        db.update_job(job_id, status="transcoding", progress=0)
        gezielt = meta.get("transcode_node") and ziel_queue != "transcode"
        db.add_log("info", "worker",
                   f"Job {job_id}: Rip fertig, Kompression eingereiht"
                   + (f" → gezielt an {meta.get('transcode_node')}" if gezielt else ""))
        transcode_files.apply_async(
            args=[job_id, raw_dir, final_dir],
            queue=ziel_queue,
        )
        return {"status": "ripped", "raw_dir": raw_dir}

    _job_abschliessen(job_id, ergebnis)
    return ergebnis


@celery_app.task(name="worker.tasks.ping_worker")
def ping_worker():
    """Winziger Task zum Beweisen des gezielten Routings: gibt Hostname +
    Encoder zurück. Wird über worker_direct an EINEN Worker geschickt — landet
    er beim richtigen, stimmt das Encoder-Routing (Encoder-Auswahl im UI)."""
    import socket

    import caps
    return {"hostname": socket.gethostname(), "encoders": caps.erkenne_encoder()}


@celery_app.task(name="worker.tasks.scan_tracks")
def scan_tracks(device_path: str):
    """Titel-Liste der eingelegten Disc erfassen (für die Auswahl-Tabelle).

    Ergebnis landet in der settings-Tabelle (key 'tracks:<device>') — die
    API pollt darauf. Läuft auf der Haupt-Queue (nur der Rip-Worker hat das
    Laufwerk). Ein Info-Lauf dauert je nach Disc 20–120 s.
    """
    db.init_db()
    key = f"tracks:{device_path}"
    db.save_settings({"status": "running"}, key)
    try:
        titel = lies_titel_info(device_path)
        db.save_settings({"status": "done", "tracks": titel}, key)
        db.add_log("info", "worker", f"Titel-Scan {device_path}: {len(titel)} Titel gefunden")
        return {"status": "done", "anzahl": len(titel)}
    except Exception as e:
        db.save_settings({"status": "error", "error": str(e)[:300]}, key)
        db.add_log("error", "worker", f"Titel-Scan {device_path} fehlgeschlagen: {e}")
        return {"status": "error", "error": str(e)}


@celery_app.task(bind=True, name="worker.tasks.transcode_files")
def transcode_files(self, job_id: str, raw_dir: str, final_dir: str):
    """Stufe 2: HandBrake komprimiert die Roh-MKVs auf Arbeitsgröße.

    Erst wenn ALLE Dateien sauber komprimiert sind, wird das Roh-Verzeichnis
    gelöscht — bricht die Kompression ab, bleibt das Original in /app/temp
    liegen (kein Datenverlust wie bei ARMs berüchtigtem Move-Bug #1530).
    Über POST /jobs/{id}/retry-transcode jederzeit neu anstoßbar.
    """
    db.init_db()
    # Native Worker (Windows) übersetzen Container-Pfade aufs Netzlaufwerk
    raw_dir = pfad_lokal(raw_dir)
    final_dir = pfad_lokal(final_dir)
    quellen = sorted(glob.glob(os.path.join(raw_dir, "*.mkv")))
    if not quellen:
        ergebnis = {"status": "error", "error": f"Keine Roh-MKVs in {raw_dir} gefunden"}
        _job_abschliessen(job_id, ergebnis)
        return ergebnis

    einstellungen = db.get_settings()
    # Preset nach Disc-Typ (Befund 25.07.2026): vorher lief JEDE Quelle durch
    # dasselbe Preset — eine 4K-UHD wurde damit auf 1080p heruntergerechnet.
    job = db.get_job(job_id) or {}
    disc_type = job.get("disc_type") or ""
    preset = preset_fuer(disc_type, einstellungen)
    original_behalten = einstellungen.get("keepOriginal", False)

    os.makedirs(final_dir, exist_ok=True)
    db.update_job(job_id, status="transcoding", progress=0, error=None)
    db.add_log(
        "info", "worker",
        f"Job {job_id}: Kompression gestartet ({len(quellen)} Datei(en), "
        f"Disc-Typ '{disc_type or 'unbekannt'}', Preset '{preset}')",
    )

    anzahl = len(quellen)
    letzter = [-1]
    letzte_abbruchpruefung = [0.0]

    def abbruch_pruefen():
        """Zeitgesteuert prüfen, ob der Nutzer abgebrochen hat. Wirft RipAbbruch.

        Befund 25.07.2026 (Commander, am laufenden Akira-Job beobachtet): Der
        Abbruch wurde NUR in `datei_fortschritt` geprüft — und diese Closure
        stieg oben sofort wieder aus, wenn sich die Prozentzahl nicht geändert
        hatte. Bei einem 4K-Encode, der pro Prozent eine halbe Stunde braucht,
        sah „Abbrechen" entsprechend lange wirkungslos aus (gemessen: 3,4 min
        zwischen Anforderung 18:30:17 und Bestätigung 18:33:38 — bei noch
        langsamerem Fortschritt entsprechend mehr).

        Deshalb hängt die Prüfung jetzt an der Zeit statt am Fortschritt und
        läuft bei JEDER Ausgabezeile von HandBrake — auch während des
        Scan-Durchlaufs, der überhaupt keine Encode-Prozente liefert.
        """
        jetzt = time.monotonic()
        if jetzt - letzte_abbruchpruefung[0] < ABBRUCH_INTERVALL_SEKUNDEN:
            return
        letzte_abbruchpruefung[0] = jetzt
        if _abbruch_angefordert(job_id):
            raise RipAbbruch()

    for index, quelle in enumerate(quellen):
        ziel = os.path.join(final_dir, os.path.basename(quelle))

        def datei_fortschritt(p, _index=index):
            gesamt = int((_index * 100 + p) / anzahl)
            if gesamt == letzter[0]:
                return
            letzter[0] = gesamt
            db.update_job(job_id, progress=min(99, gesamt))

        hb = run_handbrake(
            quelle, ziel, preset=preset,
            progress_cb=datei_fortschritt,
            abbruch_cb=abbruch_pruefen,
        )
        if hb.get("status") == "cancelled":
            _job_abschliessen(job_id, hb)
            return hb
        if hb.get("status") != "success":
            ergebnis = {
                "status": "error",
                "error": (
                    f"Kompression fehlgeschlagen bei {os.path.basename(quelle)}: "
                    f"{hb.get('error')} — Roh-Datei bleibt in /app/temp erhalten"
                ),
            }
            _job_abschliessen(job_id, ergebnis)
            return ergebnis

    if original_behalten:
        _original_aufheben(job_id, raw_dir, final_dir)
    else:
        shutil.rmtree(raw_dir, ignore_errors=True)

    ergebnis = {"status": "success", "output_dir": final_dir}
    _job_abschliessen(job_id, ergebnis)
    return ergebnis