feat: add gear provenance models and browser fetcher

2026-09-20 15:50:43 +07:00
parent b0edae98c6
commit 4e9895fbf4
19 changed files with 986 additions and 20 deletions
@@ -310,6 +310,10 @@ def main(argv: list[str] | None = None) -> int:
        default=Path(os.environ.get("RF4_COMMUNITY_FETCH_STATE", DEFAULT_STATE_FILE)),
        help="Persistent per-source cooldown state",
    )
+    parser.add_argument(
+        "--reserve-only", action="store_true",
+        help="Reserve the shared site cooldown and stop before making an HTTP request",
+    )
    args = parser.parse_args(argv)
    if args.source in DETAIL_SOURCES and not args.url:
        parser.error(f"--url is required for {args.source}")
@@ -319,6 +323,9 @@ def main(argv: list[str] | None = None) -> int:
        site_key = fetch_site_key(url)
        # Single atomic check-and-reserve before network I/O: failed attempts count toward the limit too.
        check_and_reserve(site_key, state_file=args.state_file)
+        if args.reserve_only:
+            print(json.dumps({"reserved": True, "source": args.source, "site_key": site_key}, ensure_ascii=False))
+            return 0
        html = fetch_html(url)
        parsed = (
            parse(html, source_url=url)
@@ -45,6 +45,51 @@ class EquipmentItem:
    external_id: str | None


+GEAR_CATEGORIES = frozenset({
+    "bait", "lure", "rod", "reel", "line", "hook", "rig", "float", "sinker", "other",
+})
+
+
+@dataclass(frozen=True, slots=True)
+class RF4DBGearItem:
+    source_system: str
+    source_external_id: str
+    source_url: str
+    name: str
+    category: str
+    subcategory: str | None
+    brand: str | None
+    family: str | None
+    unlock_level: int | None
+    image_url: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class RF4DBGearAttribute:
+    key: str
+    state: str
+    value: str | int | float | None
+    source_text: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class RF4DBGearDetail:
+    source_system: str
+    source_external_id: str
+    source_url: str
+    name: str
+    category: str
+    subcategory: str | None
+    brand: str | None
+    family: str | None
+    unlock_level: int | None
+    attributes: tuple[RF4DBGearAttribute, ...]
+    variants: tuple[str, ...]
+    compatible_with: tuple[str, ...]
+    rig_types: tuple[str, ...]
+    image_urls: tuple[str, ...]
+
+
@dataclass(frozen=True, slots=True)
 class RF4DBCatchDetail:
    source_external_id: str
@@ -209,6 +254,110 @@ def parse_rf4db_waterbodies(
    return result


+def _gear_level(value: str | None) -> int | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    if not re.fullmatch(r"\d+", text):
+        raise CommunityParseError(f"invalid gear unlock level: {text!r}")
+    return int(text)
+
+
+def _gear_category(value: str | None) -> str:
+    category = (value or "").strip().casefold()
+    if category not in GEAR_CATEGORIES:
+        raise CommunityParseError(f"invalid or missing gear category: {value!r}")
+    return category
+
+
+def _gear_value(node: Tag) -> str | int | float | None:
+    text = _text(node) or None
+    if text is None:
+        return None
+    value_type = str(node.get("data-type") or "text").casefold()
+    if value_type == "number":
+        try:
+            return float(text) if "." in text else int(text)
+        except ValueError as exc:
+            raise CommunityParseError(f"invalid numeric gear attribute: {text!r}") from exc
+    return text
+
+
+def parse_rf4db_gear_catalog(
+    html: str, *, source_url: str = "https://rf4db.com/ru/wiki/gear", expected_count: int | None = None,
+) -> list[RF4DBGearItem]:
+    """Parse a gear index while keeping an unknown total explicitly unknown."""
+    soup = BeautifulSoup(html, "html.parser")
+    cards = soup.select("article.gear-card, [data-gear-card]")
+    result: list[RF4DBGearItem] = []
+    seen: set[str] = set()
+    for card in cards:
+        link = card.select_one("a[href]")
+        external_id = str(card.get("data-gear-id") or _key(link.get("href") if link else None) or "")
+        name = _text(card.select_one("[data-name], h2, h3, .gear-name"))
+        category = _gear_category(card.get("data-category"))
+        if not external_id or not name:
+            raise CommunityParseError("RF4DB gear card is missing id or name")
+        if external_id in seen:
+            raise CommunityParseError(f"duplicate RF4DB gear id: {external_id}")
+        seen.add(external_id)
+        item_url = urljoin(source_url, str(link.get("href"))) if link and link.get("href") else source_url
+        image = card.select_one("img[src], img[data-src]")
+        image_url = urljoin(item_url, str(image.get("src") or image.get("data-src"))) if image else None
+        result.append(RF4DBGearItem(
+            source_system="rf4db", source_external_id=external_id, source_url=item_url,
+            name=name, category=category,
+            subcategory=_text(card.select_one("[data-subcategory], .gear-subcategory")) or None,
+            brand=_text(card.select_one("[data-brand], .gear-brand")) or None,
+            family=_text(card.select_one("[data-family], .gear-family")) or None,
+            unlock_level=_gear_level(card.get("data-unlock-level")), image_url=image_url,
+        ))
+    if not result:
+        raise CommunityParseError("RF4DB gear cards not found")
+    if expected_count is not None and len(result) != expected_count:
+        raise CommunityParseError(f"RF4DB gear catalog count mismatch: expected {expected_count}, got {len(result)}")
+    return result
+
+
+def parse_rf4db_gear_detail(
+    html: str, *, source_url: str,
+) -> RF4DBGearDetail:
+    """Parse one gear detail page with explicit missing/not-applicable/value states."""
+    soup = BeautifulSoup(html, "html.parser")
+    root = soup.select_one("main[data-gear-detail], article.gear-detail") or soup
+    external_id = _key(source_url)
+    name = _text(root.select_one("h1, [data-name]"))
+    category = _gear_category(root.get("data-category"))
+    if not external_id or not name:
+        raise CommunityParseError("RF4DB gear detail is missing id or name")
+    attributes: list[RF4DBGearAttribute] = []
+    for node in root.select("[data-gear-attribute]"):
+        key = str(node.get("data-gear-attribute") or "").strip()
+        state = str(node.get("data-state") or "value").strip().casefold()
+        if not key or state not in {"value", "not_applicable", "missing"}:
+            raise CommunityParseError("invalid RF4DB gear attribute state")
+        value = None if state != "value" else _gear_value(node)
+        attributes.append(RF4DBGearAttribute(key=key, state=state, value=value, source_text=_text(node) or None))
+    images = tuple(dict.fromkeys(
+        urljoin(source_url, str(node.get("src") or node.get("data-src")))
+        for node in root.select("img[src], img[data-src]")
+        if node.get("src") or node.get("data-src")
+    ))
+    return RF4DBGearDetail(
+        source_system="rf4db", source_external_id=external_id, source_url=source_url,
+        name=name, category=category,
+        subcategory=_text(root.select_one("[data-subcategory], .gear-subcategory")) or None,
+        brand=_text(root.select_one("[data-brand], .gear-brand")) or None,
+        family=_text(root.select_one("[data-family], .gear-family")) or None,
+        unlock_level=_gear_level(root.get("data-unlock-level")),
+        attributes=tuple(attributes),
+        variants=tuple(dict.fromkeys(_text(node) for node in root.select("[data-gear-variant]") if _text(node))),
+        compatible_with=tuple(dict.fromkeys(_text(node) for node in root.select("[data-compatible-with]") if _text(node))),
+        rig_types=tuple(dict.fromkeys(_text(node) for node in root.select("[data-rig-type]") if _text(node))),
+        image_urls=images,
+    )
+
+
 def parse_rf4db_waterbody_detail(
    html: str, *, source_url: str,
 ) -> RF4DBWaterbodyDetail: