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