""" 工作手机SDK v3.0 - 经验库系统 记录每次执行的操作，自动学习优化，让操作越来越简单 """ import json import os import time from datetime import datetime from typing import Dict, Any, List, Optional from pathlib import Path import hashlib # 经验库存储路径 EXPERIENCE_DIR = Path(__file__).parent.parent.parent / "experience_db" EXPERIENCE_DIR.mkdir(exist_ok=True) class ExperienceDB: """经验库 - 记录和学习操作经验""" def __init__(self): self.db_path = EXPERIENCE_DIR / "operations.json" self.shortcuts_path = EXPERIENCE_DIR / "shortcuts.json" self.app_elements_path = EXPERIENCE_DIR / "app_elements.json" # 加载数据 self.operations: List[Dict] = self._load_json(self.db_path, []) self.shortcuts: Dict[str, Dict] = self._load_json(self.shortcuts_path, {}) self.app_elements: Dict[str, Dict] = self._load_json(self.app_elements_path, {}) def _load_json(self, path: Path, default) -> Any: """加载JSON文件""" if path.exists(): try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except: pass return default def _save_json(self, path: Path, data: Any): """保存JSON文件""" with open(path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) # ========== 操作记录 ========== def record_operation( self, device_id: str, app: str, action: str, params: Dict, result: Dict, duration_ms: int ): """ 记录一次操作 Args: device_id: 设备ID app: APP包名或名称 action: 操作类型 (click, input, swipe, etc.) params: 操作参数 result: 执行结果 duration_ms: 耗时(毫秒) """ operation = { "id": hashlib.md5(f"{time.time()}{device_id}{action}".encode()).hexdigest()[:12], "timestamp": datetime.now().isoformat(), "device_id": device_id, "app": app, "action": action, "params": params, "result": result, "success": result.get("success", result.get("code") == 200), "duration_ms": duration_ms } self.operations.append(operation) # 只保留最近10000条记录 if len(self.operations) > 10000: self.operations = self.operations[-10000:] self._save_json(self.db_path, self.operations) # 自动学习优化 self._learn_from_operation(operation) return operation["id"] def _learn_from_operation(self, operation: Dict): """从操作中学习""" if not operation["success"]: return app = operation["app"] action = operation["action"] params = operation["params"] # 记录APP元素位置 if action in ["click", "click_text"] and operation["success"]: if app not in self.app_elements: self.app_elements[app] = {} # 如果是点击文字，记录文字对应的坐标 if action == "click_text" and "text" in params: text = params["text"] if "x" in operation.get("result", {}).get("data", {}): x = operation["result"]["data"]["x"] y = operation["result"]["data"]["y"] self.app_elements[app][text] = { "x": x, "y": y, "last_success": datetime.now().isoformat(), "success_count": self.app_elements[app].get(text, {}).get("success_count", 0) + 1 } self._save_json(self.app_elements_path, self.app_elements) # ========== 快捷操作 ========== def create_shortcut( self, name: str, description: str, steps: List[Dict] ): """ 创建快捷操作（宏） Args: name: 快捷操作名称 description: 描述 steps: 操作步骤列表 """ shortcut = { "name": name, "description": description, "steps": steps, "created_at": datetime.now().isoformat(), "use_count": 0 } self.shortcuts[name] = shortcut self._save_json(self.shortcuts_path, self.shortcuts) return shortcut def get_shortcut(self, name: str) -> Optional[Dict]: """获取快捷操作""" return self.shortcuts.get(name) def list_shortcuts(self) -> List[Dict]: """列出所有快捷操作""" return list(self.shortcuts.values()) def execute_shortcut(self, name: str, device, variables: Dict = None) -> Dict: """ 执行快捷操作 Args: name: 快捷操作名称 device: ADBDevice实例 variables: 变量替换 """ shortcut = self.shortcuts.get(name) if not shortcut: return {"success": False, "error": f"快捷操作不存在: {name}"} results = [] for step in shortcut["steps"]: action = step["action"] params = step.get("params", {}).copy() # 变量替换 if variables: for key, value in params.items(): if isinstance(value, str) and value.startswith("$"): var_name = value[1:] if var_name in variables: params[key] = variables[var_name] # 执行操作 method = getattr(device, action, None) if method: result = method(**params) results.append(result) # 如果失败，停止执行 if result.get("code") != 200: break # 步骤间延迟 if step.get("delay_ms"): time.sleep(step["delay_ms"] / 1000) # 更新使用次数 shortcut["use_count"] += 1 shortcut["last_used"] = datetime.now().isoformat() self._save_json(self.shortcuts_path, self.shortcuts) return { "success": all(r.get("code") == 200 for r in results), "results": results } # ========== 智能建议 ========== def suggest_coordinates(self, app: str, text: str) -> Optional[Dict]: """ 根据历史记录建议坐标 Args: app: APP名称 text: 要点击的文字 """ if app in self.app_elements and text in self.app_elements[app]: element = self.app_elements[app][text] return { "x": element["x"], "y": element["y"], "confidence": min(element["success_count"] / 10, 1.0), "last_success": element["last_success"] } return None def get_success_rate(self, app: str, action: str) -> float: """获取某APP某操作的成功率""" relevant = [ op for op in self.operations if op["app"] == app and op["action"] == action ] if not relevant: return 0.0 success_count = sum(1 for op in relevant if op["success"]) return success_count / len(relevant) def get_average_duration(self, app: str, action: str) -> float: """获取某APP某操作的平均耗时""" relevant = [ op for op in self.operations if op["app"] == app and op["action"] == action and op["success"] ] if not relevant: return 0.0 return sum(op["duration_ms"] for op in relevant) / len(relevant) def get_recent_operations(self, limit: int = 100) -> List[Dict]: """获取最近的操作记录""" return self.operations[-limit:] def get_statistics(self) -> Dict: """获取统计信息""" if not self.operations: return { "total_operations": 0, "success_rate": 0, "apps_used": [], "shortcuts_count": len(self.shortcuts) } apps = set(op["app"] for op in self.operations) success_count = sum(1 for op in self.operations if op["success"]) return { "total_operations": len(self.operations), "success_rate": success_count / len(self.operations), "apps_used": list(apps), "shortcuts_count": len(self.shortcuts), "elements_learned": sum(len(v) for v in self.app_elements.values()) } # 全局实例 experience_db = ExperienceDB() # ========== 预设快捷操作 ========== def init_default_shortcuts(): """初始化默认快捷操作""" # 微信发消息 if "wechat_send_message" not in experience_db.shortcuts: experience_db.create_shortcut( name="wechat_send_message", description="微信发送消息给指定联系人", steps=[ {"action": "start_app", "params": {"package": "com.tencent.mm"}, "delay_ms": 2000}, {"action": "click_text", "params": {"text": "搜索"}, "delay_ms": 500}, {"action": "input_text", "params": {"text": "$contact"}, "delay_ms": 1000}, {"action": "click_text", "params": {"text": "$contact"}, "delay_ms": 1000}, {"action": "click_text", "params": {"text": "发消息"}, "delay_ms": 500}, {"action": "input_text", "params": {"text": "$message", "clear": False}, "delay_ms": 300}, {"action": "click_text", "params": {"text": "发送"}, "delay_ms": 500}, ] ) # 豆包对话 if "doubao_chat" not in experience_db.shortcuts: experience_db.create_shortcut( name="doubao_chat", description="豆包AI对话", steps=[ {"action": "start_app", "params": {"package": "com.larus.nova"}, "delay_ms": 3000}, {"action": "click_text", "params": {"text": "输入"}, "delay_ms": 500}, {"action": "input_text", "params": {"text": "$question"}, "delay_ms": 300}, {"action": "click_text", "params": {"text": "发送"}, "delay_ms": 500}, ] ) # 打开设置 if "open_settings" not in experience_db.shortcuts: experience_db.create_shortcut( name="open_settings", description="打开系统设置", steps=[ {"action": "start_app", "params": {"package": "com.android.settings"}, "delay_ms": 1000}, ] ) # 初始化默认快捷操作 init_default_shortcuts()