feat: add Rumble new username module, rename current module to rumble_channel (#633)

* feat: add Rumble new username module, rename current module to rumble_channel

* fix(user_scan): verify Rumble namespace markers

- Distinguish 404 and 410 responses
- Require canonical URL, creator ID, and channel ID verification

Author: kristoisberg

user_scanner/user_scan/creator/rumble.py

@@ -1,51 +1,104 @@
+import html
+import json
 import re
-from user_scanner.core.orchestrator import generic_validate
+from urllib.parse import quote
+
+from curl_cffi.requests.exceptions import RequestException
+
+from user_scanner.core.impersonate import impersonate_request, impersonate_validate
 from user_scanner.core.result import Result
 
+CANONICAL_RE = re.compile(r"<link rel=canonical href=([^\s>]+)>", re.IGNORECASE)
+DESCRIPTION_RE = re.compile(
+    r'<div class="channel-about--description">.*?<p>(.*?)</p>',
+    re.IGNORECASE | re.DOTALL,
+)
+FOLLOWERS_RE = re.compile(
+    r"<span>\s*([\d.,]+[KMB]?)\s+Followers?\s*</span>", re.IGNORECASE
+)
+IMAGE_RE = re.compile(
+    r'class="channel-header--(img|backsplash-img)"[^>]+src="([^"]+)"',
+    re.IGNORECASE,
+)
+NOT_FOUND_MARKERS = {
+    404: "<title>404 Not found</title>",
+    410: "<title>410 Gone</title>",
+}
+SOCIAL_RE = re.compile(r'<a href="([^"]+)" class="channel-subheader--socials-item"')
+STAT_RE = re.compile(r"([\d,]+)\s+(views|videos)\s*</p>", re.IGNORECASE)
+VERIFIED_RE = re.compile(r'<svg class="channel-header--verified\b')
+
 
 def validate_rumble(user: str) -> Result:
-    """Validate a channel/creator on Rumble (rumble.com)."""
-    url = f"https://rumble.com/c/{user}"
-    show_url = f"https://rumble.com/c/{user}"
-    headers = {
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
-    }
+    """Validate a Rumble user account."""
+    url = f"https://rumble.com/user/{quote(user, safe='')}"
 
     def process(response):
-        response_text_lower = response.text.lower()
+        marker = NOT_FOUND_MARKERS.get(response.status_code)
+        if marker and marker in response.text:
+            return Result.available()
 
-        # 1. Explicit verification of available / not-found state (HTTP 404 or 404 title)
-        title_match = re.search(r"<title>(.*?)</title>", response.text, re.IGNORECASE)
-        page_title = title_match.group(1).strip() if title_match else ""
+        if response.status_code != 200:
+            return Result.error(f"Unexpected response status: {response.status_code}")
 
+        canonical = CANONICAL_RE.search(response.text)
+        creator_id = re.search(r'"creator_id":\s*"(\d+)"', response.text)
         if (
-            response.status_code == 404
-            or "404 not found" in page_title.lower()
-            or "page not found" in response_text_lower
+            not canonical
+            or canonical.group(1).casefold() != str(response.url).casefold()
+            or not creator_id
         ):
-            return Result.available(url=show_url)
+            return Result.error("Rumble user markers do not match the response")
 
-        # 2. Explicit verification of taken state + data extraction
-        if response.status_code == 200 and "rumble" in response_text_lower:
-            if page_title and "404" not in page_title:
-                extra = {"channel": user}
-                media = {}
+        extra = {"creator_id": creator_id.group(1)}
 
-                channel_name = page_title.replace(" - Rumble", "").replace(" on Rumble", "").strip()
-                if channel_name and channel_name.lower() != "rumble":
-                    extra["name"] = channel_name
+        profile = {}
+        profile_start = re.search(
+            rf'\{{"type":"user","url":"{re.escape(canonical.group(1))}"',
+            response.text,
+            re.IGNORECASE,
+        )
+        if profile_start:
+            try:
+                profile = json.JSONDecoder().raw_decode(
+                    response.text[profile_start.start() :]
+                )[0]
+            except json.JSONDecodeError:
+                pass
 
-                og_img_match = re.search(r'<meta property="og:image" content="(.*?)"', response.text, re.IGNORECASE)
-                if og_img_match:
-                    img_url = og_img_match.group(1).strip()
-                    if img_url and "rumble-logo" not in img_url.lower():
-                        media["avatar"] = img_url
+        for field in ("followers", "subscribers"):
+            if isinstance(profile.get(field), int):
+                extra[field] = profile[field]
+        if "followers" not in extra and (followers := FOLLOWERS_RE.search(response.text)):
+            value = followers.group(1).replace(",", "")
+            extra["followers"] = int(value) if value.isdigit() else value
+        extra["verified"] = profile.get(
+            "verified_badge", bool(VERIFIED_RE.search(response.text))
+        )
+        if badge_type := profile.get("badge_type"):
+            extra["badge_type"] = badge_type
+        if social_links := SOCIAL_RE.findall(response.text):
+            extra["social_links"] = [html.unescape(link) for link in social_links]
 
-                return Result.taken(extra=extra, media=media, url=show_url)
+        try:
+            about_response = impersonate_request(f"{url}/about")
+        except RequestException:
+            about_response = None
+        if about_response is not None and about_response.status_code == 200:
+            if description := DESCRIPTION_RE.search(about_response.text):
+                extra["description"] = " ".join(
+                    html.unescape(re.sub(r"<[^>]+>", " ", description.group(1))).split()
+                )
+            if joined := re.search(r"Joined\s+([^<]+)", about_response.text):
+                extra["joined"] = joined.group(1).strip()
+            for value, field in STAT_RE.findall(about_response.text):
+                extra[field.lower()] = int(value.replace(",", ""))
 
-            return Result.error("Could not verify channel details on Rumble", url=show_url)
+        media = {
+            "avatar" if kind == "img" else "banner": html.unescape(image_url)
+            for kind, image_url in IMAGE_RE.findall(response.text)
+        }
 
-        # 3. Graceful error for unexpected status codes (No bare else!)
-        return Result.error(f"Unexpected response status: {response.status_code}", url=show_url)
+        return Result.taken(extra=extra, media=media)
 
-    return generic_validate(url, process, headers=headers, show_url=show_url, follow_redirects=True)
+    return impersonate_validate(url, process, allow_redirects=True)

user_scanner/user_scan/creator/rumble_channel.py

@@ -0,0 +1,110 @@
+import html
+import json
+import re
+from urllib.parse import quote
+
+from curl_cffi.requests.exceptions import RequestException
+
+from user_scanner.core.impersonate import impersonate_request, impersonate_validate
+from user_scanner.core.result import Result
+
+CANONICAL_RE = re.compile(r"<link rel=canonical href=([^\s>]+)>", re.IGNORECASE)
+DESCRIPTION_RE = re.compile(
+    r'<div class="channel-about--description">.*?<p>(.*?)</p>',
+    re.IGNORECASE | re.DOTALL,
+)
+FOLLOWERS_RE = re.compile(
+    r"<span>\s*([\d.,]+[KMB]?)\s+Followers?\s*</span>", re.IGNORECASE
+)
+IMAGE_RE = re.compile(
+    r'class="channel-header--(img|backsplash-img)"[^>]+src="([^"]+)"',
+    re.IGNORECASE,
+)
+NOT_FOUND_TITLES = {404: "404 not found", 410: "410 gone"}
+SOCIAL_RE = re.compile(r'<a href="([^"]+)" class="channel-subheader--socials-item"')
+STAT_RE = re.compile(r"([\d,]+)\s+(views|videos)\s*</p>", re.IGNORECASE)
+TITLE_RE = re.compile(r"<title>([^<]*)</title>", re.IGNORECASE)
+VERIFIED_RE = re.compile(r'<svg class="channel-header--verified\b')
+
+
+def validate_rumble_channel(user: str) -> Result:
+    """Validate a Rumble channel."""
+    url = f"https://rumble.com/c/{quote(user, safe='')}"
+
+    def process(response):
+        title_match = TITLE_RE.search(response.text)
+        title = html.unescape(title_match.group(1)).strip() if title_match else ""
+
+        if title.casefold() == NOT_FOUND_TITLES.get(response.status_code):
+            return Result.available()
+
+        if response.status_code != 200:
+            return Result.error(f"Unexpected response status: {response.status_code}")
+
+        canonical = CANONICAL_RE.search(response.text)
+        creator_id = re.search(r'"creator_id":\s*"(\d+)"', response.text)
+        channel_id = re.search(r'"channel_id":\s*"(\d+)"', response.text)
+        if (
+            not canonical
+            or canonical.group(1).casefold() != str(response.url).casefold()
+            or not creator_id
+            or not channel_id
+        ):
+            return Result.error("Rumble channel markers do not match the response")
+
+        extra = {
+            "name": title,
+            "creator_id": creator_id.group(1),
+            "channel_id": channel_id.group(1),
+        }
+
+        profile = {}
+        profile_start = re.search(
+            rf'\{{"type":"channel","url":"{re.escape(canonical.group(1))}"',
+            response.text,
+            re.IGNORECASE,
+        )
+        if profile_start:
+            try:
+                profile = json.JSONDecoder().raw_decode(
+                    response.text[profile_start.start() :]
+                )[0]
+            except json.JSONDecodeError:
+                pass
+
+        for field in ("followers", "subscribers"):
+            if isinstance(profile.get(field), int):
+                extra[field] = profile[field]
+        if "followers" not in extra and (followers := FOLLOWERS_RE.search(response.text)):
+            value = followers.group(1).replace(",", "")
+            extra["followers"] = int(value) if value.isdigit() else value
+        extra["verified"] = profile.get(
+            "verified_badge", bool(VERIFIED_RE.search(response.text))
+        )
+        if badge_type := profile.get("badge_type"):
+            extra["badge_type"] = badge_type
+        if social_links := SOCIAL_RE.findall(response.text):
+            extra["social_links"] = [html.unescape(link) for link in social_links]
+
+        try:
+            about_response = impersonate_request(f"{url}/about")
+        except RequestException:
+            about_response = None
+        if about_response is not None and about_response.status_code == 200:
+            if description := DESCRIPTION_RE.search(about_response.text):
+                extra["description"] = " ".join(
+                    html.unescape(re.sub(r"<[^>]+>", " ", description.group(1))).split()
+                )
+            if joined := re.search(r"Joined\s+([^<]+)", about_response.text):
+                extra["joined"] = joined.group(1).strip()
+            for value, field in STAT_RE.findall(about_response.text):
+                extra[field.lower()] = int(value.replace(",", ""))
+
+        media = {
+            "avatar" if kind == "img" else "banner": html.unescape(image_url)
+            for kind, image_url in IMAGE_RE.findall(response.text)
+        }
+
+        return Result.taken(extra=extra, media=media)
+
+    return impersonate_validate(url, process, allow_redirects=True)