Etappe 3: Metadaten-Lookup + Pre-Scan
- SQLite-Cache für API-Rate-Limits (LRU, 10k Einträge) - TMDB/MusicBrainz/TheTVDB Clients - Pre-Scan-Modul für TOC-Lesung ohne Ripping - Metadaten-Preview UI - Jellyfin-Formatierung (NFO + Images) - API Endpoints für Lookup, Confirm, Format
This commit is contained in:
Hitonabi
@@ -0,0 +1,299 @@
|
||||
"""Pre-Scan-Modul: Liest Disc-TOC ohne Ripping."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from .clients.tmdb import TMDBClient
|
||||
from .clients.musicbrainz import MusicBrainzClient
|
||||
from .clients.thetvdb import TheTVDBClient
|
||||
from .cache import get, set
|
||||
from .config import settings
|
||||
|
||||
|
||||
class PreScanResult:
|
||||
def __init__(
|
||||
self,
|
||||
disc_type: str,
|
||||
title: str,
|
||||
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()
|
||||
|
||||
def scan(self, device_path: str) -> PreScanResult:
|
||||
"""Führe Pre-Scan durch."""
|
||||
# 1. Disc-Typ erkennen
|
||||
disc_type = self._detect_disc_type(device_path)
|
||||
|
||||
# 2. TOC lesen
|
||||
toc = self._read_toc(device_path, disc_type)
|
||||
|
||||
# 3. Metadaten lookup
|
||||
if disc_type == "CD":
|
||||
return self._scan_audio(device_path, toc)
|
||||
else:
|
||||
return self._scan_video(device_path, toc)
|
||||
|
||||
def _detect_disc_type(self, device_path: str) -> str:
|
||||
"""Erkenne Disc-Typ (CD/DVD/Blu-ray)."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["isosize", "-x", device_path],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5
|
||||
)
|
||||
|
||||
# Ausgabe parsen
|
||||
if result.returncode == 0:
|
||||
size = int(result.stdout.strip())
|
||||
if size < 700 * 1024 * 1024: # < 700MB
|
||||
return "CD"
|
||||
elif size < 15 * 1024 * 1024 * 1024: # < 15GB
|
||||
return "DVD"
|
||||
else:
|
||||
return "Blu-ray"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fallback
|
||||
return "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":
|
||||
# cdparanoia für CD-TOC
|
||||
result = subprocess.run(
|
||||
["cdparanoia", "-Q", device_path],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
# Parse cdparanoia output
|
||||
for line in result.stdout.split('\n'):
|
||||
if 'track' in line.lower():
|
||||
# Track-Info parsen
|
||||
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:
|
||||
# makemkvcon für DVD/Blu-ray (nur TOC, kein Ripping)
|
||||
result = subprocess.run(
|
||||
["makemkvcon", "--minlength=300", "--progress=off", "info", device_path],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
# Parse makemkvcon output
|
||||
for line in result.stdout.split('\n'):
|
||||
if line.startswith('DRV:'):
|
||||
# Disc-Info
|
||||
parts = line.split(',')
|
||||
if len(parts) >= 4:
|
||||
toc["title"] = parts[3].strip().strip('"')
|
||||
elif line.startswith('TINFO:'):
|
||||
# Titel-Info
|
||||
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
|
||||
cache_key = f"prescan:audio:{device_path}"
|
||||
cached = get(cache_key)
|
||||
if cached:
|
||||
return PreScanResult(**cached)
|
||||
|
||||
# Künstler und Album aus Titel extrahieren
|
||||
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]
|
||||
|
||||
# MusicBrainz Search
|
||||
if artist:
|
||||
artists = self.musicbrainz.search_artist(artist)
|
||||
if artists:
|
||||
# Best match
|
||||
best_match = artists[0]
|
||||
confidence = 0.8
|
||||
|
||||
# Album suchen
|
||||
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:
|
||||
# Tracks aus 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
|
||||
})
|
||||
|
||||
# TMDB Fallback für bekannte Soundtracks
|
||||
if not artist:
|
||||
movies = self.tmdb.search_movie(album)
|
||||
if movies:
|
||||
confidence = 0.7
|
||||
movie = movies[0]
|
||||
movie_details = self.tmdb.get_movie_details(movie["id"])
|
||||
|
||||
if movie_details:
|
||||
# Soundtrack als "Soundtrack - [Filmname]"
|
||||
artist = album
|
||||
album = f"Soundtrack - {movie_details.get('title', album)}"
|
||||
toc["title"] = album
|
||||
|
||||
# Cache result
|
||||
result = PreScanResult(
|
||||
disc_type="CD",
|
||||
title=album,
|
||||
tracks=toc["tracks"],
|
||||
confidence=confidence if 'confidence' in dir() else 0.5
|
||||
)
|
||||
|
||||
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
|
||||
cache_key = f"prescan:video:{device_path}"
|
||||
cached = get(cache_key)
|
||||
if cached:
|
||||
return PreScanResult(**cached)
|
||||
|
||||
# Disc-Titel
|
||||
title = toc["title"] or "Unknown Title"
|
||||
|
||||
# TMDB Search
|
||||
confidence = 0.0
|
||||
metadata = {}
|
||||
matched = False
|
||||
|
||||
# Versuche Film-Suche
|
||||
movies = self.tmdb.search_movie(title)
|
||||
if movies:
|
||||
for movie in movies:
|
||||
# Prüfe auf gute Übereinstimmung
|
||||
if movie.get("title", "").lower() == title.lower():
|
||||
confidence = 0.95
|
||||
movie_id = movie["id"]
|
||||
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
|
||||
|
||||
# Versuche Serien-Suche
|
||||
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_id = show["id"]
|
||||
tv_details = self.tmdb.get_tv_details(tv_id)
|
||||
|
||||
if tv_details:
|
||||
metadata = {
|
||||
"type": "tv",
|
||||
"id": tv_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 auf Titel ohne Lookup
|
||||
if not matched:
|
||||
confidence = 0.3
|
||||
metadata = {
|
||||
"type": "unknown",
|
||||
"title": title,
|
||||
"year": None
|
||||
}
|
||||
|
||||
# Cache result
|
||||
result = PreScanResult(
|
||||
disc_type=toc.get("disc_type", "DVD"),
|
||||
title=title,
|
||||
year=metadata.get("year"),
|
||||
confidence=confidence,
|
||||
metadata=metadata,
|
||||
tracks=toc.get("tracks", [])
|
||||
)
|
||||
|
||||
set(cache_key, result.to_dict())
|
||||
return result
|
||||
Reference in New Issue
Block a user