Multi account with multithreading support

[afdah]

Is it possible to support multithreading? To speed things up especially for multi accounts with proxy.

[tboy1337]

I was thinking the same thing, I don't think it necessarily needs multithreading though, just the ability to run all accounts or at most like 5 account in parallel. The problem is that when you have 10+ accounts it takes so long to get back to the first account that the amounts of taps available is always at its maximum.

[Fy0urM]

Running multiple accounts in parallel can lead to blocking of api requests for spam due to the limitation of api request frequency, only if you do not use a separate proxy for each account. if you are willing to try you can use the asyncio library and aiohttp for example to implement asynchronous startup of each account. this would require rewriting most of the main script

[Fy0urM]

You can create several folders with bots, fill the config file of each bot with different accounts and run each bot and check my assumption. it's stupid and inconvenient but to check why not?

[tboy1337]

Maybe implement it but give a warning that it should only be used that way with proxys for each account? Yea I guessed several folders would work but it's a bit messy to do it that way imo.

[Fy0urM]

I tried to implement asynchronous start of accounts, I checked on my 2 accounts, it works, but I have no proxy, so I can not guarantee the work for you. and I do not plan to update under new updates of the main script so if you triple this option you will have to update the asynchronous script yourself based on the corrections of the original script. also removed sending messages to telegram to simplify modification. also there will be a mess in the logs of several parallel accounts and the more accounts the more difficult it will be to observe the process. the aiohttp library may need to be installed additionally for operation

[Fy0urM]

I made some mistakes in the body of the main method, here is the corrected version

upd: Most likely, there is an error in the main method where previously there were return and now continue because of which the loop will start from the beginning and not reach the end. as I said before, you need to rewrite not a small part of the script logic to work correctly.

upd2: Another attempt to fix my shitty code to a working state under a spoiler

main_async.py

