"""Pre-Scan-Modul: Liest Disc-TOC ohne Ripping und schlägt Metadaten vor.
WIEDERHERGESTELLT 22.07.2026: Beim SoC-Refactoring wurde die echte Implementierung
(flache prescan.py) durch einen Stub überschattet — die Metadaten-Preview (Muss-Feature)
lieferte seitdem immer „Unknown Disc". Dies ist die echte Logik, bereinigt.
"""
import os
import shutil
import subprocess
from typing import Dict, List, Optional
from rippy import drives as _laufwerks_schicht
from clients.tmdb import TMDBClient
from clients.jikan import JikanClient
from clients.musicbrainz import MusicBrainzClient
from clients.omdb import OMDbClient
from clients.thetvdb import TheTVDBClient
from cache import get, set as cache_set
from cache.keys import generate_prescan_key
# Der Treiber DIESER Maschine statt fest der Linux-Fassung — sonst waere
# dieses Modul (und ueber es main.py) unter Windows nicht ladbar. Beide
# Treiber bieten disc_size_bytes() und detect_disc_type() an; darauf achtet
# test_treiberwahl.py.
detection = _laufwerks_schicht.treiber()
def normalize_disc_label(label: str) -> str:
"""Disc-Labels wie 'PULP_FICTION_DE' → 'Pulp Fiction De' (pure Funktion).
Volume-Labels sind fast immer GROSS_MIT_UNTERSTRICHEN — so findet keine
Metadaten-API etwas. Unterstriche zu Leerzeichen, Titel-Schreibung.
"""
if not label:
return ""
bereinigt = " ".join(label.replace("_", " ").replace(".", " ").split())
if bereinigt.isupper():
bereinigt = bereinigt.title()
return bereinigt
def read_iso_volume_label(device_path: str) -> Optional[str]:
"""Liest das Volume-Label aus dem ISO-9660 Primary Volume Descriptor.
Sektor 16 (Offset 32768), Bytes 40-71 = Volume Identifier. Funktioniert für
DVDs — Blu-rays sind oft reines UDF ohne ISO-Bridge (dann greift
read_udf_volume_label).
"""
try:
with open(device_path, "rb") as f:
f.seek(32768)
pvd = f.read(2048)
if len(pvd) < 72 or pvd[1:6] != b"CD001":
return None
label = pvd[40:72].decode("ascii", errors="replace").strip()
return label or None
except OSError:
return None
def parse_udf_dstring(data: bytes) -> str:
"""Dekodiert einen UDF d-string (ECMA-167 1/7.2.12; pure Funktion, testbar).
Letztes Byte = genutzte Länge, erstes Byte = Kompressions-ID
(8 = Latin-1, 16 = UTF-16BE), dazwischen der Text.
"""
if not data:
return ""
laenge = data[-1]
if laenge < 2 or laenge > len(data) - 1:
return ""
comp_id, text = data[0], data[1:laenge]
try:
if comp_id == 8:
return text.decode("latin-1").strip()
if comp_id == 16:
return text.decode("utf-16-be").strip()
except UnicodeDecodeError:
pass
return ""
def read_disc_title_via_mount(device_path: str) -> Optional[str]:
"""Liest den KLARTEXT-Titel einer Blu-ray aus BDMV/META/DL/bdmt_*.xml.
Das Volume-Label ist oft kryptisch (BD_EVG_D2) — der echte Titel
(„Evangelion: 2.22 …") steht in den Disc-Metadaten. Die API darf
read-only mounten (CAP_SYS_ADMIN ist für die Speicherziele ohnehin da).
Gibt None zurück, wenn kein BD-Metadatensatz existiert (z. B. DVD).
"""
import re as _re
mountpoint = "/mnt/rippy-disc"
os.makedirs(mountpoint, exist_ok=True)
gemountet = False
try:
ergebnis = subprocess.run(
["mount", "-t", "udf", "-o", "ro", device_path, mountpoint],
capture_output=True, text=True, timeout=30,
)
if ergebnis.returncode != 0:
return None
gemountet = True
meta_dir = os.path.join(mountpoint, "BDMV", "META", "DL")
if not os.path.isdir(meta_dir):
return None
kandidaten = sorted(os.listdir(meta_dir))
# bdmt_eng.xml bevorzugen, sonst erste bdmt-Datei (bdmt_ger.xml, …)
bdmt = next((k for k in kandidaten if k == "bdmt_eng.xml"), None) or next(
(k for k in kandidaten if k.startswith("bdmt_") and k.endswith(".xml")), None
)
if not bdmt:
return None
with open(os.path.join(meta_dir, bdmt), "rb") as f:
inhalt = f.read(65536).decode("utf-8", errors="replace")
# Titel — bewusst per Regex statt XML-Parser
# (Namespaces variieren je Authoring-Werkzeug)
treffer = _re.search(r"([^<]{2,120})", inhalt)
if treffer:
return treffer.group(1).strip()
return None
except Exception:
return None
finally:
if gemountet:
subprocess.run(["umount", mountpoint], capture_output=True, timeout=15)
def titel_kandidaten(titel: str) -> List[str]:
"""Suchvarianten für die Metadaten-APIs (progressive Degradation, ARM-Lehre).
'Evangelion: 2.22 You Can (Not) Advance.' → auch 'Evangelion: 2.22 …' ohne
Zusatz, ohne Doppelpunkt-Teil usw. — Disc-Titel sind selten API-freundlich.
"""
kandidaten = []
def merke(t: str):
t = " ".join(t.split()).strip(" .")
if t and t not in kandidaten:
kandidaten.append(t)
merke(titel)
if ":" in titel:
merke(titel.split(":", 1)[0]) # Haupttitel vor dem Doppelpunkt
merke(titel.replace(":", ""))
import re as _re
merke(_re.sub(r"\([^)]*\)", "", titel)) # Klammer-Zusätze raus
# Iterativ hintere Worte strippen (ARM-Lehre): "Evangelion 2.22" muss
# auch als "Evangelion" gesucht werden — Versions-/Zusatz-Tokens stehen
# fast immer hinten. Begrenzt, damit die API-Anfragen überschaubar bleiben.
worte = titel.split()
while len(worte) > 1 and len(kandidaten) < 6:
worte = worte[:-1]
merke(" ".join(worte))
return kandidaten
def disc_fingerprint(device_path: str) -> str:
"""Billiger, stabiler Disc-Fingerabdruck (Volume-Label + Größe).
Für Cache-Keys und das Merken manueller Korrekturen — bewusst OHNE den
teuren Mount-Titel, damit ihn jeder Endpunkt schnell berechnen kann.
"""
label = read_iso_volume_label(device_path) or read_udf_volume_label(device_path) or "unbekannt"
try:
groesse = detection.disc_size_bytes(device_path)
except OSError:
groesse = 0
return f"{label}|{groesse}"
def read_udf_volume_label(device_path: str) -> Optional[str]:
"""Liest das Volume-Label aus dem UDF Primary Volume Descriptor.
Weg laut ECMA-167: Anchor Volume Descriptor Pointer bei Sektor 256 →
zeigt auf die Main Volume Descriptor Sequence → darin der PVD (Tag-ID 1)
mit dem Volume Identifier (d-string, 32 Bytes ab Offset 24).
"""
sektor = 2048
try:
with open(device_path, "rb") as f:
f.seek(256 * sektor)
avdp = f.read(sektor)
if len(avdp) < 24 or int.from_bytes(avdp[0:2], "little") != 2:
return None
vds_ort = int.from_bytes(avdp[20:24], "little")
f.seek(vds_ort * sektor)
# VDS-Deskriptoren durchgehen, bis der PVD (Tag-ID 1) kommt
for _ in range(32):
block = f.read(sektor)
if len(block) < 56:
return None
tag_id = int.from_bytes(block[0:2], "little")
if tag_id == 1:
return parse_udf_dstring(block[24:56]) or None
if tag_id == 8: # Terminating Descriptor — Sequenz zu Ende
return None
except OSError:
return None
return None
class PreScanResult:
def __init__(
self,
disc_type: str = "DVD",
title: str = "Unknown",
year: int = None,
confidence: float = 0.0,
metadata: Dict = None,
tracks: List[Dict] = None,
fingerprint: str = ""
):
self.disc_type = disc_type
self.title = title
self.year = year
self.confidence = confidence
self.metadata = metadata or {}
self.tracks = tracks or []
# Fingerabdruck (Label|Größe) wandert mit ins Ergebnis — Basis der
# Duplikat-Warnung („diese Disc wurde schon gerippt")
self.fingerprint = fingerprint
def to_dict(self) -> Dict:
return {
"disc_type": self.disc_type,
"title": self.title,
"year": self.year,
"confidence": self.confidence,
"metadata": self.metadata,
"tracks": self.tracks,
"fingerprint": self.fingerprint
}
class PreScan:
def __init__(self):
self.tmdb = TMDBClient()
self.musicbrainz = MusicBrainzClient()
self.thetvdb = TheTVDBClient()
self.omdb = OMDbClient()
self.jikan = JikanClient()
def scan(self, device_path: str) -> PreScanResult:
"""Führe Pre-Scan durch."""
disc_type = self._detect_disc_type(device_path)
toc = self._read_toc(device_path, disc_type)
# Typ + Disc-Fingerabdruck (Label|Größe) an die Zweige durchreichen —
# der Typ ging vorher verloren (BDs hießen immer "DVD"), und der
# Cache-Key braucht den Fingerabdruck (sonst erbt die nächste Disc
# die Metadaten der vorherigen).
toc["disc_type"] = disc_type
toc["fingerprint"] = disc_fingerprint(device_path)
if disc_type == "CD":
return self._scan_audio(device_path, toc)
return self._scan_video(device_path, toc)
def _detect_disc_type(self, device_path: str) -> str:
"""Erkenne Disc-Typ über die zentrale ioctl-Erkennung (detection.py).
Vorher lief hier `isosize` — das Werkzeug war im Container nicht
installiert, die Erkennung fiel still immer auf "DVD" zurück.
"""
typ = detection.detect_disc_type(device_path)
return {"cd": "CD", "dvd": "DVD", "bluray": "Blu-ray", "uhd": "4K UHD"}.get(typ, "DVD")
def _read_toc(self, device_path: str, disc_type: str) -> Dict:
"""Lese TOC (Table of Contents)."""
toc = {
"title": None,
"year": None,
"tracks": [],
"duration": 0
}
try:
if disc_type == "CD":
result = subprocess.run(
["cdparanoia", "-Q", device_path],
capture_output=True,
text=True,
timeout=10
)
for line in result.stdout.split('\n'):
if 'track' in line.lower():
parts = line.split()
if len(parts) >= 4:
toc["tracks"].append({
"track_number": parts[1],
"title": " ".join(parts[3:]) if len(parts) > 3 else "Track",
"duration": 0
})
else:
# Titel-Quelle 1 (beste): Klartext-Titel aus den BD-Metadaten
# (BDMV/META — Labels wie BD_EVG_D2 taugen nicht für APIs)
klartext = read_disc_title_via_mount(device_path)
if klartext:
toc["title"] = klartext
else:
# Titel-Quelle 2: Volume-Label direkt vom Medium.
# ISO-9660 für DVDs, UDF für Blu-rays (keine ISO-Bridge).
label = read_iso_volume_label(device_path) or read_udf_volume_label(device_path)
if label:
toc["title"] = normalize_disc_label(label)
# Titel-Quelle 2 (optional, falls makemkvcon doch da ist):
if shutil.which("makemkvcon"):
result = subprocess.run(
["makemkvcon", "-r", "--noscan", "--minlength=300", "info", f"dev:{device_path}"],
capture_output=True,
text=True,
timeout=120
)
for line in result.stdout.split('\n'):
if line.startswith('TINFO:'):
parts = line.split(',')
if len(parts) >= 5:
toc["tracks"].append({
"title": parts[4].strip().strip('"') if len(parts) > 4 else "Title",
"duration": int(parts[2]) if len(parts) > 2 else 0
})
except Exception as e:
print(f"Pre-Scan TOC Error: {e}")
return toc
def _scan_audio(self, device_path: str, toc: Dict) -> PreScanResult:
"""Pre-Scan für Audio-CD."""
cache_key = generate_prescan_key(device_path, is_audio=True, fingerprint=toc.get("fingerprint", ""))
cached = get(cache_key)
if cached:
return PreScanResult(**cached)
confidence = 0.5
artist = None
album = toc["title"] or "Unknown Album"
if " - " in album:
parts = album.split(" - ", 1)
if len(parts) == 2:
artist = parts[0]
album = parts[1]
if artist:
artists = self.musicbrainz.search_artist(artist)
if artists:
confidence = 0.8
releases = self.musicbrainz.search_release(album, artist)
if releases:
confidence = 0.9
release_id = releases[0]["id"]
release_info = self.musicbrainz.get_release_info(release_id)
if release_info:
for medium in release_info.get("media", []):
for track in medium.get("tracks", []):
toc["tracks"].append({
"track_number": track.get("position", 0),
"title": track.get("title", "Unknown"),
"duration": track.get("duration", 0) // 1000
})
if not artist:
movies = self.tmdb.search_movie(album)
if movies:
confidence = 0.7
movie_details = self.tmdb.get_movie_details(movies[0]["id"])
if movie_details:
album = f"Soundtrack - {movie_details.get('title', album)}"
toc["title"] = album
result = PreScanResult(
disc_type="CD",
title=album,
tracks=toc["tracks"],
confidence=confidence,
fingerprint=toc.get("fingerprint", "")
)
cache_set(cache_key, result.to_dict())
return result
def _scan_video(self, device_path: str, toc: Dict) -> PreScanResult:
"""Pre-Scan für DVD/Blu-ray."""
cache_key = generate_prescan_key(device_path, is_audio=False, fingerprint=toc.get("fingerprint", ""))
cached = get(cache_key)
if cached:
return PreScanResult(**cached)
# Titel NICHT nochmal normalisieren: aus bdmt_*.xml kommt er sauber
# („2.22" würde die Label-Normalisierung zu „2 22" zerlegen)
title = (toc.get("title") or "").strip() or "Unknown Title"
confidence = 0.0
metadata = {}
matched = False
# Progressive Suchdegradation (ARM-Lehre): mehrere Titel-Varianten
# probieren — Disc-Titel sind selten API-freundlich formatiert.
kandidaten = titel_kandidaten(title) or [title]
movies = []
for kandidat in kandidaten:
treffer = self.tmdb.search_movie(kandidat)
if treffer and not movies:
movies = treffer # bester Rohtreffer für den Vorschlags-Fallback
for movie in treffer:
if movie.get("title", "").lower() == kandidat.lower():
confidence = 0.95
movie_details = self.tmdb.get_movie_details(movie["id"])
if movie_details:
metadata = {
"type": "movie",
"id": movie["id"],
"title": movie_details.get("title", title),
"year": int(movie_details.get("release_date", "0")[:4]) if movie_details.get("release_date") else None,
"overview": movie_details.get("overview", ""),
"poster_path": movie_details.get("poster_path", ""),
"backdrop_path": movie_details.get("backdrop_path", ""),
"runtime": movie_details.get("runtime", 0),
"genres": [g["name"] for g in movie_details.get("genres", [])],
"source": "tmdb",
}
matched = True
break
if matched:
break
if not matched:
for kandidat in kandidaten:
tv_shows = self.tmdb.search_tv(kandidat)
for show in tv_shows:
if show.get("name", "").lower() == kandidat.lower():
confidence = 0.9
tv_details = self.tmdb.get_tv_details(show["id"])
if tv_details:
metadata = {
"type": "tv",
"id": show["id"],
"title": tv_details.get("name", title),
"year": int(tv_details.get("first_air_date", "0")[:4]) if tv_details.get("first_air_date") else None,
"overview": tv_details.get("overview", ""),
"poster_path": tv_details.get("poster_path", ""),
"backdrop_path": tv_details.get("backdrop_path", ""),
"genres": [g["name"] for g in tv_details.get("genres", [])],
"source": "tmdb",
}
matched = True
break
if matched:
break
# Fallback 1: Jikan/MyAnimeList (kostenlos, KEIN Key) — für Anime die
# präziseste Quelle; wählt per Titel-Ähnlichkeit, nicht Treffer #1.
#
# NEU 26.07.2026 — „exakt schlägt unscharf", unabhängig von der
# Reihenfolge. Vorher galt der erste Jikan-Treffer über dem Gate 0,55
# sofort als sicherer Fund (confidence 0,85) und OMDb kam nie dran.
# Das ist zu großzügig, und zwar messbar (offline gerechnet mit
# titel_aehnlichkeit, echte MyAnimeList-Titel):
#
# „Alien" vs. „Alien 9" → 0,83 TRIFFT
# „Inception" vs. „Deception" → 0,78 TRIFFT
# „Hero" vs. „Heroman" → 0,73 TRIFFT
# „The Dark Knight" vs. „Dark Knight Rises" → 0,69 TRIFFT
#
# Ein Kinofilm, den TMDB verpasst, bekam damit Anime-Metadaten — und
# OMDb, das ihn kennt, wurde nie gefragt. Bemerkenswert dabei, und das
# widerlegt die Vermutung im SAVEPOINT v3.16, das Gate sei einfach zu
# niedrig: Für den Fall, für den Jikan überhaupt eingebaut wurde, ist
# es sogar zu HOCH — „Evangelion 2.22" vs. „Evangelion: 2.0 You Can
# (Not) Advance" ergibt 0,51 und fällt durch. Eine einzelne Zahl kann
# beides nicht leisten.
#
# Deshalb: Nur ein praktisch exakter Titel gilt sofort. Ein unscharfer
# Treffer wird GEMERKT, dann OMDb gefragt — und erst wenn OMDb nichts
# hat, kommt er als VORSCHLAG (niedrige Confidence) zum Zug.
jikan_unscharf = None
if not matched:
for kandidat in kandidaten:
jikan_treffer = self.jikan.lookup(kandidat)
if not jikan_treffer:
continue
if jikan_treffer.get("match_score", 0) >= JikanClient.SICHER_AEHNLICHKEIT:
confidence = 0.85
metadata = jikan_treffer
matched = True
else:
jikan_unscharf = jikan_treffer
break
# Fallback 2: OMDb (eigene Datenbasis — findet oft, was TMDB nicht
# exakt trifft; braucht OMDB_API_KEY, sonst überspringt es sich selbst)
if not matched:
for kandidat in kandidaten:
omdb_treffer = self.omdb.lookup(kandidat)
if omdb_treffer:
confidence = 0.8
metadata = omdb_treffer
title = omdb_treffer.get("title", title)
matched = True
break
# Fallback 2b: der unscharfe Jikan-Treffer — besser als nichts, aber
# ehrlich als Vorschlag ausgewiesen (0,6 = dieselbe Stufe wie ein
# TMDB-Vorschlag ohne exakten Treffer; das UI zeigt die Prozentzahl an
# und der Nutzer kann korrigieren).
if not matched and jikan_unscharf:
confidence = 0.6
metadata = jikan_unscharf
matched = True
# Fallback 2: bester TMDB-Vorschlag ohne exakten Treffer — als
# VORSCHLAG gekennzeichnet (niedrige Confidence, Nutzer korrigiert)
if not matched and movies:
vorschlag = self.tmdb.get_movie_details(movies[0]["id"])
if vorschlag:
confidence = 0.6
metadata = {
"type": "movie",
"id": movies[0]["id"],
"title": vorschlag.get("title", title),
"year": int(vorschlag.get("release_date", "0")[:4]) if vorschlag.get("release_date") else None,
"overview": vorschlag.get("overview", ""),
"poster_path": vorschlag.get("poster_path", ""),
"backdrop_path": vorschlag.get("backdrop_path", ""),
"runtime": vorschlag.get("runtime", 0),
"genres": [g["name"] for g in vorschlag.get("genres", [])],
"source": "tmdb",
}
matched = True
if not matched:
confidence = 0.3
metadata = {
"type": "unknown",
"title": title,
"year": None
}
# Deutsche Texte für OMDb-Treffer nachladen (Commander-Wunsch 24.07.):
# OMDb kann nur Englisch — über die IMDb-ID liefert TMDB /find den
# deutschen Titel, die Beschreibung und oft ein besseres Poster.
if matched and metadata.get("source") == "omdb" and str(metadata.get("id", "")).startswith("tt"):
deutsch = self.tmdb.find_by_imdb(metadata["id"])
if deutsch:
for feld in ("title", "overview", "poster_path", "year", "type"):
if deutsch.get(feld):
metadata[feld] = deutsch[feld]
# Bei Treffer den sauberen API-Titel anzeigen statt des Disc-Titels
if matched and metadata.get("title"):
title = metadata["title"]
result = PreScanResult(
disc_type=toc.get("disc_type", "DVD"),
title=title,
year=metadata.get("year"),
confidence=confidence,
metadata=metadata,
tracks=toc.get("tracks", []),
fingerprint=toc.get("fingerprint", "")
)
# Nur ECHTE Treffer cachen: ein gecachtes "unknown" würde sonst auch
# nach Key-Eintrag/Fix ewig wieder serviert (Redis ist persistent).
# Mittelgute Treffer verfallen nach einem Tag — falls eine bessere
# Quelle (Jikan-504 vorbei, TMDB-Key neu) später mehr weiß.
if confidence >= 0.9:
cache_set(cache_key, result.to_dict(), expire=7 * 86400)
elif confidence >= 0.6:
cache_set(cache_key, result.to_dict(), expire=86400)
return result