Hitonabi
bd6409038e
Ampel / ampel (push) Successful in 35s
Praxis-Befund mit echter BD-50: reines UDF, der ISO-PVD-Leser griff nicht. Jetzt ECMA-167-Weg (AVDP Sektor 256 -> Main VDS -> PVD Tag-ID 1 -> Volume Identifier als d-string), mit Tests fuer das d-string-Parsing. ROADMAP Etappe 12 aus der ARM-Vollanalyse: was wir uebernehmen (Fingerprint-DB, bdmt_eng.xml, Suchdegradation, Manual Mode, Apprise, Multi-Drive, Backup-Modus) und wo wir ARM schlagen (Serien-Episoden- Matching, Main-Feature per Laufzeitabgleich). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
369 lines
14 KiB
Python
369 lines
14 KiB
Python
"""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 shutil
|
|
import subprocess
|
|
from typing import Dict, List, Optional
|
|
|
|
import detection
|
|
from clients.tmdb import TMDBClient
|
|
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
|
|
|
|
|
|
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_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
|
|
):
|
|
self.disc_type = disc_type
|
|
self.title = title
|
|
self.year = year
|
|
self.confidence = confidence
|
|
self.metadata = metadata or {}
|
|
self.tracks = tracks or []
|
|
|
|
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
|
|
}
|
|
|
|
|
|
class PreScan:
|
|
def __init__(self):
|
|
self.tmdb = TMDBClient()
|
|
self.musicbrainz = MusicBrainzClient()
|
|
self.thetvdb = TheTVDBClient()
|
|
self.omdb = OMDbClient()
|
|
|
|
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)
|
|
|
|
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"}.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: Volume-Label direkt vom Medium — läuft
|
|
# überall. ISO-9660 für DVDs, UDF für Blu-rays (die haben
|
|
# meist keine ISO-Bridge). (makemkvcon gibt es NUR im
|
|
# Worker-Container; der alte Aufruf hier scheiterte in der
|
|
# API still und der Titel blieb ewig "Unknown Title".)
|
|
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", "--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)
|
|
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
|
|
)
|
|
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)
|
|
cached = get(cache_key)
|
|
if cached:
|
|
return PreScanResult(**cached)
|
|
|
|
title = normalize_disc_label(toc["title"] or "") or "Unknown Title"
|
|
confidence = 0.0
|
|
metadata = {}
|
|
matched = False
|
|
|
|
movies = self.tmdb.search_movie(title)
|
|
if movies:
|
|
for movie in movies:
|
|
if movie.get("title", "").lower() == title.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", [])]
|
|
}
|
|
matched = True
|
|
break
|
|
|
|
if not matched:
|
|
tv_shows = self.tmdb.search_tv(title)
|
|
if tv_shows:
|
|
for show in tv_shows:
|
|
if show.get("name", "").lower() == title.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", [])]
|
|
}
|
|
matched = True
|
|
break
|
|
|
|
# Fallback 1: OMDb (eigene Datenbasis — findet oft, was TMDB nicht
|
|
# exakt trifft; braucht OMDB_API_KEY, sonst überspringt es sich selbst)
|
|
if not matched:
|
|
omdb_treffer = self.omdb.lookup(title)
|
|
if omdb_treffer:
|
|
confidence = 0.8
|
|
metadata = omdb_treffer
|
|
title = omdb_treffer.get("title", title)
|
|
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", [])]
|
|
}
|
|
matched = True
|
|
|
|
if not matched:
|
|
confidence = 0.3
|
|
metadata = {
|
|
"type": "unknown",
|
|
"title": title,
|
|
"year": None
|
|
}
|
|
|
|
result = PreScanResult(
|
|
disc_type=toc.get("disc_type", "DVD"),
|
|
title=title,
|
|
year=metadata.get("year"),
|
|
confidence=confidence,
|
|
metadata=metadata,
|
|
tracks=toc.get("tracks", [])
|
|
)
|
|
cache_set(cache_key, result.to_dict())
|
|
return result
|