245 lines
8.6 KiB
Python
245 lines
8.6 KiB
Python
"""Scrape PSP front covers from OpenVGDB (GameFAQs) into a local bundle.
|
|
|
|
OpenVGDB (https://github.com/OpenVGDB/OpenVGDB) is a downloadable SQLite dump
|
|
whose PSP entries point at GameFAQs box art. GameFAQs sits behind a Cloudflare
|
|
TLS-fingerprint challenge, so this scraper uses ``curl_cffi`` (which impersonates
|
|
Chrome's TLS fingerprint) instead of plain urllib.
|
|
|
|
Usage::
|
|
|
|
python3 -m app.openvgdb # scrape everything
|
|
python3 -m app.openvgdb --limit 50 # test with the first 50 covers
|
|
python3 -m app.openvgdb --workers 8
|
|
|
|
Output: ``data/covers/openvgdb-covers.zip`` — entries named ``<norm(title)>.jpg``
|
|
(``norm`` = lowercased, ``&``->``and``, non-alphanumerics stripped, parenthetical
|
|
tags removed). The GUI reads this bundle with no network access.
|
|
"""
|
|
import argparse
|
|
import concurrent.futures
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from . import config
|
|
from .covers import _norm, _plain
|
|
|
|
log = logging.getLogger("openvgdb")
|
|
|
|
OVGDB_ZIP_URL = "https://github.com/OpenVGDB/OpenVGDB/releases/download/v29.0/openvgdb.zip"
|
|
BUNDLE = config.DATA_DIR / "covers" / "openvgdb-covers.zip"
|
|
|
|
# Region preference for deduping the same game across regions: USA first, then
|
|
# USA combos, then Europe, then Japan last (JP covers differ).
|
|
_REGION_RANK = {21: 0, 27: 1, 28: 1, 29: 1, 30: 1, 38: 1, 7: 2, 31: 2, 25: 3, 13: 4}
|
|
|
|
|
|
class _RateLimiter:
|
|
"""Cap total requests per second across all worker threads."""
|
|
|
|
def __init__(self, per_second):
|
|
self.interval = 1.0 / max(per_second, 0.1)
|
|
self._lock = threading.Lock()
|
|
self._next = 0.0
|
|
|
|
def wait(self):
|
|
with self._lock:
|
|
now = time.monotonic()
|
|
if now < self._next:
|
|
time.sleep(self._next - now)
|
|
self._next = time.monotonic() + self.interval
|
|
|
|
|
|
def _fetch(url, timeout=30):
|
|
"""GET `url` impersonating Chrome (passes GameFAQs' Cloudflare challenge)."""
|
|
from curl_cffi import requests
|
|
resp = requests.get(url, impersonate="chrome", timeout=timeout)
|
|
if resp.status_code != 200:
|
|
raise RuntimeError(f"HTTP {resp.status_code} for {url}")
|
|
return resp.content
|
|
|
|
|
|
def build_index(sqlite_path):
|
|
"""Return {norm_title: (display_title, cover_url)} preferring US regions."""
|
|
conn = sqlite3.connect(str(sqlite_path))
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute(
|
|
"SELECT releaseTitleName, regionLocalizedID, releaseCoverFront FROM RELEASES "
|
|
"WHERE TEMPsystemShortName='PSP' AND releaseCoverFront IS NOT NULL"
|
|
).fetchall()
|
|
conn.close()
|
|
|
|
best = {}
|
|
for r in rows:
|
|
title = (r["releaseTitleName"] or "").strip()
|
|
url = (r["releaseCoverFront"] or "").strip()
|
|
if not title or not url:
|
|
continue
|
|
key = _norm(_plain(title))
|
|
if not key:
|
|
continue
|
|
rank = _REGION_RANK.get(r["regionLocalizedID"], 3)
|
|
cur = best.get(key)
|
|
if cur is None or rank < cur[2]:
|
|
best[key] = (title, url, rank)
|
|
return best
|
|
|
|
|
|
def write_index(index, path=None):
|
|
"""Persist {key: {"title", "url"}} so the GUI can show readable titles when
|
|
searching the offline bundle (see covers._search_openvgdb)."""
|
|
path = Path(path) if path else (config.COVERS_DIR / "openvgdb-index.json")
|
|
data = {k: {"title": t, "url": u} for k, (t, u, _r) in index.items()}
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
except OSError as exc: # noqa: BLE001
|
|
log.warning("could not write index %s: %s", path, exc)
|
|
return path
|
|
|
|
|
|
def _existing_keys():
|
|
"""Titles already present in the bundle (for resume)."""
|
|
if not BUNDLE.exists():
|
|
return set()
|
|
try:
|
|
with zipfile.ZipFile(BUNDLE) as z:
|
|
return {n[:-4] for n in z.namelist() if n.endswith(".jpg")}
|
|
except Exception: # noqa: BLE001
|
|
return set()
|
|
|
|
|
|
def _download_one(key, url, dest_dir, limiter, retries=4):
|
|
from curl_cffi import requests
|
|
dest = dest_dir / f"{key}.jpg"
|
|
last = ""
|
|
for attempt in range(1, retries + 1):
|
|
limiter.wait()
|
|
try:
|
|
resp = requests.get(url, impersonate="chrome", timeout=30)
|
|
except Exception as exc: # noqa: BLE001
|
|
last = f"{type(exc).__name__}: {exc}"
|
|
time.sleep(2 * attempt)
|
|
continue
|
|
if resp.status_code == 429:
|
|
last = "HTTP 429"
|
|
time.sleep(4 * attempt) # back off and let the rate limit clear
|
|
continue
|
|
if resp.status_code != 200:
|
|
last = f"HTTP {resp.status_code}"
|
|
continue
|
|
data = resp.content
|
|
if not data or data[:3] != b"\xff\xd8\xff": # JPEG magic
|
|
last = "not a JPEG"
|
|
continue
|
|
tmp = dest.with_suffix(".tmp")
|
|
tmp.write_bytes(data)
|
|
os.replace(tmp, dest)
|
|
return True, ""
|
|
return False, last
|
|
|
|
|
|
def scrape(limit=None, workers=8, rate=3.0):
|
|
BUNDLE.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
log.info("downloading OpenVGDB dump (%s)…", OVGDB_ZIP_URL)
|
|
tmpdir = Path(tempfile.mkdtemp(prefix="openvgdb-"))
|
|
dump_zip = tmpdir / "openvgdb.zip"
|
|
req = urllib.request.Request(OVGDB_ZIP_URL, headers={"User-Agent": "psppt/0.1"})
|
|
with urllib.request.urlopen(req, timeout=60) as r, open(dump_zip, "wb") as f:
|
|
f.write(r.read())
|
|
|
|
log.info("extracting SQLite…")
|
|
sqlite_path = tmpdir / "openvgdb.sqlite"
|
|
with zipfile.ZipFile(dump_zip) as z:
|
|
z.extract("openvgdb.sqlite", tmpdir)
|
|
|
|
index = build_index(sqlite_path)
|
|
log.info("indexed %d unique PSP covers", len(index))
|
|
write_index(index)
|
|
|
|
existing = _existing_keys()
|
|
todo = [(k, v) for k, v in sorted(index.items()) if k not in existing]
|
|
if limit:
|
|
todo = todo[:limit]
|
|
log.info("%d to download (%d already in bundle)", len(todo), len(existing))
|
|
|
|
work_dir = Path(tempfile.mkdtemp(prefix="openvgdb-covers-"))
|
|
limiter = _RateLimiter(rate)
|
|
ok = fail = 0
|
|
done = 0
|
|
lock = threading.Lock()
|
|
|
|
def run(item):
|
|
key, (title, url, _rank) = item
|
|
success, err = _download_one(key, url, work_dir, limiter)
|
|
nonlocal ok, fail, done
|
|
with lock:
|
|
done += 1
|
|
if success:
|
|
ok += 1
|
|
else:
|
|
fail += 1
|
|
if done % 50 == 0 or not success:
|
|
log.info(" %d/%d (%d ok, %d failed)%s", done, len(todo), ok, fail,
|
|
f" {title!r}: {err}" if not success else "")
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
|
|
list(pool.map(run, todo))
|
|
|
|
if ok:
|
|
log.info("writing bundle…")
|
|
out_tmp = BUNDLE.with_name(BUNDLE.name + ".tmp")
|
|
existing_bytes = {}
|
|
if BUNDLE.exists():
|
|
try:
|
|
with zipfile.ZipFile(BUNDLE) as zin:
|
|
for n in zin.namelist():
|
|
if n.endswith(".jpg") and n not in existing_bytes:
|
|
existing_bytes[n] = zin.read(n)
|
|
except Exception: # noqa: BLE001
|
|
existing_bytes = {}
|
|
with zipfile.ZipFile(out_tmp, "w", zipfile.ZIP_STORED) as z:
|
|
for name, data in existing_bytes.items():
|
|
z.writestr(name, data)
|
|
for f in sorted(work_dir.glob("*.jpg")):
|
|
z.write(f, arcname=f.name)
|
|
os.replace(out_tmp, BUNDLE)
|
|
log.info("done: bundle now has %d covers (%d new, %d failed this run)",
|
|
len(existing_bytes) + ok, ok, fail)
|
|
else:
|
|
log.info("no new covers downloaded; bundle unchanged (%d failed)", fail)
|
|
return ok, fail
|
|
|
|
|
|
def main(argv=None):
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", stream=sys.stderr)
|
|
ap = argparse.ArgumentParser(description="Scrape PSP covers from OpenVGDB into a local bundle")
|
|
ap.add_argument("--limit", type=int, default=None, help="only scrape the first N covers (testing)")
|
|
ap.add_argument("--workers", type=int, default=8, help="parallel downloads")
|
|
ap.add_argument("--rate", type=float, default=3.0, help="max requests/second (avoid 429)")
|
|
args = ap.parse_args(argv)
|
|
|
|
try:
|
|
import curl_cffi # noqa: F401
|
|
except ImportError:
|
|
print("This scraper needs curl_cffi: pip install curl_cffi", file=sys.stderr)
|
|
return 1
|
|
|
|
ok, fail = scrape(limit=args.limit, workers=args.workers, rate=args.rate)
|
|
return 0 if fail == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|