Merge pull request #6 from WwwwwyDev/develop

Develop

README.md

@@ -29,9 +29,14 @@ The script for selenium in python
 You can use Crawlipt to driver the selenium by script in python.The script adopts JSON format for better cross language operations and physical storage.
 
 ## installing
-You can use pip or pip3 to install the crawlist\
+You can use pip or pip3 to install the crawlipt
+
 `pip install crawlipt` or `pip3 install crawlipt`
 
+If you have already installed crawlelip, you may need to update to the latest version
+
+`pip install --upgrade crawlipt`
+
 ## quickly start
 ```python
 from webdriver_manager.chrome import ChromeDriverManager

crawlipt/__init__.py

@@ -1,4 +1,5 @@
 from .script import Script
 from .action import Action
 from .annotation import check, alias
-from __version__ import __version__ as version
+from .pojo import Variable, VariableBase
+from .__version__ import __version__ as version

crawlipt/__version__.py

@@ -3,6 +3,6 @@
 # 88YbdP88   8P   88"""   dP__Yb  Yb      88"Yb   dP__Yb  Yb  "88 88""
 # 88 YY 88  dP    88     dP""""Yb  YboodP 88  Yb dP""""Yb  YboodP 888888
 
-VERSION = (0, 0, 4)
+VERSION = (0, 0, 5)
 
 __version__ = '.'.join(map(str, VERSION))

crawlipt/actions/__init__.py

@@ -5,3 +5,4 @@
 from .slide import Slide
 from .select import Select
 from .get import Get
+from .window import Window

crawlipt/actions/window.py

@@ -0,0 +1,45 @@
+from selenium.common import NoSuchWindowException
+from selenium.webdriver.common.by import By
+from selenium.webdriver.remote.webdriver import WebDriver
+
+from crawlipt.annotation import check, alias
+
+
+class Window:
+    @staticmethod
+    @check(exclude="driver")
+    def clear(driver: WebDriver) -> None:
+        """
+        close all windows
+        :param driver: selenium webdriver
+        """
+        for _ in range(driver.window_handles.__len__()-1):
+            driver.close()
+        driver.get("data:,")
+
+    @staticmethod
+    @check(exclude="driver")
+    def back(driver: WebDriver) -> None:
+        """
+        Goes one step backward in the browser history.
+        :param driver: selenium webdriver
+        """
+        driver.back()
+
+    @staticmethod
+    @check(exclude="driver")
+    def forword(driver: WebDriver) -> None:
+        """
+        Goes one step forward in the browser history.
+        :param driver: selenium webdriver
+        """
+        driver.forward()
+
+    @staticmethod
+    @check(exclude="driver")
+    def close(driver: WebDriver) -> None:
+        """
+        Closes the current window.
+        :param driver: selenium webdriver
+        """
+        driver.close()

crawlipt/annotation.py

@@ -54,6 +54,8 @@ def inner_wrapper(*args, **kwargs):
                     continue
                 if all_kwargs[name] == "__PRE_RETURN__":
                     continue
+                if isinstance(all_kwargs[name], str) and all_kwargs[name].startswith("__v-") and all_kwargs[name].endswith("__"):
+                    continue
                 if isinstance(all_kwargs[name], int) and type_.annotation is float:
                     all_kwargs[name] = float(all_kwargs.get(name))
                 if not isinstance(all_kwargs[name], type_.annotation):

crawlipt/pojo.py

@@ -0,0 +1,47 @@
+import json
+from typing import Any
+
+from crawlipt.annotation import check
+
+
+class VariableBase:
+
+    @check
+    def get(self, key: str) -> Any:
+        raise NotImplementedError
+
+    @check(exclude="value")
+    def set(self, key: str, value: Any) -> Any:
+        raise NotImplementedError
+
+    @check
+    def __contains__(self, key: str):
+        raise NotImplementedError
+
+
+class Variable(VariableBase):
+    @check
+    def __init__(self, values: dict | str):
+        if isinstance(values, str):
+            values: dict = json.load(values)
+        self.values = values
+
+    @check
+    def get(self, key: str) -> Any:
+        return self.values.get(key)
+
+    @check(exclude="value")
+    def set(self, key: str, value: Any):
+        self.values[key] = value
+
+    @check
+    def __contains__(self, key: str):
+        return key in self.values
+
+
+class VariableError(Exception):
+    def __init__(self, message):
+        self.message = message
+
+    def __str__(self):
+        return self.message

crawlipt/script.py

@@ -12,6 +12,8 @@
 from selenium.webdriver.support.wait import WebDriverWait
 import copy
 
+from crawlipt.pojo import VariableError, VariableBase
+
 
 class ScriptError(Exception):
     def __init__(self, e: Exception, method: str, deep: str):
@@ -34,7 +36,6 @@ def __str__(self):
         error = "[" + self.e.__class__.__name__ + "] " + self.e.__str__() + "\n"
         return "(Deep:%s method:%s) arguments is wrong\n" % (self.deep, self.method) + error + info + params + doc
 
-
 def dfs_search(obj):
     for parent in obj.__bases__:
         if parent == object:
@@ -61,7 +62,7 @@ def get_dict(obj):
 class ScriptProcess:
     ACTIONS = get_dict(Action)
     CONDITIONS = get_dict(Condition)
-    __POP_KEY = {"method", "next", "if", "check", "condition"}
+    __POP_KEY = {"method", "next", "if", "check", "condition", "loop"}
 
     @staticmethod
     @check
@@ -113,6 +114,9 @@ def syntax_check(script: dict | str, pre_deep: str = "") -> None:
                                                     pre_deep=pre_deep,
                                                     current_deep=current_deep)
                 loop_script = loop_temp.get("script")
+                if not loop_script:
+                    msg = "(Deep %s) loop must set the param of script" % (pre_deep + str(current_deep))
+                    raise ScriptError(ParamTypeError(msg), "", pre_deep + str(current_deep))
                 ScriptProcess.syntax_check(loop_script, pre_deep=pre_deep + str(current_deep) + "->")
                 script = script.get("next")
                 continue
@@ -185,7 +189,8 @@ def _process_script(script: dict, global_script: dict, webdriver: WebDriver, int
                 while_condition = loop_temp.get("while")
                 loop_script = loop_temp.get("script")
                 if while_condition and cnt:
-                    while ScriptProcess.__process_condition(temp_condition=while_condition, webdriver=webdriver) and cnt:
+                    while ScriptProcess.__process_condition(temp_condition=while_condition,
+                                                            webdriver=webdriver) and cnt:
                         ScriptProcess._process_script(script=loop_script,
                                                       global_script=global_script,
                                                       webdriver=webdriver,
@@ -250,6 +255,22 @@ def _process_script(script: dict, global_script: dict, webdriver: WebDriver, int
             time.sleep(random.uniform(interval / 2, interval))
         return pre_return
 
+    @staticmethod
+    @check
+    def _replace_variable(script: dict, variable: VariableBase) -> None:
+        while script:
+            for key in script.keys():
+                if key not in ScriptProcess.__POP_KEY and isinstance(script[key], str) and script[key].startswith("__v-") and script[key].endswith("__"):
+                    variable_name = script[key][4:-2]
+                    if variable_name not in variable:
+                        msg = f"The {variable_name} is not defined."
+                        raise VariableError(msg)
+                    script[key] = variable.get(variable_name)
+            for key in script.keys():
+                if isinstance(script[key], dict):
+                    ScriptProcess._replace_variable(script=script[key], variable=variable)
+            script = script.get("next")
+
     @staticmethod
     @check
     def generate(scripts: list | dict | str) -> dict:
@@ -323,6 +344,8 @@ def add_condition(func: callable) -> None:
             ScriptProcess.CONDITIONS[func.__crawlipt_func_name__] = func_bak
 
 
+
+
 class Script(ScriptProcess):
 
     @check
@@ -331,7 +354,7 @@ def __init__(self, script: dict | str | list, global_script: dict | str | list =
         """
         Script Parser
         :param script: Need a str of json or dict or list steps that conforms to syntax conventions
-        :param b: This script will be executed before every actions
+        :param global_script: This script will be executed before every actions
         :param interval: The direct interval between two consecutive scripts
         :param wait: The longest wait time before presence of element located
         :param is_need_syntax_check: Whether the script need a syntax check
@@ -351,15 +374,23 @@ def __init__(self, script: dict | str | list, global_script: dict | str | list =
         self.wait = wait
 
     @check
-    def process(self, webdriver: WebDriver) -> Any:
+    def process(self, webdriver: WebDriver, variable: VariableBase = None) -> Any:
         """
         process the script
         """
         if self.is_need_syntax_check:
             ScriptProcess.syntax_check(self.script)
             ScriptProcess.syntax_check(self.global_script)
-        return ScriptProcess._process_script(script=self.script,
-                                             global_script=self.global_script,
+        script = copy.deepcopy(self.script)
+        global_script = copy.deepcopy(self.global_script)
+        if variable:
+            ScriptProcess._replace_variable(script, variable)
+            ScriptProcess._replace_variable(global_script, variable)
+            if self.is_need_syntax_check:
+                ScriptProcess.syntax_check(script)
+                ScriptProcess.syntax_check(global_script)
+        return ScriptProcess._process_script(script=script,
+                                             global_script=global_script,
                                              webdriver=webdriver,
                                              interval=self.interval,
                                              wait=self.wait)

setup.py

@@ -18,7 +18,7 @@
 EMAIL = 'wwy20001014@foxmail.com'
 AUTHOR = 'WwwwwyDev'
 REQUIRES_PYTHON = '>=3.10.0'
-VERSION = '0.0.4'
+VERSION = '0.0.5'
 
 # What packages are required for this module to be executed?
 REQUIRED = [

test.py

@@ -10,6 +10,7 @@
 import crawlipt as cpt
 import ddddocr as docr
 
+
 def get_driver(is_headless=False, is_eager=False):
     option = wd.ChromeOptions()
     arguments = [
@@ -147,7 +148,7 @@ def test_04(self):
             "method": "getInnerText",
             "xpath": "//*[@id=\"trans-selection\"]/div/span",
         }]
-        result = cpt.Script(step, interval=0.1)(webdriver)
+        result = cpt.Script(step, interval=0)(webdriver)
         print(result)
         webdriver.quit()
 
@@ -170,7 +171,7 @@ def test05(self):
         webdriver.quit()
 
     def test06(self):
-        # webdriver = get_driver()
+        webdriver = get_driver()
         class A:
             @staticmethod
             @cpt.check(exclude="driver")
@@ -210,9 +211,9 @@ def crackCaptcha(driver: WebDriver, xpath: str) -> str:
             "method": "click",
             "xpath": "//*[@id=\"dosubmit\"]",
         }]
-        result = cpt.Script(step, interval=1)
-        # print(result)
-        # webdriver.quit()
+        result = cpt.Script(step, interval=1)(webdriver)
+        print(result)
+        webdriver.quit()
 
     def test07(self):
         webdriver = get_driver()
@@ -309,6 +310,43 @@ def test_conditions(self):
         }]
         cpt.Script(step, interval=3)
 
+    def test_variable(self):
+        webdriver = get_driver()
+        step = [{
+            "method": "redirect",
+            "url": "https://www.baidu.com/",
+        }, {
+            "method": "input",
+            "xpath": "//*[@id=\"kw\"]",
+            "text": "__v-searchKey__",
+            "if": {
+                "condition": "presence",
+                "xpath": "__v-button_xpath__"
+            }
+        }, {
+            "method": "clear"
+        }]
+        v1 = cpt.Variable({
+            "searchKey": "hello",
+            "button_xpath": "//*[@id=\"su\"]"
+        })
+        v2 = cpt.Variable({
+            "searchKey": "world",
+            "button_xpath": "//*[@id=\"su\"]"
+        })
+        v3 = cpt.Variable({
+            "searchKey": "world",
+            "button_xpath": "//*[@id=\"su_no_existence\"]"
+        })
+        loader = cpt.Script(step, interval=3)
+        loader.process(webdriver=webdriver,
+                       variable=v1)
+        loader.process(webdriver=webdriver,
+                       variable=v2)
+        loader.process(webdriver=webdriver,
+                       variable=v3)
+        webdriver.quit()
+
 
 if __name__ == '__main__':
     unittest.main()