diff --git a/docker/api/Dockerfile b/docker/api/Dockerfile index 26c3144..98045c9 100644 --- a/docker/api/Dockerfile +++ b/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 diff --git a/docker/api/celery_client.py b/docker/api/celery_client.py new file mode 100644 index 0000000..7bad835 --- /dev/null +++ b/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] + ) diff --git a/docker/api/db.py b/docker/api/db.py new file mode 100644 index 0000000..04406e3 --- /dev/null +++ b/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)) diff --git a/docker/api/detection.py b/docker/api/detection.py new file mode 100644 index 0000000..287d37e --- /dev/null +++ b/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" diff --git a/docker/api/devices.py b/docker/api/devices.py new file mode 100644 index 0000000..553aa0b --- /dev/null +++ b/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//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"), + } diff --git a/docker/api/main.py b/docker/api/main.py index 21322d0..d181ac3 100644 --- a/docker/api/main.py +++ b/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,96 +178,95 @@ async def root(): @app.get("/jobs", response_model=List[Job]) async def get_jobs(): - """Holt alle Jobs.""" - return [] + """Holt alle Jobs aus der Datenbank (neueste zuerst).""" + 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.""" - devices = [] - - try: - # 1. Zuerst Symlinks in /dev/disc/ prüfen - result = subprocess.run( - ["ls", "-la", "/dev/disc/"], - capture_output=True, - text=True, - timeout=5 - ) - - 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 + """Alle optischen Laufwerke mit ehrlichem Status (leer/bereit + Disc-Typ). + + Der alte Weg (udevadm + /dev/disc-Symlinks) lieferte im Container + prinzipbedingt nichts: kein udevd, keine udev-Datenbank, kein Daemon, + der Symlinks anlegt. Jetzt: /sys fürs Modell, ioctl für den Disc-Status. + """ + geraete = [] + for pfad in device_discovery.list_optical_devices(): + info = await asyncio.to_thread(device_discovery.device_info, pfad) + geraete.append(Device(**info)) + return geraete # 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: - # Job-Status aktualisieren - jobs = await get_jobs() - yield f"data: {json.dumps([j.dict() for j in jobs])}\n\n" - await asyncio.sleep(1) - + jobs = await get_jobs() + yield f"data: {json.dumps([j.dict() for j in jobs])}\n\n" + await asyncio.sleep(2) + return StreamingResponse(event_generator(), media_type="text/event-stream") diff --git a/docker/api/test_api_smoke.py b/docker/api/test_api_smoke.py new file mode 100644 index 0000000..0eba54f --- /dev/null +++ b/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