影刀

1、问题描述 最近有客户反馈影刀在操作Windwos软件的时候卡死，等待元素的指令设置了最长等待300秒但是客户那边显示在等待元素指令卡了50分钟，我远程过去排查后发现客户操作的目标软件卡死了下面图片就是当时的截图。 之前没有遇到过这种情况，我当时怀疑是否影刀操作未响应的窗口都会卡住，所以让AI写了一个能主动触发卡死的程序做测试，测试发现确实会卡住，影刀类似被阻塞了，等目标窗口结束未响应后影刀能继续操作。 2、解决方案 如果目标软件一直未响应影刀就会一直被阻塞导致整个流程卡死，我们希望的时检测到目标软件未响应了就关闭目标软件重新打开（因为未响应的情况还是比较少）。 Python可以通过窗口名称、进程PID、进程名称等获取到进程后判断进程是否卡住，所以我们思路就是给操作win元素的指令套个装饰器，每次操作前判断进程是否卡住，如果卡住则直接报错不要继续操作了。 代码如下 windows_check.py import ctypes import psutil import win32gui import win32process import fnmatch user32 = ctypes.WinDLL("user32", use_last_error=True) IsHungAppWindow = user32.IsHungAppWindow IsHungAppWindow.argtypes = [ctypes.c_void_p] IsHungAppWindow.restype = ctypes.c_bool def get_hwnd_by_window_title_pattern(pattern: str): """ 窗口标题 通配符匹配，fnmatch语法 示例：*记事本*, Notepad*, *Chrome """ hwnd_list = [] def enum_callback(hwnd, extra): wnd_title = win32gui.GetWindowText(hwnd) if win32gui.IsWindowVisible(hwnd): if fnmatch.fnmatch(wnd_title, pattern): hwnd_list.append(hwnd) win32gui.EnumWindows(enum_callback, None) return hwnd_list def is_window_hung(hwnd): """判断单个窗口是否卡死（True=卡死未响应）""" if not win32gui.IsWindow(hwnd): return None return IsHungAppWindow(hwnd) def check_by_title_pattern(title_pattern: str): hwnds = get_hwnd_by_window_title_pattern(title_pattern) if not hwnds: return {"ok": False, "msg": f"没有匹配标题【{title_pattern}】的窗口", "hung": None} res = [] for h in hwnds: hung = is_window_hung(h) res.append({"hwnd": h, "hung": hung}) return {"ok": True, "windows": res} registry.py ...

前言 最近在处理一个 SKA 客户问题的时候发现影刀自定义指令有个很坑的地方，这里记录一下。 问题描述 问题很简单，客户开发了一个自定义指令，这个指令被 100 多个应用使用，指令原本设计了 N 个参数，现在指令要跟新要增加1个参数。 类比到 python 里就是原本原本函数要3个变量，现在更新了函数多了一个变量，理论上来说只要新增的变量设置个默认值就不影响旧代码的使用。 在影刀中按理来说只要新参数设置默认值应该不会影响旧流程的调用，然而现实不是这样。 举例说明 这里我准备了一个案例，案例中的自定义指令需要传入两个参数 p1、p2 如下图 使用方调用这个指令并传入2个参数如下图 现在，更自定义指令新增加一个参数 p3 并且设置默认值为 None 如下图 这个时候更新应用中的指令，点击运行测试我们会发现一切都很顺利没有任何报错，如下图 一切都这么顺利，旧流程不用做任何修改就能使用新指令，所以问题在哪呢？ 在影刀中有 VisualFlow（可视化流程） 和 CodeFlow（代码流程） 两种方式，前面我们使用的是 VisualFlow 如果我们用代码的方式调用自定义指令会发生什么呢？ 我们会悲伤的发现使用代码调用的方式会报错，提示我们少传入了一个参数。这个情况放在客户的场景下就是，客户用代码的方式在100多个应用中调用了这个自定义指令，因为自定义指令更新现在需要去100多个应用中修改调用方式。 可以想像这绝对不是一个好的选择，哪么有办法既更新指令又可以不对旧流程做任何修改吗？有的兄弟！有的！ 解决方案 首先，我们点开可视化指令的源码，我们可以在源码中看到，影刀可视化调用自定义指令的代码是这样的 xbot_visual.process.run(process="xbot_extensions.activity_25e7d62a.process1", package=__name__, inputs={ "p1": "111", "p2": "222" }, outputs=[ ], _block=("main", 1, "a")) 可以看到是 VisualFlow 通过 xbot_visual 来调用的，并且这种调用方式支持不传参数 p3 ，在 CodeFlow 中调用是通过 xbot_extension 调用，这种调用就要求我们一定传入 p3 。 我们在深扒一下影刀自定义指令的源码，在源码中我们可以看到__init__.py 文件内容如下 import xbot_visual def process1(p1,p2,p3): """ a * @param p1， * @param p2， * @param p3， """ outputs = [] inputs = {"p1":p1,"p2":p2,"p3":p3} extension_module, activity_func = xbot_visual.process.activity_entry("xbot_extensions.activity_25e7d62a.process1", __name__) try: return xbot_visual.process.run(process="xbot_extensions.activity_25e7d62a.process1",package=__name__,inputs=inputs, outputs=outputs) finally: xbot_visual.process.replace_activity_module_to_entry_method("xbot_extensions.activity_25e7d62a.process1", extension_module, activity_func) 看到这个我们基本可以清楚 xbot_extensions.activity_25e7d62a.process1 走的就是 __init__.py 中的 process1 ，而必须传入 p3 的原因就是 process1 这个函数写死了必须要传 p3 ，所以我们只需要改一下函数定义，改成 def process1(p1, p2, p3=None) 问题迎刃而解。 ...

