forked from harrisonwhiskey/TikTok_Bot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutils.py
323 lines (267 loc) ยท 17.1 KB
/
utils.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
import logging
import time
import random, string
from datetime import datetime
import requests
from requests.exceptions import RequestException, ProxyError
from tiktok.tiktok import TikTok
from tiktok.utils import generate_random, username_to_id
import tiktok.error
# from models import ProxyModel
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
LOG_DIR = 'logs\\'
DB_DIR = 'db\\'
def setup_logger(file_name, logger, propagate=False):
formatter = logging.Formatter(
'%(asctime)s - (%(name)s %(threadName)s) - %(levelname)s - %(message)s', datefmt='%m/%d/%Y %I:%M:%S%p'
)
handler = logging.FileHandler(f'{LOG_DIR}{file_name}', encoding='utf-8')
handler.setLevel(logging.DEBUG)
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.propagate = propagate
setup_logger('app.log', logger)
def random_str(size=8):
chars = string.ascii_uppercase + string.ascii_lowercase + string.digits
return ''.join(random.choice(chars) for x in range(size))
def proxy_error(p):
from models import ProxyModel
if not p:
return
p = p.split(',')[0]
p = p.split(':')
proxy = p[0]
port = p[1]
p = ProxyModel(proxy=proxy, port=port)
if not p.is_available:
print('proxy not available')
else:
p.set('data.error', p.get('data.error') + 1).save()
def set_new_proxy(ac, country=None):
error_counter = 0
while True:
if country:
proxy = f'hp_whisk1843jmztw4181:3N8uOuBgXtowwYyN_country-{random.choice(PROXY_COUNTRIES)}_session-{random_str()}@isp2.hydraproxy.com:9989'
else:
proxy = f'hp_whisk1843jmztw4181:3N8uOuBgXtowwYyN_session-{random_str()}@isp2.hydraproxy.com:9989'
proxies = {'http': f'http://{proxy}', 'https': f'http://{proxy}'}
proxy_ip = None
for _ in range(3):
try:
proxy_ip = requests.get('https://api.ipify.org', proxies=proxies).text
break
except Exception as e:
time.sleep(1)
if not proxy_ip:
# print('%s: getting proxy ip error (%s) -> %s' %(ac.get('username'), error_counter, proxy))
error_counter += 1
if error_counter >= 4:
ac.set('proxy', proxy).save()
return
continue
ac.set('proxy', proxy).save()
return
def login(Account, folder=None, force_login=False, login_flow=True, force_login_flow=False, no_proxy=False, proxy=False):
'''
:Account: -> AccountModel
'''
if not Account.is_available:
raise Exception('Account is not available')
if Account.get('login_required') == 1 and not force_login:
raise Exception('Re-login required for %s' %Account.get('username'))
Tik = TikTok() if not folder else TikTok(folder)
if not Tik.settings.has_user(Account.get('username')):
raise Exception('Account do not exist')
if not login_flow:
Tik.login_flow = False
else:
if force_login_flow:
pass
else:
last_login_ts = Account.get('last_login')
if last_login_ts and last_login_ts.timestamp() + (3600 * 12) > time.time():
Tik.login_flow = False
else:
Account.set('last_login', datetime.now())
# Tik.login_flow = False
if not proxy:
proxy = '' if no_proxy else Account.get('proxy')
failed_counter = 0
while True:
try:
Tik.login(Account.get('username'), Account.get('password'), proxy=proxy, force_login=force_login)
if Tik.login_flow == True:
Account.set('login_required', 0)
break
except tiktok.error.BadRequestException:
logger.exception('%s. %s: Failed to login.' %(Account.get('id'), Account.get('username')))
Account.set('login_required', 1).save()
raise
except tiktok.error.AccountDisabledException:
logger.exception('%s. %s: Account disabled' %(Account.get('id'), Account.get('username')))
Account.set('login_required', 2).save()
raise
except tiktok.error.LoginExpiredException as e:
logger.error('%s. %s: Login Expired: %s' %(Account.get('id'), Account.get('username'), e))
Account.set('login_required', 1).save()
raise
except tiktok.error.CaptchaErrorException as e:
logger.debug('%s: captcha required %s' %(Account.get('username'), e))
raise
except (RequestException, ProxyError, tiktok.error.TikTokException) as e:
break
logger.error('%s. %s Failed to login. %s Requests error: %s' %(
Account.get('id'), Account.get('username'), Account.get('proxy'), e))
failed_counter += 1
if '502' in str(e):
set_new_proxy(Account)
proxy = Account.get('proxy')
if failed_counter >= 3:
set_new_proxy(Account)
raise
except Exception:
logger.exception('%s. %s: Failed to login.' %(Account.get('id'), Account.get('username')))
raise
Account.set('last_login', datetime.now()).save()
return Tik
def time_int(arg):
''' returns int representation of HH:MM time to HHMM
:arg: datetime.datetime -> will extract the hours and minutes and return HHMM
: string -> eg: '18:30' | '1830' -> 1830. '00:00' -> 0. '00:30' -> 30. '06:20' -> 620
'''
if type(arg) is datetime:
minute = str(arg.minute)
if len(minute) == 1:
minute = '0' + minute
return int('{}{}'.format(arg.hour, minute))
elif type(arg) is str:
if ':' in arg:
arg = arg.split(':')
return int('{}{}'.format(arg[0], arg[1]))
else:
return int(arg)
else:
raise TypeError('Invalid argument')
comments = [
'hey nice', 'awesome', 'hehe', 'haha', 'i ding that',
'im comin', 'lol nice', 'nice', 'oh dang', 'hell ya',
'wow lol', 'wow', 'that\'s the best', 'lol', 'lmao', 'i love it',
'i\'m a fan', 'im a fan', 'im your fan', 'fyp', 'ooohhh',
'wow this required so much talent', 'too much', 'TikTok', 'sweet',
'aaaaaah', 'NO LOVE', 'beautiful', 'this is gorgeous',
'this looks lovely', 'OMG', "omg hahahaha๐๐๐",
"so funny and awesomeโค๏ธ๐", "Omg amazing", "Hi",
"WHAT???๐๐", "not first๐", "hey be nice !!๐๐๐ถ",
"how re u feeling ๐ณ๐
", "BEST ON TIKTOK", "ok u for the win๐โค๏ธ", "Do one with just ๐ I wanna see sum ๐", "YAS ", "just got yeeted ๐", "really said ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ณ ๐", "Your dances very well๐ฅต๐ฅต๐ฅ๐ฅ๐ฅ๐ฅ๐๏ธ๐๐๏ธ", "done yeeted him", "said ๐๐๐ ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ", "POPPED OFF WITH HER SpLITS๐๐๐๐๐", "ฮmg killed that literally ๐โค๏ธ", " said โmoveโ๐๐บ๐พ", "YAAAAASS ๐๐๐ค ๐ค ๐คฉ", "looks like this ๐ฑ", "Dam bro these comments are โจfreshโจ", "that split tho", "love this yyy โค๏ธ๐", "wins", "anyone else born on their birthday??", "WHY YOU PUSHED YOUR LIKE THAT ๐ญ๐ญ", "IT WAS THE SHOVE FOR ME", "said move", " wins", "told me put a heart in a bag๐ฅบ", "need to chill๐ณ", "OMG NO NO", "nothing goes un this video๐
๐
๐
๐
๐", "anyone else born on their birthday??", "hey we look the same", "Dose the king reply me๐ณ๐ณ๐ณ", "said๐ฑ๐ฑ๏ฟผ๏ฟผ", "Your carried the video", " made me laugh so much!!!", "Canโt believe got pushed away๐คฏ๐ฅบ", "Is your ok?๐", "like ๐๐ง๐๐ง๐ I wanted to dance ๐", "perfect", "perfect", "THO ๐๐He always steals the show", "Who else saw pushed his from the bottom๐ณ๐", "Charli damelioโs leaked video ๐ณ๐ณ", "There's too much in it", "Omg this is lit โฅ๏ธโฅ๏ธ๐ฅบ", "love this FAM very much", "Michael made me laugh to the top", "1m ago gang โโโโโ>", "Omggggggggggggggg please notice meeeeeee plzzz", "how am I this early and it still has so many likes!!", "y'all went off๐", "People who donโt like peaches๐", "okay but the girl at the end popped offffff ๐ณ๐ณ๐ณ", "Does he reply tho?? Plz do", "Your sis killed it ๐ฅ", "Omg amazing", "Hi", "๐ณ๐ณ๐ณ the split", "Iโm deadddd ๐ญ๐๐๐", "First", "his reply game ๐๐ญ๐ฐ๐๐โค๏ธ", "hi", "s entry is ๐ฅ ๐๐", "Won't breathe until he replies", "who come to read comments ๐๐", "Ur fam be funny tho Lmaoo", "said ๐ฆ๐๐ฆ๐๐ฆ๐ฆ๐๐ฆ๐ฆ๐๐ฆ๐ฆ๐๐ฆ๐ฆ", "Ahhahaha !!!", "NICE DANCE HAHA๐", "Omg 73", "Hi", "How he pushed the ๐", "HI PLIS REPLY I WILL CRY CAN YOU SAY HI NANA", "Johnathonโs face at the end ๐", "I really thought Michael was about to pop off ๐", " said Gilbert", "Ur brother really said:๐ฆ๐ฆ๐ฆ๐ฆ๐", "did you guys made up this ? if you do ๐๐๐๐๐๐๐", "Does he reply to his fans?๐ฅบ killed it btwww๐ฅ", "watching at 11pm like๐", "25m late Iโm a big fan :3", " does the splits everyone ๐ฒ", "IM CRYINGGGGG ๐คฃ๐คฃ๐คฃ๐คฃ", "I'm not first or last I'm here", "Imao", "omg hahahaha๐๐๐", "Pop awf", "this has to be my fave video of all ๐๐๐", "WHAT???๐๐", "not first๐", "hey be nice !!๐๐๐ถ", "Omg nooo- he pushed mama๐๐", "omg", "your little bro hit hard doe๐ณ", "Your my favorite tiktoker I swear ๐ค๐ผ๐ค๐ผ", "โฑโฐโโโโฏโฑโฑโฑโฑโฑโฑโฑโฑ", "Omg pop off girl ๐", "woke up chris breezily", "I LOVE YOUR โค", "ROAD TO 10K WHIT ONE VIDEO๐๐", "Beatiful โค๐", "Dose the king reply me๐ณ๐ณ๐ณ", "said:๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ๐ฆ", "Hi", "Does the king replay me?", "what", "This ๐ฅฐ๐", "be like:Out of my waya", "You guys are so funny and awesomeโค๏ธ๐", "๐ฅบ Does the king reply", "lil tay๐ธ๐ค๐ซis typing...", "That kick tho ๐", "Do u replay", "Love this ๐๐", "Hahahaha", "lol", "Early", "I love it ๐ฅฐ๐คฃ๐คฃ๐คฃ๐คฃ๐คฃ", "maiko I'm ur biggest fan of all ๐ฅบ", "lol", "nice๐ฅฐ๐ฅฐ๐ฅฐ", "your funny๐๐๐๐", "sis snapped", "Is your ok looks hurt", "Lmaooo", " passed the vibe check ๐ฅบ", "That split was so straight", "Can I be a part of this ? ๐ฅบ", "Where are his fans at?", "hahaha hahaha hahaha my god I can't dstop๐๐๐๐๐", "it's funny when pushed the๐๐๐", "negative thinkers joined:", "He said ๐๐
๐ฑ", "omg๐
๐
๐
๐๐๐๐๐ SIS DID IT BETTER๐
๐
๐", "Ok I love this video ๐ณ๐คฉ", "ur whole is amazing-", "This really be a dance ", "Little man got it tho ๐ณ", "I love this lol๐ฅบ", " GET IT GIRL ๐คฉ๐คฉ๐คฃ๐คฃ๐๐๐", "Sis snapped", "Yass", "Your really said PERIODT with that split ๐๐ญ", "nice ๐ฅฐ", " won no cap๐๏ธ๐๐๏ธ", "that boi that boi sus", "I like the", "killed it", "The comments are โจoutta da oven fresh โจ", "HOW COULD YOU ๐ญ๐ญ", "lmao imagine being an involved ๐คช", "Okay so who came straight to the comments", "i loved", "whEN HE PUSHED HIS LMFAOOOO", "yeet", "Lil at the end for the win!!!!", "this is getting dangerouss", "We want to see your dance with yโall", "I went ๐ฏ๐ง๐ฒ๐ณ",
]
emojis = [
'๐','๐','๐','๐คฃ','๐','๐','๐
','๐','๐','','๐','๐','๐','๐','๐','๐ฅฐ','๐','๐','๐','๐','','๐ค','๐คฉ','๐ค','๐คจ','',
'๐ถ','๐','๐','๐ฎ','๐ค','๐ฏ','๐ช','๐ซ','๐ด','','๐','๐','๐','๐','๐คค','๐','๐','๐','๐','๐','','๐ค','๐ฒ','โน๏ธ','๐','',
'๐','๐','๐ค','๐ฆ','๐ง','๐จ','๐ฉ','๐คฏ','๐ฌ','','๐ฐ','๐ฑ','๐ฅต','๐ฅถ','๐ณ','๐คช','๐ต','๐ ','๐คฌ','๐ท','','๐ค','๐ค','๐คข','๐คฎ','',
'๐คง','๐','๐ค ','๐คก','๐ฅณ','๐ฅด','๐ฅบ','๐คฅ','๐คซ','','๐คญ','๐ง','๐ค','๐ค','๐ฉ','๐บ','๐ธ','๐น','๐ป','๐ผ','','๐ฝ','๐','๐ฟ','๐พ','',
'๐คฒ','๐','๐','๐','๐ค','๐','๐','๐','โ','','๐ค','๐ค','๐ค','โ๏ธ','๐ค','๐ค','๐','๐','๐','๐','','๐','โ๏ธ','โ','๐ค','๐','๐','๐','',
'๐ค','๐','๐','๐ฃ','๐','๐','๐',''
]
def generate_comment():
return f'{random.choice(comments)} {random.choice(emojis)}'
def device_manager(Account, tk=None, new_account=False, folder=None, db='devices.db'):
from models import DeviceModel, DevicesModel
return False
if not tk:
tk = TikTok(folder)
if new_account:
tk.settings.set_user(Account.get('username'))
else:
tk.login_flow = False
tk.login(Account.get('username'), Account.get('password'))
new_device = (
DevicesModel(db)
.where('used_count', '=', 0)
.order_by('last_action_date', 'ASC')
.limit(1)
.fetch_data()
)
if not new_device:
return False
new_device = new_device[0]
if not new_account:
# insert or update old device
old_device = DeviceModel(db=db, device_id=tk.settings.get('device_id'))
if old_device.is_available:
old_device.set('used_count', 0)
old_device.set('tags', '')
old_device.set('last_action_date', datetime.now())
else:
(
old_device.set('device_string', tk.settings.get('device_string'))
.set('google_aid', tk.settings.get('google_aid'))
.set('openudid', tk.settings.get('openudid'))
.set('uuid', tk.settings.get('uuid'))
.set('device_id', tk.settings.get('device_id'))
.set('install_id', tk.settings.get('install_id'))
)
old_device.save()
tk.settings.set('device_string', new_device.get('device_string'))
tk.settings.set('google_aid', new_device.get('google_aid'))
tk.settings.set('openudid', new_device.get('openudid'))
tk.settings.set('uuid', new_device.get('uuid'))
tk.settings.set('device_id', new_device.get('device_id'))
tk.settings.set('install_id', new_device.get('install_id'))
tk.settings.set_cookie('install_id', new_device.get('install_id'))
(
new_device.set('used_count', 1)
.set('last_action_date', datetime.now())
.set('tags', Account.get('id'))
.save()
)
return True
PROXY_COUNTRIES = [
# 'Russia',
'UnitedStates',
'Canada','Afghanistan','Albania','Algeria','Argentina','Armenia','Aruba','Australia','Austria','Azerbaijan','Bahamas','Bahrain','Bangladesh','Belarus','Belgium','BosniaandHerzegovina','Brazil','BritishVirginIslands','Brunei','Bulgaria','Cambodia','Cameroon','Chile','China','Colombia','CostaRica','Croatia','Cuba','Cyprus','Czechia','Denmark','DominicanRepublic','Ecuador','Egypt','ElSalvador','Estonia','Ethiopia','Finland','Georgia','Germany','Ghana','Greece','Guatemala','HashemiteKingdomofJordan','HongKong','Hungary','India','Indonesia','Iran','Iraq','Ireland','Israel','Italy','Jamaica','Japan','Kazakhstan','Kenya','Kosovo','Kuwait','Latvia','Liechtenstein','Luxembourg','Macedonia','Madagascar','Malaysia','Mauritius','Mexico','Mongolia','Morocco','Myanmar','Nepal','Netherlands','NewZealand','Nigeria','Norway','Oman','Pakistan','Palestine','Panama','PapuaNewGuinea','Paraguay','Peru','Philippines','Poland','PuertoRico','RepublicofLithuania','Romania','SaudiArabia','Senegal','Serbia','Seychelles','Singapore','Slovakia','Slovenia','Somalia','SouthAfrica','SouthKorea','Spain','SriLanka','Sudan','Suriname','Sweden','Switzerland','Syria','Taiwan','Thailand','TrinidadandTobago','Tunisia','Turkey','Uganda','Ukraine','UnitedArabEmirates','UnitedKingdom','Venezuela','Vietnam','Zambia'
]
'''
Test classes
'''
class Users:
def __init__(self):
self.username = generate_random()
self.user_id = generate_random()
self.sec_id = generate_random()
class Videos:
def __init__(self):
self.id = generate_random()
self.url = 'https://tiktok.com/vid/{}'.format(generate_random())
self.url_with_logo = 'https://tiktok.com/vid/{}'.format(generate_random())
class Response:
def __init__(self):
self.staus = 'ok'
self.user = Users()
self.videos = [Videos() for i in range(20)]
class Fuf:
def __init__(self):
self.status = 'ok'
self.max_time = 10
self.offset = 30
self.has_more = True
self.users = [Users() for i in range(20)]
class T:
def __init__(self):
pass
def get_user_info(self, sec_id, username=None):
return Response()
def get_followers(self, arg1, arg2, arg3, arg4):
return Fuf()
def get_followings(self, arg1, arg2, arg3, arg4):
return Fuf()
def follow_by_id(self, user_id, user_sec_id):
return Response()
def get_user_videos(self, user_sec_id):
return Response()
def like_video_by_id(self, video_id):
return Response()
def _login(Account):
return T()
if __name__ == "__main__":
pass