"""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 json import os import posixpath import shutil 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, lies_datei_dauer, lies_titel_info, preset_fuer, rip_cd, rip_video, run_handbrake, ) 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" 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 _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 normalisiert.startswith(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(). Arbeitsverzeichnis (/app/temp, Docker-Volume) und Ziel (/app/media, Bind-Mount) sind VERSCHIEDENE Dateisysteme — os.rename scheitert dort mit EXDEV, und shutil.move fällt auf Kopieren zurück. Damit lief eine 74-GB-Vollkopie auf dieselbe Platte an, bis sie mit ENOSPC voll war. Ergebnis: Platte 100 % voll, Worker-Container startete nicht mehr — und der Job wurde als FEHLGESCHLAGEN markiert, obwohl die komprimierte Datei längst fertig und in Ordnung war. Der Nutzer sah nur „nichts da". Deshalb dreifach abgesichert: Platz vorher prüfen, Fehler abfangen, und im Zweifel die Rohdatei einfach liegen lassen, wo sie ist. """ ziel_original = os.path.join(final_dir, "original") try: # Liegt beides auf demselben Dateisystem, ist es ein reines Umhängen — # sofort fertig, kein zusätzlicher Platz nötig. gleiches_dateisystem = os.stat(raw_dir).st_dev == os.stat(final_dir).st_dev if not gleiches_dateisystem: 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 — 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") transcode_an = ist_video and einstellungen.get("transcodeEnabled", True) if ist_video: # UI-Key schlägt Env-Key — Monats-Key ohne Rebuild aktualisierbar _makemkv_key_anwenden(einstellungen) # Sprechender Zielordner „ (Jahr)" statt Job-UUID — Jellyfin & Co. # erkennen den Film am Ordnernamen. UUID bleibt Fallback ohne Titel. # Serien-Rips landen stattdessen in /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) 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." ) 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:') — 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] 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: _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