forked from xBabylonia/KoniStory
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
190 lines (159 loc) · 8.43 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
import aiohttp
import asyncio
from loguru import logger
import json
import logging
from colorama import init, Fore, Back, Style
from datetime import datetime
init(autoreset=True)
class TheOrder:
def __init__(self):
self.base_url = "https://sp-odyssey-api.playnation.app/api"
self.session = None
self.headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "id-ID,id;q=0.9,en-US;q=0.8,en;q=0.7",
"Content-Type": "application/json",
"Origin": "https://story-protocol-odyssey-tele.playnation.app",
"Referer": "https://story-protocol-odyssey-tele.playnation.app/",
"Sec-Ch-Ua": '"Not A;Brand";v="99", "Chromium";v="124", "Google Chrome";v="124"',
"Sec-Ch-Ua-Mobile": "?1",
"Sec-Ch-Ua-Platform": '"Android"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"User-Agent": "Mozilla/5.0 (Linux; U; Android 4.0.2; en-us; Galaxy Nexus Build/ICL53F) AppleWebKit/534.30 (KHTML, like Gecko) Version/4.0 Mobile Safari/534.30"
}
logger.remove()
logger.add(
lambda msg: print(msg, end=""),
format="<green>{time:HH:mm:ss}</green> | <level>{level}</level> | {message}",
colorize=True,
enqueue=True,
backtrace=False,
diagnose=False
)
def print_banner(self):
banner = f"""{Fore.CYAN}
┏━━━━┳┓ ┏━━━┓ ┏┓ ┏━━━┓ ┏┓ ┏━┓ Auto KoniStory Bot
┃┏┓┏┓┃┃ ┃┏━┓┃ ┃┃ ┃┏━┓┃ ┃┃ ┃┏┛ created by @yogschannel
┗┛┃┃┗┫┗━┳━━┓┃┃ ┃┣━┳━┛┣━━┳━┓┃┃ ┃┣━━┫┗━┳━━┳┛┗┓ @AIOTelegramManager
┃┃ ┃┏┓┃┃━┫┃┃ ┃┃┏┫┏┓┃┃━┫┏┛┃┗━┛┃━━┫┏┓┃┏┓┣┓┏┛
┃┃ ┃┃┃┃┃━┫┃┗━┛┃┃┃┗┛┃┃━┫┃ ┃┏━┓┣━━┃┃┃┃┏┓┃┃┃
┗┛ ┗┛┗┻━━┛┗━━━┻┛┗━━┻━━┻┛ ┗┛ ┗┻━━┻┛┗┻┛┗┛┗┛
{Style.RESET_ALL}
"""
print(banner)
async def create_session(self):
if not self.session:
self.session = aiohttp.ClientSession(headers=self.headers)
async def close_session(self):
if self.session:
await self.session.close()
self.session = None
async def get_auth(self, init_data, address):
try:
payload = {
"address": address,
"referralCode": "C20WJXJv8",
"initData": init_data
}
async with self.session.post(f"{self.base_url}/account/login", json=payload) as response:
if response.status != 200:
return None
data = await response.json()
return {
"token": data["token"],
"username": data["info"]["telegramUsername"],
"points": data["attributes"]["point"]
}
except Exception as e:
logger.opt(colors=True).error(f"<red>Authentication error:</red> {str(e)}")
return None
async def get_tasks(self, token):
try:
headers = {**self.headers, "Authorization": f"Bearer {token}"}
async with self.session.get(f"{self.base_url}/task/history", headers=headers) as response:
if response.status == 200:
tasks = await response.json()
return [task["id"] for task in tasks if not task.get("submitted", False)]
return []
except Exception as e:
logger.opt(colors=True).error(f"<red>Error getting tasks:</red> {str(e)}")
return []
async def clear_task(self, token, task_id):
try:
headers = {**self.headers, "Authorization": f"Bearer {token}"}
payload = {
"taskId": task_id,
"extrinsicHash": "",
"network": ""
}
async with self.session.post(f"{self.base_url}/task/submit", json=payload, headers=headers) as response:
response_data = await response.json()
if response.status == 200:
return True, None
elif response.status == 400 and "Task already submitted" in str(response_data.get("error", "")):
return True, "already_submitted"
return False, None
except Exception as e:
logger.opt(colors=True).error(f"<red>Error clearing task:</red> {str(e)}")
return False, None
async def process_account(self, init_data, address):
try:
auth_data = await self.get_auth(init_data, address)
if not auth_data:
logger.opt(colors=True).error("<red>Failed to authenticate account</red>")
return False
logger.opt(colors=True).info(f"<cyan>Successfully logged in as {auth_data['username']} | Points: {auth_data['points']}</cyan>")
tasks = await self.get_tasks(auth_data['token'])
if not tasks:
logger.opt(colors=True).info(f"<yellow>No pending tasks for {auth_data['username']}</yellow>")
return True
for task_id in tasks:
success, status = await self.clear_task(auth_data['token'], task_id)
if success:
msg = "Task already completed" if status == "already_submitted" else "Successfully cleared task"
logger.opt(colors=True).info(f"<yellow>{msg}</yellow> {task_id} for {auth_data['username']}")
else:
logger.opt(colors=True).warning(f"<red>Failed to clear task</red> {task_id} for {auth_data['username']}")
await asyncio.sleep(2)
return True
except Exception as e:
logger.opt(colors=True).error(f"<red>Error processing account:</red> {str(e)}")
return False
async def run_forever(self):
iteration = 1
TWO_HOURS = 2 * 60 * 60 # 2 hours in seconds
while True:
try:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logger.opt(colors=True).info(f"\n<cyan>Starting iteration {iteration} at {current_time}</cyan>")
self.print_banner()
await self.create_session()
with open('data.txt', 'r') as f_data, open('address.txt', 'r') as f_address:
accounts = f_data.read().splitlines()
addresses = f_address.read().splitlines()
total_accounts = len(accounts)
for i, (account, address) in enumerate(zip(accounts, addresses), 1):
logger.opt(colors=True).info(f"<cyan>Processing account {i}/{total_accounts}</cyan>")
success = await self.process_account(account, address)
if i < total_accounts:
status = "Successfully processed" if success else "Failed to process"
logger.opt(colors=True).info(f"<yellow>{status} account. Waiting 3 seconds...</yellow>")
await asyncio.sleep(3)
logger.opt(colors=True).success(f"<green>Finished processing all {total_accounts} accounts in iteration {iteration}</green>")
await self.close_session()
next_run_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logger.opt(colors=True).info(f"<yellow>Waiting 2 hours before next iteration. Next run at: {next_run_time}</yellow>")
await asyncio.sleep(TWO_HOURS)
iteration += 1
except Exception as e:
logger.opt(colors=True).error(f"<red>Error in iteration {iteration}:</red> {str(e)}")
logger.opt(colors=True).info("<yellow>Attempting to restart in 60 seconds...</yellow>")
await asyncio.sleep(60)
continue
if __name__ == "__main__":
bot = TheOrder()
asyncio.run(bot.run_forever())