feat: add offline waterbody source crosswalk

2026-09-20 19:16:46 +07:00
parent d541443a1a
commit f02cb247a3
11 changed files with 311 additions and 14 deletions
@@ -127,6 +127,31 @@ class RF4DBWaterbodyDetail:
    point_urls: tuple[str, ...]


+@dataclass(frozen=True, slots=True)
+class RF4MapWaterbodyCandidate:
+    """Secondary waterbody directory facts; never a canonical catalog row."""
+
+    source_system: str
+    source_external_id: str
+    source_url: str
+    name: str
+    fish_species_count: int | None
+
+
+@dataclass(frozen=True, slots=True)
+class RF4StatWaterbodyCandidate:
+    """Secondary RF4-STAT location metrics for crosswalk and comparison."""
+
+    source_system: str
+    source_external_id: str
+    source_url: str
+    name: str
+    catches_count: int | None
+    fish_species_count: int | None
+    bait_count: int | None
+    spot_count: int | None
+
+
 def _text(node: Tag | None) -> str:
    return " ".join(node.get_text(" ", strip=True).split()) if node else ""

@@ -254,6 +279,88 @@ def parse_rf4db_waterbodies(
    return result


+def parse_rf4map_waterbodies(
+    html: str, *, base_url: str = "https://rf4map.ru/lakes",
+) -> list[RF4MapWaterbodyCandidate]:
+    """Parse the public RF4MAP directory as secondary crosswalk candidates."""
+    soup = BeautifulSoup(html, "html.parser")
+    result: list[RF4MapWaterbodyCandidate] = []
+    seen: set[str] = set()
+    for link in soup.select('a[href^="/lakes/"]'):
+        href = link.get("href")
+        external_id = _key(href)
+        if not isinstance(href, str) or not external_id or external_id in seen:
+            continue
+        label = _text(link)
+        fish_match = re.search(r"(\d+)\s+вид(?:а|ов)?\s+рыб", label, flags=re.I)
+        name = re.sub(r"\s*\d+\s+вид(?:а|ов)?\s+рыб(?:ы)?\s*$", "", label, flags=re.I).strip()
+        name = re.sub(r"^\+\s*\d+\s+", "", name).strip()
+        if not name:
+            continue
+        seen.add(external_id)
+        result.append(RF4MapWaterbodyCandidate(
+            source_system="rf4map-waterbodies",
+            source_external_id=external_id,
+            source_url=urljoin(base_url, href),
+            name=name,
+            fish_species_count=int(fish_match.group(1)) if fish_match else None,
+        ))
+    if not result:
+        raise CommunityParseError("RF4MAP waterbody directory not found")
+    return result
+
+
+def parse_rf4stat_waterbodies(
+    html: str, *, base_url: str = "https://en.rf4-stat.ru/locations/",
+) -> list[RF4StatWaterbodyCandidate]:
+    """Parse public RF4-STAT location cards without treating metrics as canonical facts."""
+    soup = BeautifulSoup(html, "html.parser")
+    result: list[RF4StatWaterbodyCandidate] = []
+    seen: set[str] = set()
+    for link in soup.select('a[href*="/locations/location/"]'):
+        href = link.get("href")
+        external_id = _key(href)
+        if not isinstance(href, str) or not external_id or external_id in seen:
+            continue
+        container = link.find_parent(["article", "li"]) or link.parent
+        text = _text(container)
+        name = _text(link)
+        metric_texts = [
+            value for node in container.select("span, div, small, strong")
+            if (value := _text(node)) and value != name
+        ]
+        metric_texts.append(text)
+        if not name:
+            continue
+
+        def metric(label: str) -> int | None:
+            patterns = (
+                rf"(?:{label})\s*:\s*([0-9]+(?: [0-9]{{3}})*)",
+                rf"([0-9]+(?: [0-9]{{3}})*)\s+(?:{label})",
+            )
+            for candidate in metric_texts:
+                for pattern in patterns:
+                    match = re.search(pattern, candidate, flags=re.I)
+                    if match:
+                        return int(match.group(1).replace(" ", ""))
+            return None
+
+        seen.add(external_id)
+        result.append(RF4StatWaterbodyCandidate(
+            source_system="rf4stat-locations",
+            source_external_id=external_id,
+            source_url=urljoin(base_url, href),
+            name=name,
+            catches_count=metric(r"catches|улов"),
+            fish_species_count=metric(r"kinds of fish|вид(?:а|ов) рыб"),
+            bait_count=metric(r"baits|нажив"),
+            spot_count=metric(r"spots|точ(?:ек|ки)"),
+        ))
+    if not result:
+        raise CommunityParseError("RF4-STAT waterbody directory not found")
+    return result
+
+
 def _gear_level(value: str | None) -> int | None:
    text = (value or "").strip()
    if not text: