272 lines
11 KiB
Python
272 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
import time as time_module
|
||
from contextlib import contextmanager
|
||
from dataclasses import asdict, dataclass
|
||
from datetime import date, datetime, time, timezone
|
||
|
||
import httpx
|
||
from sqlalchemy import select, text
|
||
from sqlalchemy.orm import Session
|
||
from rf4_research.official_parser import RecordsContractError, parse_official_records
|
||
|
||
from .models import (
|
||
Bait, BaitKind, CatchReport, Fish, ImportRecordEvent, ImportStatus, ModerationStatus,
|
||
OfficialRecordImport, SourceType, Waterbody,
|
||
)
|
||
|
||
|
||
USER_AGENT = "RF4-Spotter/0.1 (public records importer)"
|
||
|
||
|
||
class ImportSourceError(ValueError):
|
||
pass
|
||
|
||
|
||
class ImportAlreadyRunning(RuntimeError):
|
||
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_html(html: str, *, region: str, category: str) -> list[RawRecord]:
|
||
try:
|
||
rows = parse_official_records(html)
|
||
except RecordsContractError as exc:
|
||
raise ImportSourceError(str(exc)) from exc
|
||
return [RawRecord(
|
||
region=region.upper(), category=category, player=row.player, fish=row.fish,
|
||
weight_g=row.weight_g, waterbody=row.waterbody, bait=row.bait,
|
||
record_date=row.record_date,
|
||
) for row in rows]
|
||
|
||
|
||
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 _lock_key(url: str, region: str, category: str) -> int:
|
||
digest = hashlib.sha256(f"{url}|{region.upper()}|{category}".encode()).digest()
|
||
return int.from_bytes(digest[:8], byteorder="big", signed=True)
|
||
|
||
|
||
@contextmanager
|
||
def _official_import_lock(session: Session, *, url: str, region: str, category: str):
|
||
bind = session.get_bind()
|
||
if bind.dialect.name != "postgresql":
|
||
yield
|
||
return
|
||
connection = bind.connect()
|
||
key = _lock_key(url, region, category)
|
||
try:
|
||
acquired = bool(connection.scalar(text("SELECT pg_try_advisory_lock(:key)"), {"key": key}))
|
||
except Exception:
|
||
connection.close()
|
||
raise
|
||
if not acquired:
|
||
connection.close()
|
||
raise ImportAlreadyRunning("official import is already running for this source and category")
|
||
try:
|
||
yield
|
||
finally:
|
||
try:
|
||
connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key})
|
||
finally:
|
||
connection.close()
|
||
|
||
|
||
def import_records(session: Session, *, url: str, region: str, category: str, html: str | None = None) -> OfficialRecordImport:
|
||
with _official_import_lock(session, url=url, region=region, category=category):
|
||
return _import_records_locked(session, url=url, region=region, category=category, html=html)
|
||
|
||
|
||
def _import_records_locked(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)
|
||
now = datetime.now(timezone.utc)
|
||
if report is None:
|
||
report = CatchReport(fish=fish, waterbody=waterbody, bait=bait, spot=None, weight_g=raw.weight_g, caught_at=caught, reported_at=now, 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)
|
||
session.add(report)
|
||
session.add(ImportRecordEvent(
|
||
catch_report=report, import_run=run, event_type="created", created_at=now,
|
||
changes={"weight_g": raw.weight_g, "player": raw.player, "record_date": raw.record_date.isoformat()},
|
||
provenance={"source_system": "rf4-official", "source_url": url, "source_external_id": key},
|
||
))
|
||
run.rows_created += 1
|
||
else:
|
||
# A12: Only create event if actual values changed
|
||
old_payload = (report.raw_payload or {})
|
||
new_payload = asdict(raw) | {"record_date": raw.record_date.isoformat()}
|
||
changed_fields = {}
|
||
for field in ("weight_g", "player", "waterbody", "bait", "record_date"):
|
||
old_val = old_payload.get(field)
|
||
new_val = new_payload.get(field)
|
||
if old_val != new_val:
|
||
changed_fields[field] = {"old": old_val, "new": new_val}
|
||
if changed_fields:
|
||
# Keep the queryable normalized record in sync with its
|
||
# versioned source payload.
|
||
report.fish = fish
|
||
report.waterbody = waterbody
|
||
report.bait = bait
|
||
report.weight_g = raw.weight_g
|
||
report.player_name = raw.player
|
||
report.caught_at = caught
|
||
report.raw_payload = payload
|
||
report.source_url = url
|
||
session.add(ImportRecordEvent(
|
||
catch_report=report, import_run=run, event_type="updated", created_at=now,
|
||
changes=changed_fields,
|
||
provenance={"source_system": "rf4-official", "source_url": url, "source_external_id": key},
|
||
))
|
||
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
|