feat: AmorterValidator

geetest/AmorterValidator.py

@@ -10,31 +10,42 @@
 
 
 class AmorterValidator(Validator):
+    def need_api_key(self) -> bool:
+        return False
+
+    def have_gt_ui(self) -> bool:
+        return False
+
     def __init__(self):
         self.click = bili_ticket_gt_python.ClickPy()
         pass
 
-    @retry()
     def validate(self, appkey, gt, challenge, referer="http://127.0.0.1:7860/") -> str:
         try:
+            loguru.logger.info(f"gt: {gt} ; challenge: {challenge}")
             (_, _) = self.click.get_c_s(gt, challenge)
             _type = self.click.get_type(gt, challenge)
             if _type != "click":
                 raise Exception("验证码类型错误")
             (c, s, args) = self.click.get_new_c_s_args(gt, challenge)
-            before_calculate_key = time.time()
-            key = self.click.calculate_key(args)
-            # rt固定即可
-            # 此函数是使用项目目录下的click.exe生成w参数，如果文件不存在会报错，你也可以自己接入生成w的逻辑函数
-            w = self.click.generate_w(key, gt, challenge, str(c), s, "abcdefghijklmnop")
-            # 点选验证码生成w后需要等待2秒提交
-            w_use_time = time.time() - before_calculate_key
-            loguru.logger.info(f"w生成时间：{w_use_time}")
-            if w_use_time < 2:
-                time.sleep(2 - w_use_time)
-            (msg, validate) = self.click.verify(gt, challenge, w)
-            loguru.logger.info(f"msg: {msg} ; validate: {validate}")
-            return validate
+
+            @retry()
+            def inner_try():
+                before_calculate_key = time.time()
+                key = self.click.calculate_key(args)
+                # rt固定即可
+                # 此函数是使用项目目录下的click.exe生成w参数，如果文件不存在会报错，你也可以自己接入生成w的逻辑函数
+                w = self.click.generate_w(key, gt, challenge, str(c), s, "abcdefghijklmnop")
+                # 点选验证码生成w后需要等待2秒提交
+                w_use_time = time.time() - before_calculate_key
+                loguru.logger.info(f"w生成时间：{w_use_time}")
+                if w_use_time < 2:
+                    time.sleep(2 - w_use_time)
+                (msg, validate) = self.click.verify(gt, challenge, w)
+                loguru.logger.info(f"msg: {msg} ; validate: {validate}")
+                return validate
+
+            return inner_try()
         except Exception as e:
             loguru.logger.warning(e)
             raise e

geetest/CapSolverValidator.py

@@ -10,11 +10,17 @@
 
 
 class CapSolverValidator(Validator):
+    def need_api_key(self) -> bool:
+        return True
+
+    def have_gt_ui(self) -> bool:
+        return False
+
     def __init__(self):
         self.cookieManager = global_cookieManager
         pass
 
-    @retry()
+    @retry(tries=10)
     def validate(self, appkey, gt, challenge, referer="http://127.0.0.1:7860/") -> str:
         if appkey is None or appkey == "":
             appkey = self.cookieManager.get_config_value("appkey", "")
@@ -32,6 +38,7 @@ def validate(self, appkey, gt, challenge, referer="http://127.0.0.1:7860/") -> s
         resp = res.json()
         task_id = resp.get("taskId")
         if not task_id:
+            loguru.logger.info(resp)
             raise ValueError("Failed to create task: " + res.text)
         loguru.logger.info(f"Got taskId: {task_id} / Getting result...")
         while True:

geetest/NormalValidator.py

@@ -0,0 +1,15 @@
+from geetest.Validator import Validator
+
+
+class NormalValidator(Validator):
+    def need_api_key(self) -> bool:
+        return False
+
+    def have_gt_ui(self) -> bool:
+        return True
+
+    def __init__(self):
+        pass
+
+    def validate(self, appkey, gt, challenge, referer="http://www.baidu.com", ip='', host='') -> str:
+        raise Exception("No validate")

geetest/RROCRValidator.py

@@ -7,10 +7,15 @@
 from config import cookies_config_path, global_cookieManager
 from geetest.Validator import Validator
 from util.bili_request import BiliRequest
-from util.config_util import CookieManager
 
 
 class RROCRValidator(Validator):
+    def need_api_key(self) -> bool:
+        return True
+
+    def have_gt_ui(self) -> bool:
+        return True
+
     def __init__(self):
         self.url = "http://api.rrocr.com/api/recognize.html"
         self.headers = {
@@ -20,7 +25,7 @@ def __init__(self):
         }
         self.cookieManager = global_cookieManager
 
-    @retry()
+    @retry(tries=10)
     def validate(self, appkey, gt, challenge, referer="http://www.baidu.com", ip='', host='') -> str:
         if appkey is None or appkey == "":
             appkey = self.cookieManager.get_config_value("appkey", "")
