diff --git a/user_scanner/user_scan/gaming/stackb.py b/user_scanner/user_scan/gaming/stackb.py new file mode 100644 index 00000000..88c78d77 --- /dev/null +++ b/user_scanner/user_scan/gaming/stackb.py @@ -0,0 +1,126 @@ +import html +import json +import re +from urllib.parse import quote + +from user_scanner.core.helpers import get_random_user_agent +from user_scanner.core.orchestrator import make_request +from user_scanner.core.result import Result + + +def validate_stackb(user: str) -> Result: + profile_url = f"https://stackb.net/@{user}" + url = f"https://stackb.net/@{quote(user, safe='')}" + login_url = "https://stackb.net/login" + + headers = { + "User-Agent": get_random_user_agent(), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "ru,en-US;q=0.9,en;q=0.8", + } + + try: + login_response = make_request(login_url, headers=headers, follow_redirects=True) + if login_response.status_code != 200: + return Result.error(f"Unexpected login status: {login_response.status_code}", url=url) + + login_text = login_response.text + csrf_match = re.search(r']*wire:id="([^"]+)"[^>]*x-data="loginFormCaptcha', + login_text, + ) + if not csrf_match or not snapshot_match: + return Result.error("Could not find login form tokens", url=url) + + csrf_token = html.unescape(csrf_match.group(1)) + snapshot = html.unescape(snapshot_match.group(1)) + payload = { + "_token": csrf_token, + "components": [ + { + "snapshot": snapshot, + "updates": { + "identifier": user, + "password": "StackB-not-the-password-12345", + }, + "calls": [{"path": "", "method": "submitLogin", "params": []}], + } + ], + } + login_check = make_request( + "https://stackb.net/livewire/update", + method="POST", + headers={ + **headers, + "Accept": "application/json", + "Content-Type": "application/json", + "Origin": "https://stackb.net", + "Referer": login_url, + "X-CSRF-TOKEN": csrf_token, + "X-Livewire": "true", + }, + json=payload, + ) + + if login_check.status_code != 200: + return Result.error(f"Unexpected login check status: {login_check.status_code}", url=url) + + try: + login_data = login_check.json() + login_messages = " ".join( + str(dispatch.get("params", {}).get("message", "")) + for component in login_data.get("components", []) + for dispatch in component.get("effects", {}).get("dispatches", []) + ) + except (ValueError, AttributeError): + return Result.error("Unexpected login check response", url=url) + + if "Таких пользователей не нашлось" in login_messages: + return Result.available(url=url) + if "Пароль введен неверно" not in login_messages: + return Result.error("Unexpected login check response", url=url) + + extra = {} + try: + profile_response = make_request( + url, + headers=headers, + show_url=url, + follow_redirects=True, + ) + response_text = profile_response.text + profile = {} + for json_match in re.finditer(r'', response_text, re.DOTALL): + try: + data = json.loads(html.unescape(json_match.group(1))) + except json.JSONDecodeError: + continue + + if data.get("@type") == "ProfilePage": + entity = data.get("mainEntity") + if isinstance(entity, dict): + profile = entity + break + + if profile: + description_match = re.search(r']*property=["\']og:description["\'][^>]*content=["\']([^"\']*)', response_text, re.IGNORECASE) + stats_text = html.unescape(description_match.group(1)).strip() if description_match else None + + if name := profile.get("name"): extra["display_name"] = name + if description := profile.get("description"): extra["bio"] = description + if image := profile.get("image"): extra["avatar"] = image + + if stats_text: + if rank_match := re.search(r"Ранг:\s*([^\.]+)", stats_text): + extra["rank"] = rank_match.group(1).strip() + if followers_match := re.search(r"Подписчики:\s*(\d+)", stats_text): + extra["followers"] = int(followers_match.group(1)) + + extra["profile_url"] = profile_url + except Exception: + pass + + return Result.taken(extra=extra, url=url) + except Exception as exc: + return Result.error(exc, url=url) diff --git a/user_scanner/user_scan/other/pedsovet.py b/user_scanner/user_scan/other/pedsovet.py new file mode 100644 index 00000000..9c9aff9f --- /dev/null +++ b/user_scanner/user_scan/other/pedsovet.py @@ -0,0 +1,86 @@ +import html +import re + +from user_scanner.core.helpers import get_random_user_agent +from user_scanner.core.orchestrator import generic_validate +from user_scanner.core.result import Result + + +def validate_pedsovet(user: str) -> Result: + url = f"https://pedsovet.su/index/8-0-{user}" + + if not re.match(r"^[a-z0-9_@]+$", user, re.IGNORECASE): + return Result.error( + "Usernames can only contain letters, numbers, underscores and at signs", + url=url, + ) + + headers = { + "User-Agent": get_random_user_agent(), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "ru,en-US;q=0.9,en;q=0.8", + } + + def process(response): + response_text = response.text + + if "Пользователь не найден" in response_text: + return Result.available() + + if response.status_code != 200: + return Result.error(f"Unexpected response status: {response.status_code}") + + login_match = re.search(r"
Логин:\s*([^<]+)
", response_text, re.IGNORECASE) + found_login = html.unescape(login_match.group(1)).strip() if login_match else None + + if ( + found_login + and found_login.lower() == user.lower() + and "Ссылка на профиль:" in response_text + ): + extra = {"login": found_login} + + if id_match := re.search(r"/index/8-(\d+)", response_text): + extra["id"] = id_match.group(1) + extra["profile_url"] = f"https://pedsovet.su/index/8-{id_match.group(1)}" + + for label, key in ( + ("Группа пользователей", "group"), + ("Регистрация", "registered"), + ): + if field_match := re.search(rf"{re.escape(label)}:\s*(.*?)
", response_text, re.IGNORECASE | re.DOTALL): + value = re.sub(r"<[^>]+>", "", field_match.group(1)) + extra[key] = html.unescape(value).strip() + + if last_login_match := re.search(r"Последний вход\s*([^<]+)", response_text, re.IGNORECASE): + extra["last_login"] = html.unescape(last_login_match.group(1)).strip() + + if avatar_match := re.search( + r'(.*?)
', response_text, re.IGNORECASE | re.DOTALL + ): + about = re.sub(r"<[^>]+>", "", about_match.group(1)) + about = html.unescape(about).strip() + if about != "Пользователь пока ничего не сообщил о себе.": + extra["about"] = about + + return Result.taken(extra=extra) + + return Result.error("Unexpected response body") + + return generic_validate( + url, + process, + headers=headers, + show_url=url, + follow_redirects=True, + ) diff --git a/user_scanner/user_scan/shopping/yaga_co_za.py b/user_scanner/user_scan/shopping/yaga_co_za.py new file mode 100644 index 00000000..f1f34f85 --- /dev/null +++ b/user_scanner/user_scan/shopping/yaga_co_za.py @@ -0,0 +1,68 @@ +import html +import json +import re +from urllib.parse import quote + +from user_scanner.core.helpers import get_random_user_agent +from user_scanner.core.orchestrator import generic_validate +from user_scanner.core.result import Result + + +def validate_yaga_co_za(user: str) -> Result: + user = user.lower() + url = f"https://www.yaga.co.za/{quote(user, safe='')}" + + headers = { + "User-Agent": get_random_user_agent(), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + } + + def process(response): + if response.status_code == 404: + return Result.available() + + if response.status_code != 200: + return Result.error( + f"Unexpected response status: {response.status_code}", + ) + + match = re.search( + r'', + response.text, + re.DOTALL, + ) + if not match: + return Result.error("Could not find Next.js data") + + try: + data = json.loads(html.unescape(match.group(1))) + except json.JSONDecodeError: + return Result.error("Could not parse Next.js data") + + page_props = data.get("props", {}).get("pageProps", {}) + shop = page_props.get("initialShop") + + if shop is None: + return Result.available() + + if shop.get("activeSlug") != user: + return Result.error("Unexpected shop slug") + + extra = {} + owner = shop.get("owner") or {} + if shop_id := shop.get("id"): extra["id"] = shop_id + if name := shop.get("name"): extra["name"] = name + if description := shop.get("description"): extra["description"] = description + if first_name := owner.get("firstName"): extra["owner_first_name"] = first_name + if last_name := owner.get("lastName"): extra["owner_last_name"] = last_name + + return Result.taken(extra=extra) + + return generic_validate( + url, + process, + headers=headers, + show_url=url, + follow_redirects=True, + ) diff --git a/user_scanner/user_scan/shopping/yaga_ee.py b/user_scanner/user_scan/shopping/yaga_ee.py new file mode 100644 index 00000000..02adde4f --- /dev/null +++ b/user_scanner/user_scan/shopping/yaga_ee.py @@ -0,0 +1,68 @@ +import html +import json +import re +from urllib.parse import quote + +from user_scanner.core.helpers import get_random_user_agent +from user_scanner.core.orchestrator import generic_validate +from user_scanner.core.result import Result + + +def validate_yaga_ee(user: str) -> Result: + user = user.lower() + url = f"https://www.yaga.ee/{quote(user, safe='')}" + + headers = { + "User-Agent": get_random_user_agent(), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + } + + def process(response): + if response.status_code == 404: + return Result.available() + + if response.status_code != 200: + return Result.error( + f"Unexpected response status: {response.status_code}", + ) + + match = re.search( + r'', + response.text, + re.DOTALL, + ) + if not match: + return Result.error("Could not find Next.js data") + + try: + data = json.loads(html.unescape(match.group(1))) + except json.JSONDecodeError: + return Result.error("Could not parse Next.js data") + + page_props = data.get("props", {}).get("pageProps", {}) + shop = page_props.get("initialShop") + + if shop is None: + return Result.available() + + if shop.get("activeSlug") != user: + return Result.error("Unexpected shop slug") + + extra = {} + owner = shop.get("owner") or {} + if shop_id := shop.get("id"): extra["id"] = shop_id + if name := shop.get("name"): extra["name"] = name + if description := shop.get("description"): extra["description"] = description + if first_name := owner.get("firstName"): extra["owner_first_name"] = first_name + if last_name := owner.get("lastName"): extra["owner_last_name"] = last_name + + return Result.taken(extra=extra) + + return generic_validate( + url, + process, + headers=headers, + show_url=url, + follow_redirects=True, + )