106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime
|
||
from typing import Iterable
|
||
|
||
from bs4 import BeautifulSoup, Tag
|
||
|
||
|
||
class RecordsContractError(ValueError):
|
||
"""The official records page no longer matches the verified DOM contract."""
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ParsedOfficialRecord:
|
||
player: str | None
|
||
fish: str
|
||
weight_g: int
|
||
waterbody: str
|
||
bait: str | None
|
||
record_date: date
|
||
|
||
|
||
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 RecordsContractError(f"unsupported weight: {raw!r}")
|
||
number, unit = match.groups()
|
||
number = number.replace(" ", "").replace("'", "").replace("’", "")
|
||
if unit == "g":
|
||
return int(number.replace(".", "").replace(",", ""))
|
||
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 RecordsContractError(f"unsupported record date: {raw!r}") from exc
|
||
if parsed > today.replace(year=today.year + 1):
|
||
raise RecordsContractError(f"record date is implausibly far in the future: {raw!r}")
|
||
return parsed
|
||
|
||
|
||
def parse_official_records(html: str, *, today: date | None = None) -> list[ParsedOfficialRecord]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
table = soup.select_one("div.records.flex_table")
|
||
if table is None:
|
||
raise RecordsContractError("records table not found")
|
||
top_rows = _direct_child(table, ["rows"])
|
||
header = _direct_child(table, ["row", "header"])
|
||
expected = ["fish", "weight", "location", "bait", "gamername", "data"]
|
||
actual = [
|
||
next((name for name in expected if name in cell.get("class", [])), "")
|
||
for cell in (header.find_all("div", recursive=False) if header else [])
|
||
]
|
||
if actual != expected:
|
||
raise RecordsContractError(f"records columns changed: expected {expected}, got {actual}")
|
||
if top_rows is None:
|
||
raise RecordsContractError("records rows container not found")
|
||
|
||
result: list[ParsedOfficialRecord] = []
|
||
for wrapper in top_rows.find_all("div", class_="row", recursive=False):
|
||
group = _direct_child(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 ""
|
||
result.append(ParsedOfficialRecord(
|
||
player=_text(row.select_one(".gamername")) or None,
|
||
fish=fish,
|
||
weight_g=parse_weight_g(_text(row.select_one(".weight"))),
|
||
waterbody=_text(row.select_one(".location")),
|
||
bait=bait or None,
|
||
record_date=parse_record_date(_text(row.select_one(".data")), today=today),
|
||
))
|
||
if not result:
|
||
raise RecordsContractError("records table is present but contains no records")
|
||
return result
|