Code play multi account

Code play game

import requests
import random
from time import sleep
import threading
from colorama import Fore, Style, init

# Initialize Colorama
init(autoreset=True)

def countdown(seconds):
    """Countdown timer to create a delay between requests."""
    while seconds:
        mins, secs = divmod(seconds, 60)
        time_format = '{:02d}:{:02d}'.format(mins, secs)
        print(Fore.YELLOW + '  ⏱  Waiting...' + time_format + ' seconds', end='\r')
        sleep(1)
        seconds -= 1

class Vana:
    def __init__(self) -> None:
        self.game_url = 'https://www.vanadatahero.com/api'

    def common_header(self, query):
        """Sets the common headers for each request."""
        return {
            'Host': 'www.vanadatahero.com',
            'content-type': 'application/json',
            'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36 Edg/128.0.0.0',
            'accept': '*/*',
            'x-telegram-web-app-init-data': query,
            'sec-fetch-site': 'same-origin',
            'sec-fetch-mode': 'cors',
            'sec-fetch-dest': 'empty',
            'referer': f'https://www.vanadatahero.com/challenges',
            'accept-language': 'en-US,en;q=0.9',
            'priority': 'u=1, i',
        }

    def _send_request(self, session: requests.Session, method: str, url: str, headers: dict, proxies: dict, json_data: dict = {}, params: dict = {}):
        """Performs HTTP requests with error handling."""
        try:
            response = session.request(method=method, url=url, headers=headers, proxies=proxies, json=json_data, params=params, timeout=60)
            if response.status_code == 200:
                if 'https://www.vanadatahero.com/api/tasks/1' in url:
                    return 'Ok'
                try:
                    return response.json()
                except ValueError:
                    if url == 'https://www.vanadatahero.com/_vercel/insights/view': 
                        pass
                    else: 
                        print(Fore.RED + f"it returned non-JSON response: {url}\n {response.text}")
                    return response.text
            else:
                if "not found" in response.text:
                    return False
                elif 'Telegram data has expired' in response.text:
                    return Fore.RED + "Telegram data has expired"
                elif '!DOCTYPE' in response.text:
                    print(Fore.RED + "ERROR: '!DOCTYPE'")
                    return False
                elif "EDGE_FUNCTION_INVOCATION_TIMEOUT" in response.text:
                    print(Fore.RED + "  ❌ Fail.. Server lag..")
                    return False
                elif response.status_code == 504:
                    print(Fore.RED + "  ❌ ERROR 504: Server took too long to respond.")
                    return False
                print(Fore.RED + f"  ❌ ERROR: {url}\n {response.text}")
                return False  
        except requests.exceptions.RequestException as e:
            print(Fore.RED + f"  ❌ Request Exception: {e}")
            return False

    def post_request(self, session: requests.Session, url: str, headers: dict = None, proxies: dict = {}, json_data: dict = None, params: dict = None):
        return self._send_request(session, 'POST', url, headers, proxies, json_data, params)

    def get_request(self, session: requests.Session, url: str, headers: dict = None, proxies: dict = {}, json_data: dict = None, params: dict = None):
        return self._send_request(session, 'GET', url, headers, proxies, json_data, params)

    def getPlayer(self, session: requests.Session):
        """Fetch player information from the API."""
        url = f'{self.game_url}/player'
        player_info = self.get_request(session=session, url=url)
        if isinstance(player_info, dict):
            print(Fore.GREEN + f"\n  😗 Player: {player_info['tgUsername']} - Total Points: {int(player_info['points'])}🌸")
            return player_info['tgUsername']
        if isinstance(player_info, str) and "expired" in player_info:
            return player_info
        return False

    def play(self, session: requests.Session, point: int, player_name):
        """Execute a play task in the game."""
        url = f'{self.game_url}/tasks/1'
        json_data = {
            'status': 'completed',
            'points': point,
        }

        play_info = self.post_request(session=session, url=url, json_data=json_data)
        if play_info:    
            print(Fore.CYAN + f"  🎲 Player: {player_name} - Play: {play_info}: +{point} Points 🌸")
            return True  # Indicate play was successful
        else: 
            # Check for specific error message and stop account if points limit exceeded
            if isinstance(play_info, str) and "Points limit exceeded" in play_info:
                print(Fore.RED + f"  ❌ Player: {player_name} - Points limit exceeded. Stopping this account.")
                return False  # Stop processing for this account
            print(Fore.RED + f"  ❌ Player: {player_name} - Play: {play_info}")
            return False  # Indicate play was unsuccessful

    def start(self, query: str, total: int, proxies: dict):
        """Starts the game process with the provided token and proxy."""
        try:
            if not query: return
            query = query.replace("'", "")
            _header = self.common_header(query=query)

            session = requests.Session()
            session.headers.update(_header)
            session.proxies.update(proxies)  # Update session with proxy settings

            player_name = self.getPlayer(session=session)
            print('')
            if not player_name:
                print(Fore.RED + f"ERROR: {player_name}")
                return
            if isinstance(player_name, str) and "expired" in player_name:
                print(Fore.RED + "\n❌ QUERY_TOKEN IS EXPIRED. PLEASE RELOAD THE GAME AND GET YOUR TOKEN BACK")
                return
            
            for i in range(total):
                # Check if the account has been stopped
                if not self.play(session=session, point=round(random.uniform(600, 800), 1), player_name=player_name):
                    break  # Stop the loop if points limit exceeded
                
                print(Fore.YELLOW + f"\n  ➡️  Round: {i + 1}/{total}")
                countdown(20)  # Wait for 20 seconds between plays
                print('')

            self.getPlayer(session=session)
        except Exception as e:
            print(Fore.RED + f"  ❌ ERROR: {e}")

def run_account(account_info, total):
    """Function to run a single account in a separate thread."""
    proxy, port, username, password, token = account_info.split(':')
    print(Fore.BLUE + f"\n🎮 Starting account with proxy: {proxy}:{port} and token: {token}")
    proxies = {
        'http': f'http://{username}:{password}@{proxy}:{port}',  
        'https': f'http://{username}:{password}@{proxy}:{port}',  
    }
    bot = Vana()
    bot.start(token, total, proxies)

if __name__ == "__main__":
    try:
        # Enter the number of accounts
        num_accounts = int(input(Fore.WHITE + "\n🌐 Enter the number of Telegram accounts: "))
        
        # Enter the number of rounds
        total = int(input(Fore.WHITE + "\n🎮 Number of rounds to play: "))
        print(Fore.GREEN + "    - OK")

        accounts = []  # List to store account information

        # Loop to input each account's proxy, username, password, and token
        print(Fore.WHITE + f"\n🔑 Enter account details (format: proxy:port:username:password:token) one per line. Enter 'done' when finished:")
        while True:
            account_info = input()
            if account_info.lower() == 'done':
                break
            accounts.append(account_info)  # Store account information

        # Start the bot for each account in a separate thread
        threads = []
        for account_info in accounts:
            thread = threading.Thread(target=run_account, args=(account_info, total))
            thread.start()
            threads.append(thread)

        # Wait for all threads to complete
        for thread in threads:
            thread.join()

    except Exception as e:
        print(Fore.RED + f"  ❌ An error occurred: {e}")

Last updated