1. 項(xiàng)目緣起當(dāng)Python遇上AutojsPro7自動(dòng)化云控的另一種可能最近在折騰一些自動(dòng)化腳本特別是針對(duì)移動(dòng)端App的批量操作發(fā)現(xiàn)圈子里的朋友很多都在用AutojsPro7。這工具確實(shí)強(qiáng)大基于JavaScript能直接在安卓設(shè)備上跑模擬點(diǎn)擊、找圖、OCR識(shí)別都不在話下。但當(dāng)我需要管理幾十上百臺(tái)設(shè)備或者想把自動(dòng)化邏輯和更復(fù)雜的后端服務(wù)比如數(shù)據(jù)分析、任務(wù)調(diào)度、狀態(tài)監(jiān)控結(jié)合起來時(shí)純?cè)谑謾C(jī)端用Autojs寫腳本就有點(diǎn)捉襟見肘了。這時(shí)候一個(gè)很自然的想法就冒出來了能不能用Python來“遙控”這些Autojs腳本呢或者說用Python作為“大腦”和“指揮中心”去下發(fā)指令、收集結(jié)果、管理設(shè)備狀態(tài)而AutojsPro7的腳本則作為忠實(shí)執(zhí)行命令的“手腳”。這就是“Python技術(shù)下的AutojsPro7云控”這個(gè)標(biāo)題背后我想探討的核心。它不是一個(gè)現(xiàn)成的、開箱即用的龐大云控系統(tǒng)而是一種架構(gòu)思路和代碼實(shí)踐。通過Python我們可以構(gòu)建一個(gè)輕量級(jí)、高靈活度的控制層將分散的Autojs設(shè)備納入統(tǒng)一的管控體系。這對(duì)于需要做App群控、自動(dòng)化測試、數(shù)據(jù)采集或者社交媒體運(yùn)營的朋友來說意味著你可以用更熟悉的Python生態(tài)豐富的庫、成熟的Web框架、強(qiáng)大的數(shù)據(jù)處理能力來賦能移動(dòng)端自動(dòng)化實(shí)現(xiàn)“112”的效果。下面我就把自己摸索和實(shí)踐過的一套代碼方案和架構(gòu)思路分享出來希望能給有類似需求的朋友一些啟發(fā)。2. 核心架構(gòu)拆解Python如何與AutojsPro7“對(duì)話”要實(shí)現(xiàn)Python對(duì)AutojsPro7的云控關(guān)鍵在于建立兩者之間穩(wěn)定、高效的通信通道。AutojsPro7運(yùn)行在安卓設(shè)備上Python程序通常運(yùn)行在服務(wù)器或個(gè)人電腦上它們處于不同的網(wǎng)絡(luò)環(huán)境。因此整個(gè)架構(gòu)的核心是解決“指令如何下發(fā)結(jié)果如何回傳”的問題。2.1 通信橋梁的選型為何是WebSocket最直接的通信方式無外乎幾種HTTP輪詢、HTTP長連接、WebSocket、MQTT等。對(duì)于云控這種需要近乎實(shí)時(shí)雙向通信的場景WebSocket是現(xiàn)階段最合適的選擇。為什么不是HTTP輪詢讓Python端不斷向每個(gè)設(shè)備發(fā)送“有任務(wù)嗎”的HTTP請(qǐng)求或者設(shè)備不斷上報(bào)狀態(tài)。這種方式簡單但延遲高、服務(wù)器壓力大、浪費(fèi)流量不適合實(shí)時(shí)控制。為什么不是單純的HTTP APIAutojs可以調(diào)用Python提供的HTTP接口來獲取任務(wù)但這只是單向的“拉取”。Python無法主動(dòng)、即時(shí)地向設(shè)備推送一個(gè)新指令比如緊急停止或切換任務(wù)。WebSocket的優(yōu)勢它在單個(gè)TCP連接上提供全雙工通信通道。連接建立后Python服務(wù)器可以隨時(shí)向任意一臺(tái)設(shè)備發(fā)送指令設(shè)備也可以隨時(shí)將執(zhí)行狀態(tài)、日志、結(jié)果推送給服務(wù)器。這是實(shí)現(xiàn)“云控”實(shí)時(shí)性的基礎(chǔ)。所以我們的架構(gòu)簡化為一個(gè)用Python編寫的WebSocket服務(wù)器作為控制中心多臺(tái)安裝了AutojsPro7的設(shè)備作為客戶端通過WebSocket與服務(wù)端保持長連接。2.2 設(shè)備端AutojsPro7代碼骨架AutojsPro7本身支持WebSocket客戶端。我們需要在Autojs腳本中實(shí)現(xiàn)連接管理、消息接收、任務(wù)執(zhí)行和結(jié)果上報(bào)的邏輯。下面是一個(gè)高度概括但功能完整的設(shè)備端代碼框架// 說明AutojsPro7 設(shè)備端核心連接與任務(wù)處理框架 // 文件名cloud_agent.js // 1. 配置區(qū) const SERVER_WS_URL ws://你的Python服務(wù)器IP:端口/ws/device; // WebSocket服務(wù)器地址 const DEVICE_ID device.getAndroidId(); // 使用設(shè)備Android ID作為唯一標(biāo)識(shí)也可自定義 const RECONNECT_INTERVAL 5000; // 斷線重連間隔(毫秒) // 2. 全局變量 let ws null; let currentTask null; let isRunning false; // 3. 核心函數(shù) - 連接WebSocket服務(wù)器 function connectWebSocket() { try { console.log([${DEVICE_ID}] 正在連接服務(wù)器: ${SERVER_WS_URL}); // AutojsPro7的WebSocket實(shí)現(xiàn) ws new WebSocket(SERVER_WS_URL); ws.on(open, function() { console.log([${DEVICE_ID}] WebSocket連接成功); // 連接成功后立即向服務(wù)器注冊(cè)本設(shè)備 sendMessageToServer({ type: register, deviceId: DEVICE_ID, status: idle, capabilities: [click, swipe, ocr, findImage] // 上報(bào)設(shè)備支持的能力 }); }); ws.on(message, function(message) { console.log([${DEVICE_ID}] 收到服務(wù)器消息:, message); handleServerMessage(JSON.parse(message)); }); ws.on(close, function() { console.log([${DEVICE_ID}] WebSocket連接關(guān)閉${RECONNECT_INTERVAL}ms后嘗試重連); ws null; setTimeout(connectWebSocket, RECONNECT_INTERVAL); }); ws.on(error, function(err) { console.error([${DEVICE_ID}] WebSocket錯(cuò)誤:, err); }); } catch (e) { console.error([${DEVICE_ID}] 連接異常:, e); setTimeout(connectWebSocket, RECONNECT_INTERVAL); } } // 4. 核心函數(shù) - 處理服務(wù)器下發(fā)的消息 function handleServerMessage(msg) { switch(msg.type) { case task: // 收到新任務(wù) if (isRunning) { sendMessageToServer({ type: status, deviceId: DEVICE_ID, status: busy, message: 設(shè)備正忙拒絕新任務(wù) }); return; } console.log([${DEVICE_ID}] 收到新任務(wù):, msg.task); currentTask msg.task; isRunning true; // 異步執(zhí)行任務(wù)避免阻塞消息循環(huán) threads.start(function() { executeTask(currentTask); }); break; case control: // 收到控制指令如停止、重啟等 handleControlCommand(msg.command); break; case ping: // 心跳回應(yīng) sendMessageToServer({type: pong, deviceId: DEVICE_ID}); break; default: console.warn([${DEVICE_ID}] 未知消息類型:, msg.type); } } // 5. 核心函數(shù) - 執(zhí)行具體任務(wù) function executeTask(task) { sendMessageToServer({ type: status, deviceId: DEVICE_ID, status: running, taskId: task.id, message: 開始執(zhí)行任務(wù): ${task.name} }); let result { success: false, data: null, error: null }; try { // 這里是具體的自動(dòng)化邏輯根據(jù)task中的參數(shù)執(zhí)行 // 例如打開某個(gè)App進(jìn)行一系列操作 if (task.action open_app) { launch(task.packageName); sleep(2000); // ... 更多操作 result.data { screenshot: base64... }; // 示例返回截圖 } else if (task.action collect_data) { let data collectSomeData(); // 自定義的數(shù)據(jù)采集函數(shù) result.data data; } // ... 其他任務(wù)類型 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}] 任務(wù)執(zhí)行失敗:, 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: 任務(wù)執(zhí)行完畢恢復(fù)空閑 }); } } // 6. 工具函數(shù) - 發(fā)送消息到服務(wù)器 function sendMessageToServer(data) { if (ws ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify(data)); } else { console.warn([${DEVICE_ID}] WebSocket未連接消息發(fā)送失敗:, data); } } // 7. 初始化 connectWebSocket(); // 保持腳本運(yùn)行 setInterval(() {}, 1000);這個(gè)框架實(shí)現(xiàn)了設(shè)備注冊(cè)、心跳維持、任務(wù)接收與執(zhí)行、結(jié)果上報(bào)、斷線重連等云控客戶端必備的核心功能。你可以根據(jù)實(shí)際自動(dòng)化需求豐富executeTask函數(shù)中的邏輯。2.3 服務(wù)端Python代碼骨架Python端我們需要一個(gè)WebSocket服務(wù)器并管理所有連接的設(shè)備。這里選擇websockets庫和asyncio異步框架來實(shí)現(xiàn)以保證能同時(shí)處理大量設(shè)備連接。# 說明Python WebSocket 云控服務(wù)端核心代碼 # 文件名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__) # 全局設(shè)備連接管理器 class DeviceManager: def __init__(self): # 存儲(chǔ) device_id - (websocket, last_heartbeat, status, current_task) self.connected_devices: Dict[str, dict] {} # 存儲(chǔ)待執(zhí)行的任務(wù)隊(duì)列 (device_id - task_list) self.task_queues: Dict[str, list] {} async def register_device(self, device_id: str, websocket): 注冊(cè)一個(gè)新設(shè)備 if device_id in self.connected_devices: logger.warning(f設(shè)備 {device_id} 已存在覆蓋舊連接) self.connected_devices[device_id] { ws: websocket, last_heartbeat: datetime.now(), status: idle, capabilities: [] } self.task_queues[device_id] [] logger.info(f設(shè)備注冊(cè)成功: {device_id}) # 發(fā)送歡迎消息或初始配置 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): 設(shè)備斷開連接 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設(shè)備斷開連接: {device_id}) async def update_device_status(self, device_id: str, status: str, **extra_info): 更新設(shè)備狀態(tài) 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設(shè)備 {device_id} 狀態(tài)更新為: {status}) async def send_to_device(self, device_id: str, message: dict): 向指定設(shè)備發(fā)送消息 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向設(shè)備 {device_id} 發(fā)送消息時(shí)連接已關(guān)閉) await self.unregister_device(device_id) else: logger.warning(f嘗試向未連接的設(shè)備發(fā)送消息: {device_id}) async def broadcast(self, message: dict, exclude_device_id: str None): 廣播消息給所有設(shè)備除指定設(shè)備外 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): 獲取所有空閑設(shè)備 return [did for did, info in self.connected_devices.items() if info[status] idle] async def assign_task(self, device_id: str, task: dict): 向指定設(shè)備分配任務(wù) if device_id not in self.connected_devices: logger.error(f無法分配任務(wù)設(shè)備未連接: {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任務(wù) {task.get(id)} 已分配給設(shè)備 {device_id}) return True # 實(shí)例化設(shè)備管理器 device_manager DeviceManager() async def handle_device_connection(websocket, path): 處理每個(gè)設(shè)備的WebSocket連接 device_id None try: # 等待設(shè)備發(fā)送注冊(cè)消息 async for message in websocket: data json.loads(message) msg_type data.get(type) if msg_type register: # 設(shè)備注冊(cè) device_id data.get(deviceId) if not device_id: logger.error(收到注冊(cè)消息但缺少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: # 已注冊(cè)的設(shè)備 if msg_type status: # 更新設(shè)備狀態(tài) status data.get(status, unknown) await device_manager.update_device_status(device_id, status, last_reportdatetime.now()) elif msg_type task_result: # 處理任務(wù)結(jié)果 task_id data.get(taskId) result data.get(result, {}) logger.info(f收到設(shè)備 {device_id} 的任務(wù)結(jié)果任務(wù)ID: {task_id}, 成功: {result.get(success)}) # 這里可以將結(jié)果存入數(shù)據(jù)庫或進(jìn)行后續(xù)處理 # 例如save_to_database(device_id, task_id, result) # 任務(wù)完成更新設(shè)備狀態(tài)為空閑 await device_manager.update_device_status(device_id, idle, current_taskNone) elif msg_type pong: # 更新心跳時(shí)間 if device_id in device_manager.connected_devices: device_manager.connected_devices[device_id][last_heartbeat] datetime.now() elif msg_type log: # 處理設(shè)備發(fā)來的日志 log_content data.get(content, ) logger.info(f[Device {device_id} Log]: {log_content}) except websockets.exceptions.ConnectionClosed as e: logger.info(f設(shè)備連接正常關(guān)閉: {device_id}, code: {e.code}) except json.JSONDecodeError as e: logger.error(f消息JSON解析失敗: {e}) except Exception as e: logger.exception(f處理設(shè)備 {device_id} 連接時(shí)發(fā)生未預(yù)期錯(cuò)誤: {e}) finally: # 連接斷開清理資源 if device_id: await device_manager.unregister_device(device_id) async def heartbeat_checker(): 定時(shí)心跳檢查清理死連接 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秒無心跳視為失聯(lián) dead_devices.append(device_id) logger.warning(f設(shè)備 {device_id} 心跳超時(shí)即將清理) for did in dead_devices: await device_manager.unregister_device(did) async def task_scheduler(): 簡單的任務(wù)調(diào)度器示例 while True: await asyncio.sleep(10) # 每10秒調(diào)度一次 idle_devices device_manager.get_idle_devices() # 這里模擬從數(shù)據(jù)庫或隊(duì)列中獲取任務(wù) # 實(shí)際應(yīng)用中你可以連接數(shù)據(jù)庫如Redis的任務(wù)隊(duì)列來獲取真實(shí)任務(wù) if idle_devices and some_condition_to_get_task(): # 假設(shè)的獲取任務(wù)條件 task { id: ftask_{datetime.now().strftime(%Y%m%d%H%M%S)}, name: 示例自動(dòng)化任務(wù), action: open_app, packageName: com.example.app, priority: 1 } # 簡單策略分配給第一個(gè)空閑設(shè)備 target_device idle_devices[0] await device_manager.assign_task(target_device, task) async def main(): 主函數(shù)啟動(dòng)服務(wù)器和后臺(tái)任務(wù) # 啟動(dòng)心跳檢查任務(wù) asyncio.create_task(heartbeat_checker()) # 啟動(dòng)任務(wù)調(diào)度器 asyncio.create_task(task_scheduler()) # 啟動(dòng)WebSocket服務(wù)器 server await websockets.serve( handle_device_connection, 0.0.0.0, # 監(jiān)聽所有網(wǎng)絡(luò)接口 8765, # 端口號(hào) ping_interval20, # 發(fā)送Ping間隔 ping_timeout40 # Ping超時(shí)時(shí)間 ) logger.info(云控WebSocket服務(wù)器啟動(dòng)在 ws://0.0.0.0:8765) await server.wait_closed() def some_condition_to_get_task(): 模擬獲取任務(wù)的條件實(shí)際應(yīng)替換為從數(shù)據(jù)庫/隊(duì)列檢查 # 這里可以連接數(shù)據(jù)庫查詢是否有待處理任務(wù) # 例如return redis_client.llen(pending_tasks) 0 return False # 默認(rèn)返回False不分配任務(wù)僅作示例 if __name__ __main__: asyncio.run(main())這個(gè)Python服務(wù)端代碼建立了一個(gè)異步的WebSocket服務(wù)器核心是DeviceManager類它負(fù)責(zé)管理所有在線設(shè)備的狀態(tài)、連接和任務(wù)隊(duì)列。heartbeat_checker和task_scheduler是兩個(gè)后臺(tái)協(xié)程分別負(fù)責(zé)連接健康檢查和簡單的任務(wù)調(diào)度。3. 關(guān)鍵實(shí)現(xiàn)細(xì)節(jié)與避坑指南把框架搭起來只是第一步真正讓這套云控系統(tǒng)穩(wěn)定可靠地跑起來還需要注意很多細(xì)節(jié)。下面分享幾個(gè)我在實(shí)踐中踩過的坑和總結(jié)的經(jīng)驗(yàn)。3.1 網(wǎng)絡(luò)連接穩(wěn)定性斷線重連與心跳機(jī)制移動(dòng)網(wǎng)絡(luò)環(huán)境復(fù)雜Wi-Fi也可能不穩(wěn)定斷線是常態(tài)而非異常。我們的代碼必須能優(yōu)雅地處理斷線。設(shè)備端Autojs的重連策略上面的示例代碼中在on(close)和on(error)事件中都設(shè)置了重連。但這里有個(gè)細(xì)節(jié)不要立即重連。我最初設(shè)置斷線后立即重連結(jié)果在網(wǎng)絡(luò)閃斷時(shí)設(shè)備會(huì)瘋狂嘗試連接可能加劇服務(wù)器壓力。更好的做法是使用“指數(shù)退避”策略比如第一次斷線等2秒第二次等4秒第三次等8秒直到一個(gè)最大值比如60秒。服務(wù)端Python的心跳與清理僅靠WebSocket庫的ping_interval是不夠的。有些網(wǎng)絡(luò)中間設(shè)備如NAT網(wǎng)關(guān)會(huì)清理長時(shí)間無數(shù)據(jù)交互的連接。因此我們需要應(yīng)用層的心跳。設(shè)備端可以定時(shí)比如每30秒發(fā)送一個(gè){type: ping}的消息服務(wù)端收到后回復(fù)pong并更新last_heartbeat。heartbeat_checker協(xié)程定期檢查如果某個(gè)設(shè)備的last_heartbeat超過閾值如120秒就認(rèn)為它已失聯(lián)從device_manager中清理掉避免內(nèi)存泄漏和狀態(tài)不一致。連接標(biāo)識(shí)不要用websocket對(duì)象本身作為設(shè)備的唯一標(biāo)識(shí)。連接斷開重連后websocket對(duì)象是新的。必須使用設(shè)備自身上報(bào)的唯一ID如Android ID、IMEI或我們預(yù)先分配的設(shè)備碼。3.2 任務(wù)管理與狀態(tài)同步避免重復(fù)執(zhí)行與狀態(tài)丟失當(dāng)設(shè)備數(shù)量多、任務(wù)類型復(fù)雜時(shí)任務(wù)管理是關(guān)鍵。任務(wù)隊(duì)列化不要直接給設(shè)備發(fā)送任務(wù)。服務(wù)端應(yīng)該為每個(gè)設(shè)備或按設(shè)備分組維護(hù)一個(gè)任務(wù)隊(duì)列。當(dāng)設(shè)備空閑時(shí)從隊(duì)列頭部取出任務(wù)下發(fā)。這能應(yīng)對(duì)網(wǎng)絡(luò)波動(dòng)導(dǎo)致的任務(wù)丟失也方便實(shí)現(xiàn)優(yōu)先級(jí)調(diào)度。任務(wù)狀態(tài)持久化任務(wù)信息內(nèi)容、狀態(tài)、分配的設(shè)備、開始時(shí)間、結(jié)果等應(yīng)該存入數(shù)據(jù)庫如SQLite、MySQL或Redis。這樣即使Python服務(wù)重啟也能恢復(fù)任務(wù)狀態(tài)知道哪些任務(wù)已完成、哪些正在執(zhí)行、哪些失敗待重試。設(shè)備狀態(tài)同步設(shè)備執(zhí)行任務(wù)是一個(gè)過程可能耗時(shí)幾秒到幾分鐘。服務(wù)端需要知道設(shè)備當(dāng)前是“空閑”、“運(yùn)行中”還是“異?！?。設(shè)備在執(zhí)行任務(wù)開始、結(jié)束、發(fā)生關(guān)鍵步驟時(shí)都應(yīng)主動(dòng)上報(bào)狀態(tài)。服務(wù)端也要有超時(shí)機(jī)制如果一個(gè)任務(wù)分配后超過預(yù)期時(shí)間沒有收到完成或心跳報(bào)告則將其標(biāo)記為“疑似失敗”可以重新分配給其他設(shè)備。3.3 安全與認(rèn)證防止未經(jīng)授權(quán)的連接一個(gè)暴露在公網(wǎng)的WebSocket服務(wù)器是危險(xiǎn)的。必須加入簡單的認(rèn)證機(jī)制。連接時(shí)認(rèn)證可以在設(shè)備首次連接時(shí)要求其發(fā)送一個(gè)預(yù)共享的密鑰Token或進(jìn)行簡單的挑戰(zhàn)-應(yīng)答。修改設(shè)備端代碼在register消息中加入簽名或Token。// Autojs端 const DEVICE_TOKEN 你的設(shè)備預(yù)置Token; // 可以寫死在腳本里或從安全的地方讀取 // 在register消息中加入 sendMessageToServer({ type: register, deviceId: DEVICE_ID, token: DEVICE_TOKEN, // ... 其他字段 });# Python服務(wù)端 VALID_TOKENS {device_token_1, device_token_2} # 應(yīng)從數(shù)據(jù)庫或配置讀取 async def handle_device_connection(websocket, path): # ... if msg_type register: token data.get(token) if token not in VALID_TOKENS: logger.warning(f設(shè)備認(rèn)證失敗Token無效: {token}) await websocket.close(code4001, reasonUnauthorized) return # 認(rèn)證通過繼續(xù)注冊(cè)流程...數(shù)據(jù)加密可選對(duì)于敏感指令或回傳數(shù)據(jù)可以考慮在應(yīng)用層對(duì)消息體進(jìn)行對(duì)稱加密如AES。但這會(huì)增加兩端代碼的復(fù)雜性需權(quán)衡安全需求和性能開銷。3.4 性能與擴(kuò)展性當(dāng)設(shè)備量增長時(shí)初期可能只有幾臺(tái)設(shè)備但當(dāng)規(guī)模擴(kuò)大到幾十上百臺(tái)時(shí)架構(gòu)就需要調(diào)整。異步框架的選擇我們用了asynciowebsockets這是Python中處理高并發(fā)I/O的推薦方式性能比多線程模式好很多。確保你的任務(wù)處理邏輯如executeTask中的具體操作也是非阻塞的或者將其放入線程池執(zhí)行避免阻塞事件循環(huán)。服務(wù)端資源單機(jī)Python進(jìn)程能承載的連接數(shù)有限取決于系統(tǒng)資源。當(dāng)連接數(shù)過多例如上千時(shí)需要考慮水平擴(kuò)展。可以引入連接網(wǎng)關(guān)如用Nginx做WebSocket代理和負(fù)載均衡后面部署多個(gè)Python服務(wù)實(shí)例。這時(shí)設(shè)備管理器的狀態(tài)就需要外置到共享存儲(chǔ)如Redis中以便多個(gè)實(shí)例同步設(shè)備狀態(tài)。消息格式優(yōu)化WebSocket消息使用JSON很方便但文本格式有冗余。如果消息非常頻繁或體積大可以考慮使用二進(jìn)制協(xié)議如MessagePack來序列化數(shù)據(jù)能顯著減少帶寬占用和解析開銷。4. 一個(gè)完整的實(shí)戰(zhàn)案例模擬社交媒體點(diǎn)贊任務(wù)為了把上面的理論具體化我們?cè)O(shè)計(jì)一個(gè)簡單的實(shí)戰(zhàn)場景用10臺(tái)安卓設(shè)備自動(dòng)為某個(gè)社交媒體App的指定帖子批量點(diǎn)贊。4.1 任務(wù)定義與下發(fā)Python服務(wù)端從數(shù)據(jù)庫讀取一批待點(diǎn)贊的帖子ID生成任務(wù)。每個(gè)任務(wù)包含{ id: like_task_001, action: like_post, target_app: com.social.media, post_id: 1234567890, max_retry: 3 }服務(wù)端的調(diào)度器發(fā)現(xiàn)有空閑設(shè)備就將任務(wù)放入該設(shè)備的任務(wù)隊(duì)列并下發(fā)。4.2 設(shè)備端任務(wù)執(zhí)行Autojs設(shè)備收到任務(wù)后executeTask函數(shù)被觸發(fā)function executeTask(task) { if (task.action like_post) { // 1. 確保屏幕點(diǎn)亮和解鎖實(shí)際項(xiàng)目需更健壯的判斷 device.wakeUp(); sleep(500); // 2. 啟動(dòng)目標(biāo)App launch(task.target_app); sleep(3000); // 等待App啟動(dòng) // 3. 模擬操作進(jìn)入目標(biāo)帖子這里簡化實(shí)際需根據(jù)UI布局定位 // 假設(shè)通過搜索或直接打開鏈接進(jìn)入帖子詳情頁 // click(搜索框); input(帖子ID); click(搜索按鈕); sleep(2000); // click(帖子條目); sleep(2000); // 4. 尋找點(diǎn)贊按鈕并點(diǎn)擊這里需要真實(shí)的UI選擇器 let likeButton id(like_button).findOne(5000); if (likeButton) { likeButton.click(); console.log(已對(duì)帖子 ${task.post_id} 執(zhí)行點(diǎn)贊); // 5. 截圖作為執(zhí)行憑證可選 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: 未找到點(diǎn)贊按鈕 }; } // 6. 返回App主頁或退出為下一個(gè)任務(wù)準(zhǔn)備 // back(); } }執(zhí)行成功后設(shè)備將結(jié)果封裝成task_result消息發(fā)回服務(wù)器。4.3 服務(wù)端結(jié)果處理與持久化Python服務(wù)端收到task_result后不僅更新設(shè)備狀態(tài)更重要的是將結(jié)果記錄到數(shù)據(jù)庫。# 在handle_device_connection的task_result處理部分補(bǔ)充 async def handle_device_connection(websocket, path): # ... 省略其他代碼 ... elif msg_type task_result: task_id data.get(taskId) result data.get(result, {}) # 1. 更新任務(wù)狀態(tài)為完成或失敗 await update_task_in_db(task_id, statuscompleted, resultresult) # 2. 記錄設(shè)備活動(dòng)日志 await log_device_activity(device_id, f完成任務(wù) {task_id}, 結(jié)果: {result.get(success)}) # 3. 如果任務(wù)失敗且重試次數(shù)未超限可以重新放入隊(duì)列 if not result.get(success) and task_retry_count(task_id) MAX_RETRY: await requeue_task(task_id) logger.info(f任務(wù) {task_id} 處理完畢。)這樣我們就完成了一個(gè)從任務(wù)創(chuàng)建、下發(fā)、執(zhí)行到結(jié)果回收的完整閉環(huán)。通過Python后臺(tái)我們可以清晰地看到所有設(shè)備的實(shí)時(shí)狀態(tài)、任務(wù)執(zhí)行進(jìn)度和成功率實(shí)現(xiàn)了基本的云控能力。5. 進(jìn)階思路與擴(kuò)展方向基礎(chǔ)云控跑通后可以根據(jù)實(shí)際需求向不同方向深化5.1 任務(wù)編排與工作流簡單的單個(gè)任務(wù)如點(diǎn)贊不夠用?？梢栽O(shè)計(jì)“工作流”任務(wù)包含多個(gè)步驟。例如“打開App A - 搜索關(guān)鍵詞 - 截圖 - 返回主頁 - 打開App B - 發(fā)布內(nèi)容”。這需要在任務(wù)定義中支持步驟列表設(shè)備端按順序執(zhí)行并在每個(gè)步驟后向服務(wù)器報(bào)告進(jìn)度。5.2 設(shè)備分組與策略調(diào)度不是所有設(shè)備都一樣。有的設(shè)備性能好有的安裝了特定App?？梢栽谠O(shè)備注冊(cè)時(shí)上報(bào)更詳細(xì)的能力和屬性如系統(tǒng)版本、屏幕分辨率、已安裝App列表。服務(wù)端根據(jù)任務(wù)要求將任務(wù)智能分配給最合適的設(shè)備組。例如一個(gè)需要高版本Android才能運(yùn)行的任務(wù)只會(huì)分配給系統(tǒng)版本符合的設(shè)備。5.3 引入消息隊(duì)列如Redis解耦當(dāng)任務(wù)量巨大時(shí)直接在Python內(nèi)存中管理隊(duì)列可能不夠可靠?？梢砸隦edis作為消息中間件。Python服務(wù)端將任務(wù)推入Redis的List或Stream獨(dú)立的“調(diào)度器”服務(wù)從Redis消費(fèi)任務(wù)再通過WebSocket分配給設(shè)備。這樣實(shí)現(xiàn)了生產(chǎn)生成任務(wù)、消費(fèi)調(diào)度分配、執(zhí)行設(shè)備的完全解耦系統(tǒng)擴(kuò)展性更強(qiáng)。5.4 豐富的監(jiān)控與管理界面純命令行日志難以監(jiān)控上百臺(tái)設(shè)備。可以用Python的Web框架如FastAPI、Flask快速搭建一個(gè)管理后臺(tái)提供Web界面。實(shí)時(shí)展示設(shè)備在線狀態(tài)、CPU/內(nèi)存使用率需設(shè)備端上報(bào)、任務(wù)隊(duì)列情況、執(zhí)行日志搜索、手動(dòng)下發(fā)任務(wù)等功能。這會(huì)讓整個(gè)云控系統(tǒng)的易用性提升一個(gè)檔次。5.5 腳本的遠(yuǎn)程更新與熱加載當(dāng)需要修改Autojs端的業(yè)務(wù)邏輯時(shí)逐臺(tái)設(shè)備去更新腳本是噩夢?？梢詫?shí)現(xiàn)一個(gè)“腳本管理”功能。服務(wù)端可以下發(fā)一個(gè)特殊的“更新腳本”任務(wù)包含新腳本的下載地址或代碼內(nèi)容。設(shè)備端收到后下載并替換本地的執(zhí)行腳本然后重啟自動(dòng)化服務(wù)或動(dòng)態(tài)加載新腳本。這樣就實(shí)現(xiàn)了所有設(shè)備的集中式腳本更新。這套“Python AutojsPro7”的云控方案其魅力在于極高的靈活性。Python負(fù)責(zé)復(fù)雜的調(diào)度、管理和數(shù)據(jù)聚合Autojs負(fù)責(zé)在移動(dòng)端執(zhí)行精準(zhǔn)的UI自動(dòng)化。兩者通過WebSocket這個(gè)輕量級(jí)的橋梁結(jié)合讓你能用熟悉的Python生態(tài)去駕馭大量的安卓自動(dòng)化設(shè)備。當(dāng)然真正投入到生產(chǎn)環(huán)境還需要在異常處理、日志收集、安全加固等方面做大量工作。希望這個(gè)分享能為你打開一扇門剩下的精彩就靠你在自己的項(xiàng)目中去探索和實(shí)現(xiàn)了。