feat: parse RF4DB waterbody catalog
This commit is contained in:
@@ -55,6 +55,18 @@ class RF4DBCatchDetail:
|
||||
equipment: tuple[EquipmentItem, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RF4DBWaterbody:
|
||||
source_system: str
|
||||
source_external_id: str
|
||||
source_url: str
|
||||
name: str
|
||||
unlock_level: int | None
|
||||
unlock_label: str
|
||||
fish_species_count: int
|
||||
image_url: str | None
|
||||
|
||||
|
||||
def _text(node: Tag | None) -> str:
|
||||
return " ".join(node.get_text(" ", strip=True).split()) if node else ""
|
||||
|
||||
@@ -127,6 +139,56 @@ def _datetime(raw: object) -> datetime | None:
|
||||
return None
|
||||
|
||||
|
||||
def parse_rf4db_waterbodies(
|
||||
html: str, *, source_url: str = "https://rf4db.com/ru/maps", expected_count: int = 19,
|
||||
) -> list[RF4DBWaterbody]:
|
||||
"""Parse the public RF4DB waterbody index without assigning image roles."""
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
result: list[RF4DBWaterbody] = []
|
||||
seen: set[str] = set()
|
||||
for link in soup.select('a[href*="/ru/maps/level_"]'):
|
||||
href = link.get("href")
|
||||
if not isinstance(href, str):
|
||||
continue
|
||||
external_id = _key(href)
|
||||
if not external_id or external_id in seen:
|
||||
continue
|
||||
name = _text(link)
|
||||
if not name:
|
||||
continue
|
||||
card = link.find_parent(["article", "li"])
|
||||
if card is None:
|
||||
card = link.find_parent("div")
|
||||
card_text = _text(card)
|
||||
level_match = re.search(r"(?:Уровень|Level)\s*(?:Lv\.?\s*)?(Старт|Start|\d+)", card_text, re.I)
|
||||
fish_match = re.search(r"(?:Рыбы|Fish(?:es)?)\s*(\d+)", card_text, re.I)
|
||||
if not level_match or not fish_match:
|
||||
continue
|
||||
unlock_label = level_match.group(1)
|
||||
unlock_level = int(unlock_label) if unlock_label.isdigit() else None
|
||||
image = card.select_one("img[src], img[data-src]") if card else None
|
||||
image_raw = image.get("src") or image.get("data-src") if image else None
|
||||
image_url = urljoin(source_url, str(image_raw)) if image_raw else None
|
||||
result.append(RF4DBWaterbody(
|
||||
source_system="rf4db",
|
||||
source_external_id=external_id,
|
||||
source_url=urljoin(source_url, href),
|
||||
name=name,
|
||||
unlock_level=unlock_level,
|
||||
unlock_label=unlock_label,
|
||||
fish_species_count=int(fish_match.group(1)),
|
||||
image_url=image_url,
|
||||
))
|
||||
seen.add(external_id)
|
||||
if not result:
|
||||
raise CommunityParseError("RF4DB waterbody cards not found")
|
||||
if len(result) != expected_count:
|
||||
raise CommunityParseError(
|
||||
f"RF4DB waterbody catalog is incomplete: expected {expected_count}, got {len(result)}"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def parse_rf4db_catches(html: str, *, base_url: str = "https://rf4db.com") -> list[ExternalCatch]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
result: list[ExternalCatch] = []
|
||||
|
||||
Reference in New Issue
Block a user