rf4-spotter/rf4_research/records.py

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())