rippy/docker/worker/ablauf.py

"""Der Ablauf eines Jobs — rippen, komprimieren, ablegen. OHNE Celery.

## Warum es diese Datei gibt (V2-4, 28.08.2026)

Bis hierher stand der ganze Ablauf in `tasks.py`, und die Datei importierte
auf Modulebene `celery_app`. Damit war der Ablauf an einen Broker gebunden —
und im nativen Windows-Betrieb, der bewusst KEINEN Broker hat, ueberhaupt
nicht ladbar. Rippy konnte dort die Oberflaeche zeigen und Laufwerke lesen,
aber nicht rippen. Genau das war die Luecke.

Dabei hing der Ablauf an GENAU DREI Stellen an Celery — in 234 Zeilen:

    self.update_state(...)              Zwischenstand melden
    _transcode_queue(node)              Ziel-Queue waehlen
    transcode_files.apply_async(...)    Kompression weiterreichen

Alle drei sind Fragen der ZUSTELLUNG, nicht des Ablaufs. Sie sind jetzt
Rueckrufe, die der Aufrufer stellt:

    melde(zustand)                        wohin der Zwischenstand geht
    weiterreichen(job, roh, ziel, knoten) wer komprimiert

`tasks.py` reicht die Celery-Fassung herein, der Standalone-Laeufer die
lokale. **Ohne `weiterreichen` komprimiert dieselbe Funktion gleich selbst** —
genau das, was ein Ein-Prozess-Rippy braucht.

## Was das NICHT ist

Kein Neuschreiben. Der Rumpf ist der aus `tasks.py`, Zeile fuer Zeile — samt
aller Befunde, die dort ueber Monate hineingewandert sind (Platzpruefung,
Zombie-Marken, Pfad-Uebersetzung fuer native Worker, Serien-Zuordnung). Was
sich geaendert hat, sind die drei Stellen oben.

Das ist `core/pipeline.py` aus KONZEPT-V2.md § 2.1, nur noch am alten Platz:
Der Umzug nach `rippy/` kommt, wenn die flachen Worker-Importe aufgeloest
sind (`import db`, `import medien`, `import ripping`).
"""

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

import requests

from rippy import store as db
from rippy.rip import makemkv_aufruf, makemkv_daten
import medien
from rippy.core import notify

# Der Laufwerks-Treiber DIESER Maschine — Linux (ioctl) oder Windows (Win32).
#
# ⚠️ Hier stand bis V2-4:
#
#     try:
#         from rippy.drives.detection import detect_disc_type, disc_size_bytes
#     except ImportError:            # Windows: kein fcntl
#         detect_disc_type = None
#
# Das war fuer den NATIVEN WINDOWS-WORKER gedacht, der nur komprimiert. Es
# haette aber auch den eigenstaendigen Windows-Rippy getroffen: Der
# Linux-Treiber zieht ueber detection.py das Modul `fcntl` nach, das es unter
# Windows nicht gibt — also waere detect_disc_type dort IMMER None gewesen,
# und Rippen unten hart verriegelt. Rippy haette unter Windows alles anzeigen
# und nichts rippen koennen, ohne dass irgendwo ein Fehler stuende.
#
# Jetzt fragt der Ablauf den Port: Welcher Treiber es kann, entscheidet
# rippy/drives/__init__.py. `None` bleibt der ehrliche Wert fuer den Fall,
# dass diese Maschine wirklich keine Laufwerke lesen kann.
try:
    from rippy import drives as _laufwerks_schicht

    _treiber = _laufwerks_schicht.treiber()
    detect_disc_type = _treiber.detect_disc_type
    disc_size_bytes = _treiber.disc_size_bytes
except (ImportError, AttributeError):
    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,
    sprachen_zusammenfassen,
    sprachliste,
    wirf_disc_aus,
)

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

