Complete stage 0 source research

rf4_research/records.py

@@ -0,0 +1,173 @@
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from dataclasses import asdict, dataclass
+from datetime import date, datetime
+from typing import Iterable
+from urllib.request import Request, urlopen
+
+from bs4 import BeautifulSoup, Tag
+
+
+DEFAULT_URL = "https://rf4game.de/records/region/RU/"
+USER_AGENT = "RF4-Spotter-Research/0.1 (+https://github.com/)"
+
+
+class RecordsParseError(ValueError):
+    """Raised when the page no longer matches the verified records contract."""
+
+
+@dataclass(frozen=True, slots=True)
+class OfficialRecord:
+    region: str
+    category: str
+    fish: str
+    weight_g: int
+    waterbody: str
+    bait: str | None
+    player: str | None
+    record_date: date
+    source_url: str
+
+
+def _text(node: Tag | None) -> str:
+    return node.get_text(" ", strip=True) if node else ""
+
+
+def _direct_child(parent: Tag, classes: Iterable[str]) -> Tag | None:
+    wanted = set(classes)
+    for child in parent.find_all("div", recursive=False):
+        if wanted.issubset(set(child.get("class", []))):
+            return child
+    return None
+
+
+def parse_weight_g(raw: str) -> int:
+    normalized = " ".join(raw.replace("\xa0", " ").split()).lower()
+    match = re.fullmatch(r"([\d .,'’]+)\s*(kg|g)", normalized)
+    if not match:
+        raise RecordsParseError(f"unsupported weight: {raw!r}")
+
+    number, unit = match.groups()
+    number = number.replace(" ", "").replace("'", "").replace("’", "")
+    if unit == "g":
+        return int(number.replace(".", "").replace(",", ""))
+
+    # The verified pages use a dot as the kg decimal separator and spaces as
+    # thousands separators (for example, "2 519.264 kg").
+    if "," in number and "." not in number:
+        number = number.replace(",", ".")
+    return round(float(number) * 1000)
+
+
+def parse_record_date(raw: str, *, today: date | None = None) -> date:
+    today = today or date.today()
+    try:
+        parsed = datetime.strptime(raw.strip(), "%d.%m.%y").date()
+    except ValueError as exc:
+        raise RecordsParseError(f"unsupported record date: {raw!r}") from exc
+    if parsed > today.replace(year=today.year + 1):
+        raise RecordsParseError(f"record date is implausibly far in the future: {raw!r}")
+    return parsed
+
+
+def parse_records_html(
+    html: str,
+    *,
+    region: str,
+    category: str,
+    source_url: str,
+    today: date | None = None,
+) -> list[OfficialRecord]:
+    soup = BeautifulSoup(html, "html.parser")
+    table = soup.select_one("div.records.flex_table")
+    if table is None:
+        raise RecordsParseError("records table not found")
+
+    top_rows = _direct_child(table, ["rows"])
+    header = _direct_child(table, ["row", "header"])
+    expected_classes = ["fish", "weight", "location", "bait", "gamername", "data"]
+    actual_classes = [
+        next((name for name in expected_classes if name in cell.get("class", [])), "")
+        for cell in (header.find_all("div", recursive=False) if header else [])
+    ]
+    if actual_classes != expected_classes:
+        raise RecordsParseError(
+            f"records columns changed: expected {expected_classes}, got {actual_classes}"
+        )
+    if top_rows is None:
+        raise RecordsParseError("records rows container not found")
+
+    result: list[OfficialRecord] = []
+    for group_wrapper in top_rows.find_all("div", class_="row", recursive=False):
+        group = _direct_child(group_wrapper, ["records_subtable", "flex_table"])
+        if group is None:
+            continue
+        group_header = _direct_child(group, ["row", "header"])
+        more_rows = _direct_child(group, ["rows"])
+        if group_header is None:
+            continue
+
+        fish = _text(group_header.select_one(".fish .text"))
+        rows = [group_header]
+        if more_rows is not None:
+            rows.extend(more_rows.find_all("div", class_="row", recursive=False))
+
+        for row in rows:
+            bait_node = row.select_one(".bait .bait_icon")
+            bait = bait_node.get("title", "").strip() if bait_node else ""
+            player = _text(row.select_one(".gamername"))
+            result.append(
+                OfficialRecord(
+                    region=region.upper(),
+                    category=category,
+                    fish=fish,
+                    weight_g=parse_weight_g(_text(row.select_one(".weight"))),
+                    waterbody=_text(row.select_one(".location")),
+                    bait=bait or None,
+                    player=player or None,
+                    record_date=parse_record_date(_text(row.select_one(".data")), today=today),
+                    source_url=source_url,
+                )
+            )
+
+    if not result:
+        raise RecordsParseError("records table is present but contains no records")
+    return result
+
+
+def fetch_html(url: str, *, timeout: float = 20.0) -> str:
+    request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"})
+    with urlopen(request, timeout=timeout) as response:
+        content_type = response.headers.get_content_type()
+        if content_type != "text/html":
+            raise RecordsParseError(f"expected text/html, got {content_type}")
+        charset = response.headers.get_content_charset() or "utf-8"
+        return response.read().decode(charset)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Fetch and parse one RF4 records page")
+    parser.add_argument("--url", default=DEFAULT_URL)
+    parser.add_argument("--region", default="RU")
+    parser.add_argument("--category", default="records")
+    args = parser.parse_args(argv)
+    try:
+        records = parse_records_html(
+            fetch_html(args.url),
+            region=args.region,
+            category=args.category,
+            source_url=args.url,
+        )
+    except Exception as exc:
+        print(f"research import failed: {exc}", file=sys.stderr)
+        return 1
+    print(json.dumps([asdict(item) for item in records], ensure_ascii=False, default=str))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())