Adding modular support for custom validators

scylla/validator.py

@@ -5,6 +5,7 @@
 
 from .loggings import logger
 from .tcpping import ping
+from .validators import *
 
 IP_CHECKER_API = 'http://api.ipify.org/?format=json'
 IP_CHECKER_API_SSL = 'https://api.ipify.org/?format=json'
@@ -83,9 +84,22 @@ def validate_proxy(self):
             logger.debug('Catch requests.RequestException for proxy ip: {}'.format(self._host))
             logger.debug(e.__str__())
 
+    def validate_proxy_through_validators(self):
+        protocol = 'https' if self._using_https else 'http'
+        proxy_str = '{}://{}:{}'.format(protocol, self._host, self._port)
+        validators = [validator() for validator in all_validators]
+
+        is_valid = all([_validator.validate(proxy_str) for _validator in validators])
+        if is_valid:
+            self._valid = True
+        else:
+            self._valid = False
+
     def validate(self):
         self.validate_latency()
         self.validate_proxy()
+        if self._valid:
+            self.validate_proxy_through_validators()
 
     @property
     def latency(self):

scylla/validators/__init__.py

@@ -0,0 +1,5 @@
+from .bilibili_validator import BilibiliValidator
+from .twitter_validator import TwitterValidator
+
+# import Validator and add it into all_validators to enable it
+all_validators = [TwitterValidator]

scylla/validators/base_validator.py

@@ -0,0 +1,29 @@
+import requests
+from ..loggings import logger
+
+
+class BaseValidator(object):
+
+    check_url = 'https://www.google.com'
+
+    def __init__(self):
+        pass
+
+    def __str__(self):
+        return self.__class__.__name__
+
+    def validate(self, proxy_str):
+        try:
+            r = requests.get(self.check_url, proxies={'https': proxy_str, 'http': proxy_str}, verify=False, timeout=15)
+            if not r.ok:
+                logger.debug('Catch {} for proxy ip: {} when connecting to {}'.format(r.status_code, proxy_str.split(':')[0], self.check_url))
+                return False
+            return True
+        except requests.Timeout:
+            logger.debug('Catch requests.Timeout for proxy ip: {} when connecting to {}'.format(proxy_str.split(':')[0], self.check_url))
+            return False
+        except requests.RequestException as e:
+            logger.debug('Catch requests.RequestException for proxy ip: {} when connecting to {}'.format(proxy_str.split(':')[0], self.check_url))
+            logger.debug(e.__str__())
+            return False
+

scylla/validators/bilibili_validator.py

@@ -0,0 +1,8 @@
+from .base_validator import BaseValidator
+
+
+class BilibiliValidator(BaseValidator):
+
+    check_url: str = 'https://api.bilibili.com/'
+
+

scylla/validators/twitter_validator.py

@@ -0,0 +1,8 @@
+from .base_validator import BaseValidator
+
+
+class TwitterValidator(BaseValidator):
+
+    check_url: str = 'https://www.twitter.com/'
+
+