Initial release — MediaTracker v1.0
Dockerized media library scanner for NAS drives. Identifies movies and series from filenames, fetches posters via the public IMDB suggestion API (no key required), and exposes a dark-themed web UI with multilingual search support. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
178
backend/metadata.py
Normal file
178
backend/metadata.py
Normal file
@@ -0,0 +1,178 @@
|
||||
"""
|
||||
metadata.py — Poster + metadata fetching.
|
||||
- Poster: IMDB public suggestion API (no key required)
|
||||
- Alt titles (FR/EN): Wikidata SPARQL (free, no registration)
|
||||
- Local poster detection (poster.jpg / folder.jpg alongside video)
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_REQUEST_TIMEOUT = 4
|
||||
_WIKIDATA_TIMEOUT = 8
|
||||
_LOCAL_POSTER_NAMES = ["poster.jpg", "poster.png", "folder.jpg", "folder.png",
|
||||
"cover.jpg", "cover.png", "movie.jpg"]
|
||||
_ALT_TITLE_LANGS = ("fr", "en", "de", "es")
|
||||
|
||||
|
||||
@dataclass
|
||||
class MediaMetadata:
|
||||
poster_url: Optional[str] = None
|
||||
imdb_id: Optional[str] = None
|
||||
|
||||
|
||||
# ─── IMDB suggestion API ─────────────────────────────────────────────────────
|
||||
|
||||
def _imdb_suggest(query: str) -> list[dict]:
|
||||
query_clean = query.lower().strip()
|
||||
if not query_clean:
|
||||
return []
|
||||
first_char = query_clean[0] if query_clean[0].isalpha() else "a"
|
||||
encoded = urllib.parse.quote(query_clean)
|
||||
url = f"https://sg.media-imdb.com/suggests/{first_char}/{encoded}.json"
|
||||
try:
|
||||
req = urllib.request.Request(url, headers={
|
||||
"User-Agent": "Mozilla/5.0",
|
||||
"Accept": "application/json, text/javascript",
|
||||
})
|
||||
with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp:
|
||||
raw = resp.read().decode("utf-8")
|
||||
start = raw.index('(')
|
||||
end = raw.rindex(')')
|
||||
data = json.loads(raw[start + 1:end])
|
||||
return data.get("d", [])
|
||||
except (ValueError, KeyError):
|
||||
pass
|
||||
except urllib.error.HTTPError as exc:
|
||||
if exc.code != 404:
|
||||
logger.debug("IMDB suggest HTTP %d for %r", exc.code, query)
|
||||
except Exception as exc:
|
||||
logger.debug("IMDB suggest error for %r: %s", query, exc)
|
||||
return []
|
||||
|
||||
|
||||
def _extract_poster(item: dict) -> Optional[str]:
|
||||
img = item.get("i")
|
||||
if isinstance(img, list) and img:
|
||||
return str(img[0])
|
||||
if isinstance(img, dict):
|
||||
return img.get("imageUrl")
|
||||
return None
|
||||
|
||||
|
||||
def _best_match(items: list[dict], title: str, year: Optional[int], qid_filter: str) -> Optional[dict]:
|
||||
"""Return the best matching IMDB item dict, or None."""
|
||||
title_lower = title.lower()
|
||||
candidates = [it for it in items if it.get("qid") == qid_filter]
|
||||
if not candidates:
|
||||
return None
|
||||
if year:
|
||||
for it in candidates:
|
||||
if it.get("l", "").lower() == title_lower and it.get("y") == year:
|
||||
return it
|
||||
for it in candidates:
|
||||
if it.get("l", "").lower() == title_lower:
|
||||
return it
|
||||
# First candidate that has a poster
|
||||
for it in candidates:
|
||||
if _extract_poster(it):
|
||||
return it
|
||||
return candidates[0] if candidates else None
|
||||
|
||||
|
||||
def _get_metadata(title: str, year: Optional[int], qid_filter: str) -> MediaMetadata:
|
||||
items = _imdb_suggest(title)
|
||||
item = _best_match(items, title, year, qid_filter)
|
||||
if not item:
|
||||
return MediaMetadata()
|
||||
return MediaMetadata(
|
||||
poster_url=_extract_poster(item),
|
||||
imdb_id=item.get("id"),
|
||||
)
|
||||
|
||||
|
||||
def get_movie_metadata(title: str, year: Optional[int]) -> MediaMetadata:
|
||||
return _get_metadata(title, year, "movie")
|
||||
|
||||
|
||||
def get_series_metadata(title: str, year: Optional[int]) -> MediaMetadata:
|
||||
return _get_metadata(title, year, "tvSeries")
|
||||
|
||||
|
||||
# ─── Wikidata alt titles ─────────────────────────────────────────────────────
|
||||
|
||||
def get_alt_titles_batch(imdb_ids: list[str], retries: int = 3) -> dict[str, list[tuple[str, str]]]:
|
||||
"""
|
||||
Fetch FR/EN/DE/ES titles from Wikidata for a batch of IMDB IDs.
|
||||
Returns {imdb_id: [(lang, title), ...]}
|
||||
Retries up to `retries` times on 429/503 with exponential backoff.
|
||||
"""
|
||||
import time as _time
|
||||
|
||||
if not imdb_ids:
|
||||
return {}
|
||||
|
||||
values = " ".join(f'"{iid}"' for iid in imdb_ids)
|
||||
langs = " ".join(f'"{l}"' for l in _ALT_TITLE_LANGS)
|
||||
query = f"""
|
||||
SELECT ?imdbId ?label WHERE {{
|
||||
VALUES ?imdbId {{ {values} }}
|
||||
?item wdt:P345 ?imdbId .
|
||||
?item rdfs:label ?label .
|
||||
FILTER(LANG(?label) IN ({langs}))
|
||||
}}
|
||||
"""
|
||||
url = "https://query.wikidata.org/sparql?" + urllib.parse.urlencode({
|
||||
"query": query,
|
||||
"format": "json",
|
||||
})
|
||||
|
||||
for attempt in range(retries):
|
||||
try:
|
||||
req = urllib.request.Request(url, headers={
|
||||
"User-Agent": "MediaTracker/1.0",
|
||||
"Accept": "application/sparql-results+json",
|
||||
})
|
||||
with urllib.request.urlopen(req, timeout=_WIKIDATA_TIMEOUT) as resp:
|
||||
data = json.loads(resp.read().decode("utf-8"))
|
||||
|
||||
result: dict[str, list[tuple[str, str]]] = {}
|
||||
for row in data.get("results", {}).get("bindings", []):
|
||||
iid = row["imdbId"]["value"]
|
||||
label = row["label"]["value"]
|
||||
lang = row["label"].get("xml:lang", "")
|
||||
result.setdefault(iid, []).append((lang, label))
|
||||
return result
|
||||
|
||||
except urllib.error.HTTPError as exc:
|
||||
if exc.code in (429, 503) and attempt < retries - 1:
|
||||
wait = 5 * (2 ** attempt)
|
||||
logger.info("Wikidata rate-limited — retrying in %ds (attempt %d/%d)", wait, attempt + 1, retries)
|
||||
_time.sleep(wait)
|
||||
else:
|
||||
logger.warning("Wikidata batch failed (%d ids): HTTP %d", len(imdb_ids), exc.code)
|
||||
return {}
|
||||
except Exception as exc:
|
||||
logger.warning("Wikidata batch failed (%d ids): %s", len(imdb_ids), exc)
|
||||
return {}
|
||||
|
||||
return {}
|
||||
|
||||
|
||||
# ─── Local poster detection ───────────────────────────────────────────────────
|
||||
|
||||
def find_local_poster(video_path: Path) -> Optional[Path]:
|
||||
directory = video_path.parent
|
||||
for name in _LOCAL_POSTER_NAMES:
|
||||
candidate = directory / name
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
return None
|
||||
Reference in New Issue
Block a user