API: Die Job-Kette existiert jetzt — POST /jobs, Watcher, Postgres, /logs

Vorher gab es KEINEN Code-Pfad, der je einen Rip ausgelöst hat: kein
POST /jobs, kein udev-Daemon (udev_daemon.py existierte nirgends), GET /jobs
gab hart [] zurück, Postgres lag komplett brach, der SSE-Stream konnte
strukturell nie senden (sse_connections wurde nie befüllt), udevadm lieferte
ohne udevd nichts.

- POST /jobs: legt Job-Zeile an, schickt worker.tasks.rip_disc via Celery
- GET /jobs aus Postgres (running→processing fürs UI)
- Disc-Watcher: 3s-ioctl-Poll statt udev, protokolliert Einwurf/Auswurf
- /devices über /sys (vendor/model) + ioctl-Status — ehrlich statt leer
- /logs + /settings (Settings-Seite sprach vorher gegen 404)
- SSE-Fix, udev aus dem API-Image entfernt, Import-Smoke-Test

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docker/api/Dockerfile

@@ -1,8 +1,9 @@
-FROM python:3.12-slim
+FROM python:3.12-slim-bookworm
 
 WORKDIR /app
 
-RUN apt-get update && apt-get install -y --no-install-recommends gcc udev && rm -rf /var/lib/apt/lists/*
+# udev ist raus (23.07.): udevadm lieferte im Container nie Daten (kein udevd) —
+# die Geräte-Erkennung läuft jetzt über /sys + ioctls, ganz ohne Systempakete.
 
 COPY docker/api/requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt

docker/api/celery_client.py

@@ -0,0 +1,20 @@
+"""Dünner Celery-Client: die API SCHICKT Tasks an den Worker, führt sie nie aus.
+
+Bis 23.07. gab es überhaupt keinen Code-Pfad, der je einen Rip auslöste —
+kein POST /jobs, kein udev-Daemon. Dieser Client schließt die Lücke.
+"""
+
+import os
+
+from celery import Celery
+
+REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
+
+celery_client = Celery("rippy_api", broker=REDIS_URL, backend=REDIS_URL)
+
+
+def start_rip(device_path: str, job_id: str):
+    """Schickt den Rip-Task an den Worker (Task-Name aus worker/tasks.py)."""
+    return celery_client.send_task(
+        "worker.tasks.rip_disc", args=[device_path, job_id]
+    )

docker/api/db.py

@@ -0,0 +1,140 @@
+"""Job-, Log- und Settings-Persistenz in PostgreSQL (KONZEPT: Postgres für Job-Logs).
+
+Die jobs/logs-Tabellendefinition existiert bewusst identisch im Worker
+(docker/worker/db.py) — es gibt kein geteiltes Paket zwischen den Containern.
+Wer die Struktur ändert, ändert BEIDE Dateien. create_all ist idempotent.
+
+Bis 23.07. lag Postgres komplett brach: GET /jobs gab hart [] zurück, nichts
+schrieb je eine Zeile — der Job-Verlauf im UI war ein Placebo.
+"""
+
+import json
+import os
+from datetime import datetime, timezone
+
+from sqlalchemy import (
+    Column,
+    DateTime,
+    Integer,
+    MetaData,
+    String,
+    Table,
+    Text,
+    create_engine,
+    select,
+)
+
+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("error", Text),
+    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),
+)
+
+settings_table = Table(
+    "settings",
+    metadata,
+    Column("key", String(64), primary_key=True),
+    Column("value", Text),
+)
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def init_db() -> None:
+    """Legt fehlende Tabellen an (idempotent)."""
+    metadata.create_all(engine)
+
+
+def insert_job(job_id: str, device: str, disc_type: str = None, title: str = None) -> None:
+    with engine.begin() as conn:
+        conn.execute(
+            jobs.insert().values(
+                id=job_id,
+                device=device,
+                disc_type=disc_type,
+                title=title,
+                status="pending",
+                progress=0,
+                created_at=utcnow(),
+            )
+        )
+
+
+def list_jobs(limit: int = 100) -> list:
+    with engine.connect() as conn:
+        zeilen = conn.execute(
+            select(jobs).order_by(jobs.c.created_at.desc()).limit(limit)
+        ).mappings().all()
+    return [dict(z) for z in zeilen]
+
+
+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)
+        )
+
+
+def list_logs(limit: int = 200) -> list:
+    with engine.connect() as conn:
+        zeilen = conn.execute(
+            select(logs).order_by(logs.c.id.desc()).limit(limit)
+        ).mappings().all()
+    return [dict(z) for z in zeilen]
+
+
+def get_settings(key: str = "ui") -> dict:
+    with engine.connect() as conn:
+        zeile = conn.execute(
+            select(settings_table.c.value).where(settings_table.c.key == key)
+        ).first()
+    if not zeile or not zeile[0]:
+        return {}
+    try:
+        return json.loads(zeile[0])
+    except ValueError:
+        return {}
+
+
+def save_settings(werte: dict, key: str = "ui") -> None:
+    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))

docker/api/detection.py

@@ -0,0 +1,91 @@
+"""Disc-Erkennung über Kernel-ioctls — ohne `file`, ohne udev.
+
+Warum neu (Review 23.07.): Die alte Erkennung rief `file -L /dev/sr0` auf.
+`file` liest ohne `-s` aber NIE den Inhalt eines Block-Devices — die Ausgabe ist
+immer nur "block special", die Erkennung lieferte also strukturell IMMER
+"unknown". Der alte Test hatte sich die `file`-Ausgabe passend gemockt und
+bewies damit nichts.
+
+Die ioctl-Konstanten stammen aus der Kernel-UAPI (include/uapi/linux/cdrom.h
+bzw. linux/fs.h für BLKGETSIZE64) und sind seit Jahrzehnten stabil.
+"""
+
+import os
+import struct
+from fcntl import ioctl
+
+# include/uapi/linux/cdrom.h
+CDROM_DRIVE_STATUS = 0x5326
+CDROM_DISC_STATUS = 0x5327
+
+CDS_NO_DISC = 1
+CDS_TRAY_OPEN = 2
+CDS_DRIVE_NOT_READY = 3
+CDS_DISC_OK = 4
+
+CDS_AUDIO = 100
+CDS_DATA_1 = 101
+CDS_DATA_2 = 102
+CDS_XA_2_1 = 103
+CDS_XA_2_2 = 104
+CDS_MIXED = 105
+
+# include/uapi/linux/fs.h: BLKGETSIZE64 = _IOR(0x12, 114, size_t) auf 64-bit
+BLKGETSIZE64 = 0x80081272
+
+# Eine DVD9 fasst ~8,5 GB; Blu-ray beginnt bei 25 GB (Single Layer).
+# Alles ab 10 GB ist also sicher eine Blu-ray.
+BLURAY_MIN_BYTES = 10 * 1024**3
+
+
+def _open_nonblock(device_path: str) -> int:
+    """O_NONBLOCK ist Pflicht: ohne blockiert open() bis eine Disc eingelegt ist."""
+    return os.open(device_path, os.O_RDONLY | os.O_NONBLOCK)
+
+
+def drive_status(device_path: str) -> int:
+    """CDROM_DRIVE_STATUS: 1=keine Disc, 2=Schublade offen, 3=nicht bereit, 4=Disc ok."""
+    fd = _open_nonblock(device_path)
+    try:
+        return ioctl(fd, CDROM_DRIVE_STATUS, 0)
+    finally:
+        os.close(fd)
+
+
+def disc_status(device_path: str) -> int:
+    """CDROM_DISC_STATUS: 100=Audio, 101-104=Daten, 105=Mixed."""
+    fd = _open_nonblock(device_path)
+    try:
+        return ioctl(fd, CDROM_DISC_STATUS, 0)
+    finally:
+        os.close(fd)
+
+
+def disc_size_bytes(device_path: str) -> int:
+    """Größe des eingelegten Mediums in Bytes (BLKGETSIZE64)."""
+    fd = _open_nonblock(device_path)
+    try:
+        buf = bytearray(8)
+        ioctl(fd, BLKGETSIZE64, buf)
+        return struct.unpack("Q", bytes(buf))[0]
+    finally:
+        os.close(fd)
+
+
+def classify(disc_status_code: int, size_bytes: int) -> str:
+    """Pure Zuordnung (testbar): Disc-Status + Größe → cd | dvd | bluray | unknown."""
+    if disc_status_code in (CDS_AUDIO, CDS_MIXED):
+        return "cd"
+    if disc_status_code in (CDS_DATA_1, CDS_DATA_2, CDS_XA_2_1, CDS_XA_2_2):
+        return "bluray" if size_bytes >= BLURAY_MIN_BYTES else "dvd"
+    return "unknown"
+
+
+def detect_disc_type(device_path: str) -> str:
+    """Erkennt den Typ der eingelegten Disc; 'no_disc' wenn keine drin ist."""
+    try:
+        if drive_status(device_path) != CDS_DISC_OK:
+            return "no_disc"
+        return classify(disc_status(device_path), disc_size_bytes(device_path))
+    except OSError:
+        return "unknown"

docker/api/devices.py

@@ -0,0 +1,63 @@
+"""Laufwerks-Discovery über /sys und ioctls — ehrlich, ohne udev.
+
+Warum kein udevadm mehr: Im Container läuft kein udevd, die udev-Datenbank ist
+leer — `udevadm info` lieferte schlicht nichts (der alte Weg zeigte deshalb nie
+ein Laufwerk an). /sys/class/block/<name>/device/{vendor,model} kommt dagegen
+direkt vom Kernel und funktioniert überall.
+"""
+
+import glob
+import os
+
+from detection import (
+    CDS_DISC_OK,
+    classify,
+    disc_size_bytes,
+    disc_status,
+    drive_status,
+)
+
+
+def list_optical_devices() -> list:
+    """Alle optischen Laufwerke, die der Container sieht (devices: in compose)."""
+    return sorted(glob.glob("/dev/sr[0-9]*"))
+
+
+def read_sys_attr(device_name: str, attr: str) -> str:
+    pfad = f"/sys/class/block/{device_name}/device/{attr}"
+    try:
+        with open(pfad, encoding="ascii", errors="replace") as f:
+            return f.read().strip()
+    except OSError:
+        return ""
+
+
+def device_info(device_path: str) -> dict:
+    """Baut den Geräte-Eintrag fürs UI: Name aus /sys, Disc-Status per ioctl."""
+    name = os.path.basename(device_path)
+    vendor = read_sys_attr(name, "vendor")
+    model = read_sys_attr(name, "model")
+
+    try:
+        status_code = drive_status(device_path)
+    except OSError:
+        status_code = -1
+
+    disc_type = "unknown"
+    status = "empty"
+    if status_code == CDS_DISC_OK:
+        status = "ready"
+        try:
+            disc_type = classify(disc_status(device_path), disc_size_bytes(device_path))
+        except OSError:
+            disc_type = "unknown"
+
+    return {
+        "id": name,
+        "name": " ".join(teil for teil in (vendor, model) if teil) or f"Laufwerk {name}",
+        "type": disc_type,
+        "path": device_path,
+        "status": status,
+        "model": model,
+        "serial": read_sys_attr(name, "wwid"),
+    }

docker/api/main.py

@@ -4,9 +4,14 @@ from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 from typing import List, Optional, Dict
 from pathlib import Path
-import subprocess
 import asyncio
 import json
+import uuid
+
+import db
+import devices as device_discovery
+from celery_client import start_rip
+from detection import CDS_DISC_OK, CDS_NO_DISC, CDS_TRAY_OPEN, drive_status
 
 from fastapi.security import OAuth2PasswordBearer
 
@@ -41,20 +46,47 @@ app = FastAPI(
 # OAuth2 Scheme
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
-# SSE-Connections
-sse_connections: List = []
-
 
 @app.on_event("startup")
 async def startup_event():
-    """Initialisiere Cache beim Start und validiere Konfiguration."""
+    """Initialisiere Cache + Datenbank, validiere Konfiguration, starte Disc-Wache."""
     init_cache()
+    db.init_db()
 
     try:
         validate_config()
     except ConfigValidationError as e:
         print(f"⚠️  Konfigurations-Warnung: {e}")
 
+    asyncio.create_task(disc_watcher())
+
+
+async def disc_watcher():
+    """Disc-Wache: pollt die Laufwerke und protokolliert Einwurf/Auswurf.
+
+    Ersetzt den nie gebauten udev-Daemon aus dem KONZEPT: udev funktioniert im
+    Container nicht sinnvoll (kein udevd) — ein 3-Sekunden-Poll per ioctl ist
+    für den Heim-Use-Case gleichwertig und läuft überall.
+    """
+    bekannt: Dict[str, int] = {}
+    while True:
+        try:
+            for pfad in device_discovery.list_optical_devices():
+                try:
+                    status = await asyncio.to_thread(drive_status, pfad)
+                except OSError:
+                    continue
+                vorher = bekannt.get(pfad)
+                if vorher is not None and status != vorher:
+                    if status == CDS_DISC_OK:
+                        db.add_log("info", "watcher", f"Disc eingelegt: {pfad}")
+                    elif status in (CDS_NO_DISC, CDS_TRAY_OPEN) and vorher == CDS_DISC_OK:
+                        db.add_log("info", "watcher", f"Disc entfernt: {pfad}")
+                bekannt[pfad] = status
+        except Exception as e:
+            print(f"Disc-Wache: {e}")
+        await asyncio.sleep(3)
+
 
 # Middleware für Rate-Limiting
 @app.middleware("http")
@@ -102,6 +134,8 @@ class Job(BaseModel):
     startTime: str
     endTime: Optional[str] = None
     progress: int = 0
+    title: Optional[str] = None
+    error: Optional[str] = None
 
 class Device(BaseModel):
     id: str
@@ -109,6 +143,24 @@ class Device(BaseModel):
     type: str
     path: str
     status: str
+    model: Optional[str] = None
+    serial: Optional[str] = None
+
+
+def _job_row_to_model(zeile: dict) -> Job:
+    """DB-Zeile → UI-Form (Worker-Status 'running' heißt im UI 'processing')."""
+    status_map = {"running": "processing"}
+    return Job(
+        id=zeile["id"],
+        type=zeile.get("disc_type") or "unknown",
+        status=status_map.get(zeile["status"], zeile["status"]),
+        device=zeile.get("device") or "",
+        startTime=zeile["created_at"].isoformat() if zeile.get("created_at") else "",
+        endTime=zeile["finished_at"].isoformat() if zeile.get("finished_at") else None,
+        progress=zeile.get("progress") or 0,
+        title=zeile.get("title"),
+        error=zeile.get("error"),
+    )
 
 @app.get("/health")
 async def health_check():
@@ -126,95 +178,94 @@ async def root():
 
 @app.get("/jobs", response_model=List[Job])
 async def get_jobs():
-    """Holt alle Jobs."""
+    """Holt alle Jobs aus der Datenbank (neueste zuerst)."""
-    return []
+    zeilen = await asyncio.to_thread(db.list_jobs)
+    return [_job_row_to_model(z) for z in zeilen]
+
+
+class JobCreateRequest(BaseModel):
+    device_path: Optional[str] = None
+    device: Optional[str] = None  # Alias, so schickt es das UI
+    title: Optional[str] = None
+
+
+@app.post("/jobs", status_code=201)
+async def create_job(request: JobCreateRequest):
+    """Legt einen Rip-Job an und schickt ihn an den Worker.
+
+    Das war DIE fehlende Stelle: bis 23.07. gab es keinerlei Code-Pfad,
+    der je einen Rip ausgelöst hat.
+    """
+    device_path = request.device_path or request.device
+    if not device_path:
+        raise HTTPException(status_code=422, detail="device_path fehlt")
+    if device_path not in device_discovery.list_optical_devices():
+        raise HTTPException(status_code=404, detail=f"Laufwerk {device_path} nicht gefunden")
+
+    job_id = str(uuid.uuid4())
+    await asyncio.to_thread(db.insert_job, job_id, device_path, None, request.title)
+    await asyncio.to_thread(db.add_log, "info", "api", f"Job {job_id} angelegt für {device_path}")
+    start_rip(device_path, job_id)
+    return {"id": job_id, "status": "pending", "device": device_path}
+
+
+@app.get("/logs")
+async def get_logs(limit: int = 200):
+    """Echte Ereignisse aus der Datenbank (Watcher, API, Worker)."""
+    zeilen = await asyncio.to_thread(db.list_logs, min(limit, 1000))
+    return [
+        {
+            "id": str(z["id"]),
+            "timestamp": z["ts"].isoformat() if z.get("ts") else "",
+            "level": z.get("level") or "info",
+            "source": z.get("source") or "system",
+            "message": z.get("message") or "",
+        }
+        for z in zeilen
+    ]
+
+
+@app.get("/settings")
+async def get_settings():
+    """UI-Einstellungen aus der Datenbank (leeres Objekt = Defaults im UI)."""
+    return await asyncio.to_thread(db.get_settings)
+
+
+@app.post("/settings")
+async def save_settings(werte: Dict):
+    """Speichert die UI-Einstellungen als JSON in der Datenbank."""
+    await asyncio.to_thread(db.save_settings, werte)
+    return {"status": "saved"}
 
 
 @app.get("/devices", response_model=List[Device])
 async def get_devices():
-    """Holt alle Geräte."""
+    """Alle optischen Laufwerke mit ehrlichem Status (leer/bereit + Disc-Typ).
-    devices = []
 
-    try:
+    Der alte Weg (udevadm + /dev/disc-Symlinks) lieferte im Container
-        # 1. Zuerst Symlinks in /dev/disc/ prüfen
+    prinzipbedingt nichts: kein udevd, keine udev-Datenbank, kein Daemon,
-        result = subprocess.run(
+    der Symlinks anlegt. Jetzt: /sys fürs Modell, ioctl für den Disc-Status.
-            ["ls", "-la", "/dev/disc/"],
+    """
-            capture_output=True,
+    geraete = []
-            text=True,
+    for pfad in device_discovery.list_optical_devices():
-            timeout=5
+        info = await asyncio.to_thread(device_discovery.device_info, pfad)
-        )
+        geraete.append(Device(**info))
-        
+    return geraete
-        for line in result.stdout.strip().split('\n')[1:]:
-            if line and 'total' not in line:
-                parts = line.split()
-                if len(parts) >= 9:
-                    name = parts[-1]
-                    devices.append(Device(
-                        id=name,
-                        name=f"Laufwerk {name}",
-                        type="dvd",
-                        path=f"/dev/disc/{name}",
-                        status="ready"
-                    ))
-        
-        # 2. Direkte Geräte /dev/sr0, /dev/cdrom prüfen
-        import os
-        direct_devices = ['/dev/sr0', '/dev/cdrom', '/dev/dvd']
-        for device_path in direct_devices:
-            if os.path.exists(device_path):
-                # Prüfen ob es ein block device ist
-                import stat
-                mode = os.stat(device_path).st_mode
-                if stat.S_ISBLK(mode):
-                    # Geräte-Info aus udevadm holen
-                    result = subprocess.run(
-                        ["udevadm", "info", "-q", "property", "-n", device_path],
-                        capture_output=True,
-                        text=True,
-                        timeout=5
-                    )
-                    info = {}
-                    for line in result.stdout.strip().split('\n'):
-                        if '=' in line:
-                            key, value = line.split('=', 1)
-                            info[key] = value
-                    
-                    device_type = "dvd"
-                    if info.get("ID_CDROM_BD") == "1":
-                        device_type = "bluray"
-                    elif info.get("ID_CDROM_CD") == "1":
-                        device_type = "cd"
-                    
-                    model = info.get("ID_MODEL", "Laufwerk")
-                    serial = info.get("ID_SERIAL", "unknown")
-                    
-                    devices.append(Device(
-                        id=f"{model}_{serial}".replace(" ", "_"),
-                        name=model,
-                        type=device_type,
-                        path=device_path,
-                        status="ready",
-                        serial=serial,
-                        model=model
-                    ))
-    except Exception as e:
-        print(f"Fehler beim Lesen der Geräte: {e}")
-        pass
-    
-    return devices
 
 
 # SSE-Stream für Echtzeit-Updates
 @app.get("/stream/jobs")
 async def job_stream():
-    """SSE-Stream für Job-Updates."""
+    """SSE-Stream für Job-Updates.
+
+    Fix 23.07.: Der alte Generator sendete nur, wenn `sse_connections` gefüllt
+    war — aber NICHTS hat diese Liste je befüllt. Der Stream war ein Placebo.
+    """
     async def event_generator():
         while True:
-            if sse_connections:
+            jobs = await get_jobs()
-                # Job-Status aktualisieren
+            yield f"data: {json.dumps([j.dict() for j in jobs])}\n\n"
-                jobs = await get_jobs()
+            await asyncio.sleep(2)
-                yield f"data: {json.dumps([j.dict() for j in jobs])}\n\n"
-            await asyncio.sleep(1)
 
     return StreamingResponse(event_generator(), media_type="text/event-stream")
 

docker/api/test_api_smoke.py

@@ -0,0 +1,32 @@
+"""Import-Smoke-Test: bricht, wenn main.py kaputte Imports oder Verdrahtung hat.
+
+Warum: Kein anderer Test importiert main.py — ein Tippfehler dort fiele sonst
+erst beim Container-Start auf (und die Ampel bliebe fälschlich grün).
+Läuft nur unter Linux (detection.py nutzt fcntl/ioctl), also genau dort,
+wo auch die Ampel läuft.
+"""
+
+import sys
+
+import pytest
+
+if sys.platform == "win32":  # pragma: no cover
+    pytest.skip("detection.py braucht fcntl (Linux)", allow_module_level=True)
+
+
+def test_main_importierbar_und_routen_verdrahtet():
+    from main import app
+
+    routen = {route.path for route in app.routes}
+    for pfad in ("/health", "/jobs", "/devices", "/logs", "/settings", "/prescan"):
+        assert pfad in routen, f"Route {pfad} fehlt"
+
+
+def test_worker_task_name_passt_zum_celery_client():
+    """API schickt an 'worker.tasks.rip_disc' — der Name ist Vertrag mit dem Worker."""
+    import inspect
+
+    import celery_client
+
+    quelle = inspect.getsource(celery_client.start_rip)
+    assert '"worker.tasks.rip_disc"' in quelle