```python # Developed by: MasterkinG32 # Date: 2024 # Github: https://github.com/masterking32 import asyncio import datetime import json import logging import random import time import requests from colorlog import ColoredFormatter import uuid from utilities import * from promogames import * import aiohttp try: from config import * except ImportError: print("Config file not found.") print("Create a copy of config.py.example and rename it to config.py") print("And fill in the required fields.") exit() if "ConfigFileVersion" not in locals() or ConfigFileVersion != 1: print("Invalid config file version.") print("Please update the config file to the latest version.") print("Create a copy of config.py.example and rename it to config.py") print("And fill in the required fields.") exit() # ---------------------------------------------# # Logging configuration LOG_LEVEL = logging.DEBUG # Include date and time in the log format LOGFORMAT = "%(log_color)s[Master HamsterKombat Bot]%(reset)s[%(log_color)s%(levelname)s%(reset)s] %(asctime)s %(log_color)s%(message)s%(reset)s" logging.root.setLevel(LOG_LEVEL) formatter = ColoredFormatter( LOGFORMAT, "%Y-%m-%d %H:%M:%S" ) # Specify the date/time format stream = logging.StreamHandler() stream.setLevel(LOG_LEVEL) stream.setFormatter(formatter) log = logging.getLogger("pythonConfig") log.setLevel(LOG_LEVEL) log.addHandler(stream) # End of configuration # ---------------------------------------------# class HamsterKombatAccount: def __init__(self, AccountData): self.account_name = AccountData["account_name"] self.Authorization = AccountData["Authorization"] self.UserAgent = AccountData["UserAgent"] self.Proxy = AccountData["Proxy"] self.config = AccountData["config"] self.isAndroidDevice = "Android" in self.UserAgent self.balanceCoins = 0 self.availableTaps = 0 self.maxTaps = 0 self.ProfitPerHour = 0 self.earnPassivePerHour = 0 self.SpendTokens = 0 self.account_data = None self.telegram_chat_id = AccountData["telegram_chat_id"] self.totalKeys = 0 self.balanceKeys = 0 self.configVersion = "" def GetConfig(self, key, default=None): if key in self.config: return self.config[key] return default # Send HTTP requests async def HttpRequest( self, url, headers, method="POST", validStatusCodes=200, payload=None, ignore_errors=False, ): # Default headers defaultHeaders = { "Accept": "*/*", "Connection": "keep-alive", "Host": "api.hamsterkombatgame.io", "Origin": "https://hamsterkombatgame.io", "Referer": "https://hamsterkombatgame.io/", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-site", "User-Agent": self.UserAgent, } # Add default headers for Android devices to avoid detection, Not needed for iOS devices if self.isAndroidDevice: defaultHeaders["HTTP_SEC_CH_UA_PLATFORM"] = '"Android"' defaultHeaders["HTTP_SEC_CH_UA_MOBILE"] = "?1" defaultHeaders["HTTP_SEC_CH_UA"] = ( '"Android WebView";v="125", "Chromium";v="125", "Not.A/Brand";v="24"' ) defaultHeaders["HTTP_X_REQUESTED_WITH"] = "org.telegram.messenger.web" # Add and replace new headers to default headers for key, value in headers.items(): defaultHeaders[key] = value try: protocol = "https" if url.startswith("https") else "http" proxy = self.Proxy.get(protocol) async with aiohttp.ClientSession(headers=defaultHeaders) as session: if method == "GET": async with session.get(url, proxy=proxy) as response: response_data = await response.text() elif method == "POST": async with session.post(url, data=payload, proxy=proxy) as response: response_data = await response.text() elif method == "OPTIONS": async with session.options(url, proxy=proxy) as response: response_data = await response.text() else: log.error(f"[{self.account_name}] Invalid method: {method}") return None if response.status != validStatusCodes: if ignore_errors: return None log.error(f"[{self.account_name}] Status code is not {validStatusCodes}") log.error(f"[{self.account_name}] Response: {response_data}") return None if "config-version" in response.headers: self.configVersion = response.headers["config-version"] if method == "OPTIONS": return True return await response.json() except Exception as e: if ignore_errors: return None log.error(f"[{self.account_name}] Error: {e}") return None # Sending sync request async def syncRequest(self): url = "https://api.hamsterkombatgame.io/clicker/sync" headers = { "Access-Control-Request-Headers": self.Authorization, "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) # Get list of upgrades to buy async def UpgradesForBuyRequest(self): url = "https://api.hamsterkombatgame.io/clicker/upgrades-for-buy" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) # Buy an upgrade async def BuyUpgradeRequest(self, UpgradeId): url = "https://api.hamsterkombatgame.io/clicker/buy-upgrade" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, "Content-Type": "application/json", "Accept": "application/json", } payload = json.dumps( { "upgradeId": UpgradeId, "timestamp": int(datetime.datetime.now().timestamp() * 1000), } ) # Send POST request return await self.HttpRequest(url, headers, "POST", 200, payload) # Tap the hamster async def TapRequest(self, tap_count): url = "https://api.hamsterkombatgame.io/clicker/tap" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } payload = json.dumps( { "timestamp": int(datetime.datetime.now().timestamp() * 1000), "availableTaps": 0, "count": int(tap_count), } ) # Send POST request return await self.HttpRequest(url, headers, "POST", 200, payload) # Get list of boosts to buy async def BoostsToBuyListRequest(self): url = "https://api.hamsterkombatgame.io/clicker/boosts-for-buy" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) # Buy a boost async def BuyBoostRequest(self, boost_id): url = "https://api.hamsterkombatgame.io/clicker/buy-boost" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } payload = json.dumps( { "boostId": boost_id, "timestamp": int(datetime.datetime.now().timestamp() * 1000), } ) # Send POST request return await self.HttpRequest(url, headers, "POST", 200, payload) async def getAccountData(self): account_data = await self.syncRequest() if account_data is None or account_data is False: log.error(f"[{self.account_name}] Unable to get account data.") return False if "clickerUser" not in account_data: log.error(f"[{self.account_name}] Invalid account data.") return False if "balanceCoins" not in account_data["clickerUser"]: log.error(f"[{self.account_name}] Invalid balance coins.") return False self.account_data = account_data self.balanceCoins = account_data["clickerUser"]["balanceCoins"] self.availableTaps = account_data["clickerUser"]["availableTaps"] self.maxTaps = account_data["clickerUser"]["maxTaps"] self.earnPassivePerHour = account_data["clickerUser"]["earnPassivePerHour"] if "balanceKeys" in account_data["clickerUser"]: self.balanceKeys = account_data["clickerUser"]["balanceKeys"] else: self.balanceKeys = 0 if "totalKeys" in account_data["clickerUser"]: self.totalKeys = account_data["clickerUser"]["totalKeys"] else: self.totalKeys = 0 return account_data async def BuyFreeTapBoostIfAvailable(self): log.info(f"[{self.account_name}] Checking for free tap boost...") BoostList = await self.BoostsToBuyListRequest() if BoostList is None: log.error(f"[{self.account_name}] Failed to get boosts list.") return None BoostForTapList = None for boost in BoostList["boostsForBuy"]: if boost["price"] == 0 and boost["id"] == "BoostFullAvailableTaps": BoostForTapList = boost break if ( BoostForTapList is not None and "price" in BoostForTapList and "cooldownSeconds" in BoostForTapList and BoostForTapList["price"] == 0 and BoostForTapList["cooldownSeconds"] == 0 ): log.info(f"[{self.account_name}] Free boost found, attempting to buy...") await asyncio.sleep(5) await self.BuyBoostRequest(BoostForTapList["id"]) log.info(f"[{self.account_name}] Free boost bought successfully") return True else: log.info(f"\033[1;34m[{self.account_name}] No free boosts available\033[0m") return False async def IPRequest(self): url = "https://api.hamsterkombatgame.io/ip" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "GET", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 200) headers = { "Authorization": self.Authorization, } # Send GET request return await self.HttpRequest(url, headers, "GET", 200) async def GetSkins(self): url = "https://api.hamsterkombatgame.io/clicker/get-skin" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } # Send POST request return await self.HttpRequest(url, headers, "POST", 200, "{}") async def AccountInfoTelegramRequest(self): url = "https://api.hamsterkombatgame.io/auth/account-info" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) async def ListTasksRequest(self): url = "https://api.hamsterkombatgame.io/clicker/list-tasks" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) async def GetListAirDropTasksRequest(self): url = "https://api.hamsterkombatgame.io/clicker/list-airdrop-tasks" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) async def GetAccountConfigRequest(self): url = "https://api.hamsterkombatgame.io/clicker/config" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request return await self.HttpRequest(url, headers, "POST", 200) async def GetAccountConfigVersionRequest(self): if self.configVersion == "": return None url = f"https://api.hamsterkombatgame.io/clicker/config/{self.configVersion}" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "GET", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send GET request return await self.HttpRequest(url, headers, "GET", 200) async def ClaimDailyCipherRequest(self, DailyCipher): url = "https://api.hamsterkombatgame.io/clicker/claim-daily-cipher" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } payload = json.dumps( { "cipher": DailyCipher, } ) # Send POST request return await self.HttpRequest(url, headers, "POST", 200, payload) async def CheckTaskRequest(self, task_id): url = "https://api.hamsterkombatgame.io/clicker/check-task" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } payload = json.dumps( { "taskId": task_id, } ) # Send POST request return await self.HttpRequest(url, headers, "POST", 200, payload) async def BuyCard(self, card): upgradesResponse = await self.BuyUpgradeRequest(card["id"]) if upgradesResponse is None: log.error(f"[{self.account_name}] Failed to buy the card.") return False log.info(f"[{self.account_name}] Card bought successfully") await asyncio.sleep(3) self.balanceCoins -= card["price"] self.ProfitPerHour += card["profitPerHourDelta"] self.SpendTokens += card["price"] self.earnPassivePerHour += card["profitPerHourDelta"] return True async def ListBuyOptions(self, selected_upgrades): log.info( f"[{self.account_name}] List of {self.GetConfig('show_num_buy_options', 0)} best buy options:" ) count = 1 for selected_card in selected_upgrades: if ( "cooldownSeconds" in selected_card and selected_card["cooldownSeconds"] > 0 ): continue profitCoefficient = CalculateCardProfitCoefficient(selected_card) log.info( f"[{self.account_name}] {count}: {selected_card['name']}, Profit: {selected_card['profitPerHourDelta']}, Price: {number_to_string(selected_card['price'])}, Coefficient: {int(profitCoefficient)} Level: {selected_card['level']}" ) count = count + 1 if count > self.GetConfig("show_num_buy_options", 0): break async def BuyBestCard(self): log.info(f"[{self.account_name}] Checking for best card...") await asyncio.sleep(2) upgradesResponse = await self.UpgradesForBuyRequest() if upgradesResponse is None: log.error(f"[{self.account_name}] Failed to get upgrades list.") return False upgrades = [ item for item in upgradesResponse["upgradesForBuy"] if not item["isExpired"] and item["isAvailable"] and item["profitPerHourDelta"] > 0 ] if len(upgrades) == 0: log.warning(f"[{self.account_name}] No upgrades available.") return False balanceCoins = int(self.balanceCoins) log.info(f"[{self.account_name}] Searching for the best upgrades...") selected_upgrades = SortUpgrades( upgrades, 999_999_999_999 ) # Set max budget to a high number if len(selected_upgrades) == 0: log.warning(f"[{self.account_name}] No upgrades available.") return False if self.GetConfig("show_num_buy_options", 0) > 0: await self.ListBuyOptions(selected_upgrades) current_selected_card = selected_upgrades[0] for selected_card in selected_upgrades: if ( "cooldownSeconds" in selected_card and selected_card["cooldownSeconds"] > 0 and selected_card["cooldownSeconds"] < 180 ): log.warning( f"[{self.account_name}] {selected_card['name']} is on cooldown and cooldown is less than 180 seconds..." ) log.warning( f"[{self.account_name}] Waiting for {selected_card['cooldownSeconds'] + 2} seconds..." ) await asyncio.sleep(selected_card["cooldownSeconds"] + 2) selected_card["cooldownSeconds"] = 0 if ( "cooldownSeconds" in selected_card and selected_card["cooldownSeconds"] > 0 and not self.config["enable_parallel_upgrades"] ): log.warning( f"[{self.account_name}] {selected_card['name']} is on cooldown..." ) return False if ( "cooldownSeconds" in selected_card and selected_card["cooldownSeconds"] > 0 ): log.warning( f"[{self.account_name}] {selected_card['name']} is on cooldown, Checking for next card..." ) continue profitCoefficient = CalculateCardProfitCoefficient(selected_card) coefficientLimit = self.config["parallel_upgrades_max_price_per_hour"] if ( profitCoefficient > coefficientLimit and self.config["enable_parallel_upgrades"] ): log.warning( f"[{self.account_name}] {selected_card['name']} is too expensive to buy in parallel..." ) log.warning( f"[{self.account_name}] Cost is: {int(profitCoefficient)} / coin increase in profit. Cost limit: {coefficientLimit}" ) log.warning( f"[{self.account_name}] Adjust `parallel_upgrades_max_price_per_hour` to change this behaviour" ) return False current_selected_card = selected_card break log.info( f"[{self.account_name}] Best upgrade is {current_selected_card['name']} with profit {current_selected_card['profitPerHourDelta']} and price {number_to_string(current_selected_card['price'])}, Level: {current_selected_card['level']}" ) if balanceCoins < current_selected_card["price"]: log.warning( f"[{self.account_name}] Balance is too low to buy the best card." ) return False log.info(f"[{self.account_name}] Attempting to buy the best card...") buy_result = await self.BuyCard(current_selected_card) if buy_result: await asyncio.sleep(2) log.info( f"[{self.account_name}] Best card purchase completed successfully, Your profit per hour increased by {number_to_string(self.ProfitPerHour)} coins, Spend tokens: {number_to_string(self.SpendTokens)}" ) return True return False async def StartMiniGame(self, AccountConfigData, AccountID): if "dailyKeysMiniGame" not in AccountConfigData: log.error(f"[{self.account_name}] Unable to get daily keys mini game.") return if AccountConfigData["dailyKeysMiniGame"]["isClaimed"] == True: log.info( f"\033[1;34m[{self.account_name}] Daily keys mini game already claimed.\033[0m" ) return if AccountConfigData["dailyKeysMiniGame"]["remainSecondsToNextAttempt"] > 0: log.info(f"[{self.account_name}] Daily keys mini game is on cooldown...") return ## check timer. url = "https://api.hamsterkombatgame.io/clicker/start-keys-minigame" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request response = await self.HttpRequest(url, headers, "POST", 200) if response is None: log.error(f"[{self.account_name}] Unable to start mini game.") return if "dailyKeysMiniGame" not in response: log.error(f"[{self.account_name}] Unable to get daily keys mini game.") return if response["dailyKeysMiniGame"]["isClaimed"] == True: log.info( f"\033[1;34m[{self.account_name}] Daily keys mini game already claimed.\033[0m" ) return if "remainSecondsToGuess" not in response["dailyKeysMiniGame"]: log.error(f"[{self.account_name}] Unable to get daily keys mini game.") return waitTime = int( response["dailyKeysMiniGame"]["remainSecondsToGuess"] - random.randint(8, 15) ) if waitTime < 0: log.error(f"[{self.account_name}] Unable to claim mini game.") return log.info( f"[{self.account_name}] Waiting for {waitTime} seconds, Mini-game will be completed in {waitTime} seconds..." ) await asyncio.sleep(waitTime) url = "https://api.hamsterkombatgame.io/clicker/claim-daily-keys-minigame" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } cipher = ( ("0" + str(waitTime) + str(random.randint(10000000000, 99999999999)))[:10] + "|" + str(AccountID) ) cipher_base64 = base64.b64encode(cipher.encode()).decode() payload = json.dumps( { "cipher": cipher_base64, } ) # Send POST request response = await self.HttpRequest(url, headers, "POST", 200, payload) if response is None: log.error(f"[{self.account_name}] Unable to claim mini game.") return log.info(f"[{self.account_name}] Mini game claimed successfully.") async def StartPlaygroundGame(self): if not self.config["auto_playground_games"]: log.info(f"[{self.account_name}] Playground games are disabled.") return log.info(f"[{self.account_name}] Starting getting playground games...") url = "https://api.hamsterkombatgame.io/clicker/get-promos" headers = { "Access-Control-Request-Headers": "authorization", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Authorization": self.Authorization, } # Send POST request response = await self.HttpRequest(url, headers, "POST", 200) if response is None: log.error(f"[{self.account_name}] Unable to get playground games.") return if "promos" not in response: log.error(f"[{self.account_name}] Unable to get playground games.") return promo_count = 0 for promo in response["promos"]: if promo["promoId"] not in SupportedPromoGames: log.warning( f"[{self.account_name}] Detected unknown playground game: {promo['title']['en']}. Check project github for updates." ) continue if await self.CheckPlayGroundGameState(promo, response): promoData = SupportedPromoGames[promo["promoId"]] promo_count += 1 if self.GetConfig( "max_promo_games_per_round", 3 ) != 0 and promo_count > self.GetConfig("max_promo_games_per_round", 3): log.info( f"[{self.account_name}] Maximum number of playground games reached. We will retrieve other games in the next run." ) return log.info( f"[{self.account_name}] Starting {promoData['name']} Playground game..." ) await asyncio.sleep(1) promoCode = await self.GetPlayGroundGameKey(promoData) if promoCode is not None: log.info( f"\033[1;34m[{self.account_name}] {promoData['name']} key: {promoCode}\033[0m" ) await asyncio.sleep(2) log.info(f"[{self.account_name}] Claiming {promoData['name']}...") await self.ClaimPlayGroundGame(promoCode) log.info( f"[{self.account_name}] {promoData['name']} claimed successfully." ) async def ClaimPlayGroundGame(self, promoCode): url = "https://api.hamsterkombatgame.io/clicker/apply-promo" headers = { "Access-Control-Request-Headers": "authorization,content-type", "Access-Control-Request-Method": "POST", } # Send OPTIONS request await self.HttpRequest(url, headers, "OPTIONS", 204) headers = { "Accept": "application/json", "Authorization": self.Authorization, "Content-Type": "application/json", } payload = json.dumps( { "promoCode": promoCode, } ) # Send POST request return await self.HttpRequest(url, headers, "POST", 200, payload) async def GetPlayGroundGameKey(self, promoData): appToken = promoData["appToken"] clientId = f"{int(time.time() * 1000)}-{''.join(str(random.randint(0, 9)) for _ in range(19))}" if "clientIdType" in promoData and promoData["clientIdType"] == "32str": clientId = "".join( random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=32) ) if "clientIdType" in promoData and promoData["clientIdType"] == "uuid": clientId = str(uuid.uuid4()) log.info(f"[{self.account_name}] Getting {promoData['name']} key...") url = "https://api.gamepromo.io/promo/login-client" headers_option = { "Host": "api.gamepromo.io", "Origin": "", "Referer": "", "access-control-request-headers": "content-type", "access-control-request-method": "POST", } headers_post = { "Host": "api.gamepromo.io", "Origin": "", "Referer": "", "Content-Type": "application/json; charset=utf-8", } if "userAgent" in promoData and promoData["userAgent"] != None: headers_post["User-Agent"] = promoData["userAgent"] headers_option["User-Agent"] = promoData["userAgent"] if "x-unity-version" in promoData and promoData["x-unity-version"] != None: headers_post["X-Unity-Version"] = promoData["x-unity-version"] headers_option["X-Unity-Version"] = promoData["x-unity-version"] await self.HttpRequest(url, headers_option, "OPTIONS", 204, True) payloadData = { "appToken": appToken, "clientId": clientId, "clientOrigin": promoData["clientOrigin"], } if "clientVersion" in promoData and promoData["clientVersion"] != None: payloadData["clientVersion"] = promoData["clientVersion"] payload = json.dumps(payloadData) response = await self.HttpRequest(url, headers_post, "POST", 200, payload) if response is None: log.error(f"[{self.account_name}] Unable to get {promoData['name']} key.") return None if "clientToken" not in response: log.error(f"[{self.account_name}] Unable to get {promoData['name']} key.") return None clientToken = response["clientToken"] TimeSleep = promoData["delay"] + random.randint(1, 5) log.info(f"[{self.account_name}] Waiting for {TimeSleep} seconds...") await asyncio.sleep(TimeSleep) log.info( f"[{self.account_name}] Registering event for {promoData['name']} (This may take a while ~5-20 minutes)..." ) url = "https://api.gamepromo.io/promo/register-event" headers_post["Authorization"] = f"Bearer {clientToken}" response = None retryCount = 0 while retryCount < 15: retryCount += 1 eventID = str(uuid.uuid4()) if "eventIdType" in promoData: if promoData["eventIdType"] == "uuid": eventID = str(uuid.uuid4()) else: eventID = promoData["eventIdType"] headers_option["access-control-request-headers"] = ( "authorization,content-type" ) await self.HttpRequest(url, headers_option, "OPTIONS", 204, True) PayloadData = { "promoId": promoData["promoId"], "eventId": eventID, "eventOrigin": promoData["eventOrigin"], } if "eventType" in promoData and promoData["eventType"] != None: PayloadData["eventType"] = promoData["eventType"] payload = json.dumps(PayloadData) response = await self.HttpRequest(url, headers_post, "POST", 200, payload, True) if response is None or not isinstance(response, dict): await asyncio.sleep(promoData["retry_delay"] + random.randint(1, 5)) continue if not response.get("hasCode", False): await asyncio.sleep(promoData["retry_delay"] + random.randint(1, 5)) continue break if ( response is None or not isinstance(response, dict) or "hasCode" not in response ): log.error(f"[{self.account_name}] Unable to register event.") return None log.info(f"[{self.account_name}] Event registered successfully.") url = "https://api.gamepromo.io/promo/create-code" headers_option["access-control-request-headers"] = "authorization,content-type" await self.HttpRequest(url, headers_option, "OPTIONS", 204, True) payload = json.dumps( { "promoId": promoData["promoId"], } ) response = await self.HttpRequest(url, headers_post, "POST", 200, payload) if response is None: log.error(f"[{self.account_name}] Unable to get {promoData['name']} key.") return None if ( "promoCode" not in response or response.get("promoCode") is None or response.get("promoCode") == "" ): log.error(f"[{self.account_name}] Unable to get {promoData['name']} key.") return None promoCode = response["promoCode"] return promoCode async def CheckPlayGroundGameState(self, promo, promos): if not self.config["auto_playground_games"]: log.info(f"[{self.account_name}] Playground games are disabled.") return False if "states" not in promos: return True for state in promos["states"]: if ( state["promoId"] == promo["promoId"] and state["receiveKeysToday"] >= promo["keysPerDay"] ): log.info( f"\033[1;34m[{self.account_name}] Playground game {SupportedPromoGames[promo['promoId']]['name']} already claimed.\033[0m" ) return False return True async def Playing(self): log.info(f"[{self.account_name}] Starting account...") log.info(f"[{self.account_name}] Getting basic account data...") AccountBasicData = await self.AccountInfoTelegramRequest() if ( AccountBasicData is None or AccountBasicData is False or "accountInfo" not in AccountBasicData or "id" not in AccountBasicData["accountInfo"] ): log.error(f"[{self.account_name}] Unable to get account basic data.") return log.info( f"\033[1;35m[{self.account_name}] Account ID: {AccountBasicData['accountInfo']['id']}, Account Name: {AccountBasicData['accountInfo']['name']}\033[0m" ) log.info(f"[{self.account_name}] Getting account config data...") AccountConfigVersionData = None if self.configVersion != "": AccountConfigVersionData = await self.GetAccountConfigVersionRequest() log.info( f"[{self.account_name}] Account config version: {self.configVersion}" ) AccountConfigData = await self.GetAccountConfigRequest() if AccountConfigData is None or AccountConfigData is False: log.error(f"[{self.account_name}] Unable to get account config data.") return DailyCipher = "" if ( self.config["auto_get_daily_cipher"] and "dailyCipher" in AccountConfigData and "cipher" in AccountConfigData["dailyCipher"] ): log.info(f"[{self.account_name}] Decoding daily cipher...") DailyCipher = DailyCipherDecode(AccountConfigData["dailyCipher"]["cipher"]) MorseCode = TextToMorseCode(DailyCipher) log.info( f"\033[1;34m[{self.account_name}] Daily cipher: {DailyCipher} and Morse code: {MorseCode}\033[0m" ) log.info(f"[{self.account_name}] Getting account data...") getAccountDataStatus = await self.getAccountData() if getAccountDataStatus is False: return log.info( f"[{self.account_name}] Account Balance Coins: {number_to_string(self.balanceCoins)}, Available Taps: {self.availableTaps}, Max Taps: {self.maxTaps}, Total Keys: {self.totalKeys}, Balance Keys: {self.balanceKeys}" ) log.info(f"[{self.account_name}] Getting account upgrades...") upgradesResponse = await self.UpgradesForBuyRequest() if upgradesResponse is None: log.error(f"[{self.account_name}] Failed to get upgrades list.") return log.info(f"[{self.account_name}] Getting account tasks...") tasksResponse = await self.ListTasksRequest() if tasksResponse is None: log.error(f"[{self.account_name}] Failed to get tasks list.") log.info(f"[{self.account_name}] Getting account airdrop tasks...") airdropTasksResponse = await self.GetListAirDropTasksRequest() if airdropTasksResponse is None: log.error(f"[{self.account_name}] Failed to get airdrop tasks list.") log.info(f"[{self.account_name}] Getting account IP...") ipResponse = await self.IPRequest() if ipResponse is None: log.error(f"[{self.account_name}] Failed to get IP.") return log.info(f"[{self.account_name}] Getting account skins...") SkinsData = await self.GetSkins() if SkinsData is None: log.error(f"[{self.account_name}] Failed to get skins.") log.info( f"[{self.account_name}] IP: {ipResponse['ip']} Company: {ipResponse['asn_org']} Country: {ipResponse['country_code']}" ) if self.config["auto_finish_mini_game"]: log.info(f"[{self.account_name}] Attempting to finish mini game...") await asyncio.sleep(1) await self.StartMiniGame(AccountConfigData, AccountBasicData["accountInfo"]["id"]) # Start tapping if self.config["auto_tap"]: log.info(f"[{self.account_name}] Starting to tap...") await asyncio.sleep(2) await self.TapRequest(self.availableTaps) log.info(f"[{self.account_name}] Tapping completed successfully.") if self.config["auto_get_daily_cipher"] and DailyCipher != "": if AccountConfigData["dailyCipher"]["isClaimed"] == True: log.info( f"\033[1;34m[{self.account_name}] Daily cipher already claimed.\033[0m" ) else: log.info(f"[{self.account_name}] Attempting to claim daily cipher...") await asyncio.sleep(2) await self.ClaimDailyCipherRequest(DailyCipher) log.info(f"[{self.account_name}] Daily cipher claimed successfully.") if ( self.config["auto_get_daily_task"] and tasksResponse is not None and "tasks" in tasksResponse and isinstance(tasksResponse["tasks"], list) ): log.info(f"[{self.account_name}] Checking for daily task...") streak_days = None for task in tasksResponse["tasks"]: if task["id"] == "streak_days": streak_days = task break if streak_days is None: log.error(f"[{self.account_name}] Failed to get daily task.") return if streak_days["isCompleted"] == True: log.info( f"\033[1;34m[{self.account_name}] Daily task already completed.\033[0m" ) else: log.info(f"[{self.account_name}] Attempting to complete daily task...") day = streak_days["days"] rewardCoins = streak_days["rewardCoins"] await asyncio.sleep(2) await self.CheckTaskRequest("streak_days") log.info( f"[{self.account_name}] Daily task completed successfully, Day: {day}, Reward coins: {number_to_string(rewardCoins)}" ) if ( self.config["auto_get_task"] and tasksResponse is not None and "tasks" in tasksResponse and isinstance(tasksResponse["tasks"], list) ): log.info(f"[{self.account_name}] Checking for available task...") selected_task = None for task in tasksResponse["tasks"]: TaskType = task.get("type", "") if task["isCompleted"] == False and ( TaskType == "WithLink" or TaskType == "WithLocaleLink" ): log.info( f"[{self.account_name}] Attempting to complete Youtube Or Twitter task..." ) selected_task = task["id"] rewardCoins = task["rewardCoins"] await asyncio.sleep(2) await self.CheckTaskRequest(selected_task) log.info( f"[{self.account_name}] Task completed - id: {selected_task}, Reward coins: {number_to_string(rewardCoins)}" ) if selected_task is None: log.info(f"\033[1;34m[{self.account_name}] Tasks already done\033[0m") # Start buying free tap boost if ( self.config["auto_tap"] and self.config["auto_free_tap_boost"] and await self.BuyFreeTapBoostIfAvailable() ): log.info(f"[{self.account_name}] Starting to tap with free boost...") await asyncio.sleep(2) await self.TapRequest(self.availableTaps) log.info( f"[{self.account_name}] Tapping with free boost completed successfully." ) # Start Syncing account data after tapping if self.config["auto_tap"]: await self.getAccountData() log.info( f"[{self.account_name}] Account Balance Coins: {number_to_string(self.balanceCoins)}, Available Taps: {self.availableTaps}, Max Taps: {self.maxTaps}, Total Keys: {self.totalKeys}, Balance Keys: {self.balanceKeys}" ) await self.StartPlaygroundGame() # Start buying upgrades if not self.config["auto_upgrade"]: log.error(f"[{self.account_name}] Auto upgrade is disabled.") return self.ProfitPerHour = 0 self.SpendTokens = 0 if self.config["wait_for_best_card"]: while True: if not await self.BuyBestCard(): break await self.getAccountData() log.info( f"[{self.account_name}] Final account balance: {number_to_string(self.balanceCoins)} coins, Your profit per hour is {number_to_string(self.earnPassivePerHour)} (+{number_to_string(self.ProfitPerHour)}), Spent: {number_to_string(self.SpendTokens)}" ) return if self.balanceCoins < self.config["auto_upgrade_start"]: log.warning( f"[{self.account_name}] Balance is too low to start buying upgrades." ) return while self.balanceCoins >= self.config["auto_upgrade_min"]: log.info(f"[{self.account_name}] Checking for upgrades...") await asyncio.sleep(2) upgradesResponse = await self.UpgradesForBuyRequest() if upgradesResponse is None: log.warning(f"[{self.account_name}] Failed to get upgrades list.") return upgrades = [ item for item in upgradesResponse["upgradesForBuy"] if not item["isExpired"] and item["isAvailable"] and item["profitPerHourDelta"] > 0 and ("cooldownSeconds" not in item or item["cooldownSeconds"] == 0) ] if len(upgrades) == 0: log.warning(f"[{self.account_name}] No upgrades available.") return balanceCoins = int(self.balanceCoins) log.info(f"[{self.account_name}] Searching for the best upgrades...") selected_upgrades = SortUpgrades(upgrades, balanceCoins) if len(selected_upgrades) == 0: log.warning(f"[{self.account_name}] No upgrades available.") return current_selected_card = selected_upgrades[0] log.info( f"[{self.account_name}] Best upgrade is {current_selected_card['name']} with profit {current_selected_card['profitPerHourDelta']} and price {number_to_string(current_selected_card['price'])}, Level: {current_selected_card['level']}" ) balanceCoins -= current_selected_card["price"] if balanceCoins <= self.config["auto_upgrade_min"]: log.warning( f"[{self.account_name}] Upgrade purchase would decrease balance below minimum limit, aborting." ) return log.info(f"[{self.account_name}] Attempting to buy an upgrade...") await asyncio.sleep(2) upgradesResponse = await self.BuyUpgradeRequest(current_selected_card["id"]) if upgradesResponse is None: log.error(f"[{self.account_name}] Failed to buy an upgrade.") return log.info(f"[{self.account_name}] Upgrade bought successfully") await asyncio.sleep(5) self.balanceCoins = balanceCoins self.ProfitPerHour += current_selected_card["profitPerHourDelta"] self.SpendTokens += current_selected_card["price"] self.earnPassivePerHour += current_selected_card["profitPerHourDelta"] log.info(f"[{self.account_name}] Upgrades purchase completed successfully.") await self.getAccountData() log.info( f"[{self.account_name}] Final account balance: {number_to_string(self.balanceCoins)} coins, Your profit per hour is {number_to_string(self.earnPassivePerHour)} (+{number_to_string(self.ProfitPerHour)}), Spent: {number_to_string(self.SpendTokens)}" ) async def Start(self): while True: await self.Playing() if AccountsRecheckTime < 1 and MaxRandomDelay < 1: log.error( "AccountsRecheckTime and MaxRandomDelay values are set to 0, bot will close now." ) return if MaxRandomDelay > 0: randomDelay = random.randint(1, MaxRandomDelay) log.error(f"[{self.account_name}] Sleeping for {randomDelay} seconds because of random delay...") await asyncio.sleep(randomDelay) if AccountsRecheckTime > 0: log.error(f"[{self.account_name}] Rechecking account in {AccountsRecheckTime} seconds...") await asyncio.sleep(AccountsRecheckTime) async def RunAccounts(): accounts = [] for account in AccountList: accounts.append(HamsterKombatAccount(account)) log.info("\033[1;33mStarting all accounts...\033[0m") await asyncio.gather(*(account.Start() for account in accounts)) async def main(): log.info("------------------------------------------------------------------------") log.info("------------------------------------------------------------------------") log.info("\033[1;32mWelcome to [Master Hamster Kombat] Auto farming bot...\033[0m") log.info( "\033[1;34mProject Github: https://github.com/masterking32/MasterHamsterKombatBot\033[0m" ) log.info("\033[1;33mDeveloped by: MasterkinG32\033[0m") log.info("\033[1;35mVersion: 2.3\033[0m") log.info("\033[1;36mTo stop the bot, press Ctrl + C\033[0m") log.info("------------------------------------------------------------------------") log.info("------------------------------------------------------------------------") await asyncio.sleep(2) try: await RunAccounts() except KeyboardInterrupt: log.error("Stopping Master Hamster Kombat Auto farming bot...") if __name__ == "__main__": asyncio.run(main()) ```