@@ -35,6 +40,7 @@ def validate(self, appkey, gt, challenge, referer="http://www.baidu.com", ip='',
             "host": host
         }
         data = parse.urlencode(data)
+        loguru.logger.info("start rrocr validate")
         response = requests.post(self.url, headers=self.headers, data=data)
 
         if response.status_code == 200:

geetest/Validator.py

@@ -6,4 +6,10 @@ class Validator(ABC):
     def validate(self, field: str) -> str:
         pass
 
+    @abstractmethod
+    def have_gt_ui(self) -> bool:
+        pass
 
+    @abstractmethod
+    def need_api_key(self) -> bool:
+        pass

tab/go.py

@@ -14,6 +14,7 @@
 from config import cookies_config_path, global_cookieManager
 from geetest.AmorterValidator import AmorterValidator
 from geetest.CapSolverValidator import CapSolverValidator
+from geetest.NormalValidator import NormalValidator
 from geetest.RROCRValidator import RROCRValidator
 from util.bili_request import BiliRequest
 from util.error import ERRNO_DICT, withTimeString
@@ -22,7 +23,7 @@
 isRunning = False
 
 ways = ["手动", "使用 rrocr", "使用 CapSolver", "本地验证码（Amorter提供）"]
-ways_detail = [None, RROCRValidator(), CapSolverValidator(), AmorterValidator()]
+ways_detail = [NormalValidator(), RROCRValidator(), CapSolverValidator(), AmorterValidator()]
 
 
 def go_tab():
@@ -61,11 +62,11 @@ def choose_option(way):
             global select_way
             select_way = way
             # loguru.logger.info(way)
-            if way in [0, 3]:
-                # rrocr
-                return gr.update(visible=False)
-            else:
+            validator = ways_detail[select_way]
+            if  validator.need_api_key():
                 return gr.update(visible=True)
+            else:
+                return gr.update(visible=False)
 
         way_select_ui.change(choose_option, inputs=way_select_ui, outputs=api_key_input_ui)
         with gr.Row():
@@ -96,6 +97,7 @@ def choose_option(way):
                 info="设置抢票的总次数",
                 visible=False,
             )
+    validate_con = threading.Condition()
 
     def start_go(tickets_info_str, time_start, interval, mode, total_attempts, api_key):
         global isRunning, geetest_validate, geetest_seccode
@@ -152,8 +154,9 @@ def start_go(tickets_info_str, time_start, interval, mode, total_attempts, api_k
                     gt = _data["data"]["geetest"]["gt"]
                     challenge = _data["data"]["geetest"]["challenge"]
                     token = _data["data"]["token"]
+                    validator = ways_detail[select_way]
                     try:
-                        if select_way in [0, 1, 3]:
+                        if validator.have_gt_ui():
                             yield [
                                 gr.update(value=withTimeString("进行验证码验证"), visible=True),
                                 gr.update(visible=True),
@@ -163,16 +166,20 @@ def start_go(tickets_info_str, time_start, interval, mode, total_attempts, api_k
                                 gr.update(value=challenge),
                                 gr.update(value=uuid.uuid1()),
                             ]
-                        if select_way in [1, 2, 3]:
-                            # https://passport.bilibili.com/x/passport-login/captcha?source=main_web
-                            def run_validation():
-                                global geetest_validate, geetest_seccode
-                                logger.info(f"{ways[select_way]}")
-                                validator = ways_detail[select_way]
-                                geetest_validate = validator.validate(appkey=api_key, gt=gt, challenge=challenge)
-                                geetest_seccode = geetest_validate + "|jordan"
 
-                            threading.Thread(target=run_validation).start()
+                        def run_validation():
+                            global geetest_validate, geetest_seccode
+                            try:
+                                tmp = validator.validate(appkey=api_key, gt=gt, challenge=challenge)
+                            except Exception as e:
+                                return
+                            validate_con.acquire()
+                            geetest_validate = tmp
+                            geetest_seccode = geetest_validate + "|jordan"
+                            validate_con.notify()
+                            validate_con.release()
+
+                        threading.Thread(target=run_validation).start()
                     except NameError as err:
                         yield [
                             gr.update(value=withTimeString("进行验证码验证"), visible=True),
@@ -183,8 +190,10 @@ def run_validation():
                             gr.update(value=challenge),
                             gr.update(value=uuid.uuid1()),
                         ]
+                    validate_con.acquire()
                     while geetest_validate == "" or geetest_seccode == "":
-                        continue
+                        validate_con.wait()
+                    validate_con.release()
                     logger.info(
                         f"geetest_validate: {geetest_validate},geetest_seccode: {geetest_seccode}"
                     )
@@ -370,8 +379,12 @@ def run_validation():
 
     def receive_geetest_result(res):
         global geetest_validate, geetest_seccode
-        geetest_validate = res["geetest_validate"]
-        geetest_seccode = res["geetest_seccode"]
+        if "geetest_validate" in res and "geetest_seccode" in res:
+            validate_con.acquire()
+            geetest_validate = res["geetest_validate"]
+            geetest_seccode = res["geetest_seccode"]
+            validate_con.notify()
+            validate_con.release()
 
     gt_html_finish_btn.click(
         fn=None,

tab/settings.py

@@ -35,7 +35,7 @@ def on_submit_ticket_id(num):
                     gr.update(),
                     gr.update(),
                     gr.update(),
-                    gr.update(visible=True),
+                    gr.update(visible=False),
                     gr.update(value='输入无效，请输入一个有效的票ID。', visible=True),
                 ]
         bili_request = BiliRequest(cookies_config_path=cookies_config_path)

tab/train.py

@@ -45,11 +45,12 @@ def choose_option(way):
         global select_way
         select_way = way
         # loguru.logger.info(way)
-        if way in [0, 3]:
+        validator = ways_detail[select_way]
+        if validator.need_api_key():
             # rrocr
-            return gr.update(visible=False)
-        else:
             return gr.update(visible=True)
+        else:
+            return gr.update(visible=False)
 
     way_select_ui.change(choose_option, inputs=way_select_ui, outputs=api_key_input_ui)
 
@@ -71,6 +72,7 @@ def choose_option(way):
     trigger_ui = gr.Textbox(label="trigger", visible=False)
 
     geetest_result = gr.JSON(label="validate")
+    validate_con = threading.Condition()
 
     def test_get_challenge(api_key):
         global \
@@ -90,37 +92,47 @@ def test_get_challenge(api_key):
         test_csrf = _request.cookieManager.get_cookies_value("bili_jct")
         test_geetest_validate = ""
         test_geetest_seccode = ""
+        validator = ways_detail[select_way]
 
         try:
             # Capture 不支持同时
-            if select_way in [0, 1, 3]:
+            if validator.have_gt_ui():
                 yield [
                     gr.update(value=test_gt),  # test_gt_ui
                     gr.update(value=test_challenge),  # test_challenge_ui
                     gr.update(visible=True),  # test_gt_row
                     gr.update(value="重新生成"),  # test_get_challenge_btn
-                    gr.update(),
+                    gr.update(value={}),
                     gr.update(value=uuid.uuid1())
                 ]
-            if select_way in [1, 2, 3]:
-                def run_validation():
-                    global test_geetest_validate, test_geetest_seccode
-                    validator = ways_detail[select_way]
-                    test_geetest_validate = validator.validate(appkey=api_key, gt=test_gt, challenge=test_challenge)
-                    test_geetest_seccode = test_geetest_validate + "|jordan"
-
-                threading.Thread(target=run_validation).start()
+
+            def run_validation():
+                global test_geetest_validate, test_geetest_seccode
+                try:
+                    tmp = validator.validate(appkey=api_key, gt=test_gt, challenge=test_challenge)
+                except Exception as e:
+                    return
+                validate_con.acquire()
+                test_geetest_validate = tmp
+                test_geetest_seccode = test_geetest_validate + "|jordan"
+                validate_con.notify()
+                validate_con.release()
+
+            threading.Thread(target=run_validation).start()
         except NameError as err:
             yield [
                 gr.update(value=test_gt),  # test_gt_ui
                 gr.update(value=test_challenge),  # test_challenge_ui
                 gr.update(visible=True),  # test_gt_row
                 gr.update(value="重新生成"),  # test_get_challenge_btn
-                gr.update(),
+                gr.update(value={}),
                 gr.update(value=uuid.uuid1())
             ]
+        validate_con.acquire()
         while test_geetest_validate == "" or test_geetest_seccode == "":
-            continue
+            validate_con.wait()
+        validate_con.release()
+
         _url = "https://api.bilibili.com/x/gaia-vgate/v1/validate"
         _payload = {
             "challenge": test_challenge,
@@ -173,7 +185,11 @@ def run_validation():
 
     def receive_geetest_result(res):
         global test_geetest_validate, test_geetest_seccode
-        test_geetest_validate = res["geetest_validate"]
-        test_geetest_seccode = res["geetest_seccode"]
+        if "geetest_validate" in res and "geetest_seccode" in res:
+            validate_con.acquire()
+            test_geetest_validate = res["geetest_validate"]
+            test_geetest_seccode = res["geetest_seccode"]
+            validate_con.notify()
+            validate_con.release()
 
     geetest_result.change(fn=receive_geetest_result, inputs=geetest_result)