Class ScriptProcess

crawlipt/script.py:65–516

class ScriptProcess:
    ACTIONS = get_dict(Action)
    CONDITIONS = get_dict(Condition)
    __POP_KEY = {"method", "next", "if", "check", "condition", "loop", "return_flag", "while", "fail_script"}

    @staticmethod
    @check
    def __condition_check(temp_condition: dict, name: str, pre_deep: str, current_deep: int,
                          return_record: dict) -> None:
        condition = temp_condition.get("condition")
        temp_args = {"driver": None}
        if not condition:
            msg = "(Deep %s) condition of '%s' is missing" % (pre_deep + str(current_deep), name)
            raise ScriptSyntaxError(ParamTypeError(msg), "", pre_deep + str(current_deep))
        if not isinstance(condition, str):
            msg = "(Deep %s) The value of condition must be type of str" % (pre_deep + str(current_deep))
            raise ScriptSyntaxError(ParamTypeError(msg), "", pre_deep + str(current_deep))
        if condition.startswith("__not-") and condition.endswith("__"):
            condition = condition[6: -2]
        if condition not in ScriptProcess.CONDITIONS.keys():
            msg = "(Deep %s) Could not found the Condition Method in '%s'" % (pre_deep + str(current_deep), name)
            raise ScriptSyntaxError(ParamTypeError(msg), condition, pre_deep + str(current_deep))
        fail_script = temp_condition.get("fail_script")
        if fail_script:
            ScriptProcess.syntax_check(script=fail_script,
                                       pre_deep=pre_deep + str(current_deep) + "->",
                                       return_record=return_record)
        return_flag = temp_condition.get("return_flag")
        if return_flag and not isinstance(return_flag, str):
            msg = "(Deep %s) return_flag must be the type of str" % (pre_deep + str(current_deep))
            raise ScriptSyntaxError(ParamTypeError(msg), condition, pre_deep + str(current_deep))
        for key, value in temp_condition.items():
            if key.lower() not in ScriptProcess.__POP_KEY:
                temp_args[key] = value
        if "store" in signature(ScriptProcess.CONDITIONS[condition]).parameters:
            temp_args["store"] = None
        if return_record:
            for key, value in temp_args.items():
                if isinstance(value, str) and value.startswith("__rf-") and value.endswith("__"):
                    if value[5:-2] not in return_record.keys():
                        msg = f"The return_flag:{value[5:-2]} cannot be find in history record."
                        raise ScriptSyntaxError(ParamTypeError(msg), condition, pre_deep + str(current_deep))
                    annotation = signature(ScriptProcess.CONDITIONS[condition]).parameters[key].annotation
                    pre_return_flag = return_record[value[5:-2]]
                    if pre_return_flag is Any or annotation is Any:
                        continue
                    if pre_return_flag != annotation:
                        msg = f"The return_flag is {pre_return_flag}, But parameter {key} is {annotation}."
                        raise ScriptSyntaxError(ParamTypeError(msg), condition, pre_deep + str(current_deep))
        try:
            if return_flag:
                return_record[return_flag] = signature(ScriptProcess.CONDITIONS[condition]).return_annotation
            ScriptProcess.CONDITIONS[condition](**temp_args)
        except TypeError as e:
            raise ScriptSyntaxError(e, condition, pre_deep + str(current_deep))
        except AssertionError as e:
            raise ScriptSyntaxError(e, condition, pre_deep + str(current_deep))
        except ParamTypeError as e: