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進程能承載的連接數有限取決于系統資源。當連接數過多例如上千時需要考慮水平擴展?？梢砸脒B接網關如用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 任務編排與工作流簡單的單個任務如點贊不夠用?？梢栽O計“工作流”任務包含多個步驟。例如“打開App A - 搜索關鍵詞 - 截圖 - 返回主頁 - 打開App B - 發布內容”。這需要在任務定義中支持步驟列表設備端按順序執行并在每個步驟后向服務器報告進度。5.2 設備分組與策略調度不是所有設備都一樣。有的設備性能好有的安裝了特定App?？梢栽谠O備注冊時上報更詳細的能力和屬性如系統版本、屏幕分辨率、已安裝App列表。服務端根據任務要求將任務智能分配給最合適的設備組。例如一個需要高版本Android才能運行的任務只會分配給系統版本符合的設備。5.3 引入消息隊列如Redis解耦當任務量巨大時直接在Python內存中管理隊列可能不夠可靠。可以引入Redis作為消息中間件。Python服務端將任務推入Redis的List或Stream獨立的“調度器”服務從Redis消費任務再通過WebSocket分配給設備。這樣實現了生產生成任務、消費調度分配、執行設備的完全解耦系統擴展性更強。5.4 豐富的監控與管理界面純命令行日志難以監控上百臺設備?？梢杂肞ython的Web框架如FastAPI、Flask快速搭建一個管理后臺提供Web界面。實時展示設備在線狀態、CPU/內存使用率需設備端上報、任務隊列情況、執行日志搜索、手動下發任務等功能。這會讓整個云控系統的易用性提升一個檔次。5.5 腳本的遠程更新與熱加載當需要修改Autojs端的業務邏輯時逐臺設備去更新腳本是噩夢?？梢詫崿F一個“腳本管理”功能。服務端可以下發一個特殊的“更新腳本”任務包含新腳本的下載地址或代碼內容。設備端收到后下載并替換本地的執行腳本然后重啟自動化服務或動態加載新腳本。這樣就實現了所有設備的集中式腳本更新。這套“Python AutojsPro7”的云控方案其魅力在于極高的靈活性。Python負責復雜的調度、管理和數據聚合Autojs負責在移動端執行精準的UI自動化。兩者通過WebSocket這個輕量級的橋梁結合讓你能用熟悉的Python生態去駕馭大量的安卓自動化設備。當然真正投入到生產環境還需要在異常處理、日志收集、安全加固等方面做大量工作。希望這個分享能為你打開一扇門剩下的精彩就靠你在自己的項目中去探索和實現了。 Python與AutojsPro7構建云控系統：WebSocket通信與自動化任務調度實踐
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進程能承載的連接數有限取決于系統資源。當連接數過多例如上千時需要考慮水平擴展?？梢砸脒B接網關如用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 任務編排與工作流簡單的單個任務如點贊不夠用?？梢栽O計“工作流”任務包含多個步驟。例如“打開App A - 搜索關鍵詞 - 截圖 - 返回主頁 - 打開App B - 發布內容”。這需要在任務定義中支持步驟列表設備端按順序執行并在每個步驟后向服務器報告進度。5.2 設備分組與策略調度不是所有設備都一樣。有的設備性能好有的安裝了特定App?？梢栽谠O備注冊時上報更詳細的能力和屬性如系統版本、屏幕分辨率、已安裝App列表。服務端根據任務要求將任務智能分配給最合適的設備組。例如一個需要高版本Android才能運行的任務只會分配給系統版本符合的設備。5.3 引入消息隊列如Redis解耦當任務量巨大時直接在Python內存中管理隊列可能不夠可靠。可以引入Redis作為消息中間件。Python服務端將任務推入Redis的List或Stream獨立的“調度器”服務從Redis消費任務再通過WebSocket分配給設備。這樣實現了生產生成任務、消費調度分配、執行設備的完全解耦系統擴展性更強。5.4 豐富的監控與管理界面純命令行日志難以監控上百臺設備?？梢杂肞ython的Web框架如FastAPI、Flask快速搭建一個管理后臺提供Web界面。實時展示設備在線狀態、CPU/內存使用率需設備端上報、任務隊列情況、執行日志搜索、手動下發任務等功能。這會讓整個云控系統的易用性提升一個檔次。5.5 腳本的遠程更新與熱加載當需要修改Autojs端的業務邏輯時逐臺設備去更新腳本是噩夢?？梢詫崿F一個“腳本管理”功能。服務端可以下發一個特殊的“更新腳本”任務包含新腳本的下載地址或代碼內容。設備端收到后下載并替換本地的執行腳本然后重啟自動化服務或動態加載新腳本。這樣就實現了所有設備的集中式腳本更新。這套“Python AutojsPro7”的云控方案其魅力在于極高的靈活性。Python負責復雜的調度、管理和數據聚合Autojs負責在移動端執行精準的UI自動化。兩者通過WebSocket這個輕量級的橋梁結合讓你能用熟悉的Python生態去駕馭大量的安卓自動化設備。當然真正投入到生產環境還需要在異常處理、日志收集、安全加固等方面做大量工作。希望這個分享能為你打開一扇門剩下的精彩就靠你在自己的項目中去探索和實現了。