240 lines
9.7 KiB
Python
240 lines
9.7 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
import time as time_module
|
||
from dataclasses import asdict, dataclass
|
||
from datetime import date, datetime, time, timezone
|
||
|
||
import httpx
|
||
from bs4 import BeautifulSoup, Tag
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .models import (
|
||
Bait, BaitKind, CatchReport, Fish, ImportStatus, ModerationStatus,
|
||
OfficialRecordImport, SourceType, Waterbody,
|
||
)
|
||
|
||
|
||
USER_AGENT = "RF4-Spotter/0.1 (public records importer)"
|
||
|
||
|
||
class ImportSourceError(ValueError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class RawRecord:
|
||
region: str
|
||
category: str
|
||
player: str | None
|
||
fish: str
|
||
weight_g: int
|
||
waterbody: str
|
||
bait: str | None
|
||
record_date: date
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class FetchResult:
|
||
records: list[RawRecord] | None
|
||
status_code: int
|
||
etag: str | None
|
||
last_modified: str | None
|
||
content_type: str | None
|
||
response_bytes: int
|
||
|
||
|
||
def normalize(value: str) -> str:
|
||
return " ".join(value.replace("\xa0", " ").replace("–", "-").replace("—", "-").split()).casefold()
|
||
|
||
|
||
def slugify(value: str) -> str:
|
||
compact = re.sub(r"[^a-z0-9а-яё]+", "-", normalize(value), flags=re.IGNORECASE).strip("-")
|
||
return compact or hashlib.sha256(value.encode()).hexdigest()[:16]
|
||
|
||
|
||
def external_id(record: RawRecord) -> str:
|
||
parts = [record.region, record.category, record.player or "", record.fish, str(record.weight_g), record.waterbody, record.bait or "", record.record_date.isoformat()]
|
||
return hashlib.sha256("|".join(normalize(part) for part in parts).encode()).hexdigest()
|
||
|
||
|
||
def parse_weight(raw: str) -> int:
|
||
value = " ".join(raw.replace("\xa0", " ").split()).lower()
|
||
match = re.fullmatch(r"([\d .,'’]+)\s*(kg|g)", value)
|
||
if not match:
|
||
raise ImportSourceError(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 _text(node: Tag | None) -> str:
|
||
return node.get_text(" ", strip=True) if node else ""
|
||
|
||
|
||
def parse_html(html: str, *, region: str, category: str) -> list[RawRecord]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
table = soup.select_one("div.records.flex_table")
|
||
if table is None:
|
||
raise ImportSourceError("records table not found")
|
||
header = table.select_one(":scope > .row.header")
|
||
expected = ["fish", "weight", "location", "bait", "gamername", "data"]
|
||
cells = header.find_all("div", recursive=False) if header else []
|
||
actual = [next((key for key in expected if key in cell.get("class", [])), "") for cell in cells]
|
||
if actual != expected:
|
||
raise ImportSourceError(f"record columns changed: {actual}")
|
||
|
||
records: list[RawRecord] = []
|
||
for group in table.select(":scope > .rows > .row > .records_subtable"):
|
||
group_header = group.select_one(":scope > .row.header")
|
||
if group_header is None:
|
||
continue
|
||
fish = _text(group_header.select_one(".fish .text"))
|
||
rows = [group_header, *group.select(":scope > .rows > .row")]
|
||
for row in rows:
|
||
bait_node = row.select_one(".bait_icon")
|
||
bait = bait_node.get("title", "").strip() if bait_node else ""
|
||
try:
|
||
record_date = datetime.strptime(_text(row.select_one(".data")), "%d.%m.%y").date()
|
||
except ValueError as exc:
|
||
raise ImportSourceError("record date format changed") from exc
|
||
records.append(RawRecord(region.upper(), category, _text(row.select_one(".gamername")) or None, fish, parse_weight(_text(row.select_one(".weight"))), _text(row.select_one(".location")), bait or None, record_date))
|
||
if not records:
|
||
raise ImportSourceError("records table is empty")
|
||
return records
|
||
|
||
|
||
def fetch_records(
|
||
url: str, *, region: str, category: str,
|
||
etag: str | None = None, last_modified: str | None = None,
|
||
) -> FetchResult:
|
||
headers = {"User-Agent": USER_AGENT, "Accept": "text/html"}
|
||
if etag:
|
||
headers["If-None-Match"] = etag
|
||
if last_modified:
|
||
headers["If-Modified-Since"] = last_modified
|
||
with httpx.Client(timeout=20, follow_redirects=True, headers=headers) as client:
|
||
for attempt in range(3):
|
||
try:
|
||
response = client.get(url)
|
||
metadata = {
|
||
"status_code": response.status_code,
|
||
"etag": response.headers.get("etag"),
|
||
"last_modified": response.headers.get("last-modified"),
|
||
"content_type": response.headers.get("content-type"),
|
||
"response_bytes": len(response.content),
|
||
}
|
||
if response.status_code == 304:
|
||
return FetchResult(records=None, **metadata)
|
||
response.raise_for_status()
|
||
if "text/html" not in response.headers.get("content-type", ""):
|
||
raise ImportSourceError("source did not return HTML")
|
||
return FetchResult(records=parse_html(response.text, region=region, category=category), **metadata)
|
||
except (httpx.HTTPError, ImportSourceError):
|
||
if attempt == 2:
|
||
raise
|
||
time_module.sleep(2 ** attempt)
|
||
raise AssertionError("unreachable")
|
||
|
||
|
||
def import_records(session: Session, *, url: str, region: str, category: str, html: str | None = None) -> OfficialRecordImport:
|
||
run = OfficialRecordImport(started_at=datetime.now(timezone.utc), status=ImportStatus.running, source_url=url, rows_seen=0, rows_created=0, rows_updated=0)
|
||
session.add(run)
|
||
session.commit()
|
||
try:
|
||
if html is not None:
|
||
records = parse_html(html, region=region, category=category)
|
||
else:
|
||
previous = session.scalar(
|
||
select(OfficialRecordImport).where(
|
||
OfficialRecordImport.source_url == url,
|
||
OfficialRecordImport.status == ImportStatus.success,
|
||
).order_by(OfficialRecordImport.started_at.desc()).limit(1)
|
||
)
|
||
fetched = fetch_records(
|
||
url, region=region, category=category,
|
||
etag=previous.response_etag if previous else None,
|
||
last_modified=previous.response_last_modified if previous else None,
|
||
)
|
||
run.response_status = fetched.status_code
|
||
run.response_etag = fetched.etag or (previous.response_etag if previous else None)
|
||
run.response_last_modified = fetched.last_modified or (previous.response_last_modified if previous else None)
|
||
run.response_content_type = fetched.content_type
|
||
run.response_bytes = fetched.response_bytes
|
||
if fetched.records is None:
|
||
run.not_modified = True
|
||
run.status = ImportStatus.success
|
||
run.finished_at = datetime.now(timezone.utc)
|
||
session.commit()
|
||
return run
|
||
records = fetched.records
|
||
run.rows_seen = len(records)
|
||
for raw in records:
|
||
key = external_id(raw)
|
||
report = session.scalar(select(CatchReport).where(CatchReport.source_external_id == key))
|
||
fish = _fish(session, raw.fish)
|
||
waterbody = _waterbody(session, raw.waterbody)
|
||
bait = _bait(session, raw.bait) if raw.bait else None
|
||
payload = asdict(raw) | {"record_date": raw.record_date.isoformat()}
|
||
caught = datetime.combine(raw.record_date, time(), tzinfo=timezone.utc)
|
||
if report is None:
|
||
session.add(CatchReport(fish=fish, waterbody=waterbody, bait=bait, spot=None, weight_g=raw.weight_g, caught_at=caught, reported_at=datetime.now(timezone.utc), player_name=raw.player, source_type=SourceType.official_record, source_url=url, source_external_id=key, source_confidence=100, moderation_status=ModerationStatus.approved, raw_payload=payload))
|
||
run.rows_created += 1
|
||
else:
|
||
report.raw_payload = payload
|
||
report.source_url = url
|
||
run.rows_updated += 1
|
||
run.status = ImportStatus.success
|
||
run.finished_at = datetime.now(timezone.utc)
|
||
session.commit()
|
||
return run
|
||
except Exception as exc:
|
||
session.rollback()
|
||
run = session.get(OfficialRecordImport, run.id)
|
||
run.status = ImportStatus.failed
|
||
run.finished_at = datetime.now(timezone.utc)
|
||
run.error_summary = str(exc)[:1000]
|
||
session.commit()
|
||
raise
|
||
|
||
|
||
def _fish(session: Session, name: str) -> Fish:
|
||
item = session.scalar(select(Fish).where(Fish.name_ru == name))
|
||
if item is None:
|
||
item = Fish(slug=_unique_slug(session, Fish, name), name_ru=name, trophy_weight_g=None)
|
||
session.add(item)
|
||
return item
|
||
|
||
|
||
def _waterbody(session: Session, name: str) -> Waterbody:
|
||
item = session.scalar(select(Waterbody).where(Waterbody.name_ru == name))
|
||
if item is None:
|
||
item = Waterbody(slug=_unique_slug(session, Waterbody, name), name_ru=name, unlock_level=None)
|
||
session.add(item)
|
||
return item
|
||
|
||
|
||
def _bait(session: Session, name: str) -> Bait:
|
||
normalized = normalize(name)
|
||
item = session.scalar(select(Bait).where(Bait.normalized_name == normalized))
|
||
if item is None:
|
||
item = Bait(name=name, normalized_name=normalized, kind=BaitKind.unknown)
|
||
session.add(item)
|
||
return item
|
||
|
||
|
||
def _unique_slug(session: Session, model: type[Fish] | type[Waterbody], name: str) -> str:
|
||
base = slugify(name)
|
||
candidate = base
|
||
index = 2
|
||
while session.scalar(select(model.id).where(model.slug == candidate)) is not None:
|
||
candidate = f"{base}-{index}"
|
||
index += 1
|
||
return candidate
|