首先局域网中有一个调度模式电脑能访问的共享文件夹 使用 windows 的 mklink 命令将共享文件夹中指定的文件夹映射到影刀的录屏文件夹，这样就能实现所有调度模式的录屏文件统一管理 mklink /J "D:\Shadowbot\screencast" "F:\NAS上的实际文件夹路径"

测试页面（需要翻墙） 类似下图这样的弹窗，这种事 JS 原生的 Alert 弹窗，不再 HTML 标签里，无法选中弹窗的 DOM 对于获取这类弹窗的内容，可以用以下思路解决 向页面注入JS劫持 window.alert 方法，将 msg 保存到全局变量 RPA点击按钮触发弹窗，然后关闭弹窗 向页面注入JS，读取全局变量中的 msg 劫持 windw.alert 的 js function (element, input) { // 保存原始 alert window._originalAlert = window.alert; // 重写 alert，劫持弹窗内容 window.alert = function (msg) { window._lastAlertMsg = msg; // 存到全局变量 console.log("捕获到弹窗：", msg); // 调试打印 window._originalAlert(msg); // 继续弹出原弹窗 }; return null; } 读取页面全局变量的 js function (element, input) { // 读取劫持到的弹窗文本 let msg = window._lastAlertMsg || ""; console.log("最终弹窗内容：", msg); return msg; } 在影刀中的使用如下图

1. 前言 在之前尝试过 AI 驱动的RPA程序完成给定的任务，但是整个运行下来给人的感觉就是太慢了，当时跑一个淘宝搜索产品并获取产品信息的任务跑了快10分钟，总结下来慢的原因无非下面几点。 AI返回内容慢，通过OpenRouter调用gemini模型，一个请求要1-2分钟才能返回结果。 要实现程序 感知-决策-在感知-判断 就要调用多次模型，模型调用次数越多就越慢。 所以，之前AI驱动RPA属于对技术的探索，但没有实际意义， 这次我尝试用AI来处理在自动化中会遇到的一些复杂验证码问题 下面是这次需要解决的验证码的例子，这三个验证码都来自Temu平台。 2. 核心思路 整体思路很简单，把验证码截图给AI，让AI按要求返回坐标，程序解析坐标并点击 整体的代码设计如下图，首先是三个抽象类下面逐个解释一下。 Agent：用于调用AI接口，封装了图片解析、AI结果处理等功能，需要子类实现具体的AI接口 Capability：表示一种能力，就是给AI的提示词。 Handle：使用AI返回的结果处理问题 Agent有3个子类，分别是调用OpenAI规范的接口、调用影刀内置AI接口和调用影刀AIPower接口。 Capability的3个子类分别表示点选验证码处理、拖动验证码处理以及复选框处理 在做软件自动化时很容易遇到复选框无法判断是否勾选，这种情况就可以借助AI Handle有2个子类分表用于，处理点击操作和处理拖动操作 3. 代码实现 ability.py from abc import ABC, abstractmethod class Capability(ABC): """ 表示一种能力, AI能做什么操作 """ @property @abstractmethod def prompt(self): pass class ClickVerifier(Capability): @property def prompt(self): prompt = """ 请严格按照图片中给出的操作要求执行。 图片中已经明确标注了需要点击的目标及点击顺序， 你只需要根据图片内容，依次返回每一步需要点击的位置坐标。 规则： 1. 只返回图片中明确要求点击的内容，不要增加或省略步骤 2. 坐标基于当前输入图片的像素坐标 3. 坐标原点为图片左上角 (0, 0)，x 向右，y 向下 4. 每个坐标应尽量位于对应目标的可点击区域中心 5. 如果图片中某一步无法明确定位，请不要猜测，在 summary 中说明 输出要求： - 仅返回 JSON，不要输出任何多余文本 - 不要使用 markdown - JSON 结构必须严格如下： { "positions": [ { "x": xxx, "y": xxx }, { "x": xxx, "y": xxx } ], "summary": "简要说明这些坐标如何对应图片中的点击要求" } """ return prompt class DragVerifier(Capability): @property def prompt(self): prompt = """ 图片中已经明确标注了需要如何拖拽内容 你只需要根据图片内容，依次返回开始拖拽和结束拖拽的位置坐标 规则： 1. 坐标基于当前输入图片的像素坐标 2. 坐标原点为图片左上角 (0, 0)，x 向右，y 向下 3. 每个坐标应尽量位于对应目标的可点击区域中心 4. 如果图片中某一步无法明确定位，请不要猜测，在 summary 中说明 输出要求： - 仅返回 JSON，不要输出任何多余文本 - 不要使用 markdown - JSON 结构必须严格如下： { "positions": [ { "start": { "x": xxx, "y": xxx, }, "end": { "x": xxx, "y": xxx, } }, { "start": { "x": xxx, "y": xxx, }, "end": { "x": xxx, "y": xxx, } }, ], "summary": "简要说明理由" } """ return prompt class CheckboxProcess(Capability): @property def prompt(self): prompt = """ 给你一张界面截图，你需要根据任务要求返回需要点击的复选框坐标 如果一个复选框已经被点击则跳过该复选框 规则 1. 坐标基于当前输入图片的像素坐标 2. 坐标原点为图片左上角 (0, 0)，x 向右，y 向下 3. 每个坐标应尽量位于对应目标的可点击区域中心 4. 如果图片中某一步无法明确定位，请不要猜测，在 summary 中说明 输出要求: - 仅返回 JSON，不要输出任何多余文本 - 不要使用 markdown - JSON 结构必须严格如下： { "positions": [ { "x": xxx, "y": xxx }, { "x": xxx, "y": xxx } ], "summary": "简要说明这些坐标如何对应图片中的点击要求" } """ return prompt agent.py ...

定位可见元素 给元素添加属性 xbox-display 的值为 true 实例化Selector类 元素结构 { 'id': 'ifream-xpath', 'name': 'ifream-xpath', 'type': 'simple', 'xpath': { 'enable': 'true', 'segments': ['//iframe[@id="aa-challenge-whole-page-iframe"]', '//input[@name="submit_button"]'] } } 示例代码 def handle_captcha(web_page:xbot.web.WebBrowser): """处理亚马逊数英验证码""" # 构造验证码 xpath selector captcha_data = {'id': 'ifream-xpath', 'name': 'ifream-xpath', 'type': 'simple', 'xpath': {'enable': 'true', 'segments': ['//iframe[@id="aa-challenge-whole-page-iframe"]', '//img[@alt="captcha"]']}} captcha_selector = Selector(captcha_data) captcha_input_data = {'id': 'ifream-xpath', 'name': 'ifream-xpath', 'type': 'simple', 'xpath': {'enable': 'true', 'segments': ['//iframe[@id="aa-challenge-whole-page-iframe"]', '//input[@id="aa_captcha_input"]']}} captcha_input_selector = Selector(captcha_input_data) submit_data = {'id': 'ifream-xpath', 'name': 'ifream-xpath', 'type': 'simple', 'xpath': {'enable': 'true', 'segments': ['//iframe[@id="aa-challenge-whole-page-iframe"]', '//input[@name="submit_button"]']}} submit_selector = Selector(submit_data) xbot.logging.info("处理亚马逊数英验证码结束") for i in range(5): if xbot_visual.web.browser.element_display(browser=web_page, content_type="display", selector=captcha_selector): try: web_element = web_page.find(captcha_selector) result = xbot_visual.web_service.cloud_captch( captcha_engine="jfbym", captcha_type="10111", image_source="web", web_page=web_page, web_element=web_element, win_element=None, image_url=None, image_path=None, timeout="20", ) web_page.find_by_xpath(captcha_input_selector).input(result) web_page.find_by_xpath(submit_selector).click() except Exception as e: xbot.logging.error(f"验证码处理异常: {e}") sleep(3) # 控制频率，避免封号 else: break # 如果验证码元素已消失，跳出 xbot.logging.info("处理亚马逊数英验证码结束")

请求频率控制算法 常规算法 基于信号量和周期重置的限速器，它能防止并发量超标，但在高并发时可能导致请求在周期末尾阻塞，然后在周期开始时集中释放。 import asyncio import time class RateLimiter: """ 用于控制异步调用频率的类 """ def __init__(self, max_calls: int, period: float): """ 参数: - max_calls: 在指定周期内允许的最大调用次数。 - period: 限制的周期，单位为秒。 """ self.max_calls = max_calls self.period = period self.semaphore = asyncio.Semaphore(self.max_calls) self.last_reset = time.time() async def __aenter__(self): await self.wait_for_slot() return self async def __aexit__(self, exc_type, exc_val, exc_tb): self.release_slot() async def wait_for_slot(self): await self.semaphore.acquire() current_time = time.time() elapsed = current_time - self.last_reset # 如果一个周期已经过去，重置信号量 if elapsed > self.period: # 释放所有已获取的槽位 while self.semaphore._value < self.max_calls: self.semaphore.release() self.last_reset = current_time def release_slot(self): pass 令牌桶算法 令牌桶算法限速器则能提供更平滑、更精准的速率控制 ...

# 查看进程ID netstat -ano | findstr 4723 netstat -ano | findstr 4725 # 杀进程 taskkill /F /PID <PID> def get_session_id_by_udid(appium_host: str, appium_port: int, udid: str): """通过 udid 查询对应的 session_id，使用 http.client 实现""" conn = http.client.HTTPConnection(appium_host, appium_port, timeout=10) try: conn.request("GET", "/wd/hub/sessions") resp = conn.getresponse() if resp.status != 200: raise RuntimeError(f"Failed to get sessions: {resp.status} {resp.reason}") data = resp.read() sessions_info = json.loads(data.decode("utf-8")) sessions = sessions_info.get("value", []) for session in sessions: caps = session.get("capabilities", {}) session_udid = caps.get("udid") or caps.get("deviceUDID") if session_udid == udid: return session.get("id") or session.get("sessionId") finally: conn.close() return None

def encode_utf7(s): s_utf7 = s.encode('utf-7').replace(b'+', b'&').replace(b'/', b',') return s_utf7.decode('ascii') def decode_utf7(s): s_utf7 = s.replace('&', '+').replace(',', '/') return s_utf7.encode('ascii').decode('utf-7') def main(args): print(encode_utf7("我的文件夹")) print(decode_utf7("&bUuL1WWHTvZZOQ-")) """ $: &YhF2hGWHTvZZOQ- $: 测试文件夹 """

# 使用提醒: # 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 from functools import wraps from xbot_extensions.activity_photoshop import ps_back IT_BEEN_DECORATED = False _original_set_text_of_layer_by_name = ps_back.set_text_of_layer_by_name _original_change_content_of_layer_by_name_from_file = ps_back.change_content_of_layer_by_name_from_file def get_layer_by_name(ps_instance, layer_name): if ">" not in layer_name: try: active_layer = ps_instance.app.ActiveDocument.ArtLayers(layer_name) except: active_layer = ps_instance.layer(layer_name, only_visible=False) else: *layersets, layer_name = [path.strip() for path in layer_name.split(">")] current_layerset = ps_instance.doc for layerset_name in layersets: current_layerset = current_layerset.LayerSets(layerset_name) active_layer = current_layerset.Layers(layer_name) return active_layer def decorator(func): @wraps(func) def wrapper(*args, **kwargs): ps_instance = kwargs['photoshop_instance'] layer_name = kwargs['name_of_layer'] layer = get_layer_by_name(ps_instance, layer_name) before_replace_bounds = layer.Bounds func(*args, **kwargs) after_replace_bounds = layer.Bounds center_x = (before_replace_bounds[0] + before_replace_bounds[2]) // 2 center_y = (before_replace_bounds[1] + before_replace_bounds[3]) // 2 new_center_x = (after_replace_bounds[0] + after_replace_bounds[2]) // 2 new_center_y = (after_replace_bounds[1] + after_replace_bounds[3]) // 2 x_offset = center_x - new_center_x y_offset = center_y - new_center_y layer.Translate(x_offset, y_offset) return wrapper def add_patch_decorator(): global IT_BEEN_DECORATED if IT_BEEN_DECORATED: return # 应用装饰器 ps_back.set_text_of_layer_by_name = decorator(_original_set_text_of_layer_by_name) ps_back.change_content_of_layer_by_name_from_file = decorator(_original_change_content_of_layer_by_name_from_file) IT_BEEN_DECORATED = True def remove_patch_decorator(): global IT_BEEN_DECORATED if not IT_BEEN_DECORATED: return # 移除装饰器 ps_back.set_text_of_layer_by_name = _original_set_text_of_layer_by_name ps_back.change_content_of_layer_by_name_from_file = _original_change_content_of_layer_by_name_from_file IT_BEEN_DECORATED = False