refactor: share official records parser
This commit is contained in:
@@ -0,0 +1,105 @@
|
||||
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
|
||||
+8
-102
@@ -2,22 +2,19 @@ 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 datetime import date
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
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/)"
|
||||
|
||||
|
||||
class RecordsParseError(ValueError):
|
||||
"""Raised when the page no longer matches the verified records contract."""
|
||||
RecordsParseError = RecordsContractError
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
@@ -33,47 +30,6 @@ class OfficialRecord:
|
||||
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,
|
||||
*,
|
||||
@@ -82,61 +38,11 @@ def parse_records_html(
|
||||
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
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user