# Phasen-Marke in den Job-Metadaten: war der RIP fertig, als es schiefging?
#
# Sobald `status = "failed"` in der Zeile steht, ist die Phase sonst
# unwiederbringlich fort — und genau die entscheidet, was danach hilft: Nach
# einem toten Transcode genügt „Neu komprimieren"; nach einem toten Rip liegt
# nur ein Bruchstück da (Vorfall 26.07.2026: 5,1 GB von rund 40 GB) und es muss
# neu gerippt werden. Gelesen wird die Marke in api/phasen.py — der Name steht
# in beiden Dateien und wird von test_phasen.py gegeneinander geprüft, weil es
# kein geteiltes Paket zwischen den Containern gibt.
RIP_FERTIG = "rip_fertig"


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 außerhalb
    (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, wurzeln=None) -> 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.
    """
    wurzel, _vorgabe, frei = wurzeln or _betriebs_wurzeln()
    if target_dir:
        normalisiert = _normalisiert(target_dir)
        if frei or unter_wurzel(normalisiert, wurzel):
            return normalisiert
    # Ohne Wahl: die Wurzel dieses Betriebs, nicht die des Containers. Sonst
    # landete die fertige Datei unter Windows in `X:\app\media\bluray` —
    # einem Ordner, den niemand gesucht hat (gemessen 29.08.2026).
    from rippy import pfade

    return pfade.verbinden(wurzel if frei else RIP_OUTPUT_DIR, disc_type)


def _betriebs_wurzeln() -> tuple:
    """`(medien_wurzel, arbeits_vorgabe, frei)` für DIESEN Betrieb.

    ## Warum das hier gebraucht wird (Befund 29.08.2026)

    Beim Nachstellen des leeren Bildschirms lief ein echter Test-Rip durch —
    und die Rohdaten landeten in **`F:\\app\\temp\\raw`**. Also in einem
    Ordner namens `app` auf dem Laufwerk, von dem Rippy gerade lief.

    `RAW_DIR` ist `/app/temp/raw` und `MEDIA_ROOT` ist `/app/media`; unter
    Windows sind das keine Pfade, sondern Unfälle. Schlimmer noch: Die
    Prüfung `unter_wurzel(wahl, MEDIA_ROOT)` verwarf **auch eine ausdrückliche
    Wahl** — ein Arbeitsordner wie `D:\\Roh` liegt nicht unter `/app/media`,
    also fiel er still auf den Container-Standard zurück.

    Damit kam der Arbeitsordner, den der Commander am 28.08.2026 ausdrücklich
    bestellt hat („kannst du noch einbauen das man den arbeitsordner … setzen
    kann"), unter Windows nie an. Der Dialog zeigte ihn, das Setzen ging, und
    der Worker ignorierte ihn — ohne ein Wort.

    Im Container ändert sich nichts: Dort ist `/app/media` die Wurzel, und
    `frei` ist falsch.
    """
    from rippy import betrieb, config

    try:
        werte = config.laden()
    except Exception:                                # noqa: BLE001
        werte = {}
    return (betrieb.medien_wurzel(werte) or MEDIA_ROOT,
            betrieb.arbeits_vorgabe(werte) or RAW_DIR,
            betrieb.frei_blaettern(werte))


def _normalisiert(wert: str) -> str:
    """Pfad säubern — nach seiner FORM, nicht nach dem laufenden Rechner.

    `posixpath` für Container-Pfade (sonst macht Windows Backslashes daraus
    und die Wurzelprüfung greift nicht mehr), `os.path` für echte
    Windows-Pfade. Dieselbe Regel wie in `rippy.pfade`.
    """
    from rippy import pfade

    return os.path.normpath(wert) if pfade.ist_windows_pfad(wert) \
        else posixpath.normpath(wert)


def _arbeitsverzeichnis(einstellungen: dict, job_wahl: str = "",
                        wurzeln=None) -> str:
    """Basis für Roh-Rips. Reihenfolge: Wahl DIESES Rips → UI-Setting
    `workDir` → Vorgabe dieses Betriebs (Container: /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.
    """
    wurzel, vorgabe, frei = wurzeln or _betriebs_wurzeln()
    for kandidat in (job_wahl, einstellungen.get("workDir")):
        wert = (kandidat or "").strip()
        if not wert:
            continue
        normalisiert = _normalisiert(wert)
        # Nativ zählt die Wahl des Nutzers — dort IST sein Laufwerk die
        # Grenze. Im Container bleibt die Wurzelprüfung: Ein Pfad ausserhalb
        # von /app/media wäre dort ein Pfad ins Nichts.
        if frei or unter_wurzel(normalisiert, wurzel):
            return normalisiert
    return vorgabe


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
    try:
        # WOHIN, entscheidet der Rechner — unter Windows die Registry, sonst
        # die settings.conf. Begruendung und Messung in
        # `rippy.rip.makemkv_aufruf.key_ablegen`.
        #
        # UND nachsehen, ob MakeMKV ihn annimmt: Ein abgelehnter Schluessel
        # ist schlimmer als gar keiner — ohne las MakeMKV die Disc noch, mit
        # einem abgelehnten verweigerte es alles (gemessen 29.08.2026). Die
        # paar Sekunden Rueckfrage sind nichts gegen einen Rip, der Stunden
        # laeuft und sonst am ersten Titel scheitert.
        from ripping import werkzeug

        urteil = makemkv_aufruf.key_setzen_und_pruefen(key, werkzeug("makemkv"))
        if not urteil["ok"]:
            db.add_log("warning", "worker",
                       "MakeMKV nimmt den hinterlegten Beta-Key nicht an: %s"
                       % urteil["grund"])
    except (OSError, ImportError) 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(bei_fehler_leer=True)
    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(bei_fehler_leer=True)
            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")


def rippen(device_path: str, job_id: str, target_dir: str = None,
           melde=None, weiterreichen=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)
    # Ab hier gilt: der Rip läuft, ist aber NICHT fertig. Stirbt der Job jetzt,
    # ist jede Roh-Datei ein Bruchstück (siehe RIP_FERTIG oben).
    db.meta_merken(job_id, **{RIP_FERTIG: False})
    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()
        # Frueher: self.update_state(...) — der Zwischenstand ging an Celery.
        # Jetzt ein Rueckruf, den der Aufrufer stellt: im verteilten Betrieb
        # meldet er an Celery, im Standalone-Betrieb an den Ereignis-Bus.
        # Der ABLAUF muss nicht wissen, wer zuhoert.
        if melde:
            melde({"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(bei_fehler_leer=True)
    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).
        db.update_job(job_id, status="transcoding", progress=0)
        # Der Rip ist durch. Ab jetzt ist ein Fehlschlag mit „Neu komprimieren"
        # zu heilen, ohne die Disc noch einmal zu lesen.
        db.meta_merken(job_id, **{RIP_FERTIG: True})
        db.add_log("info", "worker", f"Job {job_id}: Rip fertig, Kompression eingereiht")
        # Wer die Kompression uebernimmt, entscheidet der Aufrufer:
        # Celery-Queue im verteilten Betrieb, LocalQueue im Standalone-Betrieb.
        # OHNE Rueckruf wird gleich hier komprimiert — das ist der Fall
        # "ein Prozess macht alles", also der Windows-Betrieb.
        if weiterreichen:
            weiterreichen(job_id, raw_dir, final_dir, meta.get("transcode_node"))
            return {"status": "ripped", "raw_dir": raw_dir}
        return komprimieren(job_id, raw_dir, final_dir)

    _job_abschliessen(job_id, ergebnis)
    return ergebnis


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()}


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:
        # Ein Info-Lauf liefert BEIDES: Titel-Liste und die Sprachen. Die
        # Sprach-Auskunft stand schon immer in derselben Ausgabe und wurde nur
        # weggeworfen (Commander-Anforderung 26.07.2026: vor dem Rip fragen,
        # was man haben will).
        titel, streams = lies_titel_info(device_path)
        sprachen = sprachen_zusammenfassen(streams)
        db.save_settings(
            {"status": "done", "tracks": titel, "sprachen": sprachen}, key)
        db.add_log(
            "info", "worker",
            f"Titel-Scan {device_path}: {len(titel)} Titel, "
            f"{len(sprachen['audio'])} Tonsprache(n), "
            f"{len(sprachen['untertitel'])} Untertitelsprache(n)",
        )
        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)}


def _ist_uebersetzt(container_pfad: str, lokal_pfad: str) -> bool:
    """Hat RIPPY_PATH_MAP diesen Pfad wirklich angefasst? (pure, testbar)"""
    return container_pfad != lokal_pfad


# Wie weit wird nach einem vorhandenen Elternordner gesucht. Begrenzt, damit auf
# einer toten Freigabe nicht endlos geklopft wird.
MAX_ELTERN_STUFEN = 12


def erster_vorhandener_ordner(pfad: str, isdir=None, dirname=None) -> str:
    """Der nächste EXISTIERENDE Ordner oberhalb von `pfad` — oder "".

    ⚠️ Vorfall 26.07.2026, im vollen Durchlauf aufgefallen: Der Rip lief sauber
    durch (40 GB), die Kompression brach sofort ab mit „Dieser Worker erreicht
    das Ziel nicht: \\\\NAS\\rippy\\movies\\Akira (1988)". Erreichbar war die
    Freigabe sehr wohl — es fehlten schlicht ZWEI Ordner: `movies` und der
    Filmordner darin. Die Prüfung ließ nur EINE fehlende Ebene durchgehen (sie
    sah nach `dirname`), obwohl gleich darauf `os.makedirs` die ganze Kette
    anlegt. Ein Ziel, das noch nie beschrieben wurde, war damit systematisch
    unerreichbar — also jeder erste Film in einer neuen Ablage.

    Es zählt daher, ob IRGENDEIN Vorfahre existiert: Von dem aus kann
    `makedirs` den Rest bauen. Existiert keiner, ist die Freigabe wirklich weg.

    `isdir`/`dirname` werden BEIM AUFRUF aufgelöst, nicht als Vorgabewert
    gebunden — sonst zeigt die Vorgabe für immer auf die Funktion von damals,
    und ein Ersetzen von `os.path.isdir` (im Test wie im Betrieb) ginge ins
    Leere. Genau darüber bin ich beim Schreiben der Tests gestolpert.
    """
    isdir = isdir or os.path.isdir
    dirname = dirname or os.path.dirname
    aktuell = dirname(pfad or "")
    for _ in range(MAX_ELTERN_STUFEN):
        if not aktuell:
            return ""
        if isdir(aktuell):
            return aktuell
        naechster = dirname(aktuell)
        if naechster == aktuell:  # Wurzel erreicht
            return ""
        aktuell = naechster
    return ""


def _erreichbarkeit_pruefen(raw_container: str, raw_lokal: str,
                            final_container: str, final_lokal: str) -> str:
    """Kann DIESER Worker Quelle und Ziel überhaupt sehen? Klartext oder "".

    Befund 25.07.2026 am Job 95afdc89, live: Das gezielte Routing an den
    Windows-PC des Commanders funktionierte einwandfrei — der Worker nahm die
    Aufgabe an und lehnte sie 182 ms später ab mit „Keine Roh-MKVs in
    /app/media/rippy/… gefunden". Diese Meldung klang nach einem kaputten Rip,
    obwohl der Rip vollständig war (79,6 GB lagen auf der NAS). Der Commander
    schloss daraus, das System sähe den externen Encoder nicht — verständlich
    und falsch.

    Die eigentliche Ursache: `/app/media/...` sind CONTAINER-Pfade. Ein nativer
    Worker sieht sie nur, wenn RIPPY_PATH_MAP sie auf eine Freigabe übersetzt —
    und dieses Mapping wurde von NIEMANDEM gesetzt. Es war also nicht möglich,
    dass diese Kombination je funktioniert.

    Zweiter Stolperstein im selben Fall: Die Rohdaten lagen auf der NAS (für den
    PC erreichbar), das ZIEL aber auf der VM-Platte (nicht erreichbar, dort
    läuft kein Samba). Deshalb werden hier BEIDE Pfade geprüft, nicht nur die
    Quelle — sonst scheitert es erst beim Schreiben, nach Stunden Rechenzeit.
    """
    mapping = os.getenv("RIPPY_PATH_MAP", "")
    fremder_worker = not os.path.isdir("/app")

    for zweck, container, lokal in (
        ("die Quelle (Rohdaten)", raw_container, raw_lokal),
        ("das Ziel (fertige Datei)", final_container, final_lokal),
    ):
        if os.path.isdir(lokal):
            continue
        # Das ZIEL darf fehlen — es wird gleich mit `os.makedirs` angelegt, und
        # zwar samt aller fehlenden Zwischenebenen. Es genügt also, dass
        # irgendein Vorfahre existiert (Herleitung in erster_vorhandener_ordner).
        if zweck.startswith("das Ziel") and erster_vorhandener_ordner(lokal):
            continue

        text = [f"Dieser Worker erreicht {zweck} nicht: {lokal}"]
        if fremder_worker and not mapping:
            text.append(
                "Ursache: RIPPY_PATH_MAP ist auf dieser Maschine nicht gesetzt. "
                f'„{container}" ist ein Pfad INNERHALB des Rippy-Containers — ein '
                'externer Worker sieht ihn nur, wenn er auf eine Netzwerk-Freigabe '
                'übersetzt wird.'
            )
            text.append(
                "Abhilfe: Freigabe auf diesem Rechner erreichbar machen und in "
                "start-tray.bat setzen, z. B. "
                "set RIPPY_PATH_MAP=/app/media=\\\\NAS\\rippy-media"
            )
        elif fremder_worker and mapping and not _ist_uebersetzt(container, lokal):
            text.append(
                f"RIPPY_PATH_MAP ist gesetzt ({mapping}), deckt diesen Pfad aber "
                f'nicht ab: „{container}" wurde von keinem Eintrag übersetzt — '
                'fehlt ein Präfix, oder ist es ein anderes Verzeichnis als erwartet?'
            )
        elif fremder_worker and mapping:
            # ⚠️ Hier stand bis zum 26.07.2026 dieselbe Meldung wie oben — „deckt
            # diesen Pfad nicht ab" —, obwohl die Karte ihn nachweislich übersetzt
            # hatte (der UNC-Pfad stand im selben Satz). Wer der Meldung folgte,
            # suchte den Fehler in der Karte, während die Freigabe schlicht nicht
            # verbunden war. Eine falsche Ursache ist teurer als gar keine.
            text.append(
                f'Übersetzt wurde er korrekt ({container} → {lokal}, Karte: '
                f'{mapping}) — dieser Pfad ist auf diesem Rechner nur gerade '
                'nicht erreichbar. Ist die Freigabe verbunden? Anmeldedaten noch '
                'gültig? Von Hand prüfen: den Pfad im Explorer öffnen.'
            )
        else:
            text.append(
                "Der Pfad existiert nicht. Liegt das Arbeitsverzeichnis auf einer "
                "Freigabe, die gerade nicht eingehängt ist?"
            )
        text.append(
            'WICHTIG: Die Rohdaten sind NICHT verloren — nach der Korrektur '
            'genügt „Neu komprimieren", ohne die Disc erneut zu rippen.'
        )
        return " ".join(text)
    return ""


def komprimieren(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_container, final_container = raw_dir, final_dir
    raw_dir = pfad_lokal(raw_dir)
    final_dir = pfad_lokal(final_dir)

    fehler = _erreichbarkeit_pruefen(raw_container, raw_dir, final_container, final_dir)
    if fehler:
        ergebnis = {"status": "error", "error": fehler}
        _job_abschliessen(job_id, ergebnis)
        return ergebnis

    quellen = sorted(glob.glob(os.path.join(raw_dir, "*.mkv")))
    if not quellen:
        ergebnis = {
            "status": "error",
            "error": (
                f"Das Verzeichnis {raw_dir} ist erreichbar, enthält aber keine "
                "MKV-Datei. Der Rip hat also nichts abgelegt (oder jemand hat die "
                "Datei entfernt). Die Kompression lässt sich nach einem neuen Rip "
                "erneut anstoßen."
            ),
        }
        _job_abschliessen(job_id, ergebnis)
        return ergebnis

    einstellungen = db.get_settings(bei_fehler_leer=True)
    # 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)

    # Sprachauswahl: Wahl für DIESEN Rip (aus meta) schlägt die Einstellung.
    # Leer heißt „alles behalten" — genau das Verhalten von vorher.
    try:
        job_meta = json.loads(job.get("meta") or "{}")
    except ValueError:
        job_meta = {}
    audio_sprachen = sprachliste(
        job_meta.get("audio_sprachen") or einstellungen.get("audioSprachen"))
    untertitel_sprachen = sprachliste(
        job_meta.get("untertitel_sprachen") or einstellungen.get("untertitelSprachen"))

    os.makedirs(final_dir, exist_ok=True)
    db.update_job(job_id, status="transcoding", progress=0, error=None)
    # Wer hier ankommt, hat vollständige Quelldateien — sonst wäre oben schon
    # abgebrochen worden. Das vermerkt die Marke auch für BESTANDSJOBS, die vor
    # ihrer Einführung gerippt wurden und sie darum noch nicht tragen.
    db.meta_merken(job_id, **{RIP_FERTIG: True})
    db.add_log(
        "info", "worker",
        f"Job {job_id}: Kompression gestartet ({len(quellen)} Datei(en), "
        f"Disc-Typ '{disc_type or 'unbekannt'}', Preset '{preset}'"
        + (f", Ton: {','.join(audio_sprachen)}" if audio_sprachen else ", Ton: alle")
        + (f", Untertitel: {','.join(untertitel_sprachen)}"
           if untertitel_sprachen else ", Untertitel: alle")
        + ")",
    )

    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,
            audio_sprachen=audio_sprachen,
            untertitel_sprachen=untertitel_sprachen,
        )
        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