feat: add RF4DB and RF4-STAT research parsers
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from dataclasses import asdict
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from .community_sources import parse_rf4db_catches, parse_rf4stat_fishing, parse_rf4stat_posts
|
||||
|
||||
|
||||
SOURCES = {
|
||||
"rf4db": ("https://download.rf4db.com/ru/catches", parse_rf4db_catches),
|
||||
"rf4stat-fishing": ("https://rf4-stat.ru/fishing/", parse_rf4stat_fishing),
|
||||
"rf4stat-posts": ("https://rf4-stat.ru/posts/", parse_rf4stat_posts),
|
||||
}
|
||||
USER_AGENT = "RF4-Spotter/0.1 (authorized data integration)"
|
||||
|
||||
|
||||
def fetch_html(url: str, *, timeout: float = 30) -> str:
|
||||
request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"})
|
||||
with urlopen(request, timeout=timeout) as response:
|
||||
if response.headers.get_content_type() != "text/html":
|
||||
raise ValueError(f"expected text/html, got {response.headers.get_content_type()}")
|
||||
return response.read().decode(response.headers.get_content_charset() or "utf-8")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="Fetch one authorized RF4 community source page")
|
||||
parser.add_argument("source", choices=SOURCES)
|
||||
parser.add_argument("--url", help="Override the configured public page URL")
|
||||
parser.add_argument("--limit", type=int, default=100, choices=range(1, 501), metavar="1..500")
|
||||
args = parser.parse_args(argv)
|
||||
default_url, parse = SOURCES[args.source]
|
||||
url = args.url or default_url
|
||||
try:
|
||||
records = parse(fetch_html(url))[:args.limit]
|
||||
except Exception as exc:
|
||||
print(f"community source 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())
|
||||
@@ -0,0 +1,245 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, time, timezone
|
||||
from urllib.parse import urljoin
|
||||
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
|
||||
|
||||
class CommunityParseError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ExternalCatch:
|
||||
source_system: str
|
||||
source_external_id: str
|
||||
source_url: str
|
||||
fish: str
|
||||
fish_external_id: str | None
|
||||
waterbody: str
|
||||
waterbody_external_id: str | None
|
||||
x: int | None
|
||||
y: int | None
|
||||
weight_g: int | None
|
||||
bait: str | None
|
||||
rig_type: str | None
|
||||
game_time: time | None
|
||||
published_at: datetime | None
|
||||
player_name: str | None
|
||||
weather: str | None
|
||||
water_temperature_c: float | None
|
||||
clip: str | None
|
||||
fishing_style: str | None
|
||||
evidence_urls: tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class EquipmentItem:
|
||||
kind: str
|
||||
name: str
|
||||
external_id: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RF4DBCatchDetail:
|
||||
source_external_id: str
|
||||
source_url: str
|
||||
wind: str | None
|
||||
line_release_m: float | None
|
||||
clip_m: float | None
|
||||
cast_direction_deg: float | None
|
||||
equipment: tuple[EquipmentItem, ...]
|
||||
|
||||
|
||||
def _text(node: Tag | None) -> str:
|
||||
return " ".join(node.get_text(" ", strip=True).split()) if node else ""
|
||||
|
||||
|
||||
def _key(href: str | None) -> str | None:
|
||||
if not href:
|
||||
return None
|
||||
return href.rstrip("/").rsplit("/", 1)[-1]
|
||||
|
||||
|
||||
def _coordinates(raw: str) -> tuple[int | None, int | None]:
|
||||
match = re.fullmatch(r"\s*(-?\d{1,5}):(-?\d{1,5})\s*", raw)
|
||||
return (int(match.group(1)), int(match.group(2))) if match else (None, None)
|
||||
|
||||
|
||||
def _game_time(raw: str) -> time | None:
|
||||
match = re.search(r"Игровое время\s*(\d{1,2}):(\d{2})", raw)
|
||||
return time(int(match.group(1)), int(match.group(2))) if match else None
|
||||
|
||||
|
||||
def _weight(raw: str) -> int | None:
|
||||
match = re.search(r"([\d\s.,]+)\s*(кг|kg|г|g)\b", raw.casefold())
|
||||
if not match:
|
||||
return None
|
||||
number, unit = match.groups()
|
||||
compact = number.replace(" ", "").replace(",", ".")
|
||||
return round(float(compact) * 1000) if unit in {"кг", "kg"} else int(compact.replace(".", ""))
|
||||
|
||||
|
||||
def parse_rf4db_catches(html: str, *, base_url: str = "https://rf4db.com") -> list[ExternalCatch]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
result: list[ExternalCatch] = []
|
||||
for card in soup.select("article.catch-card"):
|
||||
detail = card.select_one('a.catch-card__time[href*="/catches/"]')
|
||||
fish_link = card.select_one("h2 a")
|
||||
map_link = card.select_one('.catch-card__place a[href*="/maps/"]')
|
||||
if not detail or not fish_link or not map_link:
|
||||
continue
|
||||
external_id = _key(detail.get("href"))
|
||||
if not external_id:
|
||||
continue
|
||||
x, y = _coordinates(_text(card.select_one(".catch-card__place b")))
|
||||
bait_link = card.select_one('.catch-card__place a[href*="/wiki/baits/"]')
|
||||
badges = card.select(".catch-badge")
|
||||
weather = next((_text(b.select_one("b")) for b in badges if _text(b).startswith("Погода")), None)
|
||||
temperature_text = next((_text(b.select_one("b")) for b in badges if _text(b).startswith("Темп. воды")), "")
|
||||
temperature_match = re.search(r"-?\d+(?:[.,]\d+)?", temperature_text)
|
||||
rig = card.select_one(".catch-badge--rig")
|
||||
result.append(ExternalCatch(
|
||||
source_system="rf4db", source_external_id=external_id,
|
||||
source_url=urljoin(base_url, str(detail.get("href"))),
|
||||
fish=_text(fish_link), fish_external_id=_key(fish_link.get("href")),
|
||||
waterbody=_text(map_link), waterbody_external_id=_key(map_link.get("href")),
|
||||
x=x, y=y, weight_g=None, bait=_text(bait_link) or None,
|
||||
rig_type=_text(rig) or None,
|
||||
game_time=_game_time(_text(card.select_one(".catch-card__place"))),
|
||||
published_at=None, player_name=None, weather=weather or None,
|
||||
water_temperature_c=float(temperature_match.group().replace(",", ".")) if temperature_match else None,
|
||||
clip=None, fishing_style=None, evidence_urls=(),
|
||||
))
|
||||
if not result:
|
||||
raise CommunityParseError("RF4DB catch cards not found")
|
||||
return result
|
||||
|
||||
|
||||
def parse_rf4db_detail(html: str, *, source_url: str) -> RF4DBCatchDetail:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
root = soup.select_one("article.catch-detail")
|
||||
if root is None:
|
||||
raise CommunityParseError("RF4DB catch detail not found")
|
||||
external_id = _key(source_url)
|
||||
if not external_id:
|
||||
raise CommunityParseError("RF4DB catch detail URL has no ID")
|
||||
facts = {_text(row.select_one("dt")): _text(row.select_one("dd")) for row in root.select(".catch-facts > div")}
|
||||
|
||||
def number(label: str) -> float | None:
|
||||
match = re.search(r"-?\d+(?:[.,]\d+)?", facts.get(label, ""))
|
||||
return float(match.group().replace(",", ".")) if match else None
|
||||
|
||||
equipment: list[EquipmentItem] = []
|
||||
for item in root.select(".catch-equipment article"):
|
||||
link = item.select_one("a[href]")
|
||||
name = _text(link)
|
||||
if not name:
|
||||
continue
|
||||
equipment.append(EquipmentItem(
|
||||
kind=_text(item.select_one("small")) or "Снаряжение",
|
||||
name=name,
|
||||
external_id=_key(link.get("href")) if link else None,
|
||||
))
|
||||
return RF4DBCatchDetail(
|
||||
source_external_id=external_id,
|
||||
source_url=source_url,
|
||||
wind=facts.get("Ветер") or None,
|
||||
line_release_m=number("Выпуск лески"),
|
||||
clip_m=number("Клипса"),
|
||||
cast_direction_deg=number("Направление заброса"),
|
||||
equipment=tuple(equipment),
|
||||
)
|
||||
|
||||
|
||||
def _rf4stat_date(raw_date: str, raw_time: str, *, now: datetime) -> datetime | None:
|
||||
try:
|
||||
day, month = (int(part) for part in raw_date.split("."))
|
||||
hour, minute = (int(part) for part in raw_time.split(":"))
|
||||
value = datetime(now.year, month, day, hour, minute, tzinfo=timezone.utc)
|
||||
if value > now:
|
||||
value = value.replace(year=value.year - 1)
|
||||
return value
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def parse_rf4stat_fishing(
|
||||
html: str, *, base_url: str = "https://rf4-stat.ru/", now: datetime | None = None,
|
||||
) -> list[ExternalCatch]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
now = now or datetime.now(timezone.utc)
|
||||
result: list[ExternalCatch] = []
|
||||
for row in soup.select("tr.load-row"):
|
||||
id_link = row.select_one('a.share[href^="fishing/"]:not(.hide-print)')
|
||||
fish_link = row.select_one('.fish-icon a[href^="fish/"]')
|
||||
if not id_link or not fish_link:
|
||||
continue
|
||||
external_id = _key(id_link.get("href"))
|
||||
if not external_id:
|
||||
continue
|
||||
position = row.select_one(".list-position .position:not(.position-locked)")
|
||||
x, y = _coordinates(_text(position))
|
||||
style = row.select_one(".post-style-icon[title]")
|
||||
style_text = str(style.get("title", "")) if style else ""
|
||||
result.append(ExternalCatch(
|
||||
source_system="rf4stat-fishing", source_external_id=external_id,
|
||||
source_url=urljoin(base_url, str(id_link.get("href"))),
|
||||
fish=_text(row.select_one("td.fish a")), fish_external_id=_key(fish_link.get("href")),
|
||||
waterbody=_text(row.select_one("td.mobile-location")), waterbody_external_id=None,
|
||||
x=x, y=y, weight_g=_weight(_text(row.select_one("td.fish small"))),
|
||||
bait=_text(row.select_one('[data-filter="bait"]')) or None,
|
||||
rig_type=None, game_time=None,
|
||||
published_at=_rf4stat_date(_text(row.select_one("td.time small")), _text(row.select_one("td.time > div")), now=now),
|
||||
player_name=_text(row.select_one("td.list-gamer a")) or None,
|
||||
weather=None, water_temperature_c=None,
|
||||
clip=_text(row.select_one(".clip")) or None,
|
||||
fishing_style=style_text.removeprefix("Вид ловли:").strip() or None,
|
||||
evidence_urls=(urljoin(base_url, str(row.select_one("a.share.hide-print").get("href"))),) if row.select_one("a.share.hide-print") else (),
|
||||
))
|
||||
if not result:
|
||||
raise CommunityParseError("RF4-STAT fishing rows not found")
|
||||
return result
|
||||
|
||||
|
||||
def parse_rf4stat_posts(
|
||||
html: str, *, base_url: str = "https://rf4-stat.ru/",
|
||||
) -> list[ExternalCatch]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
result: list[ExternalCatch] = []
|
||||
for post in soup.select(".post-item.load-row"):
|
||||
post_id = str(post.get("data-post-id", "")).strip()
|
||||
if not post_id:
|
||||
continue
|
||||
published_raw = str(post.get("data-published-at", ""))
|
||||
published_at = datetime.fromtimestamp(int(published_raw), tz=timezone.utc) if published_raw.isdigit() else None
|
||||
position = post.select_one(".spot-col .position:not(.position-locked)")
|
||||
x, y = _coordinates(_text(position))
|
||||
style = post.select_one(".post-style-icon[title]")
|
||||
style_text = str(style.get("title", "")) if style else ""
|
||||
evidence = tuple(urljoin(base_url, str(image.get("src"))) for image in post.select(".screen-file-slot img[src]"))
|
||||
baits = _text(post.select_one(".bait-text")) or None
|
||||
for index, fish in enumerate(post.select(".fish-icon")):
|
||||
fish_link = fish.select_one('a[href^="fish/"]')
|
||||
fish_name = _text(fish.select_one(".fish-name"))
|
||||
if not fish_name:
|
||||
continue
|
||||
result.append(ExternalCatch(
|
||||
source_system="rf4stat-post", source_external_id=f"{post_id}:{index}",
|
||||
source_url=urljoin(base_url, f"posts/{post_id}"),
|
||||
fish=fish_name, fish_external_id=_key(fish_link.get("href")) if fish_link else None,
|
||||
waterbody=_text(post.select_one(".location")), waterbody_external_id=None,
|
||||
x=x, y=y, weight_g=_weight(_text(fish.select_one(".weight"))),
|
||||
bait=baits, rig_type=None, game_time=None, published_at=published_at,
|
||||
player_name=_text(post.select_one(".player")) or None,
|
||||
weather=None, water_temperature_c=None,
|
||||
clip=_text(post.select_one(".clip")) or None,
|
||||
fishing_style=style_text.removeprefix("Вид ловли:").strip() or None,
|
||||
evidence_urls=evidence,
|
||||
))
|
||||
if not result:
|
||||
raise CommunityParseError("RF4-STAT posts not found")
|
||||
return result
|
||||
Reference in New Issue
Block a user