ckoch/LudosData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fill developer, publisher, year and description from Wikipedia

Browse Source 
The library carried titles, systems and genres but almost nothing else:
developer was 3.8% filled, publisher 2.9%, description 0%. These are columns
the app has always had and never been able to populate.

enrich_metadata.py reads them off the same articles the cover fetcher
locates. Only empty fields are touched unless --overwrite is given.

  year         79%  -> 96%
  developer   3.8%  -> 95%
  publisher   2.9%  -> 96%
  description   0%  -> 96%

Parsing infoboxes needed several guards, each found by checking output
rather than trusting the first pass:

  * "Infobox video game" is a substring of "Infobox video game series", so
    the loose test resolved Banjo-Kazooie to the series overview. Now
    rejected, which also fixes the cover fetcher's article resolution.
  * An article spans every release and its date block leads with the
    original, so year is only filled when the article covers that platform.
    Otherwise a DS port inherits the SNES original's year.
  * A search hit that neither covers the platform nor closely matches the
    title is discarded: "Dragon Ball Z Budokai" surfaces "Shin Budokai", a
    different game on a different console. Left blank instead.
  * Values are grouped under bold platform headings, tagged with region
    codes, annotated with the platform in parentheses, and wrapped in
    templates whose named parameters leak through. Each of those read as
    the developer or publisher before being handled.

