feat(user_scan): add 5 modules

user_scanner/user_scan/community/ccm.py

@@ -0,0 +1,55 @@
+import re
+import urllib.parse
+
+from user_scanner.core.orchestrator import Result, generic_validate
+
+
+def validate_ccm(user: str) -> Result:
+    encoded_user = urllib.parse.quote(user)
+    url = f"https://ccm.net/profile/user/{encoded_user}"
+    show_url = f"https://ccm.net/profile/user/{encoded_user}"
+
+    def process(response) -> Result:
+        if response.status_code == 404:
+            if (
+                re.search(r"<title>.*?(?:404|erreur|not found).*?</title>", response.text, re.I)
+                or "404" in response.text
+            ):
+                return Result.available()
+            return Result.error("404 response received without expected error marker")
+
+        if response.status_code == 200:
+            m = re.search(
+                r"<title>(?:Profil de\s+)?([^<']+?)(?:'s profile)?\s*-\s*CCM</title>",
+                response.text,
+            )
+            if m and m.group(1).strip().lower() == user.lower():
+                extra: dict[str, str] = {}
+                s = re.search(r'pro_header__status">\s*<span>([^<]+)</span>', response.text)
+                if s:
+                    extra["status"] = s.group(1).strip()
+
+                labels = [
+                    ("Registration date", "joined"),
+                    ("Gender", "gender"),
+                    ("Location", "location"),
+                    ("Birthday", "birthday"),
+                    ("Website", "website"),
+                    ("Contributions", "contributions"),
+                ]
+                for label, key in labels:
+                    dd = re.search(
+                        rf"<dt>{label}</dt>\s*<dd[^>]*>(.*?)</dd>",
+                        response.text,
+                        re.S,
+                    )
+                    if dd:
+                        val = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", dd.group(1))).strip()
+                        if val and val.lower() not in ("not entered", "n/a"):
+                            extra[key] = val
+
+                return Result.taken(extra=extra)
+
+        return Result.error(f"Unexpected response status: {response.status_code}")
+
+    return generic_validate(url, process, show_url=show_url, follow_redirects=True)

user_scanner/user_scan/dev/hackthissite.py

@@ -0,0 +1,63 @@
+import re
+import urllib.parse
+
+from user_scanner.core.impersonate import impersonate_validate
+from user_scanner.core.orchestrator import Result
+
+
+def validate_hackthissite(user: str) -> Result:
+    encoded_user = urllib.parse.quote(user)
+    url = f"https://www.hackthissite.org/user/view/{encoded_user}"
+    show_url = f"https://www.hackthissite.org/user/view/{encoded_user}"
+
+    def process(response) -> Result:
+        if response.status_code == 200:
+            m = re.search(
+                r"<title>Hack This Site :: View User: ([^<]+)</title>",
+                response.text,
+            )
+            if m and m.group(1).strip().lower() == user.lower():
+                extra: dict[str, str] = {}
+                media: dict[str, str] = {}
+
+                r = re.search(r"Rank:\s*([^<(\n]+?)\s*\((\d+)\s*Points\)", response.text)
+                if r:
+                    extra["rank"] = r.group(1).strip()
+                    extra["points"] = r.group(2)
+
+                s = re.search(r"Status:\s*([^<\n]+)<br", response.text)
+                if s:
+                    extra["status"] = s.group(1).strip()
+
+                bad = ("", "not entered", "none", "hidden", "http://")
+                labels = [
+                    ("UserID", "id"),
+                    ("Joined", "joined"),
+                    ("Last Login", "last_login"),
+                    ("Last Active", "last_active"),
+                    ("Location", "location"),
+                    ("Website", "website"),
+                    ("TimeZone", "timezone"),
+                ]
+                for label, key in labels:
+                    v = re.search(rf"<b>{label}:</b>\s*([^<]*)<br", response.text)
+                    if v:
+                        val = v.group(1).strip()
+                        if val and val.lower() not in bad:
+                            extra[key] = val
+
+                av = re.search(
+                    r'<img src="(/pages/user/avatar\.view\.php\?id=(\d+))"',
+                    response.text,
+                )
+                if av and av.group(2) != "404":
+                    media["avatar"] = f"https://www.hackthissite.org{av.group(1)}"
+
+                return Result.taken(extra=extra, media=media)
+
+            if "<title>Hack This Site</title>" in response.text and "View User:" not in response.text:
+                return Result.available()
+
+        return Result.error(f"Unexpected response status: {response.status_code}")
+
+    return impersonate_validate(url, process, show_url=show_url, impersonate="chrome")

user_scanner/user_scan/entertainment/listal.py

@@ -0,0 +1,48 @@
+import re
+
+from user_scanner.core.impersonate import impersonate_validate
+from user_scanner.core.orchestrator import Result
+
+
+def validate_listal(user: str) -> Result:
+    if not re.match(r"^[A-Za-z0-9_-]+$", user):
+        return Result.error("Invalid username format for Listal")
+
+    url = f"https://{user}.listal.com/"
+    show_url = f"https://{user}.listal.com/"
+
+    def process(response) -> Result:
+        if response.status_code != 200:
+            return Result.error(f"Unexpected response status: {response.status_code}")
+
+        m = re.search(r"<title>([^<]+?)(?:'s stuff)? on Listal</title>", response.text)
+        if m and m.group(1).strip().lower() == user.lower():
+            extra: dict[str, str] = {}
+            media: dict[str, str] = {}
+
+            pt = re.search(r"<b>Points:</b>\s*(?:<a[^>]*>)?(\d+)", response.text)
+            if pt:
+                extra["points"] = pt.group(1)
+
+            fl = re.search(r"(\d+) Followers</a>", response.text)
+            if fl:
+                extra["followers"] = fl.group(1)
+
+            av = re.search(r'<img[^>]*src="(https://lthumb\.lisimg\.com/[^"]+)"', response.text)
+            if av:
+                media["avatar"] = av.group(1)
+
+            return Result.taken(extra=extra, media=media)
+
+        if "Listal - List the stuff you love!" in response.text:
+            return Result.available()
+
+        return Result.error("Unexpected response body from Listal")
+
+    return impersonate_validate(
+        url,
+        process,
+        show_url=show_url,
+        impersonate="chrome",
+        allow_redirects=True,
+    )

