1612 lines
53 KiB
Python
1612 lines
53 KiB
Python
"""Cover art sources.
|
|
|
|
Games: libretro-thumbnails box art (no screenshot fallback) plus IGDB and
|
|
ScreenScraper box art when credentials are configured. UMD videos: manual
|
|
covers only (no public API hosts UMD-video box art) — drop images into the
|
|
manual covers folder.
|
|
|
|
Manual covers folder: ``<data>/manual_covers/<games|videos>/<Title>.png``
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from . import config, sources
|
|
from .db import get_connection
|
|
|
|
log = logging.getLogger("covers")
|
|
|
|
_LIBRETRO_BASE = "https://raw.githubusercontent.com/libretro-thumbnails/Sony_-_PlayStation_Portable/master"
|
|
_LIBRETRO_KINDS = ("Named_Boxarts",)
|
|
_TREE_URL = "https://api.github.com/repos/libretro-thumbnails/Sony_-_PlayStation_Portable/git/trees/master?recursive=1"
|
|
|
|
_SS_API = "https://api.screenscraper.fr/api2/jeuInfos.php"
|
|
_SS_PSP_ID = "61" # Sony PSP
|
|
_IGDB_TOKEN_URL = "https://id.twitch.tv/oauth2/token"
|
|
_IGDB_API = "https://api.igdb.com/v4/games"
|
|
_IGDB_IMAGE = "https://images.igdb.com/igdb/image/upload/"
|
|
_IGDB_PSP_ID = 38 # Sony PSP
|
|
_TMDB_SEARCH = "https://api.themoviedb.org/3/search/movie"
|
|
_TMDB_IMAGE = "https://image.tmdb.org/t/p/"
|
|
_GOOGLE_CSE_API = "https://www.googleapis.com/customsearch/v1"
|
|
_MANUAL_DIR = config.DATA_DIR / "manual_covers"
|
|
_OVGDB_ZIP = config.DATA_DIR / "covers" / "openvgdb-covers.zip"
|
|
_LIBRETRO_ZIP = config.DATA_DIR / "covers" / "libretro-covers.zip"
|
|
|
|
_WIKI_SEARCH = "https://en.wikipedia.org/w/api.php"
|
|
_WIKI_SUMMARY = "https://en.wikipedia.org/api/rest_v1/page/summary/"
|
|
|
|
_PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
|
|
_JPEG_MAGIC = b"\xff\xd8\xff"
|
|
_MAX_HOPS = 3 # follow libretro symlink text bodies at most this many times
|
|
_UA = {"User-Agent": "psp-preserve/0.1 (https://github.com; local tool)"}
|
|
_igdb_token_state = {"token": "", "expires": 0.0} # IGDB OAuth2 token cache
|
|
_igdb_lock = threading.Lock()
|
|
_igdb_last = 0.0 # monotonic clock of the last IGDB API call
|
|
|
|
_index_state = {"ts": 0.0, "names": {}, "norm": {}}
|
|
_INDEX_TTL = 86400.0 # refresh the libretro file index once a day
|
|
|
|
_neg_cache = {} # cache key -> time.time() of the last failed fetch
|
|
_NEG_TTL = 1800.0 # skip re-fetching a missing cover for 30 minutes
|
|
|
|
# Persistent "no cover" cache: a DB-backed record of covers that failed so a
|
|
# restart or re-scan doesn't re-hit the network for items we already know are
|
|
# missing. Loaded lazily into memory, then queried in O(1).
|
|
_miss_db = {} # (key, system) -> time.time() of last miss
|
|
_miss_loaded = False
|
|
_MISS_TTL = 86400.0 # retry a missing cover after 1 day
|
|
|
|
|
|
def _load_miss_db():
|
|
global _miss_loaded
|
|
if _miss_loaded:
|
|
return
|
|
_miss_loaded = True
|
|
try:
|
|
conn = get_connection(config.DB_PATH)
|
|
for r in conn.execute(
|
|
"SELECT key, system, ts FROM cover_attempts WHERE status = 'missing'"
|
|
):
|
|
_miss_db[(r["key"], r["system"])] = r["ts"]
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def _mark_missing(key, system):
|
|
ts = time.time()
|
|
_miss_db[(key, system)] = ts
|
|
try:
|
|
conn = get_connection(config.DB_PATH)
|
|
conn.execute(
|
|
"INSERT INTO cover_attempts (key, system, status, ts) VALUES (?,?,?,?) "
|
|
"ON CONFLICT(key, system) DO UPDATE SET status=excluded.status, ts=excluded.ts",
|
|
(key, system, "missing", ts),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def _clear_missing(key, system):
|
|
_miss_db.pop((key, system), None)
|
|
try:
|
|
conn = get_connection(config.DB_PATH)
|
|
conn.execute(
|
|
"DELETE FROM cover_attempts WHERE key = ? AND system = ?", (key, system)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def clear_miss_cache():
|
|
"""Forget all recorded cover misses so they are retried on the next load."""
|
|
global _miss_loaded
|
|
_neg_cache.clear()
|
|
_miss_db.clear()
|
|
_miss_loaded = True
|
|
try:
|
|
conn = get_connection(config.DB_PATH)
|
|
conn.execute("DELETE FROM cover_attempts")
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def clear_cover_cache():
|
|
"""Delete all cached cover images (keeps the libretro index) so every
|
|
cover is re-fetched on the next refresh."""
|
|
clear_miss_cache()
|
|
if not config.COVERS_DIR.is_dir():
|
|
return
|
|
for root, _dirs, files in os.walk(config.COVERS_DIR):
|
|
for name in files:
|
|
if name.upper().startswith("UMD_COVER_TEMPLATE"):
|
|
continue # keep the compositing template asset
|
|
if name.lower().endswith((".png", ".jpg", ".jpeg", ".webp")):
|
|
try:
|
|
(Path(root) / name).unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _download_file(url, dest, timeout=600):
|
|
"""Stream `url` to `dest` (atomic); returns True on success."""
|
|
req = urllib.request.Request(url, headers=_UA)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
if resp.status != 200:
|
|
return False
|
|
tmp = dest.with_name(dest.name + ".tmp")
|
|
with open(tmp, "wb") as f:
|
|
while True:
|
|
chunk = resp.read(1 << 20)
|
|
if not chunk:
|
|
break
|
|
f.write(chunk)
|
|
os.replace(tmp, dest)
|
|
return True
|
|
|
|
|
|
def ensure_cover_bundles(server_url=""):
|
|
"""Download missing offline cover bundle zips from the store server.
|
|
|
|
``server_url`` is the base URL (e.g. ``http://192.168.10.100:8000``); the
|
|
zips are fetched from ``<base>/api/covers/<name>.zip`` unless a direct URL
|
|
is set in config. Returns the names of bundles downloaded.
|
|
"""
|
|
base = (server_url or config.STORE_SERVER_URL or "").rstrip("/")
|
|
downloaded = []
|
|
for name, path, direct in (
|
|
("libretro-covers.zip", _LIBRETRO_ZIP, config.LIBRETRO_COVERS_URL),
|
|
("openvgdb-covers.zip", _OVGDB_ZIP, config.OVGDB_COVERS_URL),
|
|
):
|
|
if path.exists():
|
|
continue
|
|
url = direct or (f"{base}/api/covers/{name}" if base else "")
|
|
if not url:
|
|
continue
|
|
try:
|
|
if _download_file(url, path):
|
|
downloaded.append(name)
|
|
log.info("downloaded offline cover bundle %s", name)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.warning("could not download cover bundle %s: %s", name, exc)
|
|
return downloaded
|
|
|
|
|
|
def cover_url(name, kind="Named_Boxarts"):
|
|
"""Public libretro URL of an image for a catalog name."""
|
|
return f"{_LIBRETRO_BASE}/{kind}/{urllib.parse.quote(name)}.png"
|
|
|
|
|
|
def cover_path(release_id, system):
|
|
"""Local cache path for a release's cover."""
|
|
d = config.COVERS_DIR / (system or "unknown")
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d / f"{release_id}.png"
|
|
|
|
|
|
def stable_key(name):
|
|
"""Process-independent cache key for releases without a database id."""
|
|
import hashlib
|
|
return "n-" + hashlib.md5(name.encode("utf-8")).hexdigest()[:12]
|
|
|
|
|
|
def _is_image(data):
|
|
if not isinstance(data, (bytes, bytearray)):
|
|
return False
|
|
if data.startswith(_PNG_MAGIC) or data.startswith(_JPEG_MAGIC):
|
|
return True
|
|
return len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP"
|
|
|
|
|
|
def file_is_image(path):
|
|
try:
|
|
with open(path, "rb") as f:
|
|
head = f.read(16)
|
|
if head.startswith(_PNG_MAGIC) or head.startswith(_JPEG_MAGIC):
|
|
return True
|
|
return head[:4] == b"RIFF" and head[8:12] == b"WEBP"
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _write_atomic(dest, data):
|
|
tmp = dest.with_name(dest.name + ".tmp")
|
|
tmp.write_bytes(data)
|
|
os.replace(tmp, dest)
|
|
|
|
|
|
def _http_get(url, timeout=10):
|
|
try:
|
|
req = urllib.request.Request(url, headers=_UA)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
if resp.status != 200:
|
|
return None
|
|
return resp.read()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _clean_title(s):
|
|
"""Normalize title punctuation to match libretro filename conventions."""
|
|
s = s.replace(":", " - ")
|
|
s = s.replace("\u2122", "").replace("\u00ae", "") # ™ and ®
|
|
return re.sub(r"\s+", " ", s).strip()
|
|
|
|
|
|
def _candidate_names(name):
|
|
"""Yield progressively shorter thumbnail names (exact first)."""
|
|
for base in (name, _clean_title(name)):
|
|
yield base
|
|
s = base
|
|
while True:
|
|
stripped = re.sub(r"\s*\([^()]*\)\s*$", "", s).strip()
|
|
if not stripped or stripped == s:
|
|
break
|
|
s = stripped
|
|
yield s
|
|
|
|
|
|
def _fetch_png(name, kind, timeout):
|
|
"""Fetch a real image from libretro, following symlink text bodies.
|
|
|
|
GitHub serves a repo symlink as a tiny text body containing the target
|
|
path, so a 200 response is not necessarily an image.
|
|
"""
|
|
target = name
|
|
for _ in range(_MAX_HOPS + 1):
|
|
data = _http_get(cover_url(target, kind), timeout)
|
|
if data is None:
|
|
return None
|
|
if _is_image(data):
|
|
return data
|
|
text = data.decode("utf-8", "replace").strip()
|
|
if not text or "/" in text or not text.endswith(".png"):
|
|
return None
|
|
target = text[:-4]
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# libretro filename index + fuzzy matching
|
|
# --------------------------------------------------------------------------- #
|
|
def _index_path():
|
|
return config.COVERS_DIR / "libretro-index.json"
|
|
|
|
|
|
def _load_index(force=False):
|
|
st = _index_state
|
|
if not force and st["names"] and time.time() - st["ts"] < _INDEX_TTL:
|
|
return st["names"]
|
|
path = _index_path()
|
|
if not force and path.exists():
|
|
try:
|
|
# Load the on-disk index so it doubles as a fallback if the network
|
|
# is down, but only return it immediately while it's still fresh.
|
|
names = json.loads(path.read_text(encoding="utf-8"))
|
|
_set_index(names)
|
|
if time.time() - path.stat().st_mtime < _INDEX_TTL:
|
|
return names
|
|
except Exception:
|
|
pass
|
|
data = _http_get(_TREE_URL, timeout=30)
|
|
if data:
|
|
try:
|
|
tree = json.loads(data).get("tree", [])
|
|
names = {kind: [] for kind in _LIBRETRO_KINDS}
|
|
for entry in tree:
|
|
p = entry.get("path", "")
|
|
for kind in _LIBRETRO_KINDS:
|
|
prefix = kind + "/"
|
|
if p.startswith(prefix) and p.endswith(".png"):
|
|
names[kind].append(p[len(prefix):-4])
|
|
_set_index(names)
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(names), encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
return names
|
|
except Exception:
|
|
pass
|
|
return st["names"] or {kind: [] for kind in _LIBRETRO_KINDS}
|
|
|
|
|
|
def _set_index(names):
|
|
"""Store the raw filename index plus a pre-normalized copy for fast lookup."""
|
|
st = _index_state
|
|
st["names"] = names
|
|
st["norm"] = {k: [(_norm(n), n) for n in v] for k, v in names.items()}
|
|
st["ts"] = time.time()
|
|
|
|
|
|
def _norm(s):
|
|
s = s.lower().replace("&", "and")
|
|
return re.sub(r"[^a-z0-9]+", "", s)
|
|
|
|
|
|
def _plain(s):
|
|
"""Strip parenthetical tags from a title."""
|
|
return re.sub(r"\s*\([^()]*\)", "", s).strip()
|
|
|
|
|
|
def _index_best(name, kind):
|
|
"""Return the best-matching repo filename for `name`, or None."""
|
|
st = _index_state
|
|
if not st["norm"]:
|
|
_load_index()
|
|
pairs = st["norm"].get(kind, [])
|
|
db = _norm(name)
|
|
if not pairs or len(db) < 5:
|
|
return None
|
|
base = _norm(_plain(name))
|
|
best, best_key = None, None
|
|
for nn, n in pairs:
|
|
if nn == db:
|
|
score = 5
|
|
elif nn.startswith(db) or db.startswith(nn):
|
|
score = 3
|
|
elif db in nn or nn in db:
|
|
score = 2
|
|
else:
|
|
continue
|
|
# Prefix/substring matches are only trusted when the core titles also
|
|
# match; otherwise they produce false positives (e.g. "SSX On Tour"
|
|
# matching "SSX - On Tour Portable", or "Spider-Man" matching
|
|
# "Spider-Man - Friend or Foe").
|
|
if score < 5 and _norm(_plain(n)) != base:
|
|
continue
|
|
nl = n.lower()
|
|
bad = any(t in nl for t in ("(beta)", "(demo)", "(proto)", "(alpha)", "(sample)", "(promo)"))
|
|
region = 3
|
|
if "(usa" in nl:
|
|
region = 0
|
|
elif "(europe" in nl:
|
|
region = 1
|
|
elif "(japan" in nl or "(asia" in nl or "(korea" in nl:
|
|
region = 2
|
|
key = (-score, int(bad), region, len(n))
|
|
if best_key is None or key < best_key:
|
|
best, best_key = n, key
|
|
return best
|
|
|
|
|
|
def _fetch_libretro(name, timeout, indexed_only=False):
|
|
# 0. Shipped offline bundle first (fast, no network).
|
|
data = fetch_libretro_local(name)
|
|
if data is not None:
|
|
return data
|
|
if not indexed_only:
|
|
# 1. Cheap: exact + progressively stripped names (no index needed).
|
|
for kind in _LIBRETRO_KINDS:
|
|
for cand in _candidate_names(name):
|
|
data = _fetch_png(cand, kind, timeout)
|
|
if data is not None:
|
|
return data
|
|
# 2. Fuzzy/exact via the precomputed index (covers exact names too, so the
|
|
# bulk loader skips the wasteful exact pass entirely).
|
|
for kind in _LIBRETRO_KINDS:
|
|
best = _index_best(name, kind)
|
|
if best:
|
|
data = _fetch_png(best, kind, timeout)
|
|
if data is not None:
|
|
return data
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Manual covers (user-provided UMD video / game cover images)
|
|
# --------------------------------------------------------------------------- #
|
|
def _names_match(a, b):
|
|
a, b = _norm(a), _norm(b)
|
|
if not a or not b:
|
|
return False
|
|
if len(a) < 4 or len(b) < 4:
|
|
return a == b
|
|
return a == b or a.startswith(b) or b.startswith(a)
|
|
|
|
|
|
def _title_variants(name):
|
|
base = re.sub(r"\s*\([^()]*\)", "", name).strip()
|
|
variants = [name, base]
|
|
m = re.match(r"^(.*),\s*(the|a|an)$", base, re.IGNORECASE)
|
|
if m:
|
|
variants.append(f"{m.group(2)} {m.group(1)}")
|
|
return variants
|
|
|
|
|
|
def fetch_manual(name, system):
|
|
d = _MANUAL_DIR / (system or "unknown")
|
|
if not d.is_dir():
|
|
return None
|
|
variants = _title_variants(name)
|
|
for f in d.iterdir():
|
|
if f.suffix.lower() not in (".png", ".jpg", ".jpeg", ".webp"):
|
|
continue
|
|
if not any(_names_match(f.stem, v) for v in variants):
|
|
continue
|
|
try:
|
|
data = f.read_bytes()
|
|
except OSError:
|
|
continue
|
|
if _is_image(data):
|
|
return data
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# OpenVGDB local bundle (shipped offline covers, see app/openvgdb.py)
|
|
# --------------------------------------------------------------------------- #
|
|
_ovgdb_names = None # cached set of normalized titles present in the bundle
|
|
|
|
|
|
def _ovgdb_name_list():
|
|
global _ovgdb_names
|
|
if _ovgdb_names is None:
|
|
_ovgdb_names = set()
|
|
if _OVGDB_ZIP.exists():
|
|
try:
|
|
with zipfile.ZipFile(_OVGDB_ZIP) as z:
|
|
_ovgdb_names = {n[:-4] for n in z.namelist() if n.endswith(".jpg")}
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return _ovgdb_names
|
|
|
|
|
|
_ovgdb_index = None
|
|
|
|
|
|
def _load_ovgdb_index():
|
|
"""Load {key: {"title", "url"}} written by app/openvgdb.py, or {}.
|
|
|
|
Looks in the runtime covers dir first, then the shipped ``data/covers``
|
|
copy (so a fresh install still shows readable titles without re-running
|
|
the full OpenVGDB scrape).
|
|
"""
|
|
global _ovgdb_index
|
|
if _ovgdb_index is not None:
|
|
return _ovgdb_index
|
|
_ovgdb_index = {}
|
|
for path in (
|
|
config.COVERS_DIR / "openvgdb-index.json",
|
|
config.LEGACY_COVERS_DIR / "openvgdb-index.json",
|
|
):
|
|
if path.exists():
|
|
try:
|
|
_ovgdb_index = json.loads(path.read_text(encoding="utf-8"))
|
|
break
|
|
except Exception: # noqa: BLE001
|
|
_ovgdb_index = {}
|
|
return _ovgdb_index
|
|
|
|
|
|
def fetch_openvgdb_local(name):
|
|
"""Return a bundled OpenVGDB cover image for `name`, or None."""
|
|
names = _ovgdb_name_list()
|
|
if not names:
|
|
return None
|
|
key = _norm(_plain(name))
|
|
entry = None
|
|
if key in names:
|
|
entry = key + ".jpg"
|
|
else:
|
|
for n in names:
|
|
if n.startswith(key) or key.startswith(n):
|
|
entry = n + ".jpg"
|
|
break
|
|
if entry is None:
|
|
return None
|
|
try:
|
|
with zipfile.ZipFile(_OVGDB_ZIP) as z:
|
|
data = z.read(entry)
|
|
return data if _is_image(data) else None
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
|
|
def fetch_libretro_local(name):
|
|
"""Return a bundled libretro box art for `name`, or None."""
|
|
names = _libretro_name_list()
|
|
if not names:
|
|
return None
|
|
key = _norm(_plain(name))
|
|
entry = None
|
|
if key in names:
|
|
entry = key + ".png"
|
|
else:
|
|
for n in names:
|
|
if n.startswith(key) or key.startswith(n):
|
|
entry = n + ".png"
|
|
break
|
|
if entry is None:
|
|
return None
|
|
try:
|
|
with zipfile.ZipFile(_LIBRETRO_ZIP) as z:
|
|
data = z.read(entry)
|
|
return data if _is_image(data) else None
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
|
|
def _libretro_name_list():
|
|
global _libretro_names
|
|
if _libretro_names is None:
|
|
_libretro_names = set()
|
|
if _LIBRETRO_ZIP.exists():
|
|
try:
|
|
with zipfile.ZipFile(_LIBRETRO_ZIP) as z:
|
|
_libretro_names = {n[:-4] for n in z.namelist() if n.endswith(".png")}
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return _libretro_names
|
|
|
|
|
|
_libretro_names = None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# ScreenScraper (games box art, needs devid/devpassword)
|
|
# --------------------------------------------------------------------------- #
|
|
def _screenscraper_box_urls(jeu):
|
|
medias = jeu.get("medias", [])
|
|
if isinstance(medias, dict):
|
|
medias = list(medias.values())
|
|
urls = []
|
|
for m in medias or []:
|
|
if not isinstance(m, dict):
|
|
continue
|
|
if m.get("type") not in ("box-2D", "box-3D"):
|
|
continue
|
|
url = m.get("url")
|
|
if url and url not in urls:
|
|
urls.append(url)
|
|
return urls
|
|
|
|
|
|
def _fetch_screenscraper(name, timeout):
|
|
creds = sources.load()
|
|
devid = creds.get("screenscraper_devid")
|
|
devpassword = creds.get("screenscraper_devpassword")
|
|
if not devid or not devpassword:
|
|
return None
|
|
params = {
|
|
"devid": devid,
|
|
"devpassword": devpassword,
|
|
"softname": creds.get("screenscraper_softname") or "psp-preserve",
|
|
"output": "json",
|
|
"systemeid": _SS_PSP_ID,
|
|
"romtype": "rom",
|
|
"romnom": name,
|
|
}
|
|
ssid = creds.get("screenscraper_ssid")
|
|
if ssid:
|
|
params["ssid"] = ssid
|
|
params["sspassword"] = creds.get("screenscraper_sspassword", "")
|
|
data = _http_get(_SS_API + "?" + urllib.parse.urlencode(params), timeout=timeout)
|
|
if not data:
|
|
return None
|
|
try:
|
|
jeux = json.loads(data).get("response", {}).get("jeu")
|
|
except Exception:
|
|
return None
|
|
if isinstance(jeux, dict):
|
|
jeux = [jeux]
|
|
for jeu in jeux or []:
|
|
if not isinstance(jeu, dict):
|
|
continue
|
|
for url in _screenscraper_box_urls(jeu):
|
|
img = _http_get(url, timeout=timeout)
|
|
if _is_image(img):
|
|
return img
|
|
return None
|
|
|
|
|
|
def _title_score(query, title):
|
|
q, t = _norm(query), _norm(title)
|
|
if not q or not t:
|
|
return 0
|
|
if q == t:
|
|
return 5
|
|
if t.startswith(q) or q.startswith(t):
|
|
return 4
|
|
if q in t or t in q:
|
|
return 3
|
|
return 0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Movie posters (for compositing into the UMD template)
|
|
# --------------------------------------------------------------------------- #
|
|
def movie_search_title(name):
|
|
"""'Dark Knight, The (Europe)' -> 'The Dark Knight'; 'Family Guy Vol. 1' -> 'Family Guy'."""
|
|
base = re.sub(r"\s*\([^()]*\)", "", name).strip()
|
|
base = re.sub(r"\s*vol(?:ume)?\.?\s*\d+\s*$", "", base, flags=re.IGNORECASE).strip()
|
|
m = re.match(r"^(.*),\s*(the|a|an)$", base, re.IGNORECASE)
|
|
if m:
|
|
base = f"{m.group(2)} {m.group(1)}"
|
|
return base.strip()
|
|
|
|
|
|
def _wiki_search(title, limit=5):
|
|
url = _WIKI_SEARCH + "?" + urllib.parse.urlencode({
|
|
"action": "opensearch", "search": title, "limit": limit,
|
|
"namespace": 0, "format": "json",
|
|
})
|
|
data = _http_get(url, timeout=10)
|
|
if not data:
|
|
return []
|
|
try:
|
|
arr = json.loads(data)
|
|
if isinstance(arr, list) and len(arr) >= 2:
|
|
return arr[1]
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def _wiki_thumbnail(title):
|
|
url = _WIKI_SUMMARY + urllib.parse.quote(title.replace(" ", "_"))
|
|
data = _http_get(url, timeout=10)
|
|
if not data:
|
|
return None
|
|
try:
|
|
obj = json.loads(data)
|
|
thumb = obj.get("thumbnail") or {}
|
|
src = thumb.get("source") or (obj.get("originalimage") or {}).get("source")
|
|
except Exception:
|
|
return None
|
|
if not src:
|
|
return None
|
|
return re.sub(r"/\d+px-", "/600px-", src)
|
|
|
|
|
|
def fetch_poster(title, timeout=10):
|
|
"""Return a movie poster image (bytes) from TheMovieDB then Wikipedia."""
|
|
if not title or len(title.strip()) < 2:
|
|
return None
|
|
# 1. TheMovieDB (better quality, instant key)
|
|
for r in _tmdb_results(title, timeout)[:1]:
|
|
data = _http_get(r["image_url"], timeout)
|
|
if _is_image(data):
|
|
return data
|
|
# 2. Wikipedia fallback (keyless)
|
|
bases = [title]
|
|
if "(film)" not in title.lower():
|
|
bases.append(title + " (film)")
|
|
for base in bases:
|
|
for page in _wiki_search(base):
|
|
src = _wiki_thumbnail(page)
|
|
if not src:
|
|
continue
|
|
data = _http_get(src, timeout)
|
|
if _is_image(data):
|
|
return data
|
|
return None
|
|
|
|
|
|
def save_cover(release_id, system, data):
|
|
"""Atomically store composited cover bytes in the cache; return the path."""
|
|
dest = cover_path(release_id, system)
|
|
_write_atomic(dest, data)
|
|
_clear_missing(str(release_id), system)
|
|
return dest
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Game cover compositing (IGDB key art -> full PSP cover)
|
|
# --------------------------------------------------------------------------- #
|
|
_game_template_image = None
|
|
|
|
|
|
def _transparent_rect(img):
|
|
"""Bounding box of the fully-transparent pixels in `img`."""
|
|
minx, miny, maxx, maxy = img.width(), img.height(), -1, -1
|
|
for y in range(img.height()):
|
|
for x in range(img.width()):
|
|
if ((img.pixel(x, y) >> 24) & 0xFF) == 0:
|
|
minx, miny = min(minx, x), min(miny, y)
|
|
maxx, maxy = max(maxx, x), max(maxy, y)
|
|
if maxx < minx or maxy < miny:
|
|
return (0, 0, 0, 0)
|
|
return (minx, miny, maxx - minx + 1, maxy - miny + 1)
|
|
|
|
|
|
def _game_template():
|
|
"""Load the PSP game cover template (top header + transparent window)."""
|
|
global _game_template_image
|
|
if _game_template_image is None:
|
|
try:
|
|
from PySide6.QtGui import QImage
|
|
except Exception: # noqa: BLE001 - headless API may lack PySide6
|
|
return None
|
|
img = QImage()
|
|
for path in (
|
|
config.COVERS_DIR / "UMD_GAME_COVER_TEMPLATE.png",
|
|
config.DATA_DIR / "covers" / "UMD_GAME_COVER_TEMPLATE.png",
|
|
):
|
|
if path.exists():
|
|
img.load(str(path))
|
|
if not img.isNull():
|
|
break
|
|
_game_template_image = img.convertToFormat(QImage.Format_ARGB32) if not img.isNull() else None
|
|
return _game_template_image
|
|
|
|
|
|
def compose_game_cover(key_art_bytes):
|
|
"""Paste IGDB's cropped key art into the PSP game cover template window.
|
|
|
|
Returns composited PNG bytes, or the original bytes when Qt or the template
|
|
is unavailable (the headless HTTP API may not have PySide6 installed).
|
|
"""
|
|
try:
|
|
from PySide6.QtCore import QBuffer, QByteArray, Qt
|
|
from PySide6.QtGui import QImage, QPainter
|
|
except Exception: # noqa: BLE001
|
|
return key_art_bytes
|
|
|
|
template = _game_template()
|
|
art = QImage.fromData(key_art_bytes)
|
|
if template is None or template.isNull() or art.isNull():
|
|
return key_art_bytes
|
|
art = art.convertToFormat(QImage.Format_ARGB32)
|
|
out = QImage(template.size(), QImage.Format_ARGB32)
|
|
out.fill(Qt.transparent)
|
|
x, y, w, h = _transparent_rect(template)
|
|
|
|
painter = QPainter(out)
|
|
if w > 0 and h > 0:
|
|
scaled = art.scaled(w, h, Qt.KeepAspectRatioByExpanding, Qt.SmoothTransformation)
|
|
ox = x + (w - scaled.width()) // 2
|
|
oy = y + (h - scaled.height()) // 2
|
|
painter.save()
|
|
painter.setClipRect(x, y, w, h)
|
|
painter.drawImage(ox, oy, scaled)
|
|
painter.restore()
|
|
painter.drawImage(0, 0, template)
|
|
painter.end()
|
|
|
|
ba = QByteArray()
|
|
buf = QBuffer(ba)
|
|
buf.open(QBuffer.WriteOnly)
|
|
out.save(buf, "PNG")
|
|
buf.close()
|
|
return bytes(ba)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Public entry point
|
|
# --------------------------------------------------------------------------- #
|
|
def fetch_cover(name, release_id, system, timeout=10, quick=False):
|
|
"""Download a cover into the cache; return the cached path or None.
|
|
|
|
``quick`` limits the game fallback to libretro only (index-driven and fast),
|
|
for bulk store loading; details views use the full IGDB/ScreenScraper path.
|
|
"""
|
|
dest = cover_path(release_id, system)
|
|
if dest.exists():
|
|
if file_is_image(dest):
|
|
return dest
|
|
try:
|
|
dest.unlink() # drop corrupt/partial cache and refetch
|
|
except OSError:
|
|
pass
|
|
|
|
# Reuse a cover cached under the name-derived key (the Store saves covers
|
|
# this way for releases without a DB id), so a cover fetched or manually
|
|
# chosen in the Store also applies to the same disc once it is in the Library.
|
|
alt = cover_path(stable_key(name), system)
|
|
if alt != dest and alt.exists() and file_is_image(alt):
|
|
try:
|
|
_write_atomic(dest, alt.read_bytes())
|
|
return dest
|
|
except OSError:
|
|
pass
|
|
|
|
key = str(release_id)
|
|
# In-memory + persistent negative caches: skip the network for recent misses.
|
|
last = _neg_cache.get(key)
|
|
if last is not None and time.time() - last < _NEG_TTL:
|
|
return None
|
|
_load_miss_db()
|
|
miss_ts = _miss_db.get((key, system))
|
|
if miss_ts is not None and time.time() - miss_ts < _MISS_TTL:
|
|
return None
|
|
|
|
t = min(timeout, 3 if quick else 6)
|
|
deadline = time.time() + max(timeout, 6) * (2 if quick else 4)
|
|
|
|
# 1. User-provided covers always win.
|
|
data = fetch_manual(name, system)
|
|
if data is not None:
|
|
log.info("cover %r <- manual", name)
|
|
elif system != "videos":
|
|
# 2. Games: libretro (full box art) first, then the shipped OpenVGDB
|
|
# bundle (offline), then ScreenScraper (full box art, non-quick),
|
|
# then Wikipedia (keyless, non-quick), then IGDB last — IGDB only
|
|
# stores cropped 3:4 key art.
|
|
data = _fetch_libretro(name, t, indexed_only=quick)
|
|
if data is not None:
|
|
log.info("cover %r <- libretro", name)
|
|
else:
|
|
data = fetch_openvgdb_local(name)
|
|
if data is not None:
|
|
log.info("cover %r <- openvgdb", name)
|
|
elif not quick and time.time() < deadline:
|
|
data = _fetch_screenscraper(name, t)
|
|
if data is not None:
|
|
log.info("cover %r <- screenscraper", name)
|
|
if data is None and not quick and time.time() < deadline:
|
|
data = _fetch_wikipedia_cover(name, t)
|
|
if data is not None:
|
|
log.info("cover %r <- wikipedia", name)
|
|
if data is None and time.time() < deadline:
|
|
data = _fetch_igdb_cover(name, t)
|
|
if data is not None:
|
|
# IGDB only has cropped 3:4 key art; composite it into the
|
|
# full PSP cover template (header bar + artwork).
|
|
data = compose_game_cover(data)
|
|
log.info("cover %r <- igdb", name)
|
|
|
|
if data is None:
|
|
_neg_cache[key] = time.time()
|
|
_mark_missing(key, system)
|
|
return None
|
|
_neg_cache.pop(key, None)
|
|
_clear_missing(key, system)
|
|
_write_atomic(dest, data)
|
|
return dest
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Manual search (pick a cover from a source)
|
|
# --------------------------------------------------------------------------- #
|
|
def plain_title(name):
|
|
"""Strip parenthetical tags, leaving the core title."""
|
|
return re.sub(r"\s*\([^()]*\)", "", name).strip()
|
|
|
|
|
|
def download_image(url, timeout=15):
|
|
"""Download and validate an image from a URL; return bytes or None."""
|
|
data = _http_get(url, timeout)
|
|
return data if _is_image(data) else None
|
|
|
|
|
|
def search_covers(query, source="libretro", limit=15, timeout=10):
|
|
"""Return [{title, source, thumb_url, image_url}] for a cover search."""
|
|
if source == "igdb":
|
|
return search_games_igdb(query, limit, timeout)
|
|
if source == "wikipedia":
|
|
return _search_wikipedia(query, limit, timeout)
|
|
if source == "tmdb":
|
|
return _search_tmdb(query, limit, timeout)
|
|
return _search_libretro(query, limit)
|
|
|
|
|
|
def _tmdb_results(query, timeout):
|
|
"""Return [{title, source, thumb_url, image_url}] of TMDB movie posters."""
|
|
key = sources.load().get("tmdb_api_key")
|
|
if not key:
|
|
return []
|
|
url = _TMDB_SEARCH + "?" + urllib.parse.urlencode({"api_key": key, "query": query})
|
|
data = _http_get(url, timeout=timeout)
|
|
if not data:
|
|
return []
|
|
try:
|
|
results = json.loads(data).get("results") or []
|
|
except Exception:
|
|
return []
|
|
out = []
|
|
for r in results:
|
|
poster = r.get("poster_path")
|
|
if not poster:
|
|
continue
|
|
title = r.get("title") or r.get("name") or ""
|
|
year = (r.get("release_date") or "")[:4]
|
|
out.append({
|
|
"title": f"{title} ({year})" if year else title,
|
|
"source": "tmdb",
|
|
"thumb_url": _TMDB_IMAGE + "w185" + poster,
|
|
"image_url": _TMDB_IMAGE + "w500" + poster,
|
|
})
|
|
return out
|
|
|
|
|
|
def _search_tmdb(query, limit, timeout):
|
|
return _tmdb_results(query, timeout)[:limit]
|
|
|
|
|
|
def _search_wikipedia(query, limit=15, timeout=10):
|
|
"""Return [{title, source, thumb_url, image_url}] of images on Wikipedia.
|
|
|
|
Keyless: searches the File: namespace (real poster/cover image files),
|
|
trying ``<query> cover`` and ``<query> poster`` before the raw query so both
|
|
games and movies work.
|
|
"""
|
|
results = []
|
|
for q in (f"{query} cover", f"{query} poster", query):
|
|
for r in _wiki_file_search(q, limit, timeout):
|
|
if r not in results:
|
|
results.append(r)
|
|
if len(results) >= limit:
|
|
return results
|
|
return results
|
|
|
|
|
|
def _search_google_images(query, limit=15, timeout=10):
|
|
"""Return [{title, source, thumb_url, image_url}] from Google Images via the
|
|
Custom Search JSON API. Requires google_api_key + google_cx; keyless -> [].
|
|
"""
|
|
key = sources.load().get("google_api_key")
|
|
cx = sources.load().get("google_cx")
|
|
if not key or not cx:
|
|
return []
|
|
url = _GOOGLE_CSE_API + "?" + urllib.parse.urlencode({
|
|
"key": key,
|
|
"cx": cx,
|
|
"q": query,
|
|
"searchType": "image",
|
|
"num": min(max(limit, 1), 10),
|
|
"imgSize": "xlarge",
|
|
"safe": "off",
|
|
})
|
|
data = _http_get(url, timeout=timeout)
|
|
if not data:
|
|
return []
|
|
try:
|
|
items = json.loads(data).get("items") or []
|
|
except Exception: # noqa: BLE001
|
|
return []
|
|
out = []
|
|
for it in items:
|
|
link = it.get("link") or ""
|
|
thumb = ((it.get("image") or {}).get("thumbnailLink")) or link
|
|
if not link:
|
|
continue
|
|
out.append({
|
|
"id": None,
|
|
"title": it.get("title") or query,
|
|
"thumb_url": thumb,
|
|
"image_url": link,
|
|
"source": "google",
|
|
})
|
|
return out[:limit]
|
|
|
|
|
|
def _fetch_wikipedia_cover(name, timeout):
|
|
"""Return a game cover image from Wikipedia's File: namespace (keyless)."""
|
|
base = plain_title(name)
|
|
if len(base) < 3:
|
|
return None
|
|
seen = set()
|
|
candidates = []
|
|
for q in (f"{base} cover", base):
|
|
for r in _wiki_file_search(q, limit=6, timeout=timeout):
|
|
url = r.get("image_url") or r.get("thumb_url")
|
|
if url and url not in seen:
|
|
seen.add(url)
|
|
candidates.append(r)
|
|
|
|
def rank(r):
|
|
t = (r.get("title") or "").lower()
|
|
if "cover" in t or "box" in t:
|
|
return 0
|
|
if "screenshot" in t or "logo" in t or "icon" in t:
|
|
return 2
|
|
return 1
|
|
|
|
for r in sorted(candidates, key=rank):
|
|
url = r.get("image_url") or r.get("thumb_url")
|
|
data = _http_get(url, timeout=timeout)
|
|
if _is_image(data):
|
|
return data
|
|
return None
|
|
|
|
|
|
def _wiki_file_search(query, limit, timeout):
|
|
url = _WIKI_SEARCH + "?" + urllib.parse.urlencode({
|
|
"action": "query", "list": "search", "srsearch": query,
|
|
"srnamespace": "6", "srlimit": str(max(limit, 1)), "format": "json",
|
|
})
|
|
data = _http_get(url, timeout)
|
|
if not data:
|
|
return []
|
|
try:
|
|
hits = json.loads(data).get("query", {}).get("search", [])
|
|
except Exception:
|
|
return []
|
|
titles = [h["title"] for h in hits if h.get("title", "").startswith("File:")]
|
|
if not titles:
|
|
return []
|
|
info_url = _WIKI_SEARCH + "?" + urllib.parse.urlencode({
|
|
"action": "query", "titles": "|".join(titles[:limit]),
|
|
"prop": "imageinfo", "iiprop": "url", "iiurlwidth": "400",
|
|
"format": "json",
|
|
})
|
|
data = _http_get(info_url, timeout)
|
|
if not data:
|
|
return []
|
|
try:
|
|
pages = json.loads(data).get("query", {}).get("pages", {})
|
|
except Exception:
|
|
return []
|
|
out = []
|
|
for page in pages.values():
|
|
info = (page.get("imageinfo") or [{}])[0]
|
|
full = info.get("url")
|
|
if not full:
|
|
continue
|
|
title = page.get("title", "")
|
|
if title.startswith("File:"):
|
|
title = title[5:]
|
|
if title.lower().endswith((".pdf", ".djvu", ".svg")):
|
|
continue
|
|
out.append({
|
|
"title": title,
|
|
"source": "wikipedia",
|
|
"thumb_url": info.get("thumburl") or full,
|
|
"image_url": full,
|
|
})
|
|
return out
|
|
|
|
|
|
def _search_libretro(query, limit):
|
|
names = _load_index().get("Named_Boxarts", [])
|
|
db = _norm(query)
|
|
results = []
|
|
if not db:
|
|
return results
|
|
scored = []
|
|
for n in names:
|
|
nn = _norm(n)
|
|
if not nn:
|
|
continue
|
|
if nn == db:
|
|
score = 5
|
|
elif nn.startswith(db):
|
|
score = 4
|
|
elif db in nn:
|
|
score = 3
|
|
elif nn in db:
|
|
score = 2
|
|
else:
|
|
continue
|
|
scored.append((score, n))
|
|
scored.sort(key=lambda t: (-t[0], t[1]))
|
|
for score, n in scored[:limit]:
|
|
results.append({
|
|
"title": n,
|
|
"source": "libretro",
|
|
"thumb_url": cover_url(n),
|
|
"image_url": cover_url(n),
|
|
})
|
|
return results
|
|
|
|
|
|
def _search_openvgdb(query, limit):
|
|
"""Return [{title, source, thumb_url, image_url}] from the offline OpenVGDB
|
|
bundle, using the display titles stored in openvgdb-index.json."""
|
|
names = _ovgdb_name_list()
|
|
if not names:
|
|
return []
|
|
index = _load_ovgdb_index()
|
|
db = _norm(query)
|
|
if not db:
|
|
return []
|
|
scored = []
|
|
for key in names:
|
|
if key == db:
|
|
score = 5
|
|
elif len(key) >= 4 and key.startswith(db):
|
|
score = 4
|
|
elif len(key) >= 4 and db in key:
|
|
score = 3
|
|
elif len(db) >= 4 and len(key) >= 4 and key in db:
|
|
score = 2
|
|
else:
|
|
continue
|
|
info = index.get(key) or {}
|
|
title = info.get("title") or key
|
|
scored.append((score, title, key))
|
|
scored.sort(key=lambda t: (-t[0], t[1]))
|
|
return [
|
|
{
|
|
"id": None,
|
|
"title": title,
|
|
"source": "openvgdb",
|
|
"thumb_url": "",
|
|
"image_url": "",
|
|
"_key": key,
|
|
}
|
|
for _score, title, key in scored[:limit]
|
|
]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Game metadata (details sidebar)
|
|
# --------------------------------------------------------------------------- #
|
|
_META_TTL = 86400.0
|
|
|
|
|
|
def get_game_metadata(title, timeout=10):
|
|
"""Return game metadata from IGDB."""
|
|
return get_game_metadata_igdb(title, timeout)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# IGDB (games box art + metadata; generous free tier, key via Twitch dev app)
|
|
# --------------------------------------------------------------------------- #
|
|
def _igdb_creds():
|
|
return sources.load().get("igdb_client_id"), sources.load().get("igdb_client_secret")
|
|
|
|
|
|
def _igdb_token(timeout=10):
|
|
st = _igdb_token_state
|
|
if st["token"] and time.time() < st["expires"] - 60:
|
|
return st["token"]
|
|
cid, secret = _igdb_creds()
|
|
if not cid or not secret:
|
|
return ""
|
|
body = urllib.parse.urlencode({
|
|
"client_id": cid, "client_secret": secret, "grant_type": "client_credentials",
|
|
}).encode("utf-8")
|
|
req = urllib.request.Request(_IGDB_TOKEN_URL, data=body, headers={"User-Agent": _UA["User-Agent"]})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
obj = json.loads(resp.read())
|
|
except Exception:
|
|
return ""
|
|
token = obj.get("access_token") or ""
|
|
expires_in = int(obj.get("expires_in") or 0)
|
|
st["token"] = token
|
|
st["expires"] = time.time() + max(expires_in - 60, 60)
|
|
return token
|
|
|
|
|
|
def _igdb_throttle():
|
|
global _igdb_last
|
|
with _igdb_lock:
|
|
wait = 0.26 - (time.time() - _igdb_last)
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
_igdb_last = time.time()
|
|
|
|
|
|
def _igdb_post(query, timeout=10):
|
|
_igdb_throttle()
|
|
token = _igdb_token(timeout)
|
|
if not token:
|
|
return None
|
|
cid, _ = _igdb_creds()
|
|
req = urllib.request.Request(
|
|
_IGDB_API,
|
|
data=query.encode("utf-8"),
|
|
headers={
|
|
"Client-ID": cid,
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/json",
|
|
"User-Agent": _UA["User-Agent"],
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
if resp.status != 200:
|
|
return None
|
|
return json.loads(resp.read())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _igdb_image_url(image_id, size="cover_big"):
|
|
if not image_id:
|
|
return ""
|
|
if size == "cover_small":
|
|
return f"{_IGDB_IMAGE}t_cover_small/{image_id}.jpg"
|
|
# IGDB only serves its fixed named transforms (the raw upload path 404s and
|
|
# custom Cloudinary params are disabled). t_cover_big_2x is the highest-res
|
|
# cover size (528x704). Note: IGDB stores PSP box art as 3:4, so these will
|
|
# never match libretro's tall scans — that's IGDB's data, not a crop.
|
|
return f"{_IGDB_IMAGE}t_cover_big_2x/{image_id}.jpg"
|
|
|
|
|
|
def _igdb_date(ts):
|
|
if not ts:
|
|
return ""
|
|
try:
|
|
import datetime
|
|
return datetime.datetime.utcfromtimestamp(int(ts)).strftime("%Y-%m-%d")
|
|
except (TypeError, ValueError, OSError):
|
|
return ""
|
|
|
|
|
|
def _igdb_game_meta(g):
|
|
genres = [x.get("name") for x in (g.get("genres") or []) if isinstance(x, dict) and x.get("name")]
|
|
devs, pubs = [], []
|
|
for ic in g.get("involved_companies") or []:
|
|
if not isinstance(ic, dict):
|
|
continue
|
|
company = ic.get("company")
|
|
company = company.get("name") if isinstance(company, dict) else None
|
|
if not company:
|
|
continue
|
|
if ic.get("developer"):
|
|
devs.append(company)
|
|
if ic.get("publisher"):
|
|
pubs.append(company)
|
|
videos = g.get("videos") or []
|
|
yt = ""
|
|
if videos and isinstance(videos[0], dict) and videos[0].get("video_id"):
|
|
yt = "https://www.youtube.com/watch?v=" + videos[0]["video_id"]
|
|
rating = g.get("rating")
|
|
try:
|
|
rating = round(float(rating) / 10.0, 1) if rating else None
|
|
except (TypeError, ValueError):
|
|
rating = None
|
|
return {
|
|
"title": g.get("name") or "",
|
|
"release_date": _igdb_date(g.get("first_release_date")),
|
|
"overview": (g.get("summary") or "").strip(),
|
|
"developers": list(dict.fromkeys(devs)),
|
|
"publishers": list(dict.fromkeys(pubs)),
|
|
"genres": list(dict.fromkeys(genres)),
|
|
"players": None,
|
|
"rating": rating,
|
|
"youtube": yt,
|
|
}
|
|
|
|
|
|
def _igdb_search(query, limit=20, timeout=10):
|
|
cid, _ = _igdb_creds()
|
|
if not cid:
|
|
return []
|
|
term = plain_title(query) or query
|
|
q = (
|
|
f'search "{term}"; '
|
|
"fields name, cover.image_id, first_release_date, summary, genres.name, "
|
|
"involved_companies.company.name, involved_companies.developer, involved_companies.publisher, "
|
|
"rating, videos.video_id; "
|
|
f"where platforms = ({_IGDB_PSP_ID}); "
|
|
f"limit {max(limit, 1)};"
|
|
)
|
|
return _igdb_post(q, timeout) or []
|
|
|
|
|
|
def _fetch_igdb_cover(name, timeout):
|
|
cid, _ = _igdb_creds()
|
|
if not cid:
|
|
return None
|
|
term = plain_title(name) or name
|
|
q = f'search "{term}"; fields cover.image_id; where platforms = ({_IGDB_PSP_ID}); limit 10;'
|
|
games = _igdb_post(q, timeout) or []
|
|
tried = set()
|
|
for g in games:
|
|
img_id = (g.get("cover") or {}).get("image_id")
|
|
if not img_id or img_id in tried:
|
|
continue
|
|
tried.add(img_id)
|
|
data = _http_get(_igdb_image_url(img_id, "cover_big"), timeout)
|
|
if _is_image(data):
|
|
return data
|
|
return None
|
|
|
|
|
|
def search_games_igdb(query, limit=20, timeout=10):
|
|
out = []
|
|
for g in _igdb_search(query, limit, timeout):
|
|
cover = (g.get("cover") or {}).get("image_id") or ""
|
|
out.append({
|
|
"id": g.get("id"),
|
|
"source": "igdb",
|
|
"title": g.get("name") or "",
|
|
"release_date": _igdb_date(g.get("first_release_date")),
|
|
"overview": (g.get("summary") or "").strip(),
|
|
"thumb_url": _igdb_image_url(cover, "cover_small"),
|
|
"image_url": _igdb_image_url(cover, "cover_big"),
|
|
})
|
|
out.sort(key=lambda r: -_title_score(query, r["title"]))
|
|
return out[:limit]
|
|
|
|
|
|
def get_game_metadata_igdb(title, timeout=10):
|
|
games = _igdb_search(title, limit=10, timeout=timeout)
|
|
if not games:
|
|
return None
|
|
games.sort(key=lambda g: -_title_score(title, g.get("name") or ""))
|
|
return _igdb_game_meta(games[0])
|
|
|
|
|
|
def game_metadata_by_id_igdb(game_id, timeout=10):
|
|
cid, _ = _igdb_creds()
|
|
if not cid:
|
|
return None
|
|
q = (
|
|
"fields name, cover.image_id, first_release_date, summary, genres.name, "
|
|
"involved_companies.company.name, involved_companies.developer, involved_companies.publisher, "
|
|
"rating, videos.video_id; "
|
|
f"where id = {int(game_id)}; limit 1;"
|
|
)
|
|
games = _igdb_post(q, timeout) or []
|
|
return _igdb_game_meta(games[0]) if games else None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Movie metadata (details sidebar for UMD videos)
|
|
# --------------------------------------------------------------------------- #
|
|
_tmdb_genres = {"ts": 0.0, "map": {}}
|
|
|
|
|
|
def _tmdb_genre_map(timeout):
|
|
if _tmdb_genres["map"] and time.time() - _tmdb_genres["ts"] < _META_TTL:
|
|
return _tmdb_genres["map"]
|
|
key = sources.load().get("tmdb_api_key")
|
|
if not key:
|
|
return {}
|
|
url = "https://api.themoviedb.org/3/genre/movie/list?" + urllib.parse.urlencode({"api_key": key})
|
|
data = _http_get(url, timeout=timeout)
|
|
if not data:
|
|
return _tmdb_genres["map"]
|
|
try:
|
|
genres = json.loads(data).get("genres") or []
|
|
except Exception:
|
|
return _tmdb_genres["map"]
|
|
result = {g["id"]: g["name"] for g in genres if isinstance(g, dict)}
|
|
_tmdb_genres["map"] = result
|
|
_tmdb_genres["ts"] = time.time()
|
|
return result
|
|
|
|
|
|
def _tmdb_trailer(movie_id, key, timeout):
|
|
url = "https://api.themoviedb.org/3/movie/{}/videos?".format(movie_id) + urllib.parse.urlencode({"api_key": key})
|
|
data = _http_get(url, timeout=timeout)
|
|
if not data:
|
|
return ""
|
|
try:
|
|
results = json.loads(data).get("results") or []
|
|
except Exception:
|
|
return ""
|
|
for v in results:
|
|
if v.get("site") == "YouTube" and v.get("type") == "Trailer" and v.get("key"):
|
|
return "https://www.youtube.com/watch?v=" + v["key"]
|
|
for v in results:
|
|
if v.get("site") == "YouTube" and v.get("key"):
|
|
return "https://www.youtube.com/watch?v=" + v["key"]
|
|
return ""
|
|
|
|
|
|
def _wiki_extract(title):
|
|
url = _WIKI_SUMMARY + urllib.parse.quote(title.replace(" ", "_"))
|
|
data = _http_get(url, timeout=10)
|
|
if not data:
|
|
return None
|
|
try:
|
|
return (json.loads(data).get("extract") or "").strip()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_movie_metadata(title, timeout=10):
|
|
"""Return movie metadata (release date, synopsis, trailer) for a UMD video."""
|
|
if not title:
|
|
return None
|
|
key = sources.load().get("tmdb_api_key")
|
|
if key:
|
|
url = _TMDB_SEARCH + "?" + urllib.parse.urlencode({"api_key": key, "query": title})
|
|
data = _http_get(url, timeout=timeout)
|
|
if data:
|
|
try:
|
|
results = json.loads(data).get("results") or []
|
|
except Exception:
|
|
results = []
|
|
if results:
|
|
m = results[0]
|
|
genres = _tmdb_genre_map(timeout)
|
|
vote = m.get("vote_average")
|
|
return {
|
|
"title": m.get("title") or m.get("name") or "",
|
|
"release_date": m.get("release_date") or "",
|
|
"overview": (m.get("overview") or "").strip(),
|
|
"genres": [genres.get(gid, "") for gid in (m.get("genre_ids") or []) if genres.get(gid)],
|
|
"rating": f"{vote}/10" if isinstance(vote, (int, float)) else None,
|
|
"youtube": _tmdb_trailer(m.get("id"), key, timeout),
|
|
}
|
|
# Wikipedia fallback (keyless): description only
|
|
for page in _wiki_search(title):
|
|
extract = _wiki_extract(page)
|
|
if extract:
|
|
return {
|
|
"title": page,
|
|
"release_date": "",
|
|
"overview": extract,
|
|
"genres": [],
|
|
"rating": None,
|
|
"youtube": "",
|
|
}
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Manual linking (pick the correct movie/game when the auto-match is wrong)
|
|
# --------------------------------------------------------------------------- #
|
|
def search_movies(query, limit=20, timeout=10):
|
|
"""Return movie candidates for a UMD video: TMDB first, then Wikipedia
|
|
images and Google Images (both keyless-optional). A Wikipedia page-title
|
|
fallback is added only when nothing else matched (metadata-only linking).
|
|
"""
|
|
out = []
|
|
seen = set()
|
|
|
|
def add(r):
|
|
k = (r.get("source"), _norm(r.get("title") or ""))
|
|
if k[1] and k not in seen:
|
|
seen.add(k)
|
|
out.append(r)
|
|
|
|
key = sources.load().get("tmdb_api_key")
|
|
if key:
|
|
url = _TMDB_SEARCH + "?" + urllib.parse.urlencode({"api_key": key, "query": query})
|
|
data = _http_get(url, timeout=timeout)
|
|
if data:
|
|
try:
|
|
results = json.loads(data).get("results") or []
|
|
except Exception:
|
|
results = []
|
|
# Cap TMDB results so Wikipedia/Google image picks still fit.
|
|
for r in results[: max(1, min(limit, 6))]:
|
|
poster = r.get("poster_path")
|
|
add({
|
|
"id": r.get("id"),
|
|
"title": r.get("title") or r.get("name") or "",
|
|
"year": (r.get("release_date") or "")[:4],
|
|
"overview": (r.get("overview") or "").strip(),
|
|
"thumb_url": (_TMDB_IMAGE + "w185" + poster) if poster else "",
|
|
"image_url": (_TMDB_IMAGE + "w500" + poster) if poster else "",
|
|
"source": "tmdb",
|
|
})
|
|
|
|
# Wikipedia image search (keyless): real poster/cover files.
|
|
for r in _search_wikipedia(query, min(limit, 10), timeout):
|
|
r["id"] = None
|
|
r["movie_title"] = query
|
|
add(r)
|
|
|
|
# Google Images (Custom Search JSON API, when a key + cx are configured).
|
|
for r in _search_google_images(query, min(limit, 10), timeout):
|
|
r["movie_title"] = query
|
|
add(r)
|
|
|
|
# Last resort: Wikipedia page titles (metadata-only, no image).
|
|
if not out:
|
|
for page in _wiki_search(query, limit):
|
|
out.append({
|
|
"id": None,
|
|
"title": page,
|
|
"year": "",
|
|
"overview": "",
|
|
"thumb_url": "",
|
|
"image_url": "",
|
|
"movie_title": page,
|
|
"source": "wikipedia",
|
|
})
|
|
return out[:limit]
|
|
|
|
|
|
def movie_metadata_by_id(movie_id, timeout=10):
|
|
"""Return full metadata for a specific TMDB movie id (manual linking)."""
|
|
key = sources.load().get("tmdb_api_key")
|
|
if not key:
|
|
return None
|
|
url = "https://api.themoviedb.org/3/movie/{}?".format(movie_id) + urllib.parse.urlencode({"api_key": key})
|
|
data = _http_get(url, timeout=timeout)
|
|
if not data:
|
|
return None
|
|
try:
|
|
m = json.loads(data)
|
|
except Exception:
|
|
return None
|
|
vote = m.get("vote_average")
|
|
return {
|
|
"title": m.get("title") or m.get("name") or "",
|
|
"release_date": m.get("release_date") or "",
|
|
"overview": (m.get("overview") or "").strip(),
|
|
"genres": [g.get("name") for g in (m.get("genres") or []) if isinstance(g, dict)],
|
|
"rating": f"{vote}/10" if isinstance(vote, (int, float)) else None,
|
|
"youtube": _tmdb_trailer(m.get("id"), key, timeout),
|
|
}
|
|
|
|
|
|
def movie_metadata_from_wikipedia(page_title, timeout=10):
|
|
"""Return metadata for a Wikipedia page chosen by the user (keyless)."""
|
|
extract = _wiki_extract(page_title)
|
|
if not extract:
|
|
return None
|
|
return {
|
|
"title": page_title,
|
|
"release_date": "",
|
|
"overview": extract,
|
|
"genres": [],
|
|
"rating": None,
|
|
"youtube": "",
|
|
}
|
|
|
|
|
|
def search_metadata_candidates(system, query, limit=20, timeout=10):
|
|
"""Search candidates the user can link an item to.
|
|
|
|
Games: IGDB (rich metadata) first, then libretro box art and OpenVGDB
|
|
(GameFAQs) box art so covers can be scraped even without an IGDB key.
|
|
"""
|
|
if system == "videos":
|
|
return search_movies(query, limit, timeout)
|
|
results = []
|
|
seen = set()
|
|
|
|
def add(r):
|
|
key = (r.get("source"), _norm(r.get("title") or ""))
|
|
if key[1] and key not in seen:
|
|
seen.add(key)
|
|
results.append(r)
|
|
|
|
for r in search_games_igdb(query, limit, timeout):
|
|
add(r)
|
|
for r in _search_libretro(query, limit):
|
|
r.setdefault("id", None)
|
|
add(r)
|
|
for r in _search_openvgdb(query, limit):
|
|
add(r)
|
|
return results[:limit]
|
|
|
|
|
|
def _empty_meta(title=""):
|
|
return {
|
|
"title": title,
|
|
"release_date": "",
|
|
"overview": "",
|
|
"genres": [],
|
|
"rating": None,
|
|
"youtube": "",
|
|
}
|
|
|
|
|
|
def link_metadata(system, candidate, timeout=10):
|
|
"""Fetch full metadata for a search candidate the user picked."""
|
|
src = candidate.get("source")
|
|
if src == "wikipedia":
|
|
title = candidate.get("movie_title") or candidate.get("title") or ""
|
|
meta = movie_metadata_from_wikipedia(title, timeout) if title else None
|
|
if meta:
|
|
return meta
|
|
if system == "videos":
|
|
return get_movie_metadata(title, timeout) or _empty_meta(title)
|
|
return _empty_meta(title)
|
|
if src in ("libretro", "openvgdb"):
|
|
# Cover repositories have no metadata backend; try IGDB by title so the
|
|
# user still gets synopsis/genre/etc. when a key is configured.
|
|
title = candidate.get("title") or ""
|
|
if title:
|
|
meta = get_game_metadata_igdb(title, timeout)
|
|
if meta:
|
|
return meta
|
|
return _empty_meta(title)
|
|
if system == "videos":
|
|
if src == "tmdb" and candidate.get("id") is not None:
|
|
return movie_metadata_by_id(candidate["id"], timeout)
|
|
title = candidate.get("movie_title") or candidate.get("title") or ""
|
|
return get_movie_metadata(title, timeout) or _empty_meta(title)
|
|
if src == "igdb" and candidate.get("id") is not None:
|
|
return game_metadata_by_id_igdb(candidate["id"], timeout)
|
|
return _empty_meta(candidate.get("title") or "")
|
|
|