80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
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())
|