影刀

首先局域网中有一个调度模式电脑能访问的共享文件夹 使用 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 ...

这次尝试的宗旨是 AI可替换，不和特定AI绑定 不考虑Token消耗 设计出通用的 AI 驱动框架，并不局限于 RPA 这个方面 1. 浅尝 最开始的想法是实现两个Agent分别是·、任务规划Agent和任务执行Agent 任务规划Agent根据用户需求列出步骤清单，任务执行Agent负责循环步骤清单执行步骤 1.1 任务规划Agent 我在实现完任务规划Agent后就发现，任务规划和执行分离不是一个好设计，规划的任务准确性会越来越低，所以这个方案我就没有继续实现执行Agent rask_planner_agent.py import base64 from langchain_deepseek import ChatDeepSeek from langchain_openai import ChatOpenAI from langchain.agents import create_agent from .output_schema import TaskSchema class TaskPlannerAgent: def __init__(self): self.llm = ChatOpenAI( openai_api_key="sk-or-v1-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx, openai_api_base="https://openrouter.ai/api/v1", model_name="google/gemini-3-flash-preview" ) # System prompt self.system_prompt = """ 你是一个专业的任务规划助手。用户将提供任务描述或图片，你需要结合这些信息将其拆解为结构化步骤。 """ self.agent = create_agent( model=self.llm, system_prompt=self.system_prompt, response_format=TaskSchema ) def plan(self, instruction: str, image_path: str = None) -> TaskSchema: content = [{"type": "text", "text": instruction}] if image_path: image_data = self._encode_image(image_path) content.append({ "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{image_data}" } }) result = self.agent.invoke( {"messages": [{"role": "user", "content": content}]} ) return result["structured_response"] def _encode_image(self, image_path: str): """将本地图片转为 base64编码""" with open(image_path, 'rb') as image_file: return base64.b64encode(image_file.read()).decode('utf-8') def main(args): agent = TaskPlannerAgent() for step in agent.plan("搜索熊猫，并进入搜索结果的第一项内容", r"C:\Users\Levit\Desktop\1.jpg").steps: print(step) output_schema.py 规范Agent输出格式 ...

定位可见元素 给元素添加属性 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

比影刀官方的好用 # 使用提醒: # 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 xbot.web.browser import WebBrowser import re import unicodedata import requests import io from fontTools.ttLib import TTFont class TmallFontDecoder: def __init__(self, web_page: WebBrowser): self.base_dict = { 'period': '.', 'zero': '0', 'one': '1', 'two': '2', 'three': '3', 'four': '4', 'five': '5', 'six': '6', 'seven': '7', 'eight': '8', 'nine': '9', } self.font_cmap = self._get_font_cmap(web_page) def _get_font_cmap(self, web_page: WebBrowser): font_face_js = web_page.find_by_xpath('//style[contains(text(), "AlibabaSans102CustomFont")]').get_text() pattern = f"url\('(https:\/\/[^']+\.woff)'\)" font_url = re.search(pattern, font_face_js).group(1) try: response = requests.get(font_url) response.raise_for_status() font = TTFont(io.BytesIO(response.content)) font_cmap = font['cmap'].getBestCmap() return font_cmap except Exception as e: raise Exception(f'获取字体映射出错: {e}') def decode(self, content: str): unicode_codes = [ord(c) for c in content] decoded_text = ''.join([self.base_dict[self.font_cmap[_]] for _ in unicode_codes]) return decoded_text def get_TmallFontDecoder(web_page: WebBrowser): return TmallFontDecoder(web_page) def decode(decoder: TmallFontDecoder, content: str): return decoder.decode(content)