【抛砖引玉】基于装饰器实现广告弹窗判断

前言

如何批量实现在某个指令前面调用一个前置流程？？？

因为广告是随机出现的我也不知道在那个点击元素指令前面出现

如何实现批量在某一个指令前面调用流程 看下面的视频

视频教学

装饰器去广告

https://www.bilibili.com/video/BV1YM41197bx/?spm_id_from=333.999.0.0

编码版调用可视化

https://www.bilibili.com/video/BV1tN4y1c7HF/?spm_id_from=333.999.0.0

编码版调用可视化

import xbot_visual#要引入这个包
def 调用子流程2():
    子流程2返回集合 = xbot_visual.process.run(process="process2", package=__name__, inputs={
    "输入1": "1",
    "输入2": "1",
    }, outputs=[], _block=())
    print(子流程2返回集合.输出2)

调用子流程2()


###找到可视化的文件名字 保存文件 

装饰器去广告

# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能，如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv

def main(args):
    pass

import xbot_visual#要引入这个包
def 调用去广告():
    去广告返回集合 = xbot_visual.process.run(process="process2", package=__name__, inputs={

    }, outputs=[], _block=())
    print("")




##固定写法
import xbot_visual
from xbot import print
def decorator(func):
    def wrapper(*args,**kwargs):
        # 添加你的装饰器逻辑
        print("指令前操作: -----下面写需要添加的功能比如去广告")
        ##调用去广告流程
        调用去广告()
        res=func(*args,**kwargs)
        print("指令后操作: -----下面写需要添加的功能比如去广告")
        return res
    return wrapper


#自己改
xbot_visual.web.element.click=decorator(xbot_visual.web.element.click)
xbot_visual.web.create=decorator(xbot_visual.web.create)

参考文章

https://www.yingdao.com/community/detaildiscuss?id=c26815a8-5eee-4ab4-87dd-45fb1db16170

代码2

调用流程

封装好的自定义指令

流程参数可以用字典的形式传入，支持跨域，web和win通吃，全局调用就是如此简单

python代码

import xbot
import xbot_visual
import importlib
import traceback
import json
import os
import re
from xbot_visual._core import dict_to_object
from functools import wraps

# 全局只执行一次
_PATCHED_DONE = False
# 防自调用锁
_IS_RUNNING_CLOSE = False

# 正则匹配路径
file_pattern = re.compile(r"(.+\\apps\\[^\\]+\\(xbot_robot|xbot_extensions\\[^\\]+))")

def self_code():
    with open(os.path.join(os.path.dirname(__file__),'package.json'),'r',encoding='utf-8') as f:
        json_obj = json.load(f)
    return json_obj['activity_code']

def find_process_name(source_folder, flowname):
    with open(os.path.join(source_folder, 'package.json'),'r',encoding='utf-8') as f:
        json_obj = json.load(f)
        flows = json_obj['flows']
        for flow in flows:
            if flow['name'] == flowname:
                filename = flow['filename']
                if filename == 'main':
                    raise ValueError('无法调用主流程')
                return filename
    raise ValueError(f'未找到名为【{flowname}】的流程，请检查！')

def extract_source_process():
    activity_code = self_code()
    lines = traceback.format_stack()
    for line in reversed(lines):
        results = re.findall('File "(.*?)"', line)
        if results:
            if activity_code in results[0]:
                continue
            if 'xbot_extensions' in results[0]:
                source_folder = os.path.dirname(results[0])
                return source_folder, f'xbot_extensions.{os.path.basename(source_folder)}.'
            if 'xbot_robot' in results[0]:
                source_folder = os.path.dirname(results[0])
                return source_folder, f'xbot_robot.'
    raise ValueError('发生了不可预知错误')

# ====================== 调用关闭弹窗流程 ======================
def invoke_process(flowname, inputs, file_path):
    global _IS_RUNNING_CLOSE
    if inputs is None or inputs == "":
        inputs = {}

    app_xbot_dir, prefix = extract_source_process()
    process_filename = find_process_name(app_xbot_dir, flowname)
    module_name = f"{prefix}{process_filename}"
    
    mod = importlib.import_module(module_name)
    mod.main(inputs)
    
    return dict_to_object("dynamic_process_result", inputs)

# ====================== 关闭弹窗（防自调用） ======================
def close_popup_ele(flowname, inputs):
    global _IS_RUNNING_CLOSE
    if _IS_RUNNING_CLOSE:
        return
    try:
        _IS_RUNNING_CLOSE = True
        lj = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "xbot_robot", "main.pybx")
        invoke_process(flowname, inputs, lj)
    except Exception:
        pass
    finally:
        _IS_RUNNING_CLOSE = False

# ====================== 装饰器 ======================
def auto_popup_deco(flowname, inputs):
    def decorator(raw_func):
        @wraps(raw_func)
        def wrapper(*args, **kwargs):
            if not _IS_RUNNING_CLOSE:
                try:
                    close_popup_ele(flowname, inputs)
                except:
                    pass
            return raw_func(*args, **kwargs)
        return wrapper
    return decorator

# ====================== 全覆盖精准劫持 ======================
def web_tc(flowname, inputs=None):
    global _PATCHED_DONE
    if _PATCHED_DONE:
        return
    _PATCHED_DONE = True

    deco = auto_popup_deco(flowname, inputs)
    
    # 1. web.element 方法
    WEB_ELEMENT = {
        'check', 'click', 'data_scraping', 'databook', 'download', 'drag_to',
        'get_all_elements', 'get_associated_elements', 'get_bounding',
        'get_details', 'get_element', 'get_select_item', 'hover', 'input',
        'input_password', 'iter_all_elements', 'resolve_element',
        'screenshot', 'select', 'set_attribute', 'set_value', 'utility',
        'valid', 'visual_action', 'wait'
    }

    # 2. win32 桌面方法
    WIN32 = {
        'click_mouse', 'clipboard', 'clipboard_clear', 'clipboard_get_text',
        'clipboard_set_file', 'clipboard_set_text', 'get_ime',
        'get_mouse_position', 'get_selected_text', 'lock_screen',
        'manual_motion_off', 'manual_motion_on', 'minimize_all',
        'move_mouse', 'send_keys', 'set_ime', 'unlock_screen',
        'visual_action', 'wheel_mouse', 'window'
    }

    # 3. win32.element 方法
    WIN32_ELEMENT = {
        'call_transaction', 'check', 'click', 'click_toolbar_button',
        'drag_to', 'expand_tree_and_select_node', 'get_all_elements',
        'get_associated_elements', 'get_bounding', 'get_details',
        'get_element', 'get_select_item', 'get_table_cell_by_rownum_and_columnnum',
        'get_table_cells', 'get_table_column_count', 'get_table_row_count',
        'hover', 'input', 'input_password', 'iter_all_elements',
        'read_status_bar', 'resolve_element', 'screenshot', 'select',
        'select_date_in_calendar', 'select_menu_item', 'set_value',
        'utility', 'valid', 'visual_action', 'wait', 'wait_load_completed'
    }

    # 开始劫持
    for name in WEB_ELEMENT:
        try:
            f = getattr(xbot_visual.web.element, name)
            if callable(f): setattr(xbot_visual.web.element, name, deco(f))
        except: pass

    for name in WIN32:
        try:
            f = getattr(xbot_visual.win32, name)
            if callable(f): setattr(xbot_visual.win32, name, deco(f))
        except: pass

    for name in WIN32_ELEMENT:
        try:
            f = getattr(xbot_visual.win32.element, name)
            if callable(f): setattr(xbot_visual.win32.element, name, deco(f))
        except: pass
def main(args):
    pass