[afdah]

You can create several folders with bots, fill the config file of each bot with different accounts and run each bot and check my assumption. it's stupid and inconvenient but to check why not?

i got 8 accounts with individual proxy each. 4 bots with 2 accounts each. its working but if i intend to add more accounts, it will be a hassle.

I made some mistakes in the body of the main method, here is the corrected version

upd: Most likely, there is an error in the main method where previously there were return and now continue because of which the loop will start from the beginning and not reach the end. as I said before, you need to rewrite not a small part of the script logic to work correctly.

upd2: Another attempt to fix my shitty code to a working state under a spoiler

main_async.py

i got your script to work. 8 accounts with individual proxy. i assume its running 8 accounts in parallel.

i was looking at thread pool. where i can set my max workers to 2 so i can control the number of bots running in parallel. i'm not sure if i can do that in asyncio. https://www.geeksforgeeks.org/multithreading-python-set-1/

[camelot-rus]

It's very simple! If you have, say, 50 accounts, just distribute 10 each. copy the original script folder 5 times, and add 10 accounts to each separate folder, and run 5 different scripts. that's all. you will have 10 accounts working in 5 different processes

[tboy1337]

I made some mistakes in the body of the main method, here is the corrected version

upd: Most likely, there is an error in the main method where previously there were return and now continue because of which the loop will start from the beginning and not reach the end. as I said before, you need to rewrite not a small part of the script logic to work correctly.

