from __future__ import annotations import argparse import json import sys from dataclasses import asdict, dataclass from datetime import date from urllib.request import Request, urlopen from .official_parser import RecordsContractError, parse_official_records, parse_record_date, parse_weight_g DEFAULT_URL = "https://rf4game.de/records/region/RU/" USER_AGENT = "RF4-Spotter-Research/0.1 (+https://github.com/)" RecordsParseError = RecordsContractError @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 parse_records_html( html: str, *, region: str, category: str, source_url: str, today: date | None = None, ) -> list[OfficialRecord]: return [OfficialRecord( region=region.upper(), category=category, fish=row.fish, weight_g=row.weight_g, waterbody=row.waterbody, bait=row.bait, player=row.player, record_date=row.record_date, source_url=source_url, ) for row in parse_official_records(html, today=today)] 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())