全局调用自定义指令集来咯支持跨域,装饰器版本,无需并发全局调用更简单
评论
收藏

全局调用自定义指令集来咯支持跨域,装饰器版本,无需并发全局调用更简单

经验分享
【耐家军】DC
2026-04-30 17:18·浏览量:174
【耐家军】DC
影刀专家
影刀认证工程师
发布于 2026-04-30 17:18174浏览

效果展示

调用流程

封装好的自定义指令

流程参数可以用字典的形式传入,支持跨域,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

推荐:DCpage:基于 DrissionPage 打造的影刀专属自定义指令库

收藏3
全部评论1
最新
发布评论
评论