refactor(core): V2-1 — die vier Ports, ein Store, ein Laufwerks-Treiber
Ampel / ampel (push) Successful in 40s

WAS: rippy/ports.py beschreibt Store/Queue/Bus/Drives als Protocol. Zwei
weitere Doppelungen sind zusammengelegt: db.py (API+Worker) wird
rippy/store, und der Auswurf (api/devices.py + ripping.wirf_disc_aus)
wird rippy/drives/linux. Verhalten unveraendert.

WARUM: Drei Betriebsarten tragen nur, wenn ein Modus die Auswahl der
Treiber hinter vier Nahtstellen ist statt ein eigener Codestand
(KONZEPT-V2.md §1). Diese Etappe zieht die Nahtstellen ein, ohne schon
einen zweiten Treiber zu haben — die kommen in V2-2 (SQLite/LocalQueue)
und V2-4 (Windows).

DIE UNANGENEHMERE DOPPELUNG WAR DER AUSWURF: Er stand zweimal da, mit
UNTERSCHIEDLICHEN Vertraegen — devices.eject wirft OSError, ripping.
wirf_disc_aus gibt False zurueck und wirft nie. Beides ist richtig fuer
seine Seite (Browser-Meldung gegen "ein Rip stirbt nicht an einer
klemmenden Schublade"). Jetzt liegt EINE Mechanik darunter
(auswerfen_mit_grund) und beide Vertraege unveraendert darueber.
Die API-Fassung war ausserdem NIE getestet — jetzt schon, inklusive
"reicht ENOENT/EPERM unveraendert weiter".

get_settings hatte den einzigen echten Verhaltensunterschied der beiden
db.py: die Worker-Fassung schluckte jeden Fehler und gab {} zurueck.
Nicht still entschieden, sondern sichtbar gemacht — der Parameter
bei_fehler_leer steht jetzt in der Signatur, mit der offenen Frage im
Docstring. {} heisst fuer den Aufrufer "nichts gesetzt", nicht "konnte
nicht nachsehen"; das ist dieselbe Klasse wie catch(() => []) im alten
UI. Zu entscheiden in V2-2.

ZWEITER BEINAHE-FEHLER DIESER ETAPPE: linux.py importierte detection
auf Modulebene — und das zieht fcntl. Damit waere ripping.py und ueber
es der NATIVE WINDOWS-WORKER nicht mehr ladbar gewesen. Diesmal haben
die Tests es sofort gefangen (4 Sammelfehler). Behoben an der Wurzel:
Konstanten und die reine classify() leben jetzt in drives/cdrom.py,
ganz ohne fcntl. Nebengewinn — die classify-Tests liefen bisher NUR in
der Ampel ("erst nach dem Push bewiesen") und laufen jetzt ueberall.

Ausserdem: .dockerignore-Testmuster brauchen **, sonst greifen sie nur
in der obersten Ebene. Im laufenden Container nachgezaehlt: 30 test_*.py
lagen in den Images.

GEMESSEN: ruff sauber, 301 Tests gruen + 1 uebersprungen (vorher 290;
+4 neue eject-Tests, +7 classify-Tests die jetzt lokal laufen). Kein
Modul liegt mehr doppelt im Repo.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-08-28 08:54:28 +02:00
co-authored by Claude Opus 5
parent edfe0d4313
commit dd1d0b7365
19 changed files with 1670 additions and 1436 deletions
+2 -2
View File
@@ -364,8 +364,8 @@ def werkzeug_versionen() -> dict:
info["handbrake"] = "installiert"
# Woher kommt der MakeMKV-Key? UI-Setting schlägt Env — ehrlich anzeigen.
try:
import db
ui_key = (db.get_settings().get("makemkvAppKey") or "").strip()
from rippy import store as db
ui_key = (db.get_settings(bei_fehler_leer=True).get("makemkvAppKey") or "").strip()
except Exception:
ui_key = ""
if ui_key:
+1 -1
View File
@@ -32,7 +32,7 @@ celery_app.conf.update(
# Import auf Modulebene: im worker_ready-Signal ist /app nicht mehr
# zuverlässig im sys.path (ModuleNotFoundError 'caps', Deploy 23.07.).
import caps # noqa: E402
import db # noqa: E402
from rippy import store as db # noqa: E402
import zombies # noqa: E402
-261
View File
@@ -1,261 +0,0 @@
"""Job- und Log-Persistenz in PostgreSQL (KONZEPT: Postgres für Job-Logs).
Die Tabellendefinition existiert bewusst identisch in API und Worker
(docker/api/db.py) — es gibt kein geteiltes Paket zwischen den Containern.
Wer die Struktur ändert, ändert BEIDE Dateien. create_all ist idempotent.
"""
import os
from datetime import datetime, timezone
from sqlalchemy import (
Column,
DateTime,
Integer,
MetaData,
String,
Table,
Text,
create_engine,
)
DATABASE_URL = os.getenv(
"DATABASE_URL", "postgresql://rippy:rippy@localhost:5432/rippy"
)
engine = create_engine(DATABASE_URL, pool_pre_ping=True)
metadata = MetaData()
jobs = Table(
"jobs",
metadata,
Column("id", String(36), primary_key=True),
Column("disc_type", String(16)),
Column("device", String(64)),
Column("title", String(255)),
Column("status", String(16), nullable=False, server_default="pending"),
Column("progress", Integer, nullable=False, server_default="0"),
Column("output_path", Text),
Column("target_dir", String(255)),
Column("error", Text),
Column("meta", Text), # Disc-Metadaten (JSON) — Quelle für Ordnernamen + NFO
Column("created_at", DateTime(timezone=True)),
Column("finished_at", DateTime(timezone=True)),
)
logs = Table(
"logs",
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("ts", DateTime(timezone=True)),
Column("level", String(16)),
Column("source", String(32)),
Column("message", Text),
)
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def init_db() -> None:
"""Legt fehlende Tabellen an (idempotent) und zieht Mini-Migrationen nach.
Die ALTERs stehen identisch in docker/api/db.py — wer zuerst startet,
migriert; der andere findet die Spalten dann bereits vor.
"""
metadata.create_all(engine)
with engine.begin() as conn:
conn.exec_driver_sql(
"ALTER TABLE jobs ADD COLUMN IF NOT EXISTS target_dir VARCHAR(255)"
)
conn.exec_driver_sql(
"ALTER TABLE jobs ADD COLUMN IF NOT EXISTS meta TEXT"
)
conn.exec_driver_sql(
"ALTER TABLE workers ADD COLUMN IF NOT EXISTS info TEXT"
)
settings_table = Table(
"settings",
metadata,
Column("key", String(64), primary_key=True),
Column("value", Text),
)
workers = Table(
"workers",
metadata,
Column("name", String(128), primary_key=True),
Column("encoders", Text),
Column("info", Text), # Werkzeug-Versionen (JSON: makemkv/handbrake/key-Quelle)
Column("last_seen", DateTime(timezone=True)),
)
def save_worker(name: str, encoder_liste: list, info: dict = None) -> None:
"""Worker meldet Name + Encoder-Fähigkeiten + Werkzeug-Versionen (Upsert)."""
import json
payload = json.dumps(encoder_liste)
info_payload = json.dumps(info or {})
with engine.begin() as conn:
vorhanden = conn.execute(
workers.select().where(workers.c.name == name)
).first()
if vorhanden:
conn.execute(
workers.update().where(workers.c.name == name).values(
encoders=payload, info=info_payload, last_seen=utcnow()
)
)
else:
conn.execute(
workers.insert().values(
name=name, encoders=payload, info=info_payload, last_seen=utcnow()
)
)
def get_settings(key: str = "ui") -> dict:
"""UI-Einstellungen lesen (der Worker respektiert Transcode-Optionen)."""
import json
from sqlalchemy import select
try:
with engine.connect() as conn:
zeile = conn.execute(
select(settings_table.c.value).where(settings_table.c.key == key)
).first()
if zeile and zeile[0]:
return json.loads(zeile[0])
except Exception:
pass
return {}
def get_job(job_id: str) -> dict:
"""Ganze Job-Zeile — der Worker braucht Titel + Metadaten für die
Ordner-Benennung und die Media-Server-Aufbereitung (NFO/Poster)."""
from sqlalchemy import select
with engine.connect() as conn:
zeile = conn.execute(
select(jobs).where(jobs.c.id == job_id)
).mappings().first()
return dict(zeile) if zeile else None
def save_settings(werte: dict, key: str = "ui") -> None:
"""Upsert in die settings-Tabelle — der Worker legt hier z. B. die
Track-Scan-Ergebnisse ab (key 'tracks:<device>'), die API liest sie."""
import json
from sqlalchemy import select
payload = json.dumps(werte)
with engine.begin() as conn:
vorhanden = conn.execute(
select(settings_table.c.key).where(settings_table.c.key == key)
).first()
if vorhanden:
conn.execute(
settings_table.update()
.where(settings_table.c.key == key)
.values(value=payload)
)
else:
conn.execute(settings_table.insert().values(key=key, value=payload))
def get_job_status(job_id: str) -> str:
"""Nur der Status — der Worker prüft damit kooperative Abbruch-Anfragen."""
from sqlalchemy import select
with engine.connect() as conn:
zeile = conn.execute(
select(jobs.c.status).where(jobs.c.id == job_id)
).first()
return zeile[0] if zeile else ""
def list_jobs_mit_status(stati) -> list:
"""Alle Jobs in einem der genannten Zustände (id/status/title/created_at).
Basis der Zombie-Erkennung: Jobs, die behaupten, es arbeite gerade jemand
an ihnen. Bewusst NUR diese schmale Auswahl statt der ganzen Zeile — die
Erkennung braucht nichts weiter.
"""
from sqlalchemy import select
with engine.connect() as conn:
zeilen = conn.execute(
select(jobs.c.id, jobs.c.status, jobs.c.title, jobs.c.created_at)
.where(jobs.c.status.in_(list(stati)))
).mappings().all()
return [dict(z) for z in zeilen]
def zaehle_online_worker(sekunden: int = 120) -> int:
"""Wie viele Worker gelten laut Herzschlag gerade als online?
Die Zombie-Erkennung vergleicht das mit der Zahl der Celery-Antworten:
melden sich weniger Worker als bekannt sind, ist die Auskunft
unvollständig — dann wird NICHTS als Leiche gewertet.
"""
from datetime import timedelta
from sqlalchemy import func, select
grenze = utcnow() - timedelta(seconds=sekunden)
with engine.connect() as conn:
anzahl = conn.execute(
select(func.count()).select_from(workers).where(workers.c.last_seen >= grenze)
).scalar()
return int(anzahl or 0)
def meta_merken(job_id: str, **felder) -> None:
"""Ergänzt EINZELNE Schlüssel in den Job-Metadaten. Wirft nie.
`update_job(meta=...)` würde die Spalte ersetzen — Poster, Jahr, Titel-Wahl
und Sprachwunsch dieses Rips wären damit fort. Also lesen, mischen,
schreiben.
Fehler werden geschluckt: Diese Funktion vermerkt nur, in welcher Phase ein
Job steht (api/phasen.py). Ein Rip darf daran nicht scheitern — im
schlimmsten Fall fehlt die Marke und der „Neu"-Knopf fragt nach.
"""
import json
try:
zeile = get_job(job_id)
if not zeile:
return
try:
vorher = json.loads(zeile.get("meta") or "{}")
except (ValueError, TypeError):
vorher = {}
if not isinstance(vorher, dict):
vorher = {}
vorher.update(felder)
update_job(job_id, meta=json.dumps(vorher))
except Exception as e:
try:
add_log("warning", "worker", f"Job {job_id}: Metadaten-Vermerk fehlgeschlagen: {e}")
except Exception:
pass
def update_job(job_id: str, **fields) -> None:
with engine.begin() as conn:
conn.execute(jobs.update().where(jobs.c.id == job_id).values(**fields))
def add_log(level: str, source: str, message: str) -> None:
with engine.begin() as conn:
conn.execute(
logs.insert().values(ts=utcnow(), level=level, source=source, message=message)
)
+760 -760
View File
File diff suppressed because it is too large Load Diff
+19 -105
View File
@@ -17,6 +17,25 @@ import shutil
import subprocess
import tempfile
# Auswurf und ioctl-Konstanten leben seit V2-1 im gemeinsamen Treiber
# (rippy/drives/linux.py). Vorher stand derselbe Ablauf ZWEIMAL im Repo —
# hier und in docker/api/devices.py, mit unterschiedlichen Vertraegen.
#
# Die Namen werden hier weiter angeboten, weil tasks.py und die Tests sie so
# kennen; `wirf_disc_aus` ist der Worker-Vertrag (gibt False zurueck, wirft nie).
from rippy.drives.linux import ( # noqa: F401
AUSWURF_WARTEN_SEKUNDEN,
CDROM_DRIVE_STATUS,
CDROM_LOCKDOOR,
CDROMCLOSETRAY,
CDROMEJECT,
CDS_DISC_OK,
CDS_DRIVE_NOT_READY,
CDS_NO_DISC,
CDS_TRAY_OPEN,
_auswurf_geglueckt,
)
from rippy.drives.linux import auswerfen_versuchen as wirf_disc_aus # noqa: F401
from winlauf import OHNE_FENSTER
RIP_OUTPUT_DIR = os.getenv("RIP_OUTPUT_DIR", "/app/media")
@@ -27,111 +46,6 @@ class RipAbbruch(Exception):
Nutzer den Job abgebrochen hat (Status 'canceling' in der DB)."""
# include/uapi/linux/cdrom.h — dieselben ioctls wie in api/devices.py
CDROMEJECT = 0x5309
CDROM_LOCKDOOR = 0x5329 # 1 = Tür verriegeln, 0 = entriegeln
CDROM_DRIVE_STATUS = 0x5326
CDROMCLOSETRAY = 0x5319
# Antworten von CDROM_DRIVE_STATUS (cdrom.h)
CDS_NO_DISC = 1
CDS_TRAY_OPEN = 2
CDS_DRIVE_NOT_READY = 3
CDS_DISC_OK = 4
# Wie lange auf die Schublade gewartet wird. Ein Laufwerk braucht dafür ein
# bis zwei Sekunden; fünf sind reichlich und blockieren nichts Wichtiges.
AUSWURF_WARTEN_SEKUNDEN = 5
def _auswurf_geglueckt(status: int) -> bool:
"""Ist die Disc nach dem Auswurf wirklich draußen? (pure Funktion)
Sowohl „Schublade offen" als auch „kein Datenträger" zählen: Ein
Slot-Laufwerk hat keine Schublade und meldet nach dem Auswerfen CDS_NO_DISC.
"""
return status in (CDS_TRAY_OPEN, CDS_NO_DISC)
def wirf_disc_aus(device_path: str, ioctl_fn=None, oeffnen=None,
schliessen=None, warten=None) -> bool:
"""Wirft die Disc aus und PRÜFT, ob sie draußen ist. Wirft NIE.
## Warum `CDROMEJECT` allein nicht genügt (Befund 26.07.2026, gemessen)
Der Commander meldete: „Der Button gibt es in den Settings, aber es passiert
nicht, das Laufwerk geht nicht auf." Am laufenden System nachgestellt:
wirf_disc_aus("/dev/sr0") → True
CDROM_DRIVE_STATUS danach → 4 (Disc drin)
Das ioctl wird also **angenommen und tut nichts**. Ursache: MakeMKV
verriegelt während des Rips die Laufwerkstür (`CDROM_LOCKDOOR 1`) und
entriegelt sie nicht wieder. Ein verriegeltes Laufwerk quittiert den Auswurf
trotzdem mit Erfolg. Deshalb macht das Werkzeug `eject` immer beides:
erst entriegeln, dann auswerfen. Gegenprobe an derselben Disc:
CDROM_LOCKDOOR 0 + CDROMEJECT → Status 2 (SCHUBLADE OFFEN)
## Und deshalb wird das Ergebnis geprüft, nicht geglaubt
Genau diese Sorte Fehler ist zweimal durchgerutscht: In v3.14 stand hier
„Auswurf tat nichts, jetzt entscheidet die Einstellung" — die Einstellung
wurde danach wirklich gelesen, nur ausgeworfen wurde weiterhin nicht, und im
Log stand „Disc ausgeworfen". Ein Rückgabewert eines ioctls beweist nichts;
gefragt wird jetzt das Laufwerk.
Bewusst hier und nicht in detection.py: das Modul ist ein byteweiser
Zwilling der API-Kopie. `fcntl` gibt es nur unter Linux — der native
Windows-Worker lädt ripping.py ebenfalls, rippt dort aber nie.
Die drei Parameter sind nur zum Testen einspritzbar (kein echtes Laufwerk).
"""
if ioctl_fn is None:
try:
from fcntl import ioctl
except ImportError: # Windows — dieser Worker rippt nie
return False
ioctl_fn = ioctl
if warten is None:
import time
warten = time.sleep
oeffnen = oeffnen or (lambda p: os.open(p, os.O_RDONLY | os.O_NONBLOCK))
schliessen = schliessen or os.close
try:
fd = oeffnen(device_path)
except OSError:
return False
try:
# Entriegeln ist der entscheidende Schritt. Scheitert er, wird der
# Auswurf trotzdem versucht — bei einem nicht verriegelten Laufwerk
# (oder einem, das das ioctl nicht kennt) klappt er ohnehin.
try:
ioctl_fn(fd, CDROM_LOCKDOOR, 0)
except OSError:
pass
try:
ioctl_fn(fd, CDROMEJECT, 0)
except OSError:
return False
# Nachsehen statt hoffen: Die Schublade braucht ein bis zwei Sekunden.
for _ in range(AUSWURF_WARTEN_SEKUNDEN):
try:
if _auswurf_geglueckt(ioctl_fn(fd, CDROM_DRIVE_STATUS, 0)):
return True
except OSError:
return False
warten(1)
return False
finally:
try:
schliessen(fd)
except OSError:
pass
def check_makemkv_installed() -> bool:
"""Prüft, ob makemkvcon installiert ist."""
return shutil.which("makemkvcon") is not None
+5 -5
View File
@@ -22,7 +22,7 @@ import time
import requests
import db
from rippy import store as db
from rippy.rip import makemkv_daten
import medien
from rippy.core import notify
@@ -372,7 +372,7 @@ def _serien_episoden_zuordnen(ausgabe: str, serie: str, staffel, meta: dict) ->
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()
einstellungen = db.get_settings(bei_fehler_leer=True)
url = (einstellungen.get("notificationWebhook") or "").strip()
if not url:
return
@@ -402,7 +402,7 @@ def _job_abschliessen(job_id: str, ergebnis: dict) -> None:
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()
einstellungen = db.get_settings(bei_fehler_leer=True)
try:
meta = json.loads(job.get("meta") or "{}")
except ValueError:
@@ -527,7 +527,7 @@ def rip_disc(self, device_path: str, job_id: str, target_dir: str = None):
gesehen.add(text)
db.add_log("info", "makemkv", f"Job {job_id}: {text[:300]}")
einstellungen = db.get_settings()
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.
@@ -898,7 +898,7 @@ def transcode_files(self, job_id: str, raw_dir: str, final_dir: str):
_job_abschliessen(job_id, ergebnis)
return ergebnis
einstellungen = db.get_settings()
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 {}
-84
View File
@@ -383,90 +383,6 @@ def test_fehler_ohne_preset_problem_bleibt_der_alte():
assert ergebnis["error"] == "HandBrake endete mit Code 1"
# --- Auswurf: das ioctl meldet Erfolg und tut nichts (Befund 26.07.2026) -----
def _laufwerk(verriegelt=True, kennt_lockdoor=True):
"""Ein nachgebautes Laufwerk, das sich wie das echte verhaelt.
Gemessen am BU40N der Rippy-VM: Nach einem MakeMKV-Rip ist die Tuer
verriegelt. CDROMEJECT wird dann ANGENOMMEN und tut nichts - der Status
bleibt auf 4 (Disc drin). Erst CDROM_LOCKDOOR 0 macht den Auswurf wirksam.
"""
import ripping
zustand = {"verriegelt": verriegelt, "status": ripping.CDS_DISC_OK,
"aufrufe": []}
def ioctl_fn(fd, befehl, arg=0):
zustand["aufrufe"].append(befehl)
if befehl == ripping.CDROM_LOCKDOOR:
if not kennt_lockdoor:
raise OSError("ioctl unbekannt")
zustand["verriegelt"] = bool(arg)
return 0
if befehl == ripping.CDROMEJECT:
if not zustand["verriegelt"]:
zustand["status"] = ripping.CDS_TRAY_OPEN
return 0 # <- auch verriegelt: ERFOLG, aber ohne Wirkung
if befehl == ripping.CDROM_DRIVE_STATUS:
return zustand["status"]
raise OSError("unerwartetes ioctl")
return zustand, ioctl_fn
def test_auswurf_entriegelt_zuerst_und_klappt_dann():
import ripping
zustand, ioctl_fn = _laufwerk(verriegelt=True)
ok = ripping.wirf_disc_aus(
"/dev/sr0", ioctl_fn=ioctl_fn, oeffnen=lambda p: 42,
schliessen=lambda fd: None, warten=lambda s: None)
assert ok is True
assert zustand["status"] == ripping.CDS_TRAY_OPEN
# Reihenfolge: entriegeln VOR auswerfen
assert zustand["aufrufe"][0] == ripping.CDROM_LOCKDOOR
assert zustand["aufrufe"][1] == ripping.CDROMEJECT
def test_auswurf_meldet_fehlschlag_wenn_die_disc_drin_bleibt():
"""Der eigentliche Fehler. Vorher gab wirf_disc_aus True zurueck, weil das
ioctl nicht geworfen hatte - und ins Log kam "Disc ausgeworfen", waehrend
die Schublade zu blieb. Ein ioctl-Rueckgabewert beweist nichts."""
import ripping
# Ein Laufwerk, das LOCKDOOR nicht kennt und verriegelt bleibt
zustand, ioctl_fn = _laufwerk(verriegelt=True, kennt_lockdoor=False)
ok = ripping.wirf_disc_aus(
"/dev/sr0", ioctl_fn=ioctl_fn, oeffnen=lambda p: 42,
schliessen=lambda fd: None, warten=lambda s: None)
assert ok is False
assert zustand["status"] == ripping.CDS_DISC_OK # nie aufgegangen
def test_auswurf_bei_slot_laufwerk_ohne_schublade():
"""Ein Slot-Laufwerk hat keine Schublade und meldet nach dem Auswerfen
CDS_NO_DISC. Das muss als Erfolg zaehlen."""
import ripping
assert ripping._auswurf_geglueckt(ripping.CDS_NO_DISC) is True
assert ripping._auswurf_geglueckt(ripping.CDS_TRAY_OPEN) is True
assert ripping._auswurf_geglueckt(ripping.CDS_DISC_OK) is False
assert ripping._auswurf_geglueckt(ripping.CDS_DRIVE_NOT_READY) is False
def test_auswurf_ohne_laufwerk_wirft_nicht():
import ripping
def oeffnen_kaputt(p):
raise OSError("kein Laufwerk")
assert ripping.wirf_disc_aus(
"/dev/sr9", ioctl_fn=lambda *a: 0, oeffnen=oeffnen_kaputt,
schliessen=lambda fd: None, warten=lambda s: None) is False
# --- Sprachen der Disc: gemessen an der Akira-Blu-ray (26.07.2026) -----------
#
# Woertlich aus `makemkvcon -r --noscan info dev:/dev/sr0` im Worker-Container