user_scanner/user_scan/music/mixlr.py

@@ -0,0 +1,74 @@
+import urllib.parse
+
+from user_scanner.core.helpers import get_random_user_agent
+from user_scanner.core.orchestrator import Result, generic_validate
+
+
+def validate_mixlr(user: str) -> Result:
+    encoded_user = urllib.parse.quote(user.lower())
+    url = f"https://api.mixlr.com/users/{encoded_user}"
+    show_url = f"https://mixlr.com/{user}"
+
+    headers = {
+        "User-Agent": get_random_user_agent(),
+        "Accept": "application/json",
+    }
+
+    def process(response) -> Result:
+        if response.status_code == 404:
+            try:
+                data = response.json()
+                if data.get("error") == "Resource not found":
+                    return Result.available()
+            except Exception:
+                pass
+            if "Resource not found" in response.text:
+                return Result.available()
+            return Result.error("404 response received without expected error payload")
+
+        if response.status_code == 200:
+            try:
+                data = response.json()
+            except Exception:
+                return Result.error("Invalid JSON in 200 response")
+
+            if not isinstance(data, dict) or "id" not in data:
+                return Result.error("Unexpected API response structure")
+
+            extra: dict[str, str] = {}
+            media: dict[str, str] = {}
+
+            if "id" in data:
+                extra["id"] = str(data["id"])
+
+            if role := data.get("role"):
+                extra["role"] = str(role)
+
+            if tz := data.get("time_zone"):
+                extra["time_zone"] = str(tz)
+
+            plan = data.get("plan") or {}
+            if plan.get("reference"):
+                extra["plan"] = str(plan["reference"])
+            elif plan.get("name"):
+                extra["plan"] = str(plan["name"])
+
+            if data.get("is_premium"):
+                extra["premium"] = "true"
+
+            if data.get("is_live"):
+                extra["live"] = "true"
+
+            channel = data.get("channel") or {}
+            if channel.get("name"):
+                extra["channel"] = str(channel["name"])
+
+            if avatar := data.get("profile_image_url"):
+                if "fallback-user" not in avatar:
+                    media["avatar"] = str(avatar)
+
+            return Result.taken(extra=extra, media=media)
+
+        return Result.error(f"Unexpected response status: {response.status_code}")
+
+    return generic_validate(url, process, show_url=show_url, headers=headers)

user_scanner/user_scan/music/ultimate_guitar.py

@@ -0,0 +1,69 @@
+import re
+import urllib.parse
+from datetime import datetime, timezone
+
+from user_scanner.core.impersonate import impersonate_validate
+from user_scanner.core.orchestrator import Result
+
+
+def validate_ultimate_guitar(user: str) -> Result:
+    encoded_user = urllib.parse.quote(user)
+    url = f"https://www.ultimate-guitar.com/u/{encoded_user}"
+    show_url = f"https://www.ultimate-guitar.com/u/{encoded_user}"
+
+    def process(response) -> Result:
+        if response.status_code == 410:
+            if "Oops! We couldn't find that page." in response.text:
+                return Result.available()
+            return Result.error("410 response received without expected not-found marker")
+
+        if response.status_code == 200:
+            m = re.search(r"<title>([^<]+) \| Ultimate-Guitar\.Com</title>", response.text)
+            if m and m.group(1).strip().lower() == user.lower():
+                extra: dict[str, str] = {}
+                media: dict[str, str] = {}
+
+                t = re.search(r"usertitle&quot;:&quot;([^&]+)&quot;", response.text)
+                if t:
+                    extra["usertitle"] = t.group(1)
+
+                fc = re.search(r"follower_count&quot;:(\d+)", response.text)
+                if fc:
+                    extra["followers"] = fc.group(1)
+
+                fg = re.search(r"following_count&quot;:(\d+)", response.text)
+                if fg:
+                    extra["following"] = fg.group(1)
+
+                iq = re.search(r"iq&quot;:(\d+)", response.text)
+                if iq:
+                    extra["iq"] = iq.group(1)
+
+                dr = re.search(r"date_registered&quot;:(\d+)", response.text)
+                if dr:
+                    extra["joined"] = datetime.fromtimestamp(
+                        int(dr.group(1)), tz=timezone.utc
+                    ).strftime("%Y-%m-%d")
+
+                city = re.search(r"city&quot;:&quot;([^&]*)&quot;", response.text)
+                if city and city.group(1):
+                    extra["city"] = city.group(1)
+
+                country = re.search(r"country_name&quot;:&quot;([^&]*)&quot;", response.text)
+                if country and country.group(1):
+                    extra["location"] = country.group(1)
+
+                va = re.search(r"is_avatar_exists&quot;:(true|false)", response.text)
+                if va and va.group(1) == "true":
+                    av = re.search(
+                        r"avatar&quot;:&quot;(https://www\.ultimate-guitar\.com/static[^\s&quot;]+)&quot;",
+                        response.text,
+                    )
+                    if av:
+                        media["avatar"] = av.group(1).replace("&amp;", "&")
+
+                return Result.taken(extra=extra, media=media)
+
+        return Result.error(f"Unexpected response status: {response.status_code}")
+
+    return impersonate_validate(url, process, show_url=show_url, allow_redirects=True)