Coverage for src/ph/vimm_store.py: 93.2%
193 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Vimm's Lair store implementation and HTML parsers."""
3import re
4import ssl
5import string
6from collections.abc import Mapping
7from concurrent.futures import ThreadPoolExecutor
8from dataclasses import dataclass, field
9from html.parser import HTMLParser
10from pathlib import Path
11from typing import cast
12from urllib.error import HTTPError, URLError
13from urllib.parse import parse_qs, urlencode, urljoin, urlparse
14from urllib.request import Request, urlopen
16from ph.cache_policy import DEFAULT_CATALOGUE_TTL_DAYS, catalogue_ttl_seconds
17from ph.models import Platform, SearchResult
18from ph.store import USER_AGENT, CatalogProgress, GameStore, StoreError
20CATALOG_SECTIONS = ("number", *string.ascii_uppercase)
23@dataclass(slots=True)
24class _Cell:
25 text_parts: list[str] = field(default_factory=list)
26 link: str = ""
27 image_titles: list[str] = field(default_factory=list)
29 @property
30 def text(self) -> str:
31 return " ".join(" ".join(self.text_parts).split())
34class _SearchTableParser(HTMLParser):
35 def __init__(self) -> None:
36 super().__init__(convert_charrefs=True)
37 self.rows: list[list[_Cell]] = []
38 self._table_depth = 0
39 self._target_depth: int | None = None
40 self._row: list[_Cell] | None = None
41 self._cell: _Cell | None = None
42 self._ignore_anchor = False
44 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
45 attributes = dict(attrs)
46 if tag == "table":
47 self._table_depth += 1
48 classes = (attributes.get("class") or "").split()
49 if self._target_depth is None and "rounded" in classes:
50 self._target_depth = self._table_depth
51 return
52 if self._target_depth is None:
53 return
54 if tag == "tr" and self._table_depth == self._target_depth:
55 self._row = []
56 elif tag in ("td", "th") and self._row is not None:
57 self._cell = _Cell()
58 elif tag == "a" and self._cell is not None:
59 style = (attributes.get("style") or "").replace(" ", "").casefold()
60 self._ignore_anchor = "display:none" in style
61 if not self._ignore_anchor:
62 self._cell.link = attributes.get("href") or ""
63 elif tag == "img" and self._cell is not None:
64 title = attributes.get("title")
65 if title: 65 ↛ exitline 65 didn't return from function 'handle_starttag' because the condition on line 65 was always true
66 self._cell.image_titles.append(title)
68 def handle_endtag(self, tag: str) -> None:
69 if self._target_depth is None:
70 if tag == "table" and self._table_depth: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 self._table_depth -= 1
72 return
73 if tag in ("td", "th") and self._row is not None and self._cell is not None:
74 self._row.append(self._cell)
75 self._cell = None
76 elif tag == "a":
77 self._ignore_anchor = False
78 elif tag == "tr" and self._row is not None:
79 self.rows.append(self._row)
80 self._row = None
81 elif tag == "table":
82 if self._table_depth == self._target_depth:
83 self._target_depth = None
84 self._table_depth -= 1
86 def handle_data(self, data: str) -> None:
87 if self._cell is not None and not self._ignore_anchor and data.strip():
88 self._cell.text_parts.append(data.strip())
91class _DownloadFormParser(HTMLParser):
92 def __init__(self) -> None:
93 super().__init__(convert_charrefs=True)
94 self.action = ""
95 self.media_id = ""
96 self._inside_form = False
97 self._form_depth = 0
99 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
100 attributes = dict(attrs)
101 if tag == "form" and attributes.get("id") == "dl_form":
102 self._inside_form = True
103 self._form_depth = 1
104 self.action = attributes.get("action") or ""
105 elif self._inside_form:
106 if tag == "form": 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 self._form_depth += 1
108 elif tag == "input" and not self.media_id: 108 ↛ exitline 108 didn't return from function 'handle_starttag' because the condition on line 108 was always true
109 self.media_id = attributes.get("value") or ""
111 def handle_endtag(self, tag: str) -> None:
112 if self._inside_form and tag == "form":
113 self._form_depth -= 1
114 if self._form_depth == 0: 114 ↛ exitline 114 didn't return from function 'handle_endtag' because the condition on line 114 was always true
115 self._inside_form = False
118def parse_search_results(html: str, base_url: str, system_code: str) -> list[SearchResult]:
119 """Parse both all-platform and platform-specific result table layouts."""
121 parser = _SearchTableParser()
122 parser.feed(html)
123 results: list[SearchResult] = []
125 for cells in parser.rows:
126 if system_code:
127 if len(cells) < 3 or not cells[0].link:
128 continue
129 results.append(
130 SearchResult(
131 title=cells[0].text,
132 link=urljoin(base_url, cells[0].link),
133 region=" ".join(cells[1].image_titles),
134 version=cells[2].text,
135 languages=cells[3].text if len(cells) > 3 and cells[3].text else "-",
136 rating=cells[4].text if len(cells) > 4 and cells[4].text else "-",
137 )
138 )
139 else:
140 if len(cells) < 4 or not cells[1].link:
141 continue
142 results.append(
143 SearchResult(
144 system=cells[0].text,
145 title=cells[1].text,
146 link=urljoin(base_url, cells[1].link),
147 region=" ".join(cells[2].image_titles),
148 version=cells[3].text,
149 )
150 )
151 return results
154def parse_download_url(html: str, base_url: str) -> str:
155 """Extract a media URL from a detail page download form."""
157 parser = _DownloadFormParser()
158 parser.feed(html)
159 if not parser.action or not parser.media_id: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 raise StoreError("The page does not contain an available download.")
161 action = urljoin(base_url, parser.action)
162 separator = "&" if "?" in action else "?"
163 return "{}{}{}".format(action, separator, urlencode({"mediaId": parser.media_id}))
166class VimmStore(GameStore):
167 """Vimm's Lair store implementation and HTML client."""
169 store_id = "vimm"
170 display_name = "Vimm's Lair"
171 description = "Vimm game vault"
173 def __init__(
174 self,
175 base_url: str,
176 timeout_seconds: float = 30.0,
177 cache_directory: Path | None = None,
178 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS),
179 ) -> None:
180 self._base_url = base_url.rstrip("/")
181 self.timeout_seconds = timeout_seconds
182 parsed = urlparse(self._base_url)
183 if parsed.scheme not in ("http", "https") or not parsed.netloc: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 raise ValueError("base_url must be an absolute HTTP(S) URL")
185 self._base = parsed
186 self._ssl_context = ssl.create_default_context()
187 self._configure_catalogue_cache(cache_directory, ttl_seconds)
189 @property
190 def base_url(self) -> str:
191 """Return the configured Vimm root URL."""
193 return self._base_url
195 @property
196 def download_referrer(self) -> str:
197 """Return the Vimm vault referrer used for media downloads."""
199 return f"{self.base_url}/vault/"
201 def platform_code(self, platform: Platform) -> str:
202 """Return the Vimm system code stored in shared platform metadata."""
204 return platform.code
206 @property
207 def headers(self) -> Mapping[str, str]:
208 return {
209 "User-Agent": USER_AGENT,
210 "Accept": "text/html,application/xhtml+xml,application/octet-stream;q=0.9,*/*;q=0.8",
211 "Accept-Language": "en-US,en;q=0.5",
212 "DNT": "1",
213 }
215 def validate_detail_url(self, url: str) -> bool:
216 parsed = urlparse(url.strip())
217 query = parse_qs(parsed.query, keep_blank_values=True)
218 valid_query = not query or (set(query) == {"v"} and len(query["v"]) == 1 and query["v"][0])
219 return bool(
220 parsed.scheme == self._base.scheme
221 and parsed.netloc.casefold() == self._base.netloc.casefold()
222 and re.fullmatch(r"/vault/[0-9]+/?", parsed.path)
223 and valid_query
224 and not parsed.fragment
225 )
227 def search(
228 self,
229 system_code: str,
230 query: str,
231 catalog_progress: CatalogProgress | None = None,
232 ) -> list[SearchResult]:
233 """Return case-insensitive title-prefix matches, or a platform catalogue when empty."""
235 catalogue = self._load_catalogue(system_code, catalog_progress)
236 needle = " ".join(query.split()).casefold()
237 return [result for result in catalogue if result.title.casefold().startswith(needle)]
239 def _fetch_catalogue(
240 self,
241 system_code: str,
242 catalog_progress: CatalogProgress | None,
243 ) -> list[SearchResult]:
244 def fetch(section: str) -> list[SearchResult]:
245 parameters = {"p": "list", "section": section}
246 if system_code:
247 parameters["system"] = system_code
248 query = urlencode(parameters)
249 url = f"{self.base_url}/vault/?{query}"
250 try:
251 html = self._get_text(url)
252 except StoreError as error:
253 if error.status_code == 404: 253 ↛ 255line 253 didn't jump to line 255 because the condition on line 253 was always true
254 return []
255 raise
256 return parse_search_results(html, self.base_url, system_code)
258 results: list[SearchResult] = []
259 seen_links: set[str] = set()
260 with ThreadPoolExecutor(max_workers=3, thread_name_prefix="catalog") as executor:
261 for completed, section_results in enumerate(
262 executor.map(fetch, CATALOG_SECTIONS),
263 start=1,
264 ):
265 for result in section_results:
266 if result.link not in seen_links:
267 seen_links.add(result.link)
268 results.append(result)
269 if catalog_progress is not None:
270 catalog_progress(completed, len(CATALOG_SECTIONS))
271 return results
273 def retrieve_download_url(self, detail_url: str) -> str:
274 if not self.validate_detail_url(detail_url): 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true
275 raise StoreError(f"Not a valid detail URL for {self._base.netloc}.")
276 return parse_download_url(self._get_text(detail_url), self.base_url)
278 def _get_text(self, url: str) -> str:
279 request = Request(url, headers=dict(self.headers))
280 try:
281 with urlopen(
282 request,
283 timeout=self.timeout_seconds,
284 context=self._ssl_context,
285 ) as response:
286 charset = response.headers.get_content_charset() or "utf-8"
287 return cast(str, response.read().decode(charset, errors="replace"))
288 except HTTPError as error:
289 raise StoreError("The server returned HTTP %d." % error.code, error.code) from error
290 except (URLError, TimeoutError, OSError) as error:
291 reason = getattr(error, "reason", error)
292 raise StoreError(f"Could not reach the server: {reason}") from error