-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmain.py
1221 lines (1054 loc) · 44.6 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
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
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import os
import sys # noqa
import argparse
import random
import re
import time
import copy
from DrissionPage import ChromiumOptions
from DrissionPage import ChromiumPage
from DrissionPage._elements.none_element import NoneElement
from fun_utils import ding_msg
from fun_utils import get_date
from fun_utils import load_file
from fun_utils import save2file
# from proxy_utils import change_proxy
from proxy_api import change_proxy
from proxy_api import get_proxy_current
from conf import DEF_LOCAL_PORT
from conf import DEF_USE_HEADLESS
from conf import DEF_DEBUG
from conf import DEF_PATH_USER_DATA
from conf import DEF_PWD
from conf import DEF_IP_FULL
from conf import DEF_MSG_SUCCESS
from conf import DEF_MSG_FAIL
from conf import DEF_MSG_IP_FULL
from conf import DEF_NUM_NFT
from conf import DEF_NUM_TRY_CHECKIN
from conf import DEF_NUM_TRY_PURCHASE_NFT
from conf import DEF_DING_TOKEN
from conf import DEF_BALANCE_USDG_MIN
from conf import DEF_MSG_BALANCE_ERR
from conf import DEF_PATH_BROWSER
from conf import DEF_AUTO_PROXY
from conf import DEF_PATH_DATA_PROXY
from conf import DEF_CHECKIN
from conf import DEF_UNCOMPLETE
from conf import DEF_PATH_DATA_STATUS
from conf import DEF_HEADER_STATUS
from conf import logger
"""
2024.09.06
1. Check uncompleted transaction
2024.09.05
1. 增加启动参数
2024.09.04
1. 通过 ClashX API 切换代理
2024.09.01
1. 多个账号的执行结果写到一个文件中
2024.08.30
1. 切换 IP 后，可能出现异常(与页面的连接已断开)，增加重试
2024.08.29
1. 在关键确认步骤确保页面加载完成
self.page.wait.load_start()
不追求速度，慢就是快
2. Check-in 增加 Already checked-in Toastify 检查
2024.08.28
1. 通过 pyautogui 切换 proxy
注意，锁屏下无法操作
2024.08.27
1. INSUFFICIENT BALANCE
当余额不足时，发消息提醒，手动 Deposit
2. 页面改版，适配
JOINNOW 变为了 JOIN NOW
3. 封装成 Class，便于传递成员变量
2024.08.25
1. 基于 drissionpage
https://drissionpage.cn/QandA
2. 当出现 cloudflare 5秒盾 真人验证，需手动点击
3. 一个 IP 每天只能做 20次左右的交互，提示 ip full 时，需要手动更换 IP
4. IP FULL 钉钉告警
5. 支持无头浏览器模式
"""
class ParticleTask():
def __init__(self) -> None:
self.args = None
self.page = None
self.is_update = False
self.is_checked_in = False
self.nft_purchased = -1
self.nft_limit = -1
self.usdg = -1
self.proxy_name = 'UNKNOWN(START)'
self.proxy_info = 'USING'
self.lst_proxy_cache = []
self.lst_proxy_black = []
self.s_today = get_date(is_utc=True)
self.file_proxy = None
# self.init_proxy()
# 账号执行情况
self.dic_status = {}
def set_args(self, args):
self.args = args
self.is_update = False
self.is_checked_in = False
self.nft_purchased = 0
self.nft_limit = -1
self.usdg = -1
# self.init_proxy()
self.status_load()
def __del__(self):
self.proxy_save()
self.status_save()
# logger.info(f'Exit {self.args.s_profile}')
def status_load(self):
self.file_status = f'{DEF_PATH_DATA_STATUS}/status_{self.s_today}.csv'
self.dic_status = load_file(
file_in=self.file_status,
idx_key=0,
header=DEF_HEADER_STATUS
)
def status_save(self):
if self.is_checked_in:
s_check_in = 'DONE'
else:
s_check_in = 'XXXXXXXXXX'
self.dic_status[self.args.s_profile] = [
self.args.s_profile,
s_check_in,
self.nft_purchased,
self.nft_limit,
self.usdg
]
self.file_status = f'{DEF_PATH_DATA_STATUS}/status_{self.s_today}.csv'
save2file(
file_ot=self.file_status,
dic_status=self.dic_status,
idx_key=0,
header=DEF_HEADER_STATUS
)
def init_proxy(self):
if DEF_AUTO_PROXY:
self.s_today = get_date(is_utc=True)
self.file_proxy = f'{DEF_PATH_DATA_PROXY}/proxy_{self.s_today}.csv'
self.lst_proxy_black = self.proxy_load()
self.proxy_name = change_proxy(self.lst_proxy_black)
logger.info(f'已开启自动更换 Proxy ，当前代理是 {self.proxy_name}')
def close(self):
# 在有头浏览器模式 Debug 时，不退出浏览器，用于调试
if DEF_USE_HEADLESS is False and DEF_DEBUG:
pass
else:
self.page.quit()
def proxy_update(self, proxy_update_info):
self.proxy_info = proxy_update_info
self.proxy_save()
self.lst_proxy_black = self.proxy_load()
logger.info(f'准备更换 Proxy ，更换前的代理是 {self.proxy_name}')
self.proxy_name = change_proxy(self.lst_proxy_black)
logger.info(f'完成更换 Proxy ，更换后的代理是 {self.proxy_name}')
self.proxy_info = 'USING'
def proxy_load(self):
lst_proxy_black = []
if not DEF_AUTO_PROXY:
return lst_proxy_black
try:
with open(self.file_proxy, 'r') as fp:
# Skip the header line
# next(fp)
for line in fp:
if len(line.strip()) == 0:
continue
# 逗号分隔，Proxy Info 可能包含逗号
fields = line.strip().split(',')
proxy_name = fields[0]
proxy_info = ', '.join(fields[1:])
self.lst_proxy_cache.append([proxy_name, proxy_info])
if proxy_info in [DEF_MSG_IP_FULL, DEF_MSG_FAIL]:
lst_proxy_black.append(proxy_name)
except FileNotFoundError:
pass
except Exception as e:
logger.info(f'[proxy_load] An error occurred: {str(e)}')
return lst_proxy_black
def proxy_save(self):
if not DEF_AUTO_PROXY:
return
if not self.proxy_name:
return
if not self.file_proxy:
self.s_today = get_date(is_utc=True)
self.file_proxy = f'{DEF_PATH_DATA_PROXY}/proxy_{self.s_today}.csv'
dir_file_out = os.path.dirname(self.file_proxy)
if dir_file_out and (not os.path.exists(dir_file_out)):
os.makedirs(dir_file_out)
if not os.path.exists(self.file_proxy):
with open(self.file_proxy, 'w') as fp:
fp.write('Proxy Name,Proxy Info\n')
b_new_proxy_name = True
try:
# 先读取原有内容以便更新
proxies = []
if os.path.exists(self.file_proxy):
with open(self.file_proxy, 'r') as fp:
lines = fp.readlines()
for line in lines[1:]: # 跳过头部
proxies.append(tuple(line.strip().split(',')))
with open(self.file_proxy, 'w') as fp:
fp.write('Proxy Name,Proxy Info\n')
for fields in proxies:
proxy_name = fields[0]
proxy_info = ','.join(fields[1:])
if proxy_name == self.proxy_name:
proxy_info = self.proxy_info
b_new_proxy_name = False
fp.write(f'{proxy_name},{proxy_info}\n') # noqa
if b_new_proxy_name:
fp.write(f'{self.proxy_name},{self.proxy_info}\n') # noqa
except Exception as e:
logger.info(f'[proxy_save] An error occurred: {str(e)}')
def initChrome(self, s_profile):
"""
s_profile: 浏览器数据用户目录名称
"""
profile_path = s_profile
co = ChromiumOptions()
# 设置本地启动端口
co.set_local_port(port=DEF_LOCAL_PORT)
if len(DEF_PATH_BROWSER) > 0:
co.set_paths(browser_path=DEF_PATH_BROWSER)
# co.set_paths(browser_path='/Applications/Google Chrome.app/Contents/MacOS/Google Chrome') # noqa
# 阻止“自动保存密码”的提示气泡
co.set_pref('credentials_enable_service', False)
# 阻止“要恢复页面吗？Chrome未正确关闭”的提示气泡
co.set_argument('--hide-crash-restore-bubble')
co.set_user_data_path(path=DEF_PATH_USER_DATA)
co.set_user(user=profile_path)
# https://drissionpage.cn/ChromiumPage/browser_opt
co.headless(DEF_USE_HEADLESS)
co.set_user_agent(user_agent='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36') # noqa
try:
self.page = ChromiumPage(co)
# logger.info(co.browser_path)
# page.get('https://pioneer.particle.network/zh-CN/point')
# page.get('http://DrissionPage.cn')
# page.quit()
except Exception as e:
logger.info(f'Error: {e}')
finally:
pass
def open_okx(self):
"""
https://chrome.google.com/webstore/detail/mcohilncbfahbmgdjkbpemcciiolgcge
"""
EXTENSION_ID = 'mcohilncbfahbmgdjkbpemcciiolgcge'
def get_balance():
x_path = '//*[@id="home-page-root-element-id"]/div[2]/div'
balance = self.page.ele('x:{}'.format(x_path), timeout=2)
if not isinstance(balance, NoneElement):
return balance.text
return None
max_try = 5
for i in range(1, max_try+1):
logger.info(f'Open okx to login {i}/{max_try} ...')
self.page.get('chrome-extension://{}/home.html'.format(EXTENSION_ID))
self.page.wait.load_start()
balance = get_balance()
if balance and balance != '--':
logger.info(f'okx 已经是登录状态了. 账户余额:{balance} [{self.args.s_profile}]') # noqa
return True
# 页面上如果没有Web3，可能是没有安装插件
if not self.page.ele('Web3', timeout=2):
logger.info('打开 OKX 插件页，判断插件是否已经安装')
self.page.get('https://chrome.google.com/webstore/detail/mcohilncbfahbmgdjkbpemcciiolgcge') # noqa
self.page.wait.load_start()
# 根据 button 判断插件是否已经安装(Remove from Chrome)
s_path = 'tag:span@|text():Remove from@|text():移除'
self.page.wait.eles_loaded(f'{s_path}')
button = self.page.ele(f'{s_path}', timeout=2)
if isinstance(button, NoneElement):
logger.info('插件未安装，需要手动安装 ...')
if len(DEF_DING_TOKEN) > 0:
d_cont = {
'title': 'OKX 插件未安装 [Particle]',
'text': (
'- OKX 插件未安装\n'
'- profile: {s_profile}\n'
.format(
s_profile=self.args.s_profile
)
)
}
ding_msg(d_cont, DEF_DING_TOKEN, msgtype="markdown")
return False
else:
logger.info('插件已安装，重试登录 ...')
time.sleep(3)
continue
ele_input = self.page.ele('@data-testid=okd-input', timeout=2)
if not isinstance(ele_input, NoneElement):
logger.info('OKX 输入密码')
ele_input.input(DEF_PWD)
logger.info('OKX 登录')
self.page.ele('@data-testid=okd-button').click()
balance = get_balance()
if balance and balance != '--':
logger.info(f'okx login success. 账户余额:{balance} [{self.args.s_profile}]') # noqa
return True
logger.info('未获取到账户余额，重试 ...')
# x_path = '//*[@id="home-page-root-element-id"]/div[2]/div'
# balance = self.page.ele('x:{}'.format(x_path), timeout=2)
# if not isinstance(balance, NoneElement):
# logger.info('账户余额：{}'.format(balance.text))
# if balance.text != '--':
# logger.info('okx login success')
# return True
# logger.info('未获取到账户余额，重试 ...')
# else:
# logger.info(
# 'ERROR! okx is invalid! profile:{}'
# .format(self.args.s_profile)
# )
# if DEF_USE_HEADLESS:
# self.page.quit()
time.sleep(3)
return False
def okx_confirm(self):
logger.info('准备 OKX Wallet Confirm ...')
self.page.wait.load_start()
# 当出现 cloudflare 时，勾选
try:
button = self.page.ele('x://*[@id="RlquG0"]/div/label/input')
logger.info(button.text)
button.click()
except: # noqa
pass
if len(self.page.tab_ids) == 2:
tab_id = self.page.latest_tab
tab_new = self.page.get_tab(tab_id)
try:
buttons = tab_new.eles('@data-testid=okd-button')
buttons[-1].click()
except: # noqa
pass
def check_network(self):
if len(self.page.html) == 0:
s_proxy_pre = self.proxy_name
logger.info('无法获取页面内容，请检查网络')
if DEF_AUTO_PROXY:
self.proxy_update(DEF_MSG_FAIL)
if len(DEF_DING_TOKEN) > 0:
d_cont = {
'title': '无法获取页面内容 [Particle]',
'text': (
'- 页面为空\n'
'- 请检查网络\n'
'- profile: {s_profile}\n'
'- proxy_pre: {s_proxy_pre}\n'
'- proxy_now: {s_proxy_now}\n'
.format(
s_profile=self.args.s_profile,
s_proxy_pre=s_proxy_pre,
s_proxy_now=self.proxy_name
)
)
}
ding_msg(d_cont, DEF_DING_TOKEN, msgtype="markdown")
self.page.quit()
try:
# 检查网络连接是否正常
x_path = '//*[@id="error-information-popup-content"]/div[2]'
s_info = self.page.ele('x:{}'.format(x_path), timeout=2).text
if 'ERR_CONNECTION_RESET' == s_info:
s_proxy_pre = self.proxy_name
logger.info('无法访问此网站')
if DEF_AUTO_PROXY:
self.proxy_update(DEF_MSG_FAIL)
if len(DEF_DING_TOKEN) > 0:
d_cont = {
'title': 'Network Error',
'text': (
'- 无法访问此网站\n'
'- 连接已重置\n'
'- profile: {s_profile}\n'
'- proxy_pre: {s_proxy_pre}\n'
'- proxy_now: {s_proxy_now}\n'
.format(
s_profile=self.args.s_profile,
s_proxy_pre=s_proxy_pre,
s_proxy_now=self.proxy_name
)
)
}
ding_msg(d_cont, DEF_DING_TOKEN, msgtype="markdown")
self.page.quit()
except: # noqa
pass
def check_toastify(self, s_tag=''):
try:
toastify = self.page.ele('x://*[@id="1"]/div[1]/div[2]', timeout=2).text # noqa
except: # noqa
toastify = ''
if len(toastify) > 0:
if len(s_tag) > 0:
logger.info(f'check toastify={toastify} [{s_tag}]')
else:
logger.info(f'check toastify={toastify}')
if toastify == DEF_UNCOMPLETE:
sleep_time = random.randint(30, 120)
logger.info(f'Wait transaction to complete. Sleep {sleep_time} seconds ...') # noqa
# time.sleep(sleep_time)
raise Exception(DEF_UNCOMPLETE)
def check_ip_full(self):
try:
toastify = self.page.ele('x://*[@id="1"]/div[1]/div[2]', timeout=2).text # noqa
except: # noqa
toastify = ''
if len(toastify) > 0:
logger.info(f'check toastify={toastify}')
if toastify == DEF_IP_FULL:
# s_proxy_pre = self.proxy_name
s_proxy_pre = get_proxy_current()
if DEF_AUTO_PROXY:
self.proxy_update(DEF_MSG_IP_FULL)
time.sleep(3)
if len(DEF_DING_TOKEN) > 0:
d_cont = {
'title': 'ip is full',
'text': (
'- The number of times this ip sent today is full\n'
'- please try again tomorrow\n'
'- profile: {s_profile}\n'
'- proxy_pre: {s_proxy_pre}\n'
'- proxy_now: {s_proxy_now}\n'
.format(
s_profile=self.args.s_profile,
s_proxy_pre=s_proxy_pre,
s_proxy_now=self.proxy_name
)
)
}
ding_msg(d_cont, DEF_DING_TOKEN, msgtype="markdown")
logger.info('ERROR! IP is full')
self.page.quit()
if DEF_AUTO_PROXY:
return False
else:
return True
else:
return False
def check_in(self):
"""
Return:
True: Already Checked-in
False: To Checkin-in
"""
for i in range(DEF_NUM_TRY_CHECKIN):
logger.info('check_in try_i={}'.format(i+1))
self.page.get('https://pioneer.particle.network/zh-CN/point')
logger.info('准备 CHECK-IN ...')
x_path = '//*[@id="portal"]/div[2]/div[2]/div[1]/div/div[4]/div[4]/button/div[1]' # noqa
self.page.wait.eles_loaded('x:{}'.format(x_path))
self.page.actions.move_to('x:{}'.format(x_path))
# 等页面都加载完，网速慢的时候，按钮状态未更新
self.page.wait.load_start()
# 首次不 sleep ，失败后从第二次开始，增加 sleep
if i > 0:
logger.info('sleep {} seconds...'.format(i+1))
time.sleep(i+1)
button = self.page.ele('x:{}'.format(x_path))
if isinstance(button, NoneElement):
logger.info('没有 CHECK-IN 按钮，从头重试 ...')
continue
if button.text == 'Checked in':
logger.info('Oh! Check-in is already done before!')
self.is_checked_in = True
self.is_update = True
self.status_save()
return True
if button.text == 'Check-in':
# 2024.08.27 无法点击，改为使用 js 点击，成功
button.click(by_js=True)
else:
logger.info('button.text={}'.format(button.text))
try:
toastify = self.page.ele('x://*[@id="1"]/div[1]/div[2]', timeout=2).text # noqa
except: # noqa
toastify = ''
if len(toastify) > 0:
if toastify == DEF_CHECKIN:
logger.info('Check-in is success!')
self.is_checked_in = True
self.is_update = True
self.status_save()
return True
else:
logger.info('toastify={}'.format(toastify))
# SEND TRANSACTION 窗口
x_path = '/html/body/div[4]/div/div[2]/div/div/div[1]'
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if isinstance(button, NoneElement):
logger.info('没有 SEND TRANSACTION 窗口，从头重试 ...')
continue
logger.info('准备点击 CONFIRM 按钮')
x_path = '//html/body/div[4]/div/div[2]/div/div/div[2]/div[4]/div[2]/button/div[1]' # noqa
button = self.page.ele('x:{}'.format(x_path), timeout=10)
if isinstance(button, NoneElement):
logger.info('没有 CONFIRM 按钮，从头重试 ...')
continue
button.click()
logger.info('准备 OKX CONFIRM')
self.okx_confirm()
if self.check_ip_full():
if DEF_USE_HEADLESS:
self.page.quit()
# sys.exit(-1)
break
logger.info('Check-in is Finished!')
return False
def activate(self):
"""
# noqa
激活后回到 PURCHASE 页面
# DEF_URL_ACTIVATE = 'https://pioneer.particle.network/zh-CN/accountAnimation'
# DEF_URL_SIGNUP = 'https://pioneer.particle.network/zh-CN/signup'
"""
# page.wait.load_start()
logger.info('Current url: {}'.format(self.page.url))
try:
# 图片
x_path = '/html/body/div[1]/div[1]/div/div[1]/div[3]/div[3]/img' # noqa
self.page.wait.eles_loaded('x:{}'.format(x_path))
# DrissionPage.errors.NoRectError: 该元素没有位置及大小
time.sleep(1)
button = self.page.ele('x:{}'.format(x_path)) # noqa
if isinstance(button, NoneElement):
logger.info('NO ACTIVATE PAGE')
else:
logger.info('CLICK TO ACTIVATE')
if DEF_DEBUG:
print(self.page.tab_ids)
button.click()
except: # noqa
logger.info('没有 CLICK TO ACTIVATE')
# return False
logger.info('Current url: {}'.format(self.page.url))
# CLICK TO LAUNCH
logger.info('CLICK TO LAUNCH ...')
try:
x_path = '/html/body/div[1]/div[1]/div/div[1]/a/div'
button = self.page.ele('x:{}'.format(x_path))
logger.info('button.text={}'.format(button.text))
if isinstance(button, NoneElement):
logger.info('没有 CLICK TO LAUNCH')
elif button.text == 'Click to launch':
button.click()
logger.info('CLICK TO LAUNCH, SUCCESS')
else:
logger.info('不是想要的按钮：{}'.format(button.text))
except: # noqa
logger.info('没有 CLICK TO LAUNCH')
# return False
# LAUNCH
logger.info('TO LAUNCH ...')
try:
time.sleep(1)
x_path = '//*[@id="home"]/div/div[1]/div/div[2]/a/div[1]'
button = self.page.ele('x:{}'.format(x_path))
logger.info('button.text={}'.format(button.text))
if button.text == 'LAUNCH':
button.click()
logger.info('LAUNCH SUCCESS')
return True
except: # noqa
logger.info('没有 LAUNCH')
return False
return True
def check_nft_num(self):
"""
返回值 num_ret
[1, 5]
第一个数: 今天成功购买的 NFT 数量
第二个数: 今天计算积分的 NFT 数量，超出该数量不算积分
"""
num_ret = [0, 0]
for i in range(DEF_NUM_TRY_PURCHASE_NFT):
logger.info(f'check_nft_num try_i={i+1}/{DEF_NUM_TRY_PURCHASE_NFT}') # noqa
self.page.get('https://pioneer.particle.network/zh-CN/point')
# logger.info('即将刷新页面 {}'.format(page.url))
self.page.refresh()
x_path = '//*[@id="portal"]/div[2]/div[2]/div[4]/div/div[4]/div/div' # noqa
self.page.wait.eles_loaded('x:{}'.format(x_path))
self.page.actions.move_to('x:{}'.format(x_path))
# 等页面都加载完，网速慢的时候，按钮状态未更新
self.page.wait.load_start()
# time.sleep(3)
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if isinstance(button, NoneElement):
pass
elif button.text.startswith('Attempts today'):
# Attempts today: 11 / 5
logger.info('NFT {}'.format(button.text))
# 使用正则表达式提取数字
numbers = re.findall(r'\d+', button.text)
# 将提取到的数字转换为整数
numbers = [int(num) for num in numbers]
# 输出结果
# print(numbers) # 输出: [11, 5]
if len(numbers) == 2:
self.nft_purchased = numbers[0]
self.nft_limit = numbers[1]
num_ret = numbers
break
else:
pass
return num_ret
def check_balance(self, s_balance):
"""
确认账户余额
True 余额充足
False 余额不足
"""
try:
flt_balance = float(s_balance.replace(',', ''))
except: # noqa
flt_balance = 0
self.usdg = int(flt_balance)
if flt_balance < DEF_BALANCE_USDG_MIN:
if len(DEF_DING_TOKEN) > 0:
d_cont = {
'title': 'insufficient balance',
'text': (
'- profile: {}\n'
'- balance: ${} 小于 ${}\n'
'- please deposit manually\n'
.format(
self.args.s_profile,
flt_balance, DEF_BALANCE_USDG_MIN
)
)
}
ding_msg(d_cont, DEF_DING_TOKEN, msgtype="markdown")
return False
return True
def purchase_nft(self):
"""
返回值 s_msg
"""
s_msg = DEF_MSG_FAIL
for i in range(DEF_NUM_TRY_PURCHASE_NFT):
logger.info(f'purchase_nft try_i={i+1}/{DEF_NUM_TRY_PURCHASE_NFT} [{self.args.s_profile}]') # noqa
self.page.get('https://pioneer.particle.network/zh-CN/crossChainNFT') # noqa
# logger.info('刷新页面 {}'.format(self.page.url))
# self.page.refresh()
logger.info('准备在 crossChainNFT 页面点击 PURCHASE 按钮 ...')
# time.sleep(5)
self.page.wait.load_start()
x_path = '//*[@id="content-wrapper"]/div/div[2]/div[1]/div[3]/div[2]/div[4]/button/div[1]' # noqa
self.page.wait.eles_loaded(f'x:{x_path}', timeout=3) # noqa
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if button and button.text == 'Purchase':
button.click()
else:
logger.info('没有 PURCHASE 按钮，重新开始')
continue
logger.info('准备选中 USDG 复选框 ...')
# time.sleep(5)
self.page.wait.load_start()
buttons = self.page.eles('.:polygon-small mb-3 flex cursor-pointer') # noqa
if len(buttons) >= 1:
# USDG 余额
s_balance = buttons[0].text.split('$')[-1]
if len(s_balance) == 0:
logger.info('没有获取到 USDG 余额，重新开始')
continue
logger.info(f'USDG 余额: ${s_balance} [{self.args.s_profile}]')
if not self.check_balance(s_balance):
logger.info(f'USDG 余额不足，退出 (余额: ${s_balance}) [{self.args.s_profile}]') # noqa
s_msg = DEF_MSG_BALANCE_ERR
break
buttons[0].click()
else:
self.activate()
logger.info('没有 USDG 选项，重新开始')
continue
logger.info('准备 Click Next BUTTON ...')
x_path = '/html/body/div[4]/div/div[2]/div/div/div[2]/div[6]/button/div[1]' # noqa
self.page.wait.eles_loaded(f'x:{x_path}', timeout=3) # noqa
button = self.page.ele(f'x:{x_path}', timeout=3) # noqa
if button.text == 'Next':
time.sleep(1)
if button.states.is_clickable:
button.click()
else:
logger.info('注意！Next 按钮不可点击！！！为啥呢？！')
continue
else:
logger.info('没有 Next 按钮，重新开始')
continue
self.check_toastify('S1')
logger.info('准备在 NETWORK FEE 弹窗点击 PURCHASE 按钮 ...')
x_path = '/html/body/div[4]/div/div[2]/div/div/button/div[1]'
self.page.wait.eles_loaded(f'x:{x_path}', timeout=3) # noqa
button = self.page.ele(f'x:{x_path}', timeout=3) # noqa
if isinstance(button, NoneElement):
logger.info('没有 NETWORK FEE 弹窗，重新开始')
continue
if button.text == 'Purchase':
x_path = '/html/body/div[4]/div/div[2]/div/div/div[2]/div[5]/div[1]/div[2]/div[2]' # noqa
try:
fee = self.page.ele('x:{}'.format(x_path), timeout=2).text
f_fee = float(fee.replace(',', '').replace('$', ''))
if f_fee > 10:
logger.info(f'Warning! NETWORK FEE is high! {f_fee}')
else:
logger.info(f'NETWORK FEE is ${f_fee}')
except: # noqa
pass
time.sleep(1)
button.click()
else:
logger.info('没有 PURCHASE 按钮，重新开始')
continue
self.check_toastify('S2')
# 此处可能会出现 CLOUDFLARE 5秒盾，留点时间
time.sleep(3)
self.okx_confirm()
self.check_toastify('S3')
if self.check_ip_full():
s_msg = DEF_MSG_IP_FULL
break
time.sleep(1)
# 此处最多等60秒
max_wait_sec = 60
logger.info(f'Wait SUCCESSFUL 弹窗 (最多等 {max_wait_sec} 秒) ...')
for i_wait in range(1, max_wait_sec):
x_path = '/html/body/div[4]/div/div[2]/div/div/div[1]'
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if isinstance(button, NoneElement):
logger.info(f'等待 SUCCESSFUL 弹窗 {i_wait}/{max_wait_sec}')
time.sleep(1)
continue
else:
if button.text.startswith('Preview'):
logger.info(f'当前在 Preview 窗口 {i_wait}/{max_wait_sec}')
time.sleep(1)
continue
else:
logger.info(f'离开 Preview 窗口 {i_wait}/{max_wait_sec}')
break
# self.page.wait.load_start()
logger.info('正在确认 SUCCESSFUL 弹窗 ...')
# 出现 SUCCESSFUL 弹窗，需要等待几秒
x_path = '/html/body/div[4]/div/div[2]/div/div/div[1]'
self.page.wait.eles_loaded('x:{}'.format(x_path), timeout=2)
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if isinstance(button, NoneElement):
logger.info('没有 SUCCESSFUL 弹窗，重新开始')
continue
else:
if button.text.startswith('SUCCESSFULLY'):
# logger.info('出现 SUCCESSFUL 弹窗')
# 弹窗右上角的 ×
x_path = '/html/body/div[4]/div/div[2]/div/button'
try:
self.page.ele('x:{}'.format(x_path)).click()
logger.info(f'关闭弹窗 {button.text}')
self.nft_purchased += 1
self.is_update = True
self.status_save()
except: # noqa
logger.info('未能关闭 SUCCESSFUL 弹窗，忽略')
pass
else:
logger.info('button.text:{}'.format(button.text))
logger.info('Purchase Failed，重新开始 ...')
continue
s_msg = DEF_MSG_SUCCESS
break
if s_msg == DEF_MSG_FAIL:
s_error = 'PURCHASE_NFT 连续{}次失败'.format(DEF_NUM_TRY_PURCHASE_NFT)
logger.info(s_error)
return s_msg
def particle_login(self):
try:
# 首次登录有弹窗，点击 START
logger.info('正在确认是否有 START 弹窗 ...')
time.sleep(2)
x_path = '/html/body/div[1]/button'
self.page.wait.eles_loaded('x:{}'.format(x_path))
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if not isinstance(button, NoneElement):
logger.info('点击 START')
button.click()
else:
logger.info('没有 START 弹窗')
except: # noqa
pass
self.check_network()
button = self.page.ele('.polygon-btn-text')
if isinstance(button, NoneElement):
logger.info('没有获取到页面右上角按钮')
return
logger.info('正在根据按钮[{}]确认登录状态 ...'.format(button.text))
if button.text in ['JOINNOW', 'JOIN NOW']:
logger.info('点击 JOINNOW 按钮 ...')
button.click()
# 选择登录的钱包
try:
x_path = '/html/body/div[1]/div[1]/div/div[1]/div[4]/div[3]/div[1]/button' # noqa
self.page.wait.eles_loaded('x:{}'.format(x_path))
self.page.actions.move_to('x:{}'.format(x_path))
button = self.page.ele('x:{}'.format(x_path), timeout=2)
logger.info('正在点击 OKX WALLET 连接 OKX 钱包 ...')
button.click(by_js=True)
except:
pass
# OKX Wallet 连接
logger.info('OKX Wallet 连接')
# 需要等待弹窗加载完成
self.page.wait.load_start()
if DEF_DEBUG:
print(self.page.tab_ids)
if len(self.page.tab_ids) == 2:
try:
tab_id = self.page.latest_tab
tab_new = self.page.get_tab(tab_id)
button = tab_new.ele('x://*[@id="app"]/div/div/div/div/div[5]/div[2]/button[2]', timeout=2) # noqa
logger.info('{}'.format(button.text))
button.click()
except: # noqa
pass
# OKX Wallet 请求签名
logger.info('OKX Wallet 请求签名')
self.page.wait.load_start()
if DEF_DEBUG:
print(self.page.tab_ids)
if len(self.page.tab_ids) == 2:
try:
tab_id = self.page.latest_tab
tab_new = self.page.get_tab(tab_id)
button = tab_new.ele('x://*[@id="app"]/div/div/div/div/div/div[6]/div/button[2]', timeout=2) # noqa
logger.info('{}'.format(button.text))
button.click()
except: # noqa
pass
try:
# 首次登录有弹窗，点击 LAUNCH
logger.info('确认是否有 LAUNCH 弹窗 ...')
time.sleep(2)
x_path = '//*[@id="home"]/div/div[1]/div/div[2]/a'
if self.page.ele('x:{}'.format(x_path), timeout=2).click():
logger.info('成功点击 LAUNCH')
except: # noqa
pass
def particle_init(self):
"""
登录及校验是否登录成功
"""
self.page.get('https://pioneer.particle.network/zh-CN/point')
for i in range(10):
logger.info('Page Login try_i={}'.format(i+1))
self.particle_login()
logger.info('检查是否登录成功 ...')
time.sleep(1)
# 这是已登录时的 xpath
x_path = '//*[@id="navbar"]/header/ul[3]/li/div/button'
self.page.wait.eles_loaded('x:{}'.format(x_path))
button = self.page.ele('x:{}'.format(x_path), timeout=2)
if isinstance(button, NoneElement):
# 没有获取到已登录的 xpath
pass