1. 项目缘起当Python遇上AutojsPro7自动化云控的另一种可能最近在折腾一些自动化脚本特别是针对移动端App的批量操作发现圈子里的朋友很多都在用AutojsPro7。这工具确实强大基于JavaScript能直接在安卓设备上跑模拟点击、找图、OCR识别都不在话下。但当我需要管理几十上百台设备或者想把自动化逻辑和更复杂的后端服务比如数据分析、任务调度、状态监控结合起来时纯在手机端用Autojs写脚本就有点捉襟见肘了。这时候一个很自然的想法就冒出来了能不能用Python来“遥控”这些Autojs脚本呢或者说用Python作为“大脑”和“指挥中心”去下发指令、收集结果、管理设备状态而AutojsPro7的脚本则作为忠实执行命令的“手脚”。这就是“Python技术下的AutojsPro7云控”这个标题背后我想探讨的核心。它不是一个现成的、开箱即用的庞大云控系统而是一种架构思路和代码实践。通过Python我们可以构建一个轻量级、高灵活度的控制层将分散的Autojs设备纳入统一的管控体系。这对于需要做App群控、自动化测试、数据采集或者社交媒体运营的朋友来说意味着你可以用更熟悉的Python生态丰富的库、成熟的Web框架、强大的数据处理能力来赋能移动端自动化实现“112”的效果。下面我就把自己摸索和实践过的一套代码方案和架构思路分享出来希望能给有类似需求的朋友一些启发。2. 核心架构拆解Python如何与AutojsPro7“对话”要实现Python对AutojsPro7的云控关键在于建立两者之间稳定、高效的通信通道。AutojsPro7运行在安卓设备上Python程序通常运行在服务器或个人电脑上它们处于不同的网络环境。因此整个架构的核心是解决“指令如何下发结果如何回传”的问题。2.1 通信桥梁的选型为何是WebSocket最直接的通信方式无外乎几种HTTP轮询、HTTP长连接、WebSocket、MQTT等。对于云控这种需要近乎实时双向通信的场景WebSocket是现阶段最合适的选择。为什么不是HTTP轮询让Python端不断向每个设备发送“有任务吗”的HTTP请求或者设备不断上报状态。这种方式简单但延迟高、服务器压力大、浪费流量不适合实时控制。为什么不是单纯的HTTP APIAutojs可以调用Python提供的HTTP接口来获取任务但这只是单向的“拉取”。Python无法主动、即时地向设备推送一个新指令比如紧急停止或切换任务。WebSocket的优势它在单个TCP连接上提供全双工通信通道。连接建立后Python服务器可以随时向任意一台设备发送指令设备也可以随时将执行状态、日志、结果推送给服务器。这是实现“云控”实时性的基础。所以我们的架构简化为一个用Python编写的WebSocket服务器作为控制中心多台安装了AutojsPro7的设备作为客户端通过WebSocket与服务端保持长连接。2.2 设备端AutojsPro7代码骨架AutojsPro7本身支持WebSocket客户端。我们需要在Autojs脚本中实现连接管理、消息接收、任务执行和结果上报的逻辑。下面是一个高度概括但功能完整的设备端代码框架// 说明AutojsPro7 设备端核心连接与任务处理框架 // 文件名cloud_agent.js // 1. 配置区 const SERVER_WS_URL ws://你的Python服务器IP:端口/ws/device; // WebSocket服务器地址 const DEVICE_ID device.getAndroidId(); // 使用设备Android ID作为唯一标识也可自定义 const RECONNECT_INTERVAL 5000; // 断线重连间隔(毫秒) // 2. 全局变量 let ws null; let currentTask null; let isRunning false; // 3. 核心函数 - 连接WebSocket服务器 function connectWebSocket() { try { console.log([${DEVICE_ID}] 正在连接服务器: ${SERVER_WS_URL}); // AutojsPro7的WebSocket实现 ws new WebSocket(SERVER_WS_URL); ws.on(open, function() { console.log([${DEVICE_ID}] WebSocket连接成功); // 连接成功后立即向服务器注册本设备 sendMessageToServer({ type: register, deviceId: DEVICE_ID, status: idle, capabilities: [click, swipe, ocr, findImage] // 上报设备支持的能力 }); }); ws.on(message, function(message) { console.log([${DEVICE_ID}] 收到服务器消息:, message); handleServerMessage(JSON.parse(message)); }); ws.on(close, function() { console.log([${DEVICE_ID}] WebSocket连接关闭${RECONNECT_INTERVAL}ms后尝试重连); ws null; setTimeout(connectWebSocket, RECONNECT_INTERVAL); }); ws.on(error, function(err) { console.error([${DEVICE_ID}] WebSocket错误:, err); }); } catch (e) { console.error([${DEVICE_ID}] 连接异常:, e); setTimeout(connectWebSocket, RECONNECT_INTERVAL); } } // 4. 核心函数 - 处理服务器下发的消息 function handleServerMessage(msg) { switch(msg.type) { case task: // 收到新任务 if (isRunning) { sendMessageToServer({ type: status, deviceId: DEVICE_ID, status: busy, message: 设备正忙拒绝新任务 }); return; } console.log([${DEVICE_ID}] 收到新任务:, msg.task); currentTask msg.task; isRunning true; // 异步执行任务避免阻塞消息循环 threads.start(function() { executeTask(currentTask); }); break; case control: // 收到控制指令如停止、重启等 handleControlCommand(msg.command); break; case ping: // 心跳回应 sendMessageToServer({type: pong, deviceId: DEVICE_ID}); break; default: console.warn([${DEVICE_ID}] 未知消息类型:, msg.type); } } // 5. 核心函数 - 执行具体任务 function executeTask(task) { sendMessageToServer({ type: status, deviceId: DEVICE_ID, status: running, taskId: task.id, message: 开始执行任务: ${task.name} }); let result { success: false, data: null, error: null }; try { // 这里是具体的自动化逻辑根据task中的参数执行 // 例如打开某个App进行一系列操作 if (task.action open_app) { launch(task.packageName); sleep(2000); // ... 更多操作 result.data { screenshot: base64... }; // 示例返回截图 } else if (task.action collect_data) { let data collectSomeData(); // 自定义的数据采集函数 result.data data; } // ... 其他任务类型 result.success true; sendMessageToServer({ type: task_result, deviceId: DEVICE_ID, taskId: task.id, result: result }); } catch (e) { result.error e.toString(); console.error([${DEVICE_ID}] 任务执行失败:, e); sendMessageToServer({ type: task_result, deviceId: DEVICE_ID, taskId: task.id, result: result }); } finally { currentTask null; isRunning false; sendMessageToServer({ type: status, deviceId: DEVICE_ID, status: idle, message: 任务执行完毕恢复空闲 }); } } // 6. 工具函数 - 发送消息到服务器 function sendMessageToServer(data) { if (ws ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify(data)); } else { console.warn([${DEVICE_ID}] WebSocket未连接消息发送失败:, data); } } // 7. 初始化 connectWebSocket(); // 保持脚本运行 setInterval(() {}, 1000);这个框架实现了设备注册、心跳维持、任务接收与执行、结果上报、断线重连等云控客户端必备的核心功能。你可以根据实际自动化需求丰富executeTask函数中的逻辑。2.3 服务端Python代码骨架Python端我们需要一个WebSocket服务器并管理所有连接的设备。这里选择websockets库和asyncio异步框架来实现以保证能同时处理大量设备连接。# 说明Python WebSocket 云控服务端核心代码 # 文件名cloud_server.py import asyncio import json import logging from datetime import datetime from typing import Dict, Set import websockets # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) # 全局设备连接管理器 class DeviceManager: def __init__(self): # 存储 device_id - (websocket, last_heartbeat, status, current_task) self.connected_devices: Dict[str, dict] {} # 存储待执行的任务队列 (device_id - task_list) self.task_queues: Dict[str, list] {} async def register_device(self, device_id: str, websocket): 注册一个新设备 if device_id in self.connected_devices: logger.warning(f设备 {device_id} 已存在覆盖旧连接) self.connected_devices[device_id] { ws: websocket, last_heartbeat: datetime.now(), status: idle, capabilities: [] } self.task_queues[device_id] [] logger.info(f设备注册成功: {device_id}) # 发送欢迎消息或初始配置 welcome_msg { type: system, message: fDevice {device_id} connected successfully., timestamp: datetime.now().isoformat() } await self.send_to_device(device_id, welcome_msg) async def unregister_device(self, device_id: str): 设备断开连接 if device_id in self.connected_devices: del self.connected_devices[device_id] if device_id in self.task_queues: del self.task_queues[device_id] logger.info(f设备断开连接: {device_id}) async def update_device_status(self, device_id: str, status: str, **extra_info): 更新设备状态 if device_id in self.connected_devices: self.connected_devices[device_id][status] status self.connected_devices[device_id].update(extra_info) logger.debug(f设备 {device_id} 状态更新为: {status}) async def send_to_device(self, device_id: str, message: dict): 向指定设备发送消息 if device_id in self.connected_devices: try: ws self.connected_devices[device_id][ws] await ws.send(json.dumps(message, ensure_asciiFalse)) except websockets.exceptions.ConnectionClosed: logger.error(f向设备 {device_id} 发送消息时连接已关闭) await self.unregister_device(device_id) else: logger.warning(f尝试向未连接的设备发送消息: {device_id}) async def broadcast(self, message: dict, exclude_device_id: str None): 广播消息给所有设备除指定设备外 tasks [] for did, info in self.connected_devices.items(): if did ! exclude_device_id: tasks.append(self.send_to_device(did, message)) if tasks: await asyncio.gather(*tasks, return_exceptionsTrue) def get_idle_devices(self): 获取所有空闲设备 return [did for did, info in self.connected_devices.items() if info[status] idle] async def assign_task(self, device_id: str, task: dict): 向指定设备分配任务 if device_id not in self.connected_devices: logger.error(f无法分配任务设备未连接: {device_id}) return False task_msg { type: task, task: task, assign_time: datetime.now().isoformat() } await self.send_to_device(device_id, task_msg) await self.update_device_status(device_id, busy, current_tasktask.get(id)) logger.info(f任务 {task.get(id)} 已分配给设备 {device_id}) return True # 实例化设备管理器 device_manager DeviceManager() async def handle_device_connection(websocket, path): 处理每个设备的WebSocket连接 device_id None try: # 等待设备发送注册消息 async for message in websocket: data json.loads(message) msg_type data.get(type) if msg_type register: # 设备注册 device_id data.get(deviceId) if not device_id: logger.error(收到注册消息但缺少deviceId) await websocket.close() return await device_manager.register_device(device_id, websocket) capabilities data.get(capabilities, []) device_manager.connected_devices[device_id][capabilities] capabilities elif device_id: # 已注册的设备 if msg_type status: # 更新设备状态 status data.get(status, unknown) await device_manager.update_device_status(device_id, status, last_reportdatetime.now()) elif msg_type task_result: # 处理任务结果 task_id data.get(taskId) result data.get(result, {}) logger.info(f收到设备 {device_id} 的任务结果任务ID: {task_id}, 成功: {result.get(success)}) # 这里可以将结果存入数据库或进行后续处理 # 例如save_to_database(device_id, task_id, result) # 任务完成更新设备状态为空闲 await device_manager.update_device_status(device_id, idle, current_taskNone) elif msg_type pong: # 更新心跳时间 if device_id in device_manager.connected_devices: device_manager.connected_devices[device_id][last_heartbeat] datetime.now() elif msg_type log: # 处理设备发来的日志 log_content data.get(content, ) logger.info(f[Device {device_id} Log]: {log_content}) except websockets.exceptions.ConnectionClosed as e: logger.info(f设备连接正常关闭: {device_id}, code: {e.code}) except json.JSONDecodeError as e: logger.error(f消息JSON解析失败: {e}) except Exception as e: logger.exception(f处理设备 {device_id} 连接时发生未预期错误: {e}) finally: # 连接断开清理资源 if device_id: await device_manager.unregister_device(device_id) async def heartbeat_checker(): 定时心跳检查清理死连接 while True: await asyncio.sleep(60) # 每分钟检查一次 now datetime.now() dead_devices [] for device_id, info in device_manager.connected_devices.items(): last_beat info[last_heartbeat] if (now - last_beat).total_seconds() 120: # 超过120秒无心跳视为失联 dead_devices.append(device_id) logger.warning(f设备 {device_id} 心跳超时即将清理) for did in dead_devices: await device_manager.unregister_device(did) async def task_scheduler(): 简单的任务调度器示例 while True: await asyncio.sleep(10) # 每10秒调度一次 idle_devices device_manager.get_idle_devices() # 这里模拟从数据库或队列中获取任务 # 实际应用中你可以连接数据库如Redis的任务队列来获取真实任务 if idle_devices and some_condition_to_get_task(): # 假设的获取任务条件 task { id: ftask_{datetime.now().strftime(%Y%m%d%H%M%S)}, name: 示例自动化任务, action: open_app, packageName: com.example.app, priority: 1 } # 简单策略分配给第一个空闲设备 target_device idle_devices[0] await device_manager.assign_task(target_device, task) async def main(): 主函数启动服务器和后台任务 # 启动心跳检查任务 asyncio.create_task(heartbeat_checker()) # 启动任务调度器 asyncio.create_task(task_scheduler()) # 启动WebSocket服务器 server await websockets.serve( handle_device_connection, 0.0.0.0, # 监听所有网络接口 8765, # 端口号 ping_interval20, # 发送Ping间隔 ping_timeout40 # Ping超时时间 ) logger.info(云控WebSocket服务器启动在 ws://0.0.0.0:8765) await server.wait_closed() def some_condition_to_get_task(): 模拟获取任务的条件实际应替换为从数据库/队列检查 # 这里可以连接数据库查询是否有待处理任务 # 例如return redis_client.llen(pending_tasks) 0 return False # 默认返回False不分配任务仅作示例 if __name__ __main__: asyncio.run(main())这个Python服务端代码建立了一个异步的WebSocket服务器核心是DeviceManager类它负责管理所有在线设备的状态、连接和任务队列。heartbeat_checker和task_scheduler是两个后台协程分别负责连接健康检查和简单的任务调度。3. 关键实现细节与避坑指南把框架搭起来只是第一步真正让这套云控系统稳定可靠地跑起来还需要注意很多细节。下面分享几个我在实践中踩过的坑和总结的经验。3.1 网络连接稳定性断线重连与心跳机制移动网络环境复杂Wi-Fi也可能不稳定断线是常态而非异常。我们的代码必须能优雅地处理断线。设备端Autojs的重连策略上面的示例代码中在on(close)和on(error)事件中都设置了重连。但这里有个细节不要立即重连。我最初设置断线后立即重连结果在网络闪断时设备会疯狂尝试连接可能加剧服务器压力。更好的做法是使用“指数退避”策略比如第一次断线等2秒第二次等4秒第三次等8秒直到一个最大值比如60秒。服务端Python的心跳与清理仅靠WebSocket库的ping_interval是不够的。有些网络中间设备如NAT网关会清理长时间无数据交互的连接。因此我们需要应用层的心跳。设备端可以定时比如每30秒发送一个{type: ping}的消息服务端收到后回复pong并更新last_heartbeat。heartbeat_checker协程定期检查如果某个设备的last_heartbeat超过阈值如120秒就认为它已失联从device_manager中清理掉避免内存泄漏和状态不一致。连接标识不要用websocket对象本身作为设备的唯一标识。连接断开重连后websocket对象是新的。必须使用设备自身上报的唯一ID如Android ID、IMEI或我们预先分配的设备码。3.2 任务管理与状态同步避免重复执行与状态丢失当设备数量多、任务类型复杂时任务管理是关键。任务队列化不要直接给设备发送任务。服务端应该为每个设备或按设备分组维护一个任务队列。当设备空闲时从队列头部取出任务下发。这能应对网络波动导致的任务丢失也方便实现优先级调度。任务状态持久化任务信息内容、状态、分配的设备、开始时间、结果等应该存入数据库如SQLite、MySQL或Redis。这样即使Python服务重启也能恢复任务状态知道哪些任务已完成、哪些正在执行、哪些失败待重试。设备状态同步设备执行任务是一个过程可能耗时几秒到几分钟。服务端需要知道设备当前是“空闲”、“运行中”还是“异常”。设备在执行任务开始、结束、发生关键步骤时都应主动上报状态。服务端也要有超时机制如果一个任务分配后超过预期时间没有收到完成或心跳报告则将其标记为“疑似失败”可以重新分配给其他设备。3.3 安全与认证防止未经授权的连接一个暴露在公网的WebSocket服务器是危险的。必须加入简单的认证机制。连接时认证可以在设备首次连接时要求其发送一个预共享的密钥Token或进行简单的挑战-应答。修改设备端代码在register消息中加入签名或Token。// Autojs端 const DEVICE_TOKEN 你的设备预置Token; // 可以写死在脚本里或从安全的地方读取 // 在register消息中加入 sendMessageToServer({ type: register, deviceId: DEVICE_ID, token: DEVICE_TOKEN, // ... 其他字段 });# Python服务端 VALID_TOKENS {device_token_1, device_token_2} # 应从数据库或配置读取 async def handle_device_connection(websocket, path): # ... if msg_type register: token data.get(token) if token not in VALID_TOKENS: logger.warning(f设备认证失败Token无效: {token}) await websocket.close(code4001, reasonUnauthorized) return # 认证通过继续注册流程...数据加密可选对于敏感指令或回传数据可以考虑在应用层对消息体进行对称加密如AES。但这会增加两端代码的复杂性需权衡安全需求和性能开销。3.4 性能与扩展性当设备量增长时初期可能只有几台设备但当规模扩大到几十上百台时架构就需要调整。异步框架的选择我们用了asynciowebsockets这是Python中处理高并发I/O的推荐方式性能比多线程模式好很多。确保你的任务处理逻辑如executeTask中的具体操作也是非阻塞的或者将其放入线程池执行避免阻塞事件循环。服务端资源单机Python进程能承载的连接数有限取决于系统资源。当连接数过多例如上千时需要考虑水平扩展。可以引入连接网关如用Nginx做WebSocket代理和负载均衡后面部署多个Python服务实例。这时设备管理器的状态就需要外置到共享存储如Redis中以便多个实例同步设备状态。消息格式优化WebSocket消息使用JSON很方便但文本格式有冗余。如果消息非常频繁或体积大可以考虑使用二进制协议如MessagePack来序列化数据能显著减少带宽占用和解析开销。4. 一个完整的实战案例模拟社交媒体点赞任务为了把上面的理论具体化我们设计一个简单的实战场景用10台安卓设备自动为某个社交媒体App的指定帖子批量点赞。4.1 任务定义与下发Python服务端从数据库读取一批待点赞的帖子ID生成任务。每个任务包含{ id: like_task_001, action: like_post, target_app: com.social.media, post_id: 1234567890, max_retry: 3 }服务端的调度器发现有空闲设备就将任务放入该设备的任务队列并下发。4.2 设备端任务执行Autojs设备收到任务后executeTask函数被触发function executeTask(task) { if (task.action like_post) { // 1. 确保屏幕点亮和解锁实际项目需更健壮的判断 device.wakeUp(); sleep(500); // 2. 启动目标App launch(task.target_app); sleep(3000); // 等待App启动 // 3. 模拟操作进入目标帖子这里简化实际需根据UI布局定位 // 假设通过搜索或直接打开链接进入帖子详情页 // click(搜索框); input(帖子ID); click(搜索按钮); sleep(2000); // click(帖子条目); sleep(2000); // 4. 寻找点赞按钮并点击这里需要真实的UI选择器 let likeButton id(like_button).findOne(5000); if (likeButton) { likeButton.click(); console.log(已对帖子 ${task.post_id} 执行点赞); // 5. 截图作为执行凭证可选 let screenshot captureScreen(); let base64Img images.toBase64(screenshot, png, 50); return { success: true, data: { post_id: task.post_id, screenshot: base64Img, timestamp: new Date().toISOString() } }; } else { return { success: false, error: 未找到点赞按钮 }; } // 6. 返回App主页或退出为下一个任务准备 // back(); } }执行成功后设备将结果封装成task_result消息发回服务器。4.3 服务端结果处理与持久化Python服务端收到task_result后不仅更新设备状态更重要的是将结果记录到数据库。# 在handle_device_connection的task_result处理部分补充 async def handle_device_connection(websocket, path): # ... 省略其他代码 ... elif msg_type task_result: task_id data.get(taskId) result data.get(result, {}) # 1. 更新任务状态为完成或失败 await update_task_in_db(task_id, statuscompleted, resultresult) # 2. 记录设备活动日志 await log_device_activity(device_id, f完成任务 {task_id}, 结果: {result.get(success)}) # 3. 如果任务失败且重试次数未超限可以重新放入队列 if not result.get(success) and task_retry_count(task_id) MAX_RETRY: await requeue_task(task_id) logger.info(f任务 {task_id} 处理完毕。)这样我们就完成了一个从任务创建、下发、执行到结果回收的完整闭环。通过Python后台我们可以清晰地看到所有设备的实时状态、任务执行进度和成功率实现了基本的云控能力。5. 进阶思路与扩展方向基础云控跑通后可以根据实际需求向不同方向深化5.1 任务编排与工作流简单的单个任务如点赞不够用。可以设计“工作流”任务包含多个步骤。例如“打开App A - 搜索关键词 - 截图 - 返回主页 - 打开App B - 发布内容”。这需要在任务定义中支持步骤列表设备端按顺序执行并在每个步骤后向服务器报告进度。5.2 设备分组与策略调度不是所有设备都一样。有的设备性能好有的安装了特定App。可以在设备注册时上报更详细的能力和属性如系统版本、屏幕分辨率、已安装App列表。服务端根据任务要求将任务智能分配给最合适的设备组。例如一个需要高版本Android才能运行的任务只会分配给系统版本符合的设备。5.3 引入消息队列如Redis解耦当任务量巨大时直接在Python内存中管理队列可能不够可靠。可以引入Redis作为消息中间件。Python服务端将任务推入Redis的List或Stream独立的“调度器”服务从Redis消费任务再通过WebSocket分配给设备。这样实现了生产生成任务、消费调度分配、执行设备的完全解耦系统扩展性更强。5.4 丰富的监控与管理界面纯命令行日志难以监控上百台设备。可以用Python的Web框架如FastAPI、Flask快速搭建一个管理后台提供Web界面。实时展示设备在线状态、CPU/内存使用率需设备端上报、任务队列情况、执行日志搜索、手动下发任务等功能。这会让整个云控系统的易用性提升一个档次。5.5 脚本的远程更新与热加载当需要修改Autojs端的业务逻辑时逐台设备去更新脚本是噩梦。可以实现一个“脚本管理”功能。服务端可以下发一个特殊的“更新脚本”任务包含新脚本的下载地址或代码内容。设备端收到后下载并替换本地的执行脚本然后重启自动化服务或动态加载新脚本。这样就实现了所有设备的集中式脚本更新。这套“Python AutojsPro7”的云控方案其魅力在于极高的灵活性。Python负责复杂的调度、管理和数据聚合Autojs负责在移动端执行精准的UI自动化。两者通过WebSocket这个轻量级的桥梁结合让你能用熟悉的Python生态去驾驭大量的安卓自动化设备。当然真正投入到生产环境还需要在异常处理、日志收集、安全加固等方面做大量工作。希望这个分享能为你打开一扇门剩下的精彩就靠你在自己的项目中去探索和实现了。