



流程参数可以用字典的形式传入,支持跨域,web和win通吃,全局调用就是如此简单
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