rippy/docker/api/devices.py

"""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 fcntl import ioctl

from detection import (
    CDS_DISC_OK,
    classify,
    disc_size_bytes,
    disc_status,
    drive_status,
)

# include/uapi/linux/cdrom.h
CDROMEJECT = 0x5309
CDROM_LOCKDOOR = 0x5329       # 1 = Tür verriegeln, 0 = entriegeln
CDROM_DRIVE_STATUS = 0x5326
CDS_NO_DISC = 1
CDS_TRAY_OPEN = 2

AUSWURF_WARTEN_SEKUNDEN = 5


def eject(device_path: str) -> None:
    """Wirft die Disc aus und prüft es nach. Wirft OSError, wenn sie drin bleibt.

    ⚠️ ERST ENTRIEGELN (Befund 26.07.2026, am laufenden System gemessen): Ein
    nacktes CDROMEJECT wird von einem verriegelten Laufwerk mit ERFOLG quittiert
    und tut nichts. MakeMKV verriegelt die Tür während des Rips
    (`CDROM_LOCKDOOR 1`) und entriegelt sie nicht wieder — danach blieb die
    Schublade zu, während Rippy „Disc ausgeworfen" ins Log schrieb. Gegenprobe
    an derselben Disc: mit `CDROM_LOCKDOOR 0` davor geht sie auf (Status 2).
    Deshalb macht das Werkzeug `eject` immer beides.

    Gleichlautend in worker/ripping.wirf_disc_aus — es gibt kein geteiltes Paket
    zwischen den Containern; dort steht die ausführliche Herleitung.
    """
    import time

    fd = os.open(device_path, os.O_RDONLY | os.O_NONBLOCK)
    try:
        try:
            ioctl(fd, CDROM_LOCKDOOR, 0)
        except OSError:
            pass  # nicht verriegelt oder ioctl unbekannt — Auswurf trotzdem versuchen
        ioctl(fd, CDROMEJECT, 0)
        # Nachsehen statt hoffen. „Kein Datenträger" zählt mit: ein
        # Slot-Laufwerk hat keine Schublade.
        for _ in range(AUSWURF_WARTEN_SEKUNDEN):
            if ioctl(fd, CDROM_DRIVE_STATUS, 0) in (CDS_TRAY_OPEN, CDS_NO_DISC):
                return
            time.sleep(1)
        raise OSError(
            "Das Laufwerk hat den Auswurf angenommen, die Disc ist aber noch "
            "drin. Blockiert etwas die Schublade, oder läuft noch ein Zugriff?"
        )
    finally:
        os.close(fd)


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"),
    }