Coverage for src/ph/minerva_store.py: 98.8%
192 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Minerva Archive RetroAchievements store and selected-torrent downloads."""
3import re
4import ssl
5import time
6from collections.abc import Mapping
7from concurrent.futures import ThreadPoolExecutor
8from dataclasses import dataclass
9from html.parser import HTMLParser
10from pathlib import Path, PurePosixPath
11from typing import cast
12from urllib.error import HTTPError, URLError
13from urllib.parse import parse_qs, quote, urljoin, urlparse
14from urllib.request import Request, urlopen
16from ph.cache_policy import DEFAULT_CATALOGUE_TTL_DAYS, catalogue_ttl_seconds
17from ph.models import MediaDownload, Platform, SearchResult
18from ph.platforms import DARKOS_PLATFORMS
19from ph.store import USER_AGENT, CatalogProgress, GameStore, StoreError
21PLATFORM_DIRECTORIES: dict[str, str] = {
22 "3do": "RA - 3DO Interactive Multiplayer",
23 "amstrad-cpc": "RA - Amstrad CPC",
24 "apple-ii": "RA - Apple II",
25 "arcade": "RA - Arcade",
26 "arduboy": "RA - Arduboy",
27 "atari-2600": "RA - Atari 2600",
28 "atari-7800": "RA - Atari 7800",
29 "atari-jaguar": "RA - Atari Jaguar",
30 "atari-lynx": "RA - Atari Lynx",
31 "colecovision": "RA - Colecovision",
32 "dreamcast": "RA - Sega Dreamcast",
33 "fairchild-channel-f": "RA - Fairchild Channel F",
34 "famicom-disk-system": "RA - Nintendo Famicom Disk System",
35 "game-boy": "RA - Nintendo Game Boy",
36 "game-boy-advance": "RA - Nintendo Game Boy Advance",
37 "game-boy-color": "RA - Nintendo Game Boy Color",
38 "game-gear": "RA - Sega Game Gear",
39 "genesis": "RA - Sega Genesis",
40 "intellivision": "RA - Mattel Intellivision",
41 "master-system": "RA - Sega Master System",
42 "mega-duck": "RA - Mega Duck",
43 "msx": "RA - Microsoft MSX",
44 "neo-geo-cd": "RA - SNK Neo Geo CD",
45 "neo-geo-pocket": "RA - SNK Neo Geo Pocket",
46 "neo-geo-pocket-color": "RA - SNK Neo Geo Pocket",
47 "nintendo-64": "RA - Nintendo 64",
48 "nintendo-ds": "RA - Nintendo DS",
49 "nintendo": "RA - Nintendo Entertainment System",
50 "odyssey-2": "RA - Magnavox Odyssey 2",
51 "pc-engine": "RA - NEC TurboGrafx-16",
52 "pc-engine-cd": "RA - NEC TurboGrafx-CD",
53 "pc-fx": "RA - NEC PC-FX",
54 "playstation": "RA - Sony Playstation",
55 "ps-portable": "RA - Sony PSP",
56 "pokemon-mini": "RA - Nintendo Pokemon Mini",
57 "sega-32x": "RA - Sega 32X",
58 "sega-cd": "RA - Sega CD",
59 "saturn": "RA - Sega Saturn",
60 "sg-1000": "RA - Sega SG-1000",
61 "super-nintendo": "RA - Super Nintendo Entertainment System",
62 "uzebox": "RA - Uzebox",
63 "vectrex": "RA - GCE Vectrex",
64 "virtual-boy": "RA - Nintendo Virtual Boy",
65 "wasm-4": "RA - WASM-4",
66 "watara-supervision": "RA - Watara Supervision",
67 "wonderswan": "RA - WonderSwan",
68 "wonderswan-color": "RA - WonderSwan",
69}
70RA_DIRECTORIES = tuple(dict.fromkeys(PLATFORM_DIRECTORIES.values()))
71_PLATFORMS_BY_SLUG = {platform.slug: platform for platform in DARKOS_PLATFORMS}
72_DISPLAY_SYSTEM = {
73 directory: _PLATFORMS_BY_SLUG[slug].alias for slug, directory in PLATFORM_DIRECTORIES.items()
74}
75_REGION_WORDS = (
76 "Australia",
77 "Brazil",
78 "Canada",
79 "China",
80 "Europe",
81 "France",
82 "Germany",
83 "Italy",
84 "Japan",
85 "Korea",
86 "Spain",
87 "Taiwan",
88 "USA",
89 "World",
90)
93@dataclass(frozen=True, slots=True)
94class MinervaEntry:
95 """One file and its one-based position inside a platform torrent."""
97 filename: str
98 link: str
99 file_index: int
102class _DirectoryParser(HTMLParser):
103 def __init__(self, base_url: str) -> None:
104 super().__init__(convert_charrefs=True)
105 self.base_url = base_url
106 self.entries: list[MinervaEntry] = []
107 self._entry_depth = 0
108 self._link = ""
109 self._text: list[str] = []
111 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
112 attributes = dict(attrs)
113 if tag == "div" and "entry" in (attributes.get("class") or "").split():
114 self._entry_depth = 1
115 self._link = ""
116 self._text = []
117 elif self._entry_depth: 117 ↛ exitline 117 didn't return from function 'handle_starttag' because the condition on line 117 was always true
118 if tag == "div":
119 self._entry_depth += 1
120 elif tag == "a":
121 self._link = attributes.get("href") or ""
123 def handle_endtag(self, tag: str) -> None:
124 if not self._entry_depth or tag != "div":
125 return
126 self._entry_depth -= 1
127 if self._entry_depth:
128 return
129 if urlparse(self._link).path.rstrip("/") != "/rom":
130 return
131 query = parse_qs(urlparse(self._link).query)
132 paths = query.get("name", [])
133 if len(paths) != 1:
134 return
135 filename = PurePosixPath(paths[0]).name
136 if not filename:
137 filename = " ".join(" ".join(self._text).split())
138 if filename: 138 ↛ exitline 138 didn't return from function 'handle_endtag' because the condition on line 138 was always true
139 self.entries.append(
140 MinervaEntry(filename, urljoin(self.base_url, self._link), len(self.entries) + 1)
141 )
143 def handle_data(self, data: str) -> None:
144 if self._entry_depth and data.strip(): 144 ↛ exitline 144 didn't return from function 'handle_data' because the condition on line 144 was always true
145 self._text.append(data.strip())
148def parse_directory(html: str, base_url: str) -> list[MinervaEntry]:
149 """Parse one Minerva browse directory in torrent file order."""
151 parser = _DirectoryParser(base_url)
152 parser.feed(html)
153 return parser.entries
156def _title_metadata(filename: str) -> tuple[str, str, str, str]:
157 title = Path(filename).stem
158 groups = re.findall(r"\(([^()]*)\)", title)
159 region = next(
160 (group for group in groups if any(word in group for word in _REGION_WORDS)),
161 "",
162 )
163 version = next(
164 (group for group in groups if re.match(r"(?i)(?:rev(?:ision)?\s*|v)\d", group)),
165 "",
166 )
167 languages = next(
168 (group for group in groups if re.fullmatch(r"(?:[A-Z][a-z](?:,[A-Z][a-z])*)", group)),
169 "-",
170 )
171 return title, region, version, languages
174class MinervaStore(GameStore):
175 """Minerva's RetroAchievements collection, downloaded one torrent file at a time."""
177 store_id = "minerva"
178 display_name = "Minerva Archive"
179 description = "RetroAchievements torrents (native Python)"
181 def __init__(
182 self,
183 base_url: str,
184 torrent_base_url: str,
185 timeout_seconds: float = 30.0,
186 cache_directory: Path | None = None,
187 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS),
188 ) -> None:
189 self._base_url = base_url.rstrip("/")
190 self._torrent_base_url = torrent_base_url.rstrip("/")
191 self.timeout_seconds = timeout_seconds
192 self._base = urlparse(self._base_url)
193 torrent_base = urlparse(self._torrent_base_url)
194 if self._base.scheme not in ("http", "https") or not self._base.netloc:
195 raise ValueError("base_url must be an absolute HTTP(S) URL")
196 if torrent_base.scheme not in ("http", "https") or not torrent_base.netloc:
197 raise ValueError("torrent_base_url must be an absolute HTTP(S) URL")
198 self._ssl_context = ssl.create_default_context()
199 self._entry_cache: dict[str, tuple[MinervaEntry, ...]] = {}
200 self._configure_catalogue_cache(cache_directory, ttl_seconds)
202 @property
203 def base_url(self) -> str:
204 return self._base_url
206 @property
207 def download_referrer(self) -> str:
208 return f"{self.base_url}/browse/RetroAchievements/"
210 def platform_code(self, platform: Platform) -> str:
211 if platform.slug == "all":
212 return ""
213 return PLATFORM_DIRECTORIES.get(platform.slug, "")
215 def supports_platform(self, platform: Platform) -> bool:
216 return platform.slug == "all" or platform.slug in PLATFORM_DIRECTORIES
218 @property
219 def headers(self) -> Mapping[str, str]:
220 return {
221 "User-Agent": USER_AGENT,
222 "Accept": "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8",
223 "Accept-Language": "en-US,en;q=0.5",
224 }
226 def search(
227 self,
228 system_code: str,
229 query: str,
230 catalog_progress: CatalogProgress | None = None,
231 ) -> list[SearchResult]:
232 catalogue = self._load_catalogue(system_code, catalog_progress)
233 normalized_query = " ".join(query.split()).casefold()
234 return [
235 result for result in catalogue if result.title.casefold().startswith(normalized_query)
236 ]
238 def _fetch_catalogue(
239 self,
240 system_code: str,
241 catalog_progress: CatalogProgress | None,
242 ) -> list[SearchResult]:
243 if system_code:
244 if system_code not in RA_DIRECTORIES:
245 raise StoreError("Minerva does not provide this platform in RetroAchievements.")
246 directories = (system_code,)
247 else:
248 directories = RA_DIRECTORIES
250 def fetch(directory: str) -> tuple[str, tuple[MinervaEntry, ...]]:
251 return directory, self._entries(directory)
253 results: list[SearchResult] = []
254 with ThreadPoolExecutor(max_workers=2, thread_name_prefix="minerva") as executor:
255 for completed, (directory, entries) in enumerate(
256 executor.map(fetch, directories),
257 start=1,
258 ):
259 for entry in entries:
260 title, region, version, languages = _title_metadata(entry.filename)
261 results.append(
262 SearchResult(
263 title=title,
264 link=entry.link,
265 system=_DISPLAY_SYSTEM[directory] if not system_code else "",
266 region=region,
267 version=version,
268 languages=languages,
269 )
270 )
271 if catalog_progress is not None:
272 catalog_progress(completed, len(directories))
273 return sorted(
274 results, key=lambda result: (result.title.casefold(), result.system.casefold())
275 )
277 def validate_detail_url(self, url: str) -> bool:
278 target = self._detail_target(url)
279 return target is not None
281 def retrieve_download_url(self, detail_url: str) -> str:
282 directory, _filename, _file_index = self._download_metadata(detail_url)
283 torrent_name = f"Minerva_Myrient - RetroAchievements - {directory}.torrent"
284 return f"{self._torrent_base_url}/{quote(torrent_name)}"
286 def download_request(self, detail_url: str) -> MediaDownload:
287 directory, filename, file_index = self._download_metadata(detail_url)
288 torrent_name = f"Minerva_Myrient - RetroAchievements - {directory}.torrent"
289 return MediaDownload(
290 url=f"{self._torrent_base_url}/{quote(torrent_name)}",
291 torrent_file_index=file_index,
292 expected_filename=filename,
293 )
295 def _download_metadata(self, detail_url: str) -> tuple[str, str, int]:
296 target = self._detail_target(detail_url)
297 if target is None:
298 raise StoreError(f"Not a valid RetroAchievements detail URL for {self._base.netloc}.")
299 directory, filename = target
300 entry = next(
301 (item for item in self._entries(directory) if item.filename == filename),
302 None,
303 )
304 if entry is None:
305 raise StoreError("The selected game is no longer present in Minerva's catalogue.")
306 return directory, filename, entry.file_index
308 def _detail_target(self, url: str) -> tuple[str, str] | None:
309 parsed = urlparse(url.strip())
310 if (
311 parsed.scheme != self._base.scheme
312 or parsed.netloc.casefold() != self._base.netloc.casefold()
313 or parsed.path.rstrip("/") != "/rom"
314 or parsed.fragment
315 ):
316 return None
317 query = parse_qs(parsed.query, keep_blank_values=True)
318 names = query.get("name", [])
319 if set(query) != {"name"} or len(names) != 1:
320 return None
321 raw_path = names[0].removeprefix("./")
322 if "\\" in raw_path:
323 return None
324 parts = PurePosixPath(raw_path).parts
325 if len(parts) != 3 or parts[0] != "RetroAchievements" or ".." in parts:
326 return None
327 directory, filename = parts[1], parts[2]
328 if directory not in RA_DIRECTORIES or not filename:
329 return None
330 return directory, filename
332 def _entries(self, directory: str) -> tuple[MinervaEntry, ...]:
333 cached = self._entry_cache.get(directory)
334 if cached is not None:
335 return cached
336 url = f"{self.base_url}/browse/RetroAchievements/{quote(directory)}/"
337 entries = tuple(parse_directory(self._get_text(url), self.base_url))
338 if not entries:
339 raise StoreError(f"Minerva returned an empty catalogue for {directory}.")
340 self._entry_cache[directory] = entries
341 return entries
343 def _get_text(self, url: str) -> str:
344 request = Request(url, headers=dict(self.headers))
345 last_error: BaseException | None = None
346 for attempt in range(3):
347 try:
348 with urlopen(
349 request,
350 timeout=self.timeout_seconds,
351 context=self._ssl_context,
352 ) as response:
353 charset = response.headers.get_content_charset() or "utf-8"
354 return cast(str, response.read().decode(charset, errors="replace"))
355 except HTTPError as error:
356 if error.code < 500 and error.code != 429:
357 raise StoreError(
358 "Minerva returned HTTP %d." % error.code,
359 error.code,
360 ) from error
361 last_error = error
362 except (URLError, TimeoutError, OSError) as error:
363 last_error = error
364 if attempt < 2:
365 time.sleep(0.25 * (2**attempt))
366 reason = getattr(last_error, "reason", last_error)
367 raise StoreError(f"Could not reach Minerva after 3 attempts: {reason}") from last_error