diff --git a/user_scanner/user_scan/community/ccm.py b/user_scanner/user_scan/community/ccm.py new file mode 100644 index 000000000..48508aac6 --- /dev/null +++ b/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".*?(?:404|erreur|not found).*?", 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"(?:Profil de\s+)?([^<']+?)(?:'s profile)?\s*-\s*CCM", + response.text, + ) + if m and m.group(1).strip().lower() == user.lower(): + extra: dict[str, str] = {} + s = re.search(r'pro_header__status">\s*([^<]+)', 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"
{label}
\s*]*>(.*?)", + 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) diff --git a/user_scanner/user_scan/dev/hackthissite.py b/user_scanner/user_scan/dev/hackthissite.py new file mode 100644 index 000000000..b48588928 --- /dev/null +++ b/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"Hack This Site :: View User: ([^<]+)", + 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]+){label}:\s*([^<]*)Hack This Site" 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") diff --git a/user_scanner/user_scan/entertainment/listal.py b/user_scanner/user_scan/entertainment/listal.py new file mode 100644 index 000000000..ee8fab57a --- /dev/null +++ b/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"([^<]+?)(?:'s stuff)? on Listal", response.text) + if m and m.group(1).strip().lower() == user.lower(): + extra: dict[str, str] = {} + media: dict[str, str] = {} + + pt = re.search(r"Points:\s*(?:]*>)?(\d+)", response.text) + if pt: + extra["points"] = pt.group(1) + + fl = re.search(r"(\d+) Followers", response.text) + if fl: + extra["followers"] = fl.group(1) + + av = re.search(r']*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, + ) diff --git a/user_scanner/user_scan/music/mixlr.py b/user_scanner/user_scan/music/mixlr.py new file mode 100644 index 000000000..be72234c3 --- /dev/null +++ b/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) diff --git a/user_scanner/user_scan/music/ultimate_guitar.py b/user_scanner/user_scan/music/ultimate_guitar.py new file mode 100644 index 000000000..aab7c4e56 --- /dev/null +++ b/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"([^<]+) \| Ultimate-Guitar\.Com", 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":"([^&]+)"", response.text) + if t: + extra["usertitle"] = t.group(1) + + fc = re.search(r"follower_count":(\d+)", response.text) + if fc: + extra["followers"] = fc.group(1) + + fg = re.search(r"following_count":(\d+)", response.text) + if fg: + extra["following"] = fg.group(1) + + iq = re.search(r"iq":(\d+)", response.text) + if iq: + extra["iq"] = iq.group(1) + + dr = re.search(r"date_registered":(\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":"([^&]*)"", response.text) + if city and city.group(1): + extra["city"] = city.group(1) + + country = re.search(r"country_name":"([^&]*)"", response.text) + if country and country.group(1): + extra["location"] = country.group(1) + + va = re.search(r"is_avatar_exists":(true|false)", response.text) + if va and va.group(1) == "true": + av = re.search( + r"avatar":"(https://www\.ultimate-guitar\.com/static[^\s"]+)"", + response.text, + ) + if av: + media["avatar"] = av.group(1).replace("&", "&") + + 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)