upd2: Another attempt to fix my shitty code to a working state under a spoiler

main_async.py

If its fully working can you create a pull request?

[Fy0urM]

If its fully working can you create a pull request?

I do not plan to commit this version, if the author of the repository wants he will add my asynchronous version himself. I believe that I do not have the moral right to interfere so much with the original project and do not plan to further support this asynchronous version. this is just an example of a theoretically working variant but since I do not have several accounts and enough proxies to check the full correctness of this variant.

[vashamamasha2014]

i got 8 accounts with individual proxy each. 4 bots with 2 accounts each. its working but if i intend to add more accounts, it will be a hassle.

Can you please share proxy addresses or instructions on where to get working addresses?

[tboy1337]

i got 8 accounts with individual proxy each. 4 bots with 2 accounts each. its working but if i intend to add more accounts, it will be a hassle.

Can you please share proxy addresses or instructions on where to get working addresses?

https://github.com/masterking32/MasterHamsterKombatBot/blob/6323c888a0e00165f1dcd1fc7d367d813be36575/README.MD?plain=1#L89

[MeninSun]

It's very simple! If you have, say, 50 accounts, just distribute 10 each. copy the original script folder 5 times, and add 10 accounts to each separate folder, and run 5 different scripts. that's all. you will have 10 accounts working in 5 different processes

i think this is best solution for now, if using linux/vps you can using multiple screen to run it, or using service (systemd), you can see here #12

me using too: