fixed: #39

Handle challenge `seaplane` gracefully.

Co-Authored-By: Bingjie Yan <44976445+beiyuouo@users.noreply.github.com>

src/apis/scaffold/install.py

@@ -10,7 +10,12 @@
 from webdriver_manager.chrome import ChromeType
 from webdriver_manager.utils import get_browser_version_from_os
 
-from services.hcaptcha_challenger import YOLO, SKRecognition, ElephantsDrawnWithLeaves
+from services.hcaptcha_challenger import (
+    YOLO,
+    SKRecognition,
+    ElephantsDrawnWithLeaves,
+    ResNetSeaplane,
+)
 from services.settings import DIR_MODEL, logger, PATH_RAINBOW
 
 
@@ -20,6 +25,7 @@ def _download_model(onnx_prefix: Optional[str] = None):
 
     YOLO(dir_model=DIR_MODEL, onnx_prefix=onnx_prefix).download_model()
     ElephantsDrawnWithLeaves(dir_model=DIR_MODEL).download_model()
+    ResNetSeaplane(dir_model=DIR_MODEL).download_model()
 
 
 def _download_rainbow():

src/services/hcaptcha_challenger/__init__.py

@@ -4,8 +4,16 @@
 # Github     : https://github.com/QIN2DIM
 # Description:
 from .core import ArmorCaptcha, ArmorUtils
-from .solutions.de_stylized import ElephantsDrawnWithLeaves
+from .solutions.resnet import ElephantsDrawnWithLeaves
+from .solutions.resnet import ResNetSeaplane
 from .solutions.sk_recognition import SKRecognition
 from .solutions.yolo import YOLO
 
-__all__ = ["SKRecognition", "YOLO", "ArmorCaptcha", "ArmorUtils", "ElephantsDrawnWithLeaves"]
+__all__ = [
+    "SKRecognition",
+    "YOLO",
+    "ArmorCaptcha",
+    "ArmorUtils",
+    "ElephantsDrawnWithLeaves",
+    "ResNetSeaplane",
+]

src/services/hcaptcha_challenger/core.py

@@ -26,7 +26,7 @@
     AssertTimeout,
     ChallengeLangException,
 )
-from .solutions import sk_recognition, de_stylized, yolo, kernel
+from .solutions import sk_recognition, resnet, yolo
 
 
 class ArmorCaptcha:
@@ -118,6 +118,8 @@ def __init__(self, dir_workspace: str = None, lang: Optional[str] = "zh", debug=
         # 运行缓存
         self.dir_workspace = dir_workspace if dir_workspace else "."
 
+        self.threat = 0
+
     def _init_workspace(self):
         """初始化工作目录，存放缓存的挑战图片"""
         _prefix = (
@@ -200,20 +202,19 @@ def tactical_retreat(self) -> Optional[str]:
     def switch_solution(self, dir_model, onnx_prefix):
         """模型卸载"""
         label = self.label_alias.get(self.label)
-
+        if label in ["seaplane"]:
+            return resnet.ResNetSeaplane(dir_model)
+        if label in ["elephants drawn with leaves"]:
+            return resnet.ElephantsDrawnWithLeaves(dir_model, path_rainbow=PATH_RAINBOW)
         if label in ["vertical river"]:
             return sk_recognition.VerticalRiverRecognition(path_rainbow=PATH_RAINBOW)
         if label in ["airplane in the sky flying left"]:
             return sk_recognition.LeftPlaneRecognition(path_rainbow=PATH_RAINBOW)
         if label in ["airplanes in the sky that are flying to the right"]:
             return sk_recognition.RightPlaneRecognition(path_rainbow=PATH_RAINBOW)
-        if label in ["elephants drawn with leaves"]:
-            return de_stylized.ElephantsDrawnWithLeaves(dir_model, path_rainbow=PATH_RAINBOW)
         if label in ["horses drawn with flowers"]:
-            return de_stylized.HorsesDrawnWithFlowers(dir_model, path_rainbow=PATH_RAINBOW)
-        if label in ["seaplane"]:
-            return kernel.RainbowSeaplane(path_rainbow=PATH_RAINBOW)
-        return yolo.YOLO(dir_model, onnx_prefix=onnx_prefix)
+            return resnet.HorsesDrawnWithFlowers(dir_model, path_rainbow=PATH_RAINBOW)
+        return yolo.YOLO(dir_model, onnx_prefix)
 
     def mark_samples(self, ctx: Chrome):
         """
@@ -272,8 +273,6 @@ async def control_driver(self, context, session=None):
                     with open(path_challenge_img, "wb") as file:
                         file.write(await response.read())
 
-        # self.log(message="Download the challenge image")
-
         # 初始化挑战图片下载目录
         workspace_ = self._init_workspace()
 
@@ -292,6 +291,7 @@ async def control_driver(self, context, session=None):
             loop = asyncio.get_event_loop()
             loop.run_until_complete(ImageDownloader(docker=docker_).subvert(workers="fast"))
 
+        self.log(message="Download the challenge image")
         self.runtime_workspace = workspace_
 
     def challenge(self, ctx: Chrome, model):
@@ -381,7 +381,8 @@ def is_flagged_flow():
                 WebDriverWait(ctx, 1, 0.1).until(
                     EC.visibility_of_element_located((By.XPATH, "//div[@class='error-text']"))
                 )
-                if urllib.request.getproxies():
+                self.threat += 1
+                if urllib.request.getproxies() and self.threat >1:
                     logger.warning("Your proxy IP may have been flagged.")
                 return True
             except TimeoutException:

src/services/hcaptcha_challenger/solutions/kernel.py

@@ -12,9 +12,9 @@
 
 
 class Solutions:
-    def __init__(self, flag: str, path_rainbow: str):
+    def __init__(self, name: str, path_rainbow: str = None):
         self.path_rainbow = "rainbow.yaml" if path_rainbow is None else path_rainbow
-        self.flag = flag
+        self.flag = name
         self.rainbow_table = self.build_rainbow(path_rainbow=self.path_rainbow)
 
     @staticmethod
@@ -75,10 +75,10 @@ def match_rainbow(self, img_stream: bytes, rainbow_key: str) -> Optional[bool]:
     @staticmethod
     def download_model_(dir_model, path_model, model_src, model_name):
         """Download the de-stylized binary classification model"""
-        if not os.path.exists(dir_model):
-            os.mkdir(dir_model)
         if os.path.exists(path_model):
             return
+        if not os.path.exists(dir_model):
+            os.mkdir(dir_model)
 
         if not model_src.lower().startswith("http"):
             raise ValueError from None
@@ -92,16 +92,3 @@ def download_model_(dir_model, path_model, model_src, model_name):
     def solution(self, img_stream, **kwargs) -> bool:
         """Implementation process of solution"""
         raise NotImplementedError
-
-
-class RainbowSeaplane(Solutions):
-    """Handle challenge 「seaplane」"""
-
-    def __init__(self, path_rainbow=None):
-        super(RainbowSeaplane, self).__init__(flag="Rainbow", path_rainbow=path_rainbow)
-
-        self.rainbow_key = "seaplane"
-
-    def solution(self, img_stream, **kwargs) -> bool:
-        """Implementation process of solution"""
-        return self.match_rainbow(img_stream, self.rainbow_key)

src/services/hcaptcha_challenger/solutions/de_stylized.py → src/services/hcaptcha_challenger/solutions/resnet.py

@@ -6,7 +6,7 @@
 import os
 import time
 import warnings
-from typing import Optional
+from typing import List, Callable, Union
 
 import cv2
 import numpy as np
@@ -17,66 +17,84 @@
 warnings.filterwarnings("ignore", category=UserWarning)
 
 
-class DeStylized(Solutions):
-    def __init__(self, dir_model: str, onnx_prefix: str, path_rainbow: Optional[str] = None):
-        super(DeStylized, self).__init__(flag="de-stylized", path_rainbow=path_rainbow)
-
-        self.dir_model = "./model" if dir_model is None else dir_model
-
-        self.onnx_prefix = onnx_prefix
+class ResNetFactory(Solutions):
+    def __init__(self, _onnx_prefix, _name, _dir_model: str, path_rainbow=None):
+        """
 
+        :param _name: 日志打印显示的标记
+        :param _dir_model: 模型所在的本地目录
+        :param _onnx_prefix: 模型文件名，远程仓库文件和本地的一致。也用于拼接下载链接，因此该参数不允许用户自定义，
+          仅支持在范围内选择。
+        :param path_rainbow: 彩虹表本地路径，可选。
+        """
+        super().__init__(_name, path_rainbow=path_rainbow)
+        self.dir_model = _dir_model
         self.onnx_model = {
-            "name": f"{self.onnx_prefix}(de-stylized)_model",
-            "path": os.path.join(self.dir_model, f"{self.onnx_prefix}.onnx"),
-            "src": f"https://github.com/QIN2DIM/hcaptcha-challenger/releases/download/model/{self.onnx_prefix}.onnx",
+            "name": _name,
+            "path": os.path.join(_dir_model, f"{_onnx_prefix}.onnx"),
+            "src": f"https://github.com/QIN2DIM/hcaptcha-challenger/releases/download/model/{_onnx_prefix}.onnx",
         }
 
-        self.flag = self.onnx_model["name"]
-
     def download_model(self):
-        """Download the de-stylized binary classification model"""
+        """Download the ResNet ONNX classification model"""
         Solutions.download_model_(
             dir_model=self.dir_model,
             path_model=self.onnx_model["path"],
             model_src=self.onnx_model["src"],
             model_name=self.onnx_model["name"],
         )
 
-    def solution(self, img_stream, **kwargs) -> bool:
-        """Implementation process of solution"""
-        raise NotImplementedError
+    def classifier(self, img_stream, feature_filters: Union[Callable, List[Callable]] = None):
+        img_arr = np.frombuffer(img_stream, np.uint8)
+        img = cv2.imdecode(img_arr, flags=1)
 
+        if feature_filters is not None:
+            if not isinstance(feature_filters, list):
+                feature_filters = [feature_filters]
+            for tnt in feature_filters:
+                if not tnt(img):
+                    return False
 
-class HorsesDrawnWithFlowers(DeStylized):
-    """Handle challenge「Please select all the horses drawn with flowers」"""
+        self.download_model()
 
-    def __init__(self, dir_model, path_rainbow=None):
-        super().__init__(
-            dir_model=dir_model, path_rainbow=path_rainbow, onnx_prefix="horses_drawn_with_flowers"
-        )
-        self.rainbow_key = "horses drawn with flowers"
+        img = cv2.resize(img, (64, 64))
+        blob = cv2.dnn.blobFromImage(img, 1 / 255.0, (64, 64), (0, 0, 0), swapRB=True, crop=False)
 
-    @staticmethod
-    def is_drawn_with_flowers(img) -> bool:
-        """Work in progress"""
+        net = cv2.dnn.readNetFromONNX(self.onnx_model["path"])
+
+        net.setInput(blob)
+
+        out = net.forward()
+
+        if not np.argmax(out, axis=1)[0]:
+            return True
+        return False
+
+    def solution(self, img_stream, **kwargs) -> bool:
+        """Implementation process of solution"""
+
+
+class ResNetSeaplane(ResNetFactory):
+    """Handle challenge 「seaplane」"""
 
-    def classifier(self, img_stream) -> bool:
-        """Work in progress"""
+    def __init__(self, dir_model: str, path_rainbow=None):
+        _onnx_prefix = "seaplane"
+        super().__init__(_onnx_prefix, f"{_onnx_prefix}(resnet)_model", dir_model, path_rainbow)
 
     def solution(self, img_stream, **kwargs) -> bool:
         """Implementation process of solution"""
+        return self.classifier(img_stream, feature_filters=None)
 
 
-class ElephantsDrawnWithLeaves(DeStylized):
+class ElephantsDrawnWithLeaves(ResNetFactory):
     """Handle challenge 「Please select all the elephants drawn with leaves」"""
 
     def __init__(self, dir_model, path_rainbow=None):
+        _onnx_prefix = "elephants_drawn_with_leaves"
+        self.rainbow_key = _onnx_prefix
         super().__init__(
-            dir_model=dir_model,
-            path_rainbow=path_rainbow,
-            onnx_prefix="elephants_drawn_with_leaves",
+            _onnx_prefix, f"{_onnx_prefix}(de-stylized)_model", dir_model, path_rainbow
         )
-        self.rainbow_key = "elephants drawn with leaves"
 
     @staticmethod
     def is_drawn_with_leaves(img) -> bool:
@@ -95,32 +113,24 @@ def is_drawn_with_leaves(img) -> bool:
             return True
         return False
 
-    def classifier(self, img_stream) -> bool:
-        img_arr = np.frombuffer(img_stream, np.uint8)
-        img = cv2.imdecode(img_arr, flags=1)
-
-        # de-stylized
-        if not self.is_drawn_with_leaves(img):
-            return False
-        self.download_model()
-
-        img = cv2.resize(img, (64, 64))
-        blob = cv2.dnn.blobFromImage(img, 1 / 255.0, (64, 64), (0, 0, 0), swapRB=True, crop=False)
-
-        net = cv2.dnn.readNetFromONNX(self.onnx_model["path"])
-
-        net.setInput(blob)
-
-        out = net.forward()
-
-        if not np.argmax(out, axis=1)[0]:
-            return True
-        return False
-
     def solution(self, img_stream, **kwargs) -> bool:
         """Implementation process of solution"""
         match_output = self.match_rainbow(img_stream, rainbow_key=self.rainbow_key)
         if match_output is not None:
             time.sleep(0.3)
             return match_output
-        return self.classifier(img_stream)
+        return self.classifier(img_stream, feature_filters=self.is_drawn_with_leaves)
+
+
+class HorsesDrawnWithFlowers(ResNetFactory):
+    """Handle challenge「Please select all the horses drawn with flowers」"""
+
+    def __init__(self, dir_model, path_rainbow=None):
+        _onnx_prefix = "horses_drawn_with_flowers"
+        self.rainbow_key = _onnx_prefix
+        super().__init__(
+            _onnx_prefix, f"{_onnx_prefix}(de-stylized)_model", dir_model, path_rainbow
+        )
+
+    def solution(self, img_stream, **kwargs) -> bool:
+        """Implementation process of solution"""

src/services/hcaptcha_challenger/solutions/sk_recognition.py

@@ -17,7 +17,7 @@
 
 class SKRecognition(Solutions):
     def __init__(self, path_rainbow: Optional[str] = None):
-        super().__init__(flag="skimage_model", path_rainbow=path_rainbow)
+        super().__init__("skimage_model", path_rainbow)
 
     @staticmethod
     def _weight_mean_color(graph_, src: int, dst: int, n: int):  # noqa