-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathbot.py
356 lines (311 loc) · 13.4 KB
/
bot.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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
import os
import sys
import json
import time
import random
import argparse
import requests
from base64 import b64decode, urlsafe_b64decode
from datetime import datetime
from urllib.parse import parse_qs
from colorama import init, Fore, Style
merah = Fore.LIGHTRED_EX
kuning = Fore.LIGHTYELLOW_EX
hijau = Fore.LIGHTGREEN_EX
biru = Fore.LIGHTBLUE_EX
putih = Fore.LIGHTWHITE_EX
hitam = Fore.LIGHTBLACK_EX
reset = Style.RESET_ALL
line = putih + "~" * 50
class Tomartod:
def __init__(self):
self.headers = {
"host": "api-web.tomarket.ai",
"connection": "keep-alive",
"accept": "application/json, text/plain, */*",
"user-agent": "Mozilla/5.0 (Linux; Android 10; Redmi 4A / 5A Build/QQ3A.200805.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/86.0.4240.185 Mobile Safari/537.36",
"content-type": "application/json",
"origin": "https://mini-app.tomarket.ai",
"x-requested-with": "tw.nekomimi.nekogram",
"sec-fetch-site": "same-site",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
"referer": "https://mini-app.tomarket.ai/",
"accept-language": "en-US,en;q=0.9",
}
self.marinkitagawa = lambda data: {
key: value[0] for key, value in parse_qs(data).items()
}
def set_proxy(self, proxy=None):
self.ses = requests.Session()
if proxy is not None:
self.ses.proxies.update({"http": proxy, "https": proxy})
def set_authorization(self, auth):
self.headers["authorization"] = auth
def del_authorization(self):
if "authorization" in self.headers.keys():
self.headers.pop("authorization")
def login(self, data):
url = "https://api-web.tomarket.ai/tomarket-game/v1/user/login"
data = json.dumps(
{
"init_data": data,
"invite_code": "",
}
)
self.del_authorization()
res = self.http(url, self.headers, data)
if res.status_code != 200:
self.log(f"{merah}failed fetch token authorization, check http.log !")
return None
data = res.json().get("data")
token = data.get("access_token")
if token is None:
self.log(f"{merah}failed fetch token authorization, check http.log !")
return None
return token
def start_farming(self):
data = json.dumps({"game_id": "53b22103-c7ff-413d-bc63-20f6fb806a07"})
url = "https://api-web.tomarket.ai/tomarket-game/v1/farm/start"
res = self.http(url, self.headers, data)
if res.status_code != 200:
self.log(f"{merah}failed start farming, check http.log last line !")
return False
data = res.json().get("data")
end_farming = data["end_at"]
format_end_farming = (
datetime.fromtimestamp(end_farming).isoformat(" ").split(".")[0]
)
self.log(f"{hijau}success start farming !")
def end_farming(self):
data = json.dumps({"game_id": "53b22103-c7ff-413d-bc63-20f6fb806a07"})
url = "https://api-web.tomarket.ai/tomarket-game/v1/farm/claim"
res = self.http(url, self.headers, data)
if res.status_code != 200:
self.log(f"{merah}failed start farming, check http.log last line !")
return False
poin = res.json()["data"]["claim_this_time"]
self.log(f"{hijau}success claim farming !")
self.log(f"{hijau}reward : {putih}{poin}")
def daily_claim(self):
url = "https://api-web.tomarket.ai/tomarket-game/v1/daily/claim"
data = json.dumps({"game_id": "fa873d13-d831-4d6f-8aee-9cff7a1d0db1"})
res = self.http(url, self.headers, data)
if res.status_code != 200:
self.log(f"{merah}failed claim daily sign,check http.log last line !")
return False
data = res.json().get("data")
if isinstance(data, str):
self.log(f"{kuning}maybe already sign in")
return
poin = data.get("today_points")
self.log(
f"{hijau}success claim {biru}daily sign {hijau}reward : {putih}{poin} !"
)
return
def play_game_func(self, amount_pass):
data_game = json.dumps({"game_id": "59bcd12e-04e2-404c-a172-311a0084587d"})
start_url = "https://api-web.tomarket.ai/tomarket-game/v1/game/play"
claim_url = "https://api-web.tomarket.ai/tomarket-game/v1/game/claim"
for i in range(amount_pass):
res = self.http(start_url, self.headers, data_game)
if res.status_code != 200:
self.log(f"{merah}failed start game !")
return
self.log(f"{hijau}success {biru}start{hijau} game !")
self.countdown(30)
point = random.randint(self.game_low_point, self.game_high_point)
data_claim = json.dumps(
{"game_id": "59bcd12e-04e2-404c-a172-311a0084587d", "points": point}
)
res = self.http(claim_url, self.headers, data_claim)
if res.status_code != 200:
self.log(f"{merah}failed claim game point !")
continue
self.log(f"{hijau}success {biru}claim{hijau} game point : {putih}{point}")
def get_balance(self):
url = "https://api-web.tomarket.ai/tomarket-game/v1/user/balance"
while True:
res = self.http(url, self.headers, "")
if res.status_code != 200:
self.log(f"{merah}failed fetch balance !")
continue
data = res.json().get("data")
if data is None:
self.log(f"{merah}failed get data !")
return None
timestamp = data["timestamp"]
balance = data["available_balance"]
self.log(f"{hijau}balance : {putih}{balance}")
if "daily" not in data.keys():
self.daily_claim()
continue
if data["daily"] is None:
self.daily_claim()
continue
next_daily = data["daily"]["next_check_ts"]
if timestamp > next_daily:
self.daily_claim()
if "farming" not in data.keys():
self.log(f"{kuning}farming not started !")
result = self.start_farming()
continue
end_farming = data["farming"]["end_at"]
format_end_farming = (
datetime.fromtimestamp(end_farming).isoformat(" ").split(".")[0]
)
if timestamp > end_farming:
self.end_farming()
continue
self.log(f"{kuning}not time to claim !")
self.log(f"{kuning}end farming at : {putih}{format_end_farming}")
if self.play_game:
self.log(f"{hijau}auto play game is enable !")
play_pass = data.get("play_passes")
self.log(f"{hijau}game ticket : {putih}{play_pass}")
if int(play_pass) > 0:
self.play_game_func(play_pass)
continue
_next = end_farming - timestamp
return _next + random.randint(self.add_time_min, self.add_time_max)
def load_data(self, file):
datas = [i for i in open(file).read().splitlines() if len(i) > 0]
if len(datas) <= 0:
print(
f"{merah}0 account detected from {file}, fill your data in {file} first !{reset}"
)
sys.exit()
return datas
def load_config(self, file):
config = json.loads(open(file).read())
self.interval = config["interval"]
self.play_game = config["play_game"]
self.game_low_point = config["game_point"]["low"]
self.game_high_point = config["game_point"]["high"]
self.add_time_min = config["additional_time"]["min"]
self.add_time_max = config["additional_time"]["max"]
def save(self, id, token):
tokens = json.loads(open("tokens.json").read())
tokens[str(id)] = token
open("tokens.json", "w").write(json.dumps(tokens, indent=4))
def get(self, id):
tokens = json.loads(open("tokens.json").read())
if str(id) not in tokens.keys():
return None
return tokens[str(id)]
def is_expired(self, token):
header, payload, sign = token.split(".")
deload = urlsafe_b64decode(payload + "==").decode()
jeload = json.loads(deload)
now = int(datetime.now().timestamp())
if now > jeload["exp"]:
return True
return False
def http(self, url, headers, data=None):
while True:
try:
now = datetime.now().isoformat(" ").split(".")[0]
if data is None:
res = self.ses.get(url, headers=headers, timeout=100)
elif data == "":
res = self.ses.post(url, headers=headers, timeout=100)
else:
res = self.ses.post(url, headers=headers, data=data, timeout=100)
open("http.log", "a", encoding="utf-8").write(
f"{now} - {res.status_code} - {res.text}\n"
)
return res
except requests.exceptions.ProxyError:
print(f"{merah}bad proxy !")
time.sleep(1)
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
print(f"{merah}connection error / connection timeout !")
time.sleep(1)
continue
def countdown(self, t):
for i in range(t, 0, -1):
menit, detik = divmod(i, 60)
jam, menit = divmod(menit, 60)
jam = str(jam).zfill(2)
menit = str(menit).zfill(2)
detik = str(detik).zfill(2)
print(f"{putih}waiting {jam}:{menit}:{detik} ", flush=True, end="\r")
time.sleep(1)
print(" ", flush=True, end="\r")
def log(self, msg):
now = datetime.now().isoformat(" ").split(".")[0]
print(f"{hitam}[{now}]{reset} {msg}{reset}")
def main(self):
banner = f"""
{hijau}Auto Claim {biru}Tomarket_ai
_ _
/\ /\___| | | ___
/ /_/ / _ \ | |/ _ \
/ __ / __/ | | (_) |
\/ /_/ \___|_|_|\___/
__ __ _ _
/ / /\ \ \___ _ __| | __| |
\ \/ \/ / _ \| '__| |/ _` |
\ /\ / (_) | | | | (_| |
\/ \/ \___/|_| |_|\__,_|
Tomarket BOT V 1.1
Prepared and Developed by: https://github.com/kucingeros
{hijau}Message: {putih}dont't forget to 'git pull' maybe the script is updated
"""
arg = argparse.ArgumentParser()
arg.add_argument("--data", default="data.txt")
arg.add_argument("--config", default="config.json")
arg.add_argument("--proxy", default="proxies.txt")
arg.add_argument("--marinkitagawa", action="store_true")
args = arg.parse_args()
if not args.marinkitagawa:
os.system("cls" if os.name == "nt" else "clear")
print(banner)
self.load_config(args.config)
datas = self.load_data(args.data)
proxies = open(args.proxy).read().splitlines()
self.log(f"{biru}total account : {putih}{len(datas)}")
self.log(f"{biru}total proxies detected : {putih}{len(proxies)}")
use_proxy = True if len(proxies) > 0 else False
self.log(f"{hijau}use proxy : {putih}{use_proxy}")
print(line)
while True:
list_countdown = []
_start = int(time.time())
for no, data in enumerate(datas):
if use_proxy:
proxy = proxies[no % len(proxies)]
self.set_proxy(proxy if use_proxy else None)
parser = self.marinkitagawa(data)
user = json.loads(parser["user"])
id = user["id"]
self.log(
f"{hijau}account number : {putih}{no+1}{hijau}/{putih}{len(datas)}"
)
self.log(f"{hijau}name : {putih}{user['first_name']}")
token = self.get(id)
if token is None:
token = self.login(data)
if token is None:
continue
self.save(id, token)
if self.is_expired(token):
token = self.login(data)
if token is None:
continue
self.save(id, token)
self.set_authorization(token)
result = self.get_balance()
print(line)
self.countdown(self.interval)
list_countdown.append(result)
_end = int(time.time())
_tot = _end - _start
_min = min(list_countdown) - _tot
self.countdown(_min)
if __name__ == "__main__":
try:
app = Tomartod()
app.main()
except KeyboardInterrupt:
sys.exit()