Coverage for src/ph/compatibility.py: 99.0%
249 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Optional compatibility ratings and cached title matching."""
3import json
4import logging
5import os
6import re
7import ssl
8import unicodedata
9from collections.abc import Callable, Mapping, Sequence
10from concurrent.futures import ThreadPoolExecutor
11from dataclasses import dataclass
12from difflib import SequenceMatcher
13from html import unescape
14from pathlib import Path
15from time import time
16from urllib.parse import urljoin, urlparse
17from urllib.request import Request, urlopen
19from ph.cache_policy import DEFAULT_CATALOGUE_TTL_DAYS, catalogue_ttl_seconds
20from ph.models import Platform, SearchResult
21from ph.store import USER_AGENT
23COMPATIBILITY_CATALOGUE_URL = "https://r36sgamelist.com"
24LOGGER = logging.getLogger(__name__)
26_SCRIPT_PATTERN = re.compile(r'(?:src|href)="([^"?#]+\.js(?:\?[^"#]*)?)"')
27_GAME_PATTERN = re.compile(
28 r'\{"name":"((?:\\.|[^"\\])*)","console":"((?:\\.|[^"\\])*)",'
29 r'"slug":"((?:\\.|[^"\\])*)"'
30)
31_BRACKETED_TEXT = re.compile(r"\s*[\[(][^\])]*[\])]\s*")
32_NON_ALPHANUMERIC = re.compile(r"[^a-z0-9]+")
33_MATCH_THRESHOLD = 0.82
34_AMBIGUITY_MARGIN = 0.04
35_TITLE_NOISE = frozenset(
36 {
37 "beta",
38 "demo",
39 "en",
40 "eng",
41 "europe",
42 "eur",
43 "fr",
44 "fre",
45 "germany",
46 "hack",
47 "ita",
48 "italy",
49 "japan",
50 "jpn",
51 "korea",
52 "proto",
53 "prototype",
54 "rev",
55 "revision",
56 "spa",
57 "spain",
58 "uk",
59 "usa",
60 "version",
61 }
62)
64# The upstream source assigns compatibility at console level. A title match
65# confirms that the game is listed; actual performance can vary by device.
66_CONSOLE_LEVELS: dict[str, str] = {
67 "arcade": "Perfect",
68 "capcom play system i": "Perfect",
69 "capcom play system ii": "Perfect",
70 "dreamcast": "Limited",
71 "famicom": "Perfect",
72 "gameboy advance": "Perfect",
73 "gameboy color": "Perfect",
74 "mega drive": "Perfect",
75 "nes": "Perfect",
76 "neogeo": "Perfect",
77 "neogeo pocket": "Perfect",
78 "neogeo pocket color": "Perfect",
79 "nintendo 64": "Playable",
80 "nintendo ds": "Limited",
81 "pc engine": "Perfect",
82 "playstation 1": "Perfect",
83 "psp": "Limited",
84 "sega genesis": "Perfect",
85 "sega naomi": "Limited",
86 "sfc": "Perfect",
87 "snes": "Perfect",
88}
90_CONSOLE_ALIASES: dict[str, str] = {
91 "arcade": "arcade",
92 "mame": "arcade",
93 "mame 2003": "arcade",
94 "mame 2010": "arcade",
95 "cps1": "capcom play system i",
96 "cps 1": "capcom play system i",
97 "capcom play system 1": "capcom play system i",
98 "cps2": "capcom play system ii",
99 "cps 2": "capcom play system ii",
100 "capcom play system 2": "capcom play system ii",
101 "dc": "dreamcast",
102 "dreamcast": "dreamcast",
103 "famicom": "famicom",
104 "gba": "gameboy advance",
105 "game boy advance": "gameboy advance",
106 "gameboy advance": "gameboy advance",
107 "gbc": "gameboy color",
108 "game boy color": "gameboy color",
109 "gameboy color": "gameboy color",
110 "gen": "mega drive",
111 "genesis": "sega genesis",
112 "mega drive": "mega drive",
113 "sega genesis": "sega genesis",
114 "nes": "nes",
115 "nintendo": "nes",
116 "nintendo entertainment system": "nes",
117 "neogeo": "neogeo",
118 "neo geo": "neogeo",
119 "ngp": "neogeo pocket",
120 "neo geo pocket": "neogeo pocket",
121 "neogeo pocket": "neogeo pocket",
122 "ngpc": "neogeo pocket color",
123 "neo geo pocket color": "neogeo pocket color",
124 "neogeo pocket color": "neogeo pocket color",
125 "n64": "nintendo 64",
126 "nintendo 64": "nintendo 64",
127 "nds": "nintendo ds",
128 "ds": "nintendo ds",
129 "nintendo ds": "nintendo ds",
130 "pce": "pc engine",
131 "pc engine": "pc engine",
132 "turbografx 16": "pc engine",
133 "ps1": "playstation 1",
134 "psx": "playstation 1",
135 "playstation": "playstation 1",
136 "playstation 1": "playstation 1",
137 "psp": "psp",
138 "playstation portable": "psp",
139 "naomi": "sega naomi",
140 "sega naomi": "sega naomi",
141 "sfc": "sfc",
142 "snes": "snes",
143 "super famicom": "sfc",
144 "super nintendo": "snes",
145}
147_UNSUPPORTED_NAMES = frozenset(
148 {
149 "3ds",
150 "gamecube",
151 "nintendo 3ds",
152 "nintendo switch",
153 "nintendo wii",
154 "nintendo wii u",
155 "playstation 2",
156 "playstation 3",
157 "playstation 4",
158 "playstation 5",
159 "playstation vita",
160 "ps2",
161 "ps3",
162 "ps4",
163 "ps5",
164 "ps vita",
165 "psvita",
166 "switch",
167 "vita",
168 "wii",
169 "wii u",
170 "wiiu",
171 "xbox",
172 "xbox 360",
173 "xbox one",
174 "xbox360",
175 }
176)
178type GameKey = tuple[str, str]
179type FetchText = Callable[[str], str]
180type TitleIndex = Mapping[str, Sequence[str]]
183class CompatibilityError(RuntimeError):
184 """The game compatibility catalogue could not be downloaded or stored."""
187@dataclass(frozen=True, slots=True)
188class CompatibilityInfo:
189 """Compatibility displayed next to one remote search result."""
191 level: str
192 title_listed: bool
193 match_score: float | None = None
195 @property
196 def short_label(self) -> str:
197 """Return the compact results-list label."""
199 if self.level == "Not listed":
200 return self.level
201 if self.title_listed and self.match_score is not None:
202 return f"{self.level} - {round(self.match_score * 100)}% match"
203 return f"{self.level}{' - listed' if self.title_listed else ''}"
205 @property
206 def detail_label(self) -> str:
207 """Explain whether the live catalogue reliably matched this title."""
209 if self.level == "Not listed":
210 return "Not listed by r36sgamelist.com"
211 qualifier = (
212 f"title match {round(self.match_score * 100)}%"
213 if self.title_listed and self.match_score is not None
214 else "title listed"
215 if self.title_listed
216 else "platform rating"
217 )
218 return f"{self.level} ({qualifier})"
221def normalize_title(value: str) -> str:
222 """Normalize regional/version title variants for conservative exact matching."""
224 without_brackets = _BRACKETED_TEXT.sub(" ", value)
225 decomposed = unicodedata.normalize("NFKD", without_brackets.casefold())
226 ascii_text = "".join(
227 character for character in decomposed if not unicodedata.combining(character)
228 )
229 return " ".join(_NON_ALPHANUMERIC.sub(" ", ascii_text).split())
232def normalize_console(value: str) -> str | None:
233 """Map remote and ROM-folder console spellings to the compatibility catalogue."""
235 normalized = " ".join(_NON_ALPHANUMERIC.sub(" ", value.casefold()).split())
236 return _CONSOLE_ALIASES.get(normalized)
239def is_unsupported_system(value: str) -> bool:
240 """Return whether a console family is outside the supported handheld profile."""
242 normalized = " ".join(_NON_ALPHANUMERIC.sub(" ", value.casefold()).split())
243 return normalized in _UNSUPPORTED_NAMES or any(
244 normalized.startswith(f"{name} ") or normalized.endswith(f" {name}")
245 for name in _UNSUPPORTED_NAMES
246 )
249def filter_supported_results(results: Sequence[SearchResult]) -> list[SearchResult]:
250 """Remove explicitly unsupported consoles from all-platform remote results."""
252 return [result for result in results if not is_unsupported_system(result.system)]
255class GameCompatibilityClient:
256 """Fetch and cache the frontend-only game compatibility index."""
258 def __init__(
259 self,
260 cache_path: Path,
261 base_url: str = COMPATIBILITY_CATALOGUE_URL,
262 timeout_seconds: float = 30.0,
263 fetch_text: FetchText | None = None,
264 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS),
265 ) -> None:
266 self.cache_path = cache_path
267 self.base_url = base_url.rstrip("/")
268 self.timeout_seconds = timeout_seconds
269 self._fetch_text_override = fetch_text
270 self.ttl_seconds = ttl_seconds
271 self._game_index: frozenset[GameKey] | None = None
273 def lookup_many(
274 self,
275 results: Sequence[SearchResult],
276 platform: Platform,
277 ) -> list[CompatibilityInfo]:
278 """Return one compatibility record for each result, preserving order."""
280 consoles = [self._result_console(result, platform) for result in results]
281 should_load_index = any(console is not None for console in consoles)
282 index = self._load_game_index() if should_load_index else frozenset()
283 match_indexes = _build_match_indexes(index)
284 return [
285 self._lookup(result.title, console, index, match_indexes.get(console, {}))
286 for result, console in zip(results, consoles, strict=True)
287 ]
289 def load(self) -> frozenset[GameKey] | None:
290 """Load the local catalogue, including a stale but still usable copy."""
292 cached = self._read_cache()
293 if cached is not None: 293 ↛ 295line 293 didn't jump to line 295 because the condition on line 293 was always true
294 self._game_index = cached
295 return cached
297 def refresh(self) -> int:
298 """Explicitly replace the cache with the current frontend catalogue."""
300 try:
301 index = self._download_game_index()
302 self._write_cache(index, strict=True)
303 except CompatibilityError:
304 raise
305 except (OSError, TimeoutError, ValueError) as error:
306 raise CompatibilityError(
307 f"Could not update the compatibility catalogue: {error}"
308 ) from error
309 self._game_index = index
310 LOGGER.info("Compatibility catalogue cached games=%d path=%s", len(index), self.cache_path)
311 return len(index)
313 def cache_age_seconds(self) -> float | None:
314 """Return the age of a valid local cache, or ``None`` when unavailable."""
316 payload = self._read_cache_payload()
317 return None if payload is None else max(0.0, time() - payload[0])
319 def cache_is_stale(self) -> bool:
320 """Return whether the local catalogue exceeds its configured lifetime."""
322 age = self.cache_age_seconds()
323 return age is not None and age > self.ttl_seconds
325 @staticmethod
326 def _result_console(result: SearchResult, platform: Platform) -> str | None:
327 candidates = (
328 result.system,
329 platform.alias,
330 platform.name,
331 platform.slug,
332 *(platform.rom_folders),
333 )
334 return next(
335 (console for value in candidates if value and (console := normalize_console(value))),
336 None,
337 )
339 @staticmethod
340 def _lookup(
341 title: str,
342 console: str | None,
343 index: frozenset[GameKey],
344 candidate_index: TitleIndex,
345 ) -> CompatibilityInfo:
346 if console is None:
347 return CompatibilityInfo("Not listed", False)
348 level = _CONSOLE_LEVELS[console]
349 normalized = normalize_title(title)
350 if (console, normalized) in index:
351 return CompatibilityInfo(level, True, 1.0)
352 candidates = {
353 candidate
354 for key in _title_match_keys(normalized)
355 for candidate in candidate_index.get(key, ())
356 }
357 ranked = sorted(
358 (title_match_score(normalized, candidate), candidate) for candidate in candidates
359 )
360 if not ranked:
361 return CompatibilityInfo(level, False)
362 best_score, _best_title = ranked[-1]
363 second_score = ranked[-2][0] if len(ranked) > 1 else 0.0
364 accepted = best_score >= _MATCH_THRESHOLD and (
365 best_score >= 0.97 or best_score - second_score >= _AMBIGUITY_MARGIN
366 )
367 return CompatibilityInfo(level, accepted, best_score if accepted else None)
369 def _load_game_index(self) -> frozenset[GameKey]:
370 if self._game_index is not None:
371 return self._game_index
372 cached = self._read_cache()
373 if cached is not None:
374 self._game_index = cached
375 return cached
376 try:
377 index = self._download_game_index()
378 except OSError, TimeoutError, ValueError:
379 index = frozenset()
380 if index:
381 self._write_cache(index)
382 self._game_index = index
383 return index
385 def _download_game_index(self) -> frozenset[GameKey]:
386 home = self._fetch_text(self.base_url)
387 script_urls = tuple(
388 dict.fromkeys(
389 urljoin(f"{self.base_url}/", unescape(match))
390 for match in _SCRIPT_PATTERN.findall(home)
391 if self._same_origin(urljoin(f"{self.base_url}/", unescape(match)))
392 )
393 )
394 with ThreadPoolExecutor(max_workers=4) as executor:
395 documents = (home, *executor.map(self._fetch_optional, script_urls))
396 index = frozenset(game for document in documents for game in parse_game_index(document))
397 if len(index) < 100:
398 raise ValueError("r36sgamelist.com did not expose a usable frontend game index")
399 return index
401 def _same_origin(self, url: str) -> bool:
402 expected = urlparse(self.base_url)
403 parsed = urlparse(url)
404 return parsed.scheme == expected.scheme and parsed.netloc == expected.netloc
406 def _fetch_text(self, url: str) -> str:
407 if self._fetch_text_override is not None:
408 return self._fetch_text_override(url)
409 request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html,*/*"})
410 context = ssl.create_default_context()
411 with urlopen(request, timeout=self.timeout_seconds, context=context) as response:
412 return response.read().decode("utf-8", errors="replace")
414 def _fetch_optional(self, url: str) -> str:
415 try:
416 return self._fetch_text(url)
417 except OSError, TimeoutError, ValueError:
418 return ""
420 def _read_cache(self) -> frozenset[GameKey] | None:
421 payload = self._read_cache_payload()
422 return None if payload is None else payload[1]
424 def _read_cache_payload(self) -> tuple[float, frozenset[GameKey]] | None:
425 try:
426 payload = json.loads(self.cache_path.read_text(encoding="utf-8"))
427 fetched_at = float(payload["fetched_at"])
428 games = payload["games"]
429 if not isinstance(games, list):
430 return None
431 index = frozenset((str(item[0]), str(item[1])) for item in games)
432 except IndexError, KeyError, OSError, TypeError, ValueError, json.JSONDecodeError:
433 return None
434 return (fetched_at, index) if index else None
436 def _write_cache(self, index: frozenset[GameKey], *, strict: bool = False) -> None:
437 payload = {"fetched_at": time(), "games": sorted(index)}
438 temporary = self.cache_path.with_name(self.cache_path.name + ".tmp")
439 try:
440 self.cache_path.parent.mkdir(parents=True, exist_ok=True)
441 temporary.write_text(
442 json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
443 encoding="utf-8",
444 )
445 os.replace(temporary, self.cache_path)
446 except OSError as error:
447 temporary.unlink(missing_ok=True)
448 if strict:
449 raise CompatibilityError(
450 f"Could not save the compatibility catalogue: {error}"
451 ) from error
452 LOGGER.warning("Could not cache the compatibility catalogue: %s", error)
455def parse_game_index(document: str) -> frozenset[GameKey]:
456 """Extract the title/console objects embedded in Next.js frontend chunks."""
458 games: set[GameKey] = set()
459 for raw_title, raw_console, _raw_slug in _GAME_PATTERN.findall(document):
460 try:
461 title = json.loads(f'"{raw_title}"')
462 console_value = json.loads(f'"{raw_console}"')
463 except json.JSONDecodeError:
464 continue
465 console = normalize_console(console_value)
466 normalized_title = normalize_title(title)
467 if console is not None and normalized_title:
468 games.add((console, normalized_title))
469 return frozenset(games)
472def title_match_score(left: str, right: str) -> float:
473 """Score normalized titles while discounting common release metadata."""
475 left = normalize_title(left)
476 right = normalize_title(right)
477 if not left or not right:
478 return 0.0
479 if left == right:
480 return 1.0
481 left_tokens = left.split()
482 right_tokens = right.split()
483 left_core = _core_title_tokens(left_tokens, right_tokens)
484 right_core = _core_title_tokens(right_tokens, left_tokens)
485 if left_core and left_core == right_core:
486 return 0.98
488 left_value = " ".join(left_core or left_tokens)
489 right_value = " ".join(right_core or right_tokens)
490 sequence = SequenceMatcher(None, left_value, right_value, autojunk=False).ratio()
491 compact = SequenceMatcher(
492 None,
493 left_value.replace(" ", ""),
494 right_value.replace(" ", ""),
495 autojunk=False,
496 ).ratio()
497 left_set = set(left_core or left_tokens)
498 right_set = set(right_core or right_tokens)
499 token_score = 2 * len(left_set & right_set) / (len(left_set) + len(right_set))
500 return max(sequence * 0.7 + token_score * 0.3, compact * 0.85 + token_score * 0.15)
503def _core_title_tokens(tokens: Sequence[str], other_tokens: Sequence[str]) -> tuple[str, ...]:
504 core: list[str] = []
505 shared = set(other_tokens)
506 skip_revision_number = False
507 for token in tokens:
508 if token in ("rev", "revision"):
509 skip_revision_number = True
510 continue
511 if token in _TITLE_NOISE and token not in shared:
512 continue
513 if skip_revision_number and re.fullmatch(r"v?\d+(?:\.\d+)*", token):
514 skip_revision_number = False
515 continue
516 skip_revision_number = False
517 core.append(token)
518 return tuple(core)
521def _build_match_indexes(index: frozenset[GameKey]) -> dict[str, dict[str, tuple[str, ...]]]:
522 mutable: dict[str, dict[str, list[str]]] = {}
523 for console, title in index:
524 console_index = mutable.setdefault(console, {})
525 for key in _title_match_keys(title):
526 console_index.setdefault(key, []).append(title)
527 return {
528 console: {key: tuple(titles) for key, titles in title_index.items()}
529 for console, title_index in mutable.items()
530 }
533def _title_match_keys(title: str) -> frozenset[str]:
534 tokens = normalize_title(title).split()
535 meaningful = {token for token in tokens if len(token) >= 3 and token not in {"and", "the"}}
536 compact = "".join(tokens)
537 if len(compact) >= 4: 537 ↛ 539line 537 didn't jump to line 539 because the condition on line 537 was always true
538 meaningful.add(f"#{compact[:4]}")
539 return frozenset(meaningful)