rippy/docker/worker/detection.py

"""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
# 4K-UHD-Discs sind BD-66 (66 GB) oder BD-100 — eine normale BD-50 bleibt
# unter ~47 GiB. Ab 55 GiB ist es also sicher eine UHD. (Seltene 50-GB-UHDs
# laufen als "bluray" — der Rip-Weg ist ohnehin identisch.)
UHD_MIN_BYTES = 55 * 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 | uhd | 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):
        if size_bytes >= UHD_MIN_BYTES:
            return "uhd"
        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"