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 glob
import os
import shutil

import db
from celery_app import celery_app
from detection import detect_disc_type
from ripping import (
    DEFAULT_HB_PRESET,
    RIP_OUTPUT_DIR,
    RipAbbruch,
    rip_cd,
    rip_video,
    run_handbrake,
)

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


def _zielbasis(target_dir, disc_type: str) -> str:
    """Ablagebasis: vom Nutzer gewähltes Ziel (validiert) oder Standard."""
    if target_dir:
        normalisiert = os.path.normpath(target_dir)
        if normalisiert.startswith(MEDIA_ROOT):
            return normalisiert
    return os.path.join(RIP_OUTPUT_DIR, disc_type)


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 _job_abschliessen(job_id: str, ergebnis: dict) -> None:
    """Schreibt den Endzustand eines Jobs (completed/failed) nach Postgres."""
    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")
        return
    if ergebnis.get("status") == "success":
        db.update_job(
            job_id,
            status="completed",
            progress=100,
            output_path=ergebnis.get("output_dir"),
            finished_at=db.utcnow(),
        )
        db.add_log("success", "worker", f"Job {job_id}: abgeschlossen → {ergebnis.get('output_dir')}")
    else:
        db.update_job(
            job_id,
            status="failed",
            error=ergebnis.get("error", "unbekannter Fehler"),
            finished_at=db.utcnow(),
        )
        db.add_log("error", "worker", f"Job {job_id}: {ergebnis.get('error', 'unbekannter Fehler')}")


@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 _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)

    einstellungen = db.get_settings()
    transcode_an = (
        disc_type in ("dvd", "bluray")
        and einstellungen.get("transcodeEnabled", True)
    )

    final_dir = os.path.join(_zielbasis(target_dir, disc_type), job_id)

    if disc_type == "cd":
        ergebnis = rip_cd(device_path, job_id, progress_cb=fortschritt, output_dir=final_dir)
    elif transcode_an:
        # Roh-Rip nach /app/temp (wird nach erfolgreicher Kompression gelöscht)
        ergebnis = rip_video(
            device_path, job_id, disc_type,
            progress_cb=fortschritt,
            output_dir=os.path.join(RAW_DIR, job_id),
        )
    else:
        ergebnis = rip_video(
            device_path, job_id, disc_type,
            progress_cb=fortschritt, output_dir=final_dir,
        )

    if ergebnis.get("status") == "success" and transcode_an:
        # Kompression als eigener Task auf der transcode-Queue — kann vom
        # lokalen Worker ODER einem Remote-GPU-Worker übernommen werden.
        db.update_job(job_id, status="transcoding", progress=0)
        db.add_log("info", "worker", f"Job {job_id}: Rip fertig, Kompression eingereiht")
        transcode_files.apply_async(
            args=[job_id, os.path.join(RAW_DIR, job_id), final_dir],
            queue="transcode",
        )
        return {"status": "ripped", "raw_dir": os.path.join(RAW_DIR, job_id)}

    _job_abschliessen(job_id, ergebnis)
    return ergebnis


@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()
    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 = einstellungen.get("transcodePreset") or DEFAULT_HB_PRESET
    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), Preset '{preset}')",
    )

    anzahl = len(quellen)
    letzter = [-1]
    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
            if _abbruch_angefordert(job_id):
                raise RipAbbruch()
            db.update_job(job_id, progress=min(99, gesamt))

        hb = run_handbrake(quelle, ziel, preset=preset, progress_cb=datei_fortschritt)
        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:
        ziel_original = os.path.join(final_dir, "original")
        shutil.move(raw_dir, ziel_original)
        db.add_log("info", "worker", f"Job {job_id}: Original behalten unter {ziel_original}")
    else:
        shutil.rmtree(raw_dir, ignore_errors=True)

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