Developer, publisher and year are facts and written verbatim. Descriptions
are article summaries under CC BY-SA, stored with an attribution line.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
ckoch
2026-08-04 12:34:55 -04:00
co-authored by Claude Opus 5
parent 130921cf89
commit 771b34bb4b
7 changed files with 503 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/library/__pycache__/enrich_metadata.cpython-314.pyc
BIN
View File
Binary file not shown.
tools/library/__pycache__/fetch_art.cpython-314.pyc
BIN
View File
Binary file not shown.
tools/library/enrich_metadata.py
+436
View File
@@ -0,0 +1,436 @@
#!/usr/bin/env python3
"""Fill in developer, publisher, year and description from Wikipedia.
The library's own data is thin in exactly these columns — the 2018 dump carried
titles, systems and genres, but almost no developer or publisher and no
descriptions at all. This reads the values off the same English Wikipedia
articles the cover fetcher already locates.
Developer, publisher and release year are facts and are written verbatim.
Descriptions come from the article summary, which is CC BY-SA text, so each one
is stored with an attribution line naming the source article.
Only empty fields are filled unless --overwrite is given: anything typed by hand
outranks anything guessed here.
python3 enrich_metadata.py --password '...' --dry-run
python3 enrich_metadata.py --password '...'
"""
from __future__ import annotations
import argparse
import difflib
import json
import re
import sys
import urllib.parse
from pathlib import Path
# The cover fetcher already owns the throttled, cached Wikipedia client.
from fetch_art import (
WIKI_UA, api_json, http, is_game_article, normalise, wiki_api, wiki_article,
)
FIELDS = ("developer", "publisher", "year", "description")
# Infobox values for multi-platform games are frequently grouped under a bold
# platform heading ("'''PlayStation'''{{vgrelease|JP|Square|...}}") or annotated
# with the platform in parentheses ("Rare (N64)<br>Nintendo (DS)"). Both would
# otherwise be read as the developer or publisher itself.
PLATFORM_WORDS = (
"nes", "snes", "super nes", "super nintendo", "nintendo entertainment system",
"n64", "nintendo 64", "gamecube", "gc", "wii", "wii u", "switch",
"nintendo switch", "game boy", "game boy color", "game boy advance", "gb",
"gbc", "gba", "nintendo ds", "ds", "3ds", "nintendo 3ds",
"playstation", "ps1", "ps2", "ps3", "ps4", "ps5", "playstation 2",
"playstation 3", "playstation 4", "playstation 5", "psp",
"playstation portable", "playstation vita", "vita",
"xbox", "xbox 360", "xbox one", "xbox series x", "x360",
"windows", "win", "pc", "ms-dos", "dos", "mac", "macos", "osx", "linux",
"ios", "android", "arcade", "arc", "genesis", "mega drive", "md",
"dreamcast", "saturn", "amiga", "psx", "xbla", "wiiu", "ngc", "3do",
"steam", "microsoft windows",
)
# Whether an article covers the platform a row claims. Used to keep a port's
# entry from inheriting the original release's year.
PLATFORM_PATTERNS = {
"NES": r"Nintendo Entertainment System|\bNES\b",
"SNES": r"Super Nintendo|Super NES|\bSNES\b",
"N64": r"Nintendo 64",
"GB": r"Game Boy(?! Advance)", # plain Game Boy or Game Boy Color
"GBA": r"Game Boy Advance",
"DS": r"Nintendo DS",
"GC": r"GameCube",
"WII": r"\bWii\b(?! ?U)",
"PS1": r"PlayStation(?!\s*(?:2|3|4|5|Portable|Vita))",
"PS2": r"PlayStation 2",
"PSP": r"PlayStation Portable|\bPSP\b",
"360": r"Xbox 360",
}
# Release templates tag each entry with a region, so those codes sit between the
# heading and the value we actually want.
REGION_CODES = {
"jp", "na", "eu", "au", "ww", "uk", "us", "ca", "kr", "cn", "br", "ru", "in",
"sea", "int", "pal", "ntsc", "row", "as", "hk", "tw", "mx", "nz", "fr", "de",
"es", "it", "nl", "se", "no", "dk", "fi", "pt", "pl",
}
DATE_LIKE = re.compile(
r"^\s*(?:\d{1,2}\s+)?(?:January|February|March|April|May|June|July|August|"
r"September|October|November|December)\b|^\s*\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\s*$"
r"|^\s*(?:19|20)\d{2}\s*$",
re.I,
)
def is_platform_label(text: str) -> bool:
return text.strip().lower().strip(":;,") in PLATFORM_WORDS
def is_noise_entry(text: str) -> bool:
"""Region tags, dates and platform headings are scaffolding, not values."""
stripped = text.strip().lower().strip(":;,")
return (
not stripped
or stripped in REGION_CODES
or is_platform_label(stripped)
or bool(DATE_LIKE.match(text))
)
def article_covers_system(wikitext: str, system: str | None) -> bool:
"""True when the article's platform list mentions the row's system."""
if not system:
return False
pattern = PLATFORM_PATTERNS.get(system.upper())
if not pattern:
return False
platforms = infobox_field(wikitext, "platforms") or infobox_field(wikitext, "platform") or ""
return re.search(pattern, platforms, flags=re.I) is not None
# --------------------------------------------------------------------------
# Article resolution
# --------------------------------------------------------------------------
def resolve_article(game: dict, cache_dir: Path) -> tuple[str, str] | None:
"""Find the article about this specific game.
Direct title guesses first, then a search. The search matters for titles
whose bare name is the series overview rather than a game — "Mortal Kombat"
and "SimCity" both are, with the games at "(1992 video game)" and
"(1989 video game)" — and for small punctuation differences, such as our
"Dragon Ball Z Budokai" against the catalogued "Dragon Ball Z: Budokai".
"""
title, year, system = game["title"], game.get("year"), game.get("system")
attempts = [title, f"{title} (video game)"]
if year:
attempts.append(f"{title} ({year} video game)")
for candidate in attempts:
found = wiki_article(candidate, cache_dir)
if found and is_game_article(found[1]):
return found
try:
results = wiki_api(list="search", srsearch=f"{title} video game", srlimit=5)
titles = [r["title"] for r in results.get("query", {}).get("search", [])]
except Exception: # noqa: BLE001
return None
target = normalise(title)
scored: list[tuple[bool, float, tuple[str, str]]] = []
for candidate in titles:
found = wiki_article(candidate, cache_dir)
if not found or not is_game_article(found[1]):
continue
# Strip the disambiguator before comparing: "Mortal Kombat (1992 video
# game)" should read as "Mortal Kombat".
bare = re.sub(r"\s*\([^)]*\)\s*$", "", found[0])
similarity = difflib.SequenceMatcher(None, target, normalise(bare)).ratio()
scored.append((article_covers_system(found[1], system), similarity, found))
if not scored:
return None
scored.sort(key=lambda s: (s[0], s[1]), reverse=True)
covers, similarity, article = scored[0]
# A search hit that neither covers the platform nor closely matches the name
# is more likely a sibling in the series than this game — our "Dragon Ball Z
# Budokai" surfaces "Dragon Ball Z: Shin Budokai", a different game on a
# different console. A blank field beats a confidently wrong one.
if not covers and similarity < 0.90:
return None
return article
# --------------------------------------------------------------------------
# Wikitext cleaning
# --------------------------------------------------------------------------
def clean_value(value: str) -> str:
"""Reduce an infobox field to plain text.
Infobox values are wikitext, so a developer might arrive as
``[[Rare (company)|Rare]]`` or as a ``{{ubl|...}}`` list with references
attached. This unwraps the common forms and keeps the first entry.
"""
text = value
# Footnotes and comments carry nothing we want.
text = re.sub(r"<ref[^>]*/>", "", text)
text = re.sub(r"<ref.*?</ref>", "", text, flags=re.S | re.I)
text = re.sub(r"<!--.*?-->", "", text, flags=re.S)
# List and release templates carry the value we want as their arguments —
# {{vgrelease|JP|Square|NA|Square Electronic Arts}} holds the publisher —
# so keep the payload and drop only the wrapper.
text = re.sub(
r"\{\{\s*(?:ubl|unbulleted list|plainlist|flatlist|hlist|"
r"vgrelease|vgr|video game release|collapsible list|nobold|nowrap)\s*\|",
"|", text, flags=re.I)
# Anything still wrapped in braces is scaffolding for these fields.
text = re.sub(r"\{\{[^{}]*\}\}", "|", text)
text = text.replace("{{", "|").replace("}}", "|")
# [[Target|Label]] -> Label, [[Target]] -> Target
text = re.sub(r"\[\[([^\]|]+)\|([^\]]+)\]\]", r"\2", text)
text = re.sub(r"\[\[([^\]]+)\]\]", r"\1", text)
text = re.sub(r"<br\s*/?>", "|", text, flags=re.I)
text = re.sub(r"<[^>]+>", "", text)
text = text.replace("'''", "").replace("''", "")
# Take the first real entry of whatever list survived, skipping any leading
# platform heading that was grouping the list.
parts = [p.strip(" *\n\t") for p in re.split(r"[|\n]", text)]
parts = [re.sub(r"\s+", " ", p).strip(" ,;") for p in parts if p.strip(" *\n\t")]
# A named template parameter leaks through unwrapping as "title=Foo";
# keep the value, drop the key.
parts = [re.sub(r"^[a-z_][a-z0-9_]*\s*=\s*", "", p, flags=re.I) for p in parts]
parts = [p for p in parts if p]
first = next((p for p in parts if not is_noise_entry(p)), "")
# Drop a trailing platform annotation: "Rare (N64)" -> "Rare".
match = re.match(r"^(.*?)\s*\(([^()]*)\)$", first)
if match and is_platform_label(match.group(2)):
first = match.group(1).strip()
return first.strip(" ,;")
def infobox_field(wikitext: str, field: str) -> str | None:
"""Raw value of one infobox field, brace- and bracket-aware."""
match = re.search(rf"^\s*\|\s*{field}\s*=", wikitext, flags=re.M | re.I)
if not match:
return None
index = match.end()
depth_brace = depth_bracket = 0
out: list[str] = []
while index < len(wikitext):
two = wikitext[index:index + 2]
if two == "{{":
depth_brace += 1
elif two == "}}":
depth_brace -= 1
elif two == "[[":
depth_bracket += 1
elif two == "]]":
depth_bracket -= 1
char = wikitext[index]
# A pipe or newline-pipe at depth zero ends this field.
if depth_brace <= 0 and depth_bracket <= 0 and char == "|":
break
if depth_brace <= 0 and depth_bracket <= 0 and char == "\n" and \
re.match(r"\s*[|}]", wikitext[index:index + 3]):
break
out.append(char)
index += 1
return "".join(out).strip() or None
def extract_year(value: str) -> str | None:
"""First plausible release year in a {{Video game release}} block."""
years = re.findall(r"\b(19[5-9]\d|20[0-4]\d)\b", value)
return years[0] if years else None
# --------------------------------------------------------------------------
# Summary
# --------------------------------------------------------------------------
def wiki_summary(page_title: str) -> str | None:
"""Short article summary via the REST endpoint, which is built for reuse."""
quoted = urllib.parse.quote(page_title.replace(" ", "_"), safe="")
raw = http(f"https://en.wikipedia.org/api/rest_v1/page/summary/{quoted}",
headers={"User-Agent": WIKI_UA})
extract = (json.loads(raw) or {}).get("extract")
if not extract:
return None
text = re.sub(r"\s+", " ", extract).strip()
# CC BY-SA requires attribution, so it travels with the text.
return f"{text}\n\nSource: Wikipedia — {page_title}"
# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------
def main() -> int:
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--api", default="http://localhost:8080")
parser.add_argument("--user", default="ckoch")
parser.add_argument("--password", required=True)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--overwrite", action="store_true",
help="replace fields that already have a value")
parser.add_argument("--fields", default=",".join(FIELDS),
help=f"comma-separated subset of: {','.join(FIELDS)}")
parser.add_argument("--cache", default=str(Path(__file__).parent / ".cache"))
args = parser.parse_args()
wanted = [f.strip() for f in args.fields.split(",") if f.strip() in FIELDS]
if not wanted:
print("No valid fields requested.", file=sys.stderr)
return 2
base = args.api.rstrip("/")
cache_dir = Path(args.cache)
print("Signing in…")
auth = api_json(base, "/api/auth/login",
data={"userName": args.user, "password": args.password})
token = auth["token"]
print("Fetching library…")
games: list[dict] = []
page = 1
while True:
result = api_json(base, f"/api/games?page={page}&pageSize=100", token)
games.extend(result["items"])
if page >= result["totalPages"] or not result["items"]:
break
page += 1
print(f" {len(games)} games\n")
filled = {field: 0 for field in wanted}
touched = failed = 0
no_article: list[dict] = []
skipped_year: list[dict] = []
for game in sorted(games, key=lambda g: g["title"].lower()):
missing = [f for f in wanted if args.overwrite or not game.get(f)]
if not missing:
continue
try:
article = resolve_article(game, cache_dir)
except Exception as exc: # noqa: BLE001
print(f" ! {game['title'][:44]:46} lookup failed: {exc}")
article = None
if not article:
no_article.append(game)
continue
page_title, wikitext = article
updates: dict[str, str] = {}
for field in missing:
if field == "description":
try:
summary = wiki_summary(page_title)
except Exception: # noqa: BLE001
summary = None
if summary:
updates["description"] = summary
continue
if field == "year":
# An article covers every platform a game shipped on, and its
# release block leads with the original. Taking that year for a
# later port would date a DS re-release to the SNES original, so
# only fill it when the article actually covers this platform.
if not article_covers_system(wikitext, game.get("system")):
skipped_year.append(game)
continue
raw = infobox_field(wikitext, "released")
value = extract_year(raw) if raw else None
else:
raw = infobox_field(wikitext, field)
value = clean_value(raw) if raw else None
if value:
updates[field] = value
if not updates:
continue
summary_line = ", ".join(
f"{k}={v[:26]}…" if len(v) > 26 else f"{k}={v}"
for k, v in updates.items() if k != "description")
if "description" in updates:
summary_line = (summary_line + ", " if summary_line else "") + "description"
print(f" + {game['system'] or '-':4} {game['title'][:40]:42} {summary_line[:60]}")
for field in updates:
filled[field] += 1
touched += 1
if args.dry_run:
continue
payload = {k: game.get(k) for k in (
"title", "system", "genre", "year", "developer", "publisher",
"art", "description", "own", "dumped", "played", "finished")}
payload.update(updates)
try:
api_json(base, f"/api/games/{game['id']}", token, data=payload, method="PUT")
except Exception as exc: # noqa: BLE001
print(f" -> FAILED: {exc}")
failed += 1
touched -= 1
print("\n" + "=" * 72)
verb = "would update" if args.dry_run else "updated"
print(f"{verb}: {touched} games failed: {failed}")
for field in wanted:
print(f" {field:12} {filled[field]:3}")
if skipped_year:
print(f"\nYear left alone — the article does not cover this platform, so its "
f"release date belongs to a different version ({len(skipped_year)}):")
for game in skipped_year[:15]:
print(f" {game['system'] or '-':4} {game['title']}")
if no_article:
print(f"\nNo Wikipedia article found ({len(no_article)}):")
for game in no_article[:20]:
print(f" {game['system'] or '-':4} {game['title']}")
return 0
if __name__ == "__main__":
sys.exit(main())
tools/library/fetch_art.py
+559
View File
@@ -0,0 +1,559 @@
#!/usr/bin/env python3
"""Fetch box art for the library and attach it to each game.
Two sources, tried in order, neither needing an account:
1. libretro-thumbnails (https://thumbnails.libretro.com) — scanned retail
boxes named to the No-Intro / Redump conventions. Best art where it has
any, but its coverage is the retro consoles.
2. English Wikipedia — a cover on essentially every notable game article,
which is what fills the Xbox 360 shelf.
Images are pushed through the app's own POST /api/images endpoint, so they get
the same validation, WebP re-encoding and per-user filing as a manual upload.
Standard library only, so it runs without a virtualenv.
python3 fetch_art.py --password '...' --dry-run # report, change nothing
python3 fetch_art.py --password '...' # download and attach
"""
from __future__ import annotations
import argparse
import difflib
import json
import re
import sys
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
THUMBNAIL_HOST = "https://thumbnails.libretro.com"
# Our system codes to libretro's DAT-derived directory names.
# Several directories may back one system code. Our "GB" covers both Game Boy
# and Game Boy Color titles — Pokemon Trading Card Game, for instance, is only in
# the Color set — so both are searched and the better match wins.
SYSTEM_DIRS = {
"NES": ["Nintendo - Nintendo Entertainment System"],
"SNES": ["Nintendo - Super Nintendo Entertainment System"],
"N64": ["Nintendo - Nintendo 64"],
"GB": ["Nintendo - Game Boy", "Nintendo - Game Boy Color"],
"GBA": ["Nintendo - Game Boy Advance"],
"DS": ["Nintendo - Nintendo DS"],
"GC": ["Nintendo - GameCube"],
"WII": ["Nintendo - Wii"],
"PS1": ["Sony - PlayStation"],
"PS2": ["Sony - PlayStation 2"],
"PSP": ["Sony - PlayStation Portable"],
# libretro does publish a "Microsoft - Xbox 360" set, but it is a stub of
# about a dozen entries. It is searched anyway in case it fills out later;
# in practice these titles fall through to the Wikipedia pass.
"360": ["Microsoft - Xbox 360"],
}
# Preferred release region, best first. The library is a US collection, so a USA
# release wins; a European box is better than nothing.
REGION_RANK = ["usa", "world", "usa, europe", "europe", "japan, usa", "japan"]
# Tags that mark a variant we would rather not pick when a plain release exists.
UNDESIRABLE_TAGS = (
"beta", "proto", "demo", "sample", "virtual console", "switch online",
"classic mini", "rev ", "alt", "unl", "aftermarket", "competition",
)
ROMAN = {
"i": 1, "ii": 2, "iii": 3, "iv": 4, "v": 5, "vi": 6, "vii": 7, "viii": 8,
"ix": 9, "x": 10, "xi": 11, "xii": 12, "xiii": 13, "xiv": 14, "xv": 15,
}
# --------------------------------------------------------------------------
# Title normalisation
# --------------------------------------------------------------------------
def strip_accents(text: str) -> str:
"""'Pokémon' -> 'Pokemon', so our un-accented rows still match."""
return "".join(
c for c in unicodedata.normalize("NFKD", text) if not unicodedata.combining(c)
)
def normalise(title: str) -> str:
"""Reduce a title to a comparable form.
Roman numerals become digits, which is what makes our SNES 'Final Fantasy 2'
line up with the catalogued 'Final Fantasy II', and the PS1 'Final Fantasy V'
with 'Final Fantasy V' rather than drifting to a different entry.
"""
text = strip_accents(title).lower()
text = text.replace("&", " and ")
# The catalogue moves a leading article to the end: "Sims 2, The",
# "Legend of Zelda, The - Ocarina of Time". Drop it before anything else,
# while the comma that marks it is still there to find.
text = re.sub(r",\s*(the|a|an)\b", " ", text)
# libretro renders a subtitle colon as " - "; flatten both to a space.
text = re.sub(r"\s+-\s+", " ", text)
text = re.sub(r"[^a-z0-9]+", " ", text)
words = [str(ROMAN.get(w, w)) for w in text.split()]
# Leading articles carry no signal and differ between catalogues.
while words and words[0] in ("the", "a", "an"):
words.pop(0)
return " ".join(words).strip()
@dataclass
class Candidate:
filename: str
base: str # title with all parenthetical tags removed
tags: str # the tags, lowercased, for region and variant ranking
directory: str # libretro set it came from, needed to build the download URL
@property
def region_rank(self) -> int:
for i, region in enumerate(REGION_RANK):
if region in self.tags:
return i
return len(REGION_RANK)
@property
def variant_penalty(self) -> int:
return sum(1 for tag in UNDESIRABLE_TAGS if tag in self.tags)
def parse_candidate(filename: str, directory: str) -> Candidate:
stem = filename[:-4] if filename.lower().endswith(".png") else filename
tags = " ".join(re.findall(r"\(([^)]*)\)", stem)).lower()
base = re.sub(r"\s*\([^)]*\)", "", stem).strip()
return Candidate(filename=filename, base=base, tags=tags, directory=directory)
# --------------------------------------------------------------------------
# HTTP
# --------------------------------------------------------------------------
def http(url: str, *, data=None, headers=None, method=None, timeout=90) -> bytes:
request = urllib.request.Request(url, data=data, method=method)
for key, value in (headers or {}).items():
request.add_header(key, value)
last_error: Exception | None = None
for attempt in range(4):
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.read()
except urllib.error.HTTPError as exc:
# 429 means we asked too fast, so back off hard and try again.
if exc.code == 429:
last_error = exc
time.sleep(5 * (attempt + 1))
continue
# Other 4xx will not improve on retry; surface immediately.
if exc.code < 500:
raise
last_error = exc
except (urllib.error.URLError, TimeoutError) as exc:
last_error = exc
time.sleep(1.5 * (attempt + 1))
raise RuntimeError(f"GET {url} failed after 4 attempts: {last_error}")
def api_json(base: str, path: str, token: str | None = None, *, data=None, method=None):
headers = {"Accept": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
body = None
if data is not None:
body = json.dumps(data).encode()
headers["Content-Type"] = "application/json"
raw = http(base + path, data=body, headers=headers, method=method)
return json.loads(raw) if raw else None
def upload_image(base: str, token: str, filename: str, blob: bytes) -> dict:
"""multipart/form-data POST, hand-rolled to avoid a requests dependency."""
boundary = "----LudosArt" + str(int(time.time() * 1000))
body = b"".join([
f'--{boundary}\r\n'.encode(),
f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'.encode(),
b"Content-Type: image/png\r\n\r\n",
blob,
f"\r\n--{boundary}--\r\n".encode(),
])
raw = http(
base + "/api/images",
data=body,
headers={
"Authorization": f"Bearer {token}",
"Content-Type": f"multipart/form-data; boundary={boundary}",
},
)
return json.loads(raw)
# --------------------------------------------------------------------------
# Wikipedia fallback
# --------------------------------------------------------------------------
#
# libretro's coverage stops at the retro consoles — its "Microsoft - Xbox 360"
# set exists but holds only a dozen entries, none of them ours. English
# Wikipedia carries a cover image on essentially every notable game article and
# needs no account.
#
# The image is read from the article's infobox rather than guessed from file
# names: filtering names for "box" also matches "Xbox-360-Pro-wController.png".
WIKI_API = "https://en.wikipedia.org/w/api.php"
WIKI_UA = "LudosData/1.0 (personal game library; https://github.com/)"
WIKI_MIN_INTERVAL = 1.0 # seconds between calls; the API 429s if pushed
_wiki_last_call = 0.0
def wiki_api(**params):
"""Throttled Wikipedia API call."""
global _wiki_last_call
wait = WIKI_MIN_INTERVAL - (time.monotonic() - _wiki_last_call)
if wait > 0:
time.sleep(wait)
params.setdefault("action", "query")
params.setdefault("format", "json")
raw = http(f"{WIKI_API}?{urllib.parse.urlencode(params)}", headers={"User-Agent": WIKI_UA})
_wiki_last_call = time.monotonic()
return json.loads(raw)
def is_game_article(wikitext: str) -> bool:
"""True for an article about a single game.
A substring test for "Infobox video game" also matches "Infobox video game
series", which would resolve Banjo-Kazooie to the series overview rather
than the 1998 game — and give the wrong cover and the wrong summary.
"""
return re.search(r"\{\{\s*Infobox video game\b(?!\s*series)", wikitext, re.I) is not None
def wiki_article(title: str, cache_dir: Path) -> tuple[str, str] | None:
"""(resolved title, wikitext) for a page, or None when it does not exist."""
safe = re.sub(r"[^A-Za-z0-9._ -]", "_", title)[:120]
cache = cache_dir / "wiki" / f"{safe}.json"
if cache.exists():
blob = json.loads(cache.read_text())
return (blob["title"], blob["text"]) if blob else None
data = wiki_api(prop="revisions", rvprop="content", rvslots="main",
titles=title, redirects=1)
page = next(iter(data["query"]["pages"].values()))
result = None
if "missing" not in page and page.get("revisions"):
result = (page["title"], page["revisions"][0]["slots"]["main"]["*"])
cache.parent.mkdir(parents=True, exist_ok=True)
cache.write_text(json.dumps({"title": result[0], "text": result[1]} if result else None))
return result
def wiki_cover_url(game: dict, cache_dir: str | Path) -> tuple[str, str] | None:
"""(image URL, source description) for a game's cover, or None."""
cache_dir = Path(cache_dir)
title, year, system = game["title"], game.get("year"), game.get("system")
# Bare title first, then the disambiguated forms Wikipedia actually uses.
attempts = [title, f"{title} (video game)"]
if year:
attempts.append(f"{title} ({year} video game)")
article = None
for candidate in attempts:
found = wiki_article(candidate, cache_dir)
if found and is_game_article(found[1]):
article = found
break
if not article:
return None
page_title, text = article
# Guard against landing on the film/book of the same name, or a different
# entry in the series: the article should mention the platform we hold.
if system == "360" and "Xbox 360" not in text:
return None
match = re.search(r"\|\s*image\s*=\s*([^\n|]+)", text)
if not match:
return None
filename = match.group(1).strip()
filename = re.sub(r"<!--.*?-->", "", filename).strip()
filename = filename.strip("[]| ")
# The field is written both ways: bare ("Halo 3 final boxshot.JPG") and
# prefixed ("File:Lost-Planet-New.jpg"), with or without [[ ]] around it.
# Strip any prefix so exactly one is added back below.
filename = re.sub(r"^\s*(?:File|Image)\s*:\s*", "", filename, flags=re.I).strip()
if not filename:
return None
info = wiki_api(prop="imageinfo", titles=f"File:{filename}",
iiprop="url|mime", iiurlwidth=600)
page = next(iter(info["query"]["pages"].values()))
image_info = (page.get("imageinfo") or [{}])[0]
url = image_info.get("thumburl") or image_info.get("url")
return (url, f"wikipedia:{page_title}") if url else None
# --------------------------------------------------------------------------
# Listings
# --------------------------------------------------------------------------
def load_listing(system: str, cache_dir: Path) -> list[Candidate]:
"""Every candidate for a system, across all of its libretro sets.
Listings are cached on disk, so reruns and dry-runs cost nothing.
"""
candidates: list[Candidate] = []
for directory in SYSTEM_DIRS[system]:
cache = cache_dir / f"{directory}.json"
if cache.exists():
names = json.loads(cache.read_text())
else:
quoted = urllib.parse.quote(directory)
html = http(f"{THUMBNAIL_HOST}/{quoted}/Named_Boxarts/").decode(
"utf-8", errors="replace"
)
names = sorted(
{urllib.parse.unquote(m) for m in re.findall(r'href="([^"]+\.png)"', html)}
)
cache.parent.mkdir(parents=True, exist_ok=True)
cache.write_text(json.dumps(names, indent=0))
candidates.extend(parse_candidate(n, directory) for n in names)
return candidates
def contains_tokens(haystack: list[str], needle: list[str]) -> bool:
"""True when `needle` appears as a contiguous run inside `haystack`."""
if not needle or len(needle) > len(haystack):
return False
return any(
haystack[i:i + len(needle)] == needle
for i in range(len(haystack) - len(needle) + 1)
)
def best_match(title: str, candidates: list[Candidate]) -> tuple[Candidate | None, float]:
"""Highest-scoring candidate, with region and variant used as tie-breakers."""
target = normalise(title)
if not target:
return None, 0.0
target_tokens = target.split()
scored: list[tuple[float, int, int, int, Candidate]] = []
for candidate in candidates:
base = normalise(candidate.base)
score = difflib.SequenceMatcher(None, target, base).ratio()
base_tokens = base.split()
extra = len(base_tokens) - len(target_tokens)
# Subtitles go missing in both directions. Our rows sometimes omit what
# the catalogue carries ("Wave Race 64" vs "Wave Race 64 - Kawasaki Jet
# Ski") and sometimes carry what it omits (our "Donkey Kong Country 2:
# Diddy's Kong Quest" vs the GBA set's "Donkey Kong Country 2"). Either
# way a clean token-run containment is a strong signal, so compare
# whichever is shorter against whichever is longer.
#
# Capped below 0.95 so a genuine exact title always outranks it, and
# scaled by coverage so the closest-length candidate wins among several
# ("Donkey Kong Country 3" beats a bare "Donkey Kong Country").
#
# Guard against collapsing a sequel onto its base game. If we are asking
# for a number the candidate does not have — "Donkey Kong Country 2"
# against a plain "Donkey Kong Country" — containment would happily match
# the wrong box. Extra numbers on the candidate side are fine, since that
# is just a series prefix ("Super Mario World 2 - Yoshi's Island").
target_numbers = {t for t in target_tokens if t.isdigit()}
base_numbers = {t for t in base_tokens if t.isdigit()}
sequel_mismatch = bool(target_numbers - base_numbers)
shorter, longer = sorted((target_tokens, base_tokens), key=len)
if extra != 0 and not sequel_mismatch and contains_tokens(longer, shorter):
coverage = len(shorter) / len(longer)
score = max(score, 0.88 + 0.06 * coverage)
if score >= 0.80:
scored.append(
(score, -candidate.region_rank, -abs(extra), -candidate.variant_penalty, candidate)
)
if not scored:
return None, 0.0
# Similarity first, then the US release, the closest-length title, and the
# plain variant over a revision or re-release.
scored.sort(key=lambda t: (round(t[0], 3), t[1], t[2], t[3]), reverse=True)
score, _, _, _, candidate = scored[0]
return candidate, score
# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--api", default="http://localhost:8080", help="API base URL")
parser.add_argument("--user", default="ckoch")
parser.add_argument("--password", required=True)
parser.add_argument("--dry-run", action="store_true", help="report matches, change nothing")
parser.add_argument("--overwrite", action="store_true", help="replace art that is already set")
parser.add_argument("--min-score", type=float, default=0.80,
help="similarity below which a match is not applied (0-1)")
parser.add_argument("--no-wikipedia", action="store_true",
help="skip the Wikipedia fallback; use libretro only")
parser.add_argument("--cache", default=str(Path(__file__).parent / ".cache"))
args = parser.parse_args()
base = args.api.rstrip("/")
cache_dir = Path(args.cache)
print("Signing in…")
auth = api_json(base, "/api/auth/login", data={"userName": args.user, "password": args.password})
token = auth["token"]
print("Fetching library…")
games: list[dict] = []
page = 1
while True:
result = api_json(base, f"/api/games?page={page}&pageSize=100", token)
games.extend(result["items"])
if page >= result["totalPages"] or not result["items"]:
break
page += 1
print(f" {len(games)} games\n")
systems = sorted({g["system"] for g in games if g["system"]})
listings: dict[str, list[Candidate]] = {}
for system in systems:
if system not in SYSTEM_DIRS:
continue
print(f"Loading {system} catalogue…", end=" ", flush=True)
listings[system] = load_listing(system, cache_dir)
print(f"{len(listings[system])} covers")
print()
applied = skipped = failed = 0
from_wiki = 0
weak: list[tuple[dict, str, float]] = []
unmatched: list[dict] = []
def attach(game: dict, blob: bytes, filename: str) -> None:
"""Upload the image and point the game at it."""
uploaded = upload_image(base, token, filename, blob)
payload = {k: game.get(k) for k in (
"title", "system", "genre", "year", "developer", "publisher",
"description", "own", "dumped", "played", "finished")}
payload["art"] = uploaded["fileName"]
api_json(base, f"/api/games/{game['id']}", token, data=payload, method="PUT")
for game in sorted(games, key=lambda g: g["title"].lower()):
title, system = game["title"], game["system"]
if game.get("art") and not args.overwrite:
skipped += 1
continue
# --- pass 1: libretro, which has the better art where it has any ----
candidate, score = (None, 0.0)
if system in SYSTEM_DIRS:
candidate, score = best_match(title, listings.get(system, []))
if candidate and score >= args.min_score:
flag = " " if score >= 0.95 else "~"
print(f" {flag} {system:4} {title[:46]:48} {score:.2f} {candidate.filename[:50]}")
if score < 0.95:
weak.append((game, candidate.filename, score))
if args.dry_run:
applied += 1
continue
try:
directory = urllib.parse.quote(candidate.directory)
name = urllib.parse.quote(candidate.filename)
attach(game, http(f"{THUMBNAIL_HOST}/{directory}/Named_Boxarts/{name}"),
candidate.filename)
applied += 1
except Exception as exc: # noqa: BLE001 - report and continue
print(f" -> FAILED: {exc}")
failed += 1
continue
# --- pass 2: Wikipedia, for anything libretro does not carry --------
if args.no_wikipedia:
print(f" ? {system:4} {title[:46]:48} no match")
unmatched.append(game)
failed += 1
continue
try:
found = wiki_cover_url(game, cache_dir)
except Exception as exc: # noqa: BLE001
print(f" ! {system:4} {title[:46]:48} wikipedia error: {exc}")
found = None
if not found:
print(f" ? {system:4} {title[:46]:48} no match")
unmatched.append(game)
failed += 1
continue
url, source = found
print(f" W {system:4} {title[:46]:48} {source[:50]}")
if args.dry_run:
applied += 1
from_wiki += 1
continue
try:
attach(game, http(url, headers={"User-Agent": WIKI_UA}), url.split("/")[-1])
applied += 1
from_wiki += 1
except Exception as exc: # noqa: BLE001
print(f" -> FAILED: {exc}")
failed += 1
# ---- report ----------------------------------------------------------
print("\n" + "=" * 72)
verb = "would attach" if args.dry_run else "attached"
print(f"{verb}: {applied} (libretro {applied - from_wiki}, wikipedia {from_wiki}) "
f"already had art: {skipped} still missing: {failed}")
if weak:
print(f"\nWorth eyeballing in the UI — matched below 0.95 similarity ({len(weak)}):")
for game, filename, score in sorted(weak, key=lambda w: w[2]):
print(f" {score:.2f} {game['system']:4} {game['title'][:40]:42} -> {filename[:50]}")
if unmatched:
print(f"\nNo cover found from either source ({len(unmatched)}):")
for game in unmatched:
print(f" {game['system'] or '-':4} {game['title']}")
return 0
if __name__ == "__main__":
sys.exit(main())