""" 微信 ADB 后端静默操作引擎 v2.0 通过 ADB + UI 自动化完整控制微信的所有功能（无需 Root） 适配微信 8.0.56 版本 功能覆盖: 1. 消息管理 - 发送/获取/转发/撤回/名片/文件 2. 好友管理 - 添加/通过/备注/删除/搜索/批量 3. 群聊管理 - 创建/邀请/移除/公告/群名/群消息/退群 4. 标签管理 - 创建/删除/列表/打标/取标/按标签查人 5. 朋友圈 - 发布/点赞/评论/获取/删除 6. 个人设置 - 昵称/签名/头像/性别/地区/资料 7. 账号安全 - 状态检查/解封/安全中心/改密码 8. 收藏管理 - 收藏/列表 9. 支付/红包 - 发红包/转账 10. 聊天设置 - 置顶/免打扰/清空记录 11. 小程序 - 打开 12. 公众号 - 关注 13. 解封完善 - 自助解封/申诉/限制检查/短信解封 14. 视频号 - 打开/点赞/评论/关注/分享/列表 15. 扫一扫 - 扫码/加好友/我的二维码/图片识别 16. 支付完善 - 付款码/收款码/钱包/账单/领红包 17. 通话 - 语音通话/视频通话 18. 群发助手 - 群发消息 19. 搜一搜/看一看 - 搜一搜/看一看 20. 微信运动 - 获取步数/点赞步数 21. 位置分享 - 发送位置/共享实时位置 22. 表情管理 - 发送表情/表情包列表 23. 朋友圈增强 - 设置封面/可见天数/分享链接 24. 语音消息 - 发送语音消息 25. 文件管理 - 发送文件/下载文件 26. 设置管理 - 勿扰模式/清理缓存/检查更新/退出/切换账号 全局容错: _retry / _recover_state / _verify_action """ import time import re import logging import subprocess import base64 from typing import Dict, Any, List, Optional, Tuple logger = logging.getLogger(__name__) WX_PKG = "com.tencent.mm" WX_LAUNCHER = f"{WX_PKG}/.ui.LauncherUI" WX_SEARCH = f"{WX_PKG}/.plugin.subapp.ui.friend.FTSMainUI" class WeChatADBEngine: """微信 ADB 后端静默操作引擎 — 全功能（带导航缓存 & 操作日志）""" def __init__(self, device): self.d = device self._w = 1080 self._h = 2400 self._detect_screen() self._init_nav_cache() def _init_nav_cache(self): """初始化导航缓存和操作日志""" try: from services.wechat_nav_cache import WeChatNavCache, OperationLogger self.nav = WeChatNavCache("8.0.56", self._w, self._h) self.op_log = OperationLogger( getattr(self.d, 'serial', getattr(self.d, 'device_id', ''))) except Exception as e: logger.warning(f"导航缓存初始化失败（降级为无缓存模式）: {e}") self.nav = None self.op_log = None # ═══════════════════ 底层工具方法 ═══════════════════ def _detect_screen(self): try: size = self.d._shell("wm size") if "Physical size:" in size: dims = size.split("Physical size:")[-1].strip() if "x" in dims: self._w, self._h = map(int, dims.split("x")) except Exception: pass def _sh(self, cmd: str, timeout: int = 10) -> str: return self.d._shell(cmd, timeout=timeout) def _tap(self, x: int, y: int): self._sh(f"input tap {x} {y}") time.sleep(0.3) def _swipe(self, x1, y1, x2, y2, ms=400): self._sh(f"input swipe {x1} {y1} {x2} {y2} {ms}") time.sleep(0.5) def _scroll_up(self): cx = self._w // 2 self._swipe(cx, self._h * 3 // 4, cx, self._h // 4) def _scroll_down(self): cx = self._w // 2 self._swipe(cx, self._h // 4, cx, self._h * 3 // 4) def _back(self): self._sh("input keyevent KEYCODE_BACK") time.sleep(0.3) def _enter(self): self._sh("input keyevent KEYCODE_ENTER") time.sleep(0.3) def _long_press(self, x, y, duration_ms=1000): self._swipe(x, y, x, y, duration_ms) def _wait(self, sec: float = 1.0): time.sleep(sec) def _get_xml(self) -> str: try: self._sh("uiautomator dump /sdcard/ui.xml", timeout=5) return self._sh("cat /sdcard/ui.xml", timeout=3) except Exception: return "" def _find_text(self, text: str, xml: str = None) -> Optional[Tuple[int, int]]: if xml is None: xml = self._get_xml() if not xml: return None for pat in [ f'text="{re.escape(text)}"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"', f'content-desc="{re.escape(text)}"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"', f'text="[^"]*{re.escape(text)}[^"]*"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"', ]: m = re.search(pat, xml) if m: x1, y1, x2, y2 = map(int, m.groups()) return ((x1 + x2) // 2, (y1 + y2) // 2) return None def _click_text(self, text: str, timeout: float = 5) -> bool: end = time.time() + timeout while time.time() < end: pos = self._find_text(text) if pos: self._tap(*pos) return True time.sleep(0.5) return False def _input(self, text: str, clear: bool = False): if clear: try: self._sh("input keyevent KEYCODE_CTRL_A", timeout=2) time.sleep(0.1) self._sh("input keyevent KEYCODE_DEL", timeout=2) time.sleep(0.1) except Exception as e: logger.info(f"清空输入框失败，继续输入: {e}") has_cjk = any(ord(c) > 127 for c in text) if has_cjk: self._input_cjk(text) else: escaped = text.replace(" ", "%s").replace("'", "\\'").replace('"', '\\"') escaped = escaped.replace("&", "\\&").replace("|", "\\|").replace(";", "\\;") self._sh(f"input text '{escaped}'") time.sleep(0.3) def _input_cjk(self, text: str): escaped = text.replace("'", "'\\''").replace('"', '\\"') try: result = self._sh( f"am broadcast -a ADB_INPUT_TEXT --es msg '{escaped}' 2>&1", timeout=2, ) if "Broadcast completed" in result: return except Exception as e: logger.info(f"ADB_INPUT_TEXT 不可用，回退系统输入: {e}") # 部分 MIUI/HyperOS 设备支持直接 input text 中文，这里作为兜底。 self._sh(f"input text '{escaped}'", timeout=5) def _ensure_wechat(self) -> bool: cur = self._sh("dumpsys window | grep -E 'mCurrentFocus'") if WX_PKG not in cur: self._sh(f"am start -n {WX_LAUNCHER}") time.sleep(2) if "NotificationShade" in cur: self._sh("input swipe 540 1200 540 100 200") time.sleep(0.5) self._sh("input keyevent KEYCODE_HOME") time.sleep(0.5) self._sh(f"am start -n {WX_LAUNCHER}") time.sleep(2) return True def _go_home(self): self._sh(f"am start -n {WX_LAUNCHER}") time.sleep(1.5) def _tab_positions(self) -> dict: """底部 4 个 tab 的中心坐标（优先从缓存取精确坐标）""" if self.nav: tabs = self.nav.element_cache.get("tabs", {}) if tabs: return {k: tuple(v["center"]) for k, v in tabs.items()} tw = self._w // 4 ty = self._h - 252 return { "微信": (tw // 2, ty), "通讯录": (tw + tw // 2, ty), "发现": (tw * 2 + tw // 2, ty), "我": (tw * 3 + tw // 2, ty), } def _tap_tab(self, name: str): """点击微信底部 tab（缓存优先）""" if self.nav: coord = self.nav.get_tab_coord(name) if coord: self._tap(*coord) time.sleep(1) if self.op_log: self.op_log.log_step("tap_tab", name, coord) return pos = self._tab_positions().get(name) if pos: self._tap(*pos) time.sleep(1) def _navigate_cached(self, path_name: str) -> bool: """用缓存的路径快速导航到目标页面""" if not self.nav: return False path = self.nav.get_path(path_name) if not path: return False self._ensure_wechat() self._go_home() time.sleep(0.5) for step in path.steps: try: if step.action == "tap_tab": self._tap_tab(step.target) elif step.action == "tap_text": if step.coords: self._tap(*step.coords) elif not self._click_text(step.target, timeout=3): self.nav.record_path_result(path_name, False) return False elif step.action == "tap_coord": if step.coords: self._tap(*step.coords) elif step.action == "scroll_up": self._scroll_up() elif step.action == "scroll_down": self._scroll_down() elif step.action == "scroll_to_top": for _ in range(3): self._swipe(self._w // 2, 400, self._w // 2, 1600, 200) time.sleep(0.2) elif step.action == "back": self._back() elif step.action == "swipe": pass time.sleep(step.wait) if self.op_log: self.op_log.log_step( step.action, step.target or "", step.coords, True) except Exception as e: logger.warning(f"导航步骤失败 [{step.action}]: {e}") self.nav.record_path_result(path_name, False) return False self.nav.record_path_result(path_name, True) return True def _record_page_elements(self, page_name: str) -> Dict[str, Any]: """录制当前页面所有元素到缓存""" xml = self._get_xml() items = re.findall( r'text="([^"]+)"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml) elements = {} for t, x1, y1, x2, y2 in items: if t.strip() and int(y1) > 50 and int(y1) < 2100: cx = (int(x1) + int(x2)) // 2 cy = (int(y1) + int(y2)) // 2 elements[t] = {"center": [cx, cy], "bounds": [int(x1), int(y1), int(x2), int(y2)]} if self.nav and elements: self.nav.record_page(page_name, elements) return elements def _search_and_enter(self, keyword: str) -> bool: self._go_home() time.sleep(0.5) if not self._click_text("搜索", timeout=3): self._sh(f"am start -n {WX_SEARCH}") time.sleep(1) time.sleep(0.5) self._input(keyword, clear=True) time.sleep(1.5) xml = self._get_xml() pos = self._find_text(keyword, xml) if pos: self._tap(*pos) time.sleep(1) return True self._tap(self._w // 2, 350) time.sleep(1) return True def _ok(self, **kw) -> dict: r = {"success": True, "mode": "adb", "timestamp": int(time.time() * 1000)} r.update(kw) return r def _fail(self, msg: str) -> dict: return {"success": False, "error": msg, "mode": "adb"} def _retry(self, fn, max_retries=3, delay=1.0): """通用重试包装器""" for i in range(max_retries): try: result = fn() if isinstance(result, dict) and result.get("success"): return result if isinstance(result, dict) and not result.get("success") and i < max_retries - 1: logger.warning(f"操作失败，重试 {i+1}/{max_retries}: {result.get('error')}") self._recover_state() time.sleep(delay) continue return result except Exception as e: if i < max_retries - 1: logger.warning(f"异常，重试 {i+1}/{max_retries}: {e}") self._recover_state() time.sleep(delay) else: return self._fail(f"重试{max_retries}次后仍失败: {e}") return self._fail("重试耗尽") def _recover_state(self): """状态恢复：关闭弹窗/对话框，回到微信主界面""" try: self._click_text("取消", timeout=1) except Exception: pass try: self._click_text("关闭", timeout=1) except Exception: pass self._back() time.sleep(0.3) self._back() time.sleep(0.3) self._go_home() def _verify_action(self, expected_text: str = None) -> bool: """验证操作是否成功（通过检查当前页面）""" xml = self._get_xml() if expected_text: return self._find_text(expected_text, xml) is not None return len(xml) > 100 # ═══════════════════ 1. 消息管理 ═══════════════════ def send_message(self, to_id: str, content: str, msg_type: str = "text", media_url: str = None, at_list: list = None) -> dict: try: self._ensure_wechat() if not self._search_and_enter(to_id): return self._fail(f"未找到联系人: {to_id}") time.sleep(0.5) if msg_type == "image" and media_url: return self._send_media_in_chat("相册") if msg_type == "file" and media_url: return self._send_media_in_chat("文件") msg = content if at_list: msg = " ".join(f"@{a}" for a in at_list) + " " + content self._input(msg) time.sleep(0.3) if not self._click_text("发送", timeout=3): self._enter() return self._ok(message_id=f"adb_{int(time.time()*1000)}") except Exception as e: return self._fail(str(e)) def _send_media_in_chat(self, media_type: str) -> dict: self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) self._click_text(media_type) or self._click_text("图片") time.sleep(1) self._tap(self._w // 4, self._h // 3) time.sleep(0.5) self._click_text("发送") or self._click_text("完成") return self._ok(message_id=f"media_{int(time.time()*1000)}") def get_messages(self, conversation_id: str = None, limit: int = 20) -> dict: try: self._ensure_wechat() if conversation_id: self._search_and_enter(conversation_id) time.sleep(0.5) xml = self._get_xml() messages = self._parse_chat_messages(xml, limit) return self._ok(messages=messages, count=len(messages)) except Exception as e: return self._fail(str(e)) def _parse_chat_messages(self, xml: str, limit: int) -> list: msgs = [] if not xml: return msgs all_texts = re.findall(r'text="([^"]{2,})"', xml) skip = {"微信", "搜索", "发送", "返回", "更多", "+", "语音", "表情", "切换到键盘", "按住说话", "通讯录", "发现", "我"} for t in all_texts: if t not in skip: msgs.append({"content": t, "timestamp": int(time.time())}) if len(msgs) >= limit: break return msgs def forward_message(self, to_id: str, content: str) -> dict: try: self._ensure_wechat() xml = self._get_xml() pos = self._find_text(content, xml) if pos: self._long_press(*pos) time.sleep(0.5) if self._click_text("转发", timeout=3): time.sleep(0.5) self._input(to_id, clear=True) time.sleep(1) self._click_text(to_id, timeout=3) time.sleep(0.5) self._click_text("发送", timeout=3) return self._ok() return self._fail("未找到要转发的消息") except Exception as e: return self._fail(str(e)) def recall_message(self) -> dict: try: self._ensure_wechat() xml = self._get_xml() all_msgs = re.findall( r'text="([^"]+)"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml ) for text, x1, y1, x2, y2 in reversed(all_msgs): if int(x1) > self._w // 3: cx, cy = (int(x1) + int(x2)) // 2, (int(y1) + int(y2)) // 2 self._long_press(cx, cy) time.sleep(0.5) if self._click_text("撤回", timeout=3): time.sleep(0.5) self._click_text("确定") or self._click_text("确认") return self._ok() break return self._fail("未找到可撤回的消息") except Exception as e: return self._fail(str(e)) def send_card(self, to_id: str, card_wxid: str) -> dict: try: self._ensure_wechat() self._search_and_enter(to_id) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) if not self._click_text("名片", timeout=2): self._scroll_up() self._click_text("名片", timeout=2) time.sleep(1) self._input(card_wxid, clear=True) time.sleep(1) self._click_text(card_wxid, timeout=3) time.sleep(0.5) self._click_text("发送", timeout=3) return self._ok() except Exception as e: return self._fail(str(e)) # ═══════════════════ 2. 好友管理 ═══════════════════ def add_friend(self, user_id: str, message: str = "", source: str = None) -> dict: try: self._ensure_wechat() self._go_home() self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("添加朋友", timeout=3) time.sleep(1) self._click_text("微信号/手机号", timeout=3) or self._tap(self._w // 2, 200) time.sleep(0.3) self._input(user_id, clear=True) time.sleep(0.5) self._click_text("搜索", timeout=3) or self._enter() time.sleep(2) if self._click_text("添加到通讯录", timeout=3): time.sleep(1) if message: self._input(message, clear=True) self._click_text("发送", timeout=3) return self._ok(note=f"好友请求已发送: {user_id}") return self._fail(f"未找到用户或已是好友: {user_id}") except Exception as e: return self._fail(str(e)) def accept_friend(self, user_id: str = None) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(0.5) self._click_text("新的朋友", timeout=3) time.sleep(1) if user_id: self._click_text(user_id, timeout=2) time.sleep(0.5) if self._click_text("接受", timeout=3) or self._click_text("通过", timeout=2): return self._ok(note="好友请求已通过") return self._fail("未找到待处理的好友请求") except Exception as e: return self._fail(str(e)) def set_remark(self, user_id: str, remark: str) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._tap(self._w - 60, 80) time.sleep(1) if not self._click_text("备注", timeout=3): self._click_text("设置备注", timeout=2) time.sleep(0.5) self._input(remark, clear=True) time.sleep(0.3) self._click_text("完成") or self._click_text("保存") or self._click_text("确定") return self._ok(note=f"备注已设为: {remark}") except Exception as e: return self._fail(str(e)) def delete_friend(self, user_id: str) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._tap(self._w - 60, 80) time.sleep(1) self._tap(self._w // 2, 200) time.sleep(1) self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("删除", timeout=3) time.sleep(0.5) self._click_text("删除", timeout=3) return self._ok(note=f"已删除好友: {user_id}") except Exception as e: return self._fail(str(e)) def get_contacts(self, limit: int = 100) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(1) contacts, seen = [], set() for _ in range(min(limit // 10, 20)): xml = self._get_xml() texts = re.findall(r'text="([^"]{2,15})"', xml) skip = {"微信", "通讯录", "发现", "我", "搜索", "新的朋友", "群聊", "标签", "公众号", "返回", "添加"} for n in texts: if n not in skip and n not in seen: seen.add(n) contacts.append({"nickname": n}) if len(contacts) >= limit: break self._scroll_up() time.sleep(0.5) return self._ok(contacts=contacts[:limit], count=len(contacts[:limit])) except Exception as e: return self._fail(str(e)) def search_contact(self, keyword: str) -> dict: try: self._ensure_wechat() self._go_home() self._click_text("搜索", timeout=3) or self._sh(f"am start -n {WX_SEARCH}") time.sleep(0.5) self._input(keyword, clear=True) time.sleep(1.5) xml = self._get_xml() results = [] texts = re.findall(r'text="([^"]+)"', xml) skip = {"搜索", "返回", keyword, "联系人", "群聊", "聊天记录", "公众号", "朋友圈", "文章", "小程序"} for n in texts: if n not in skip and len(n) > 1: results.append({"nickname": n}) self._back() return self._ok(results=results, count=len(results)) except Exception as e: return self._fail(str(e)) def batch_add_friend(self, user_ids: list, message: str = "", interval: float = 5) -> dict: added, failed = [], [] for uid in user_ids: r = self.add_friend(uid, message) if r.get("success"): added.append(uid) else: failed.append({"user_id": uid, "error": r.get("error")}) time.sleep(max(interval, 3)) return self._ok(added=added, failed=failed, total=len(user_ids)) def get_friend_info(self, user_id: str) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._tap(self._w - 60, 80) time.sleep(1) self._tap(self._w // 2, 200) time.sleep(1) xml = self._get_xml() info = {"user_id": user_id} texts = re.findall(r'text="([^"]+)"', xml) for t in texts: if t.startswith("微信号"): info["wechat_id"] = t.replace("微信号：", "").replace("微信号:", "").strip() elif t.startswith("地区"): info["region"] = t.replace("地区：", "").replace("地区:", "").strip() if texts: for t in texts: if t not in {"返回", "更多", "发消息", "音视频通话"} and len(t) > 1: info.setdefault("nickname", t) break self._back() return self._ok(info=info) except Exception as e: return self._fail(str(e)) # ═══════════════════ 3. 群聊管理 ═══════════════════ def create_group(self, group_name: str, member_ids: list) -> dict: try: self._ensure_wechat() self._go_home() self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("发起群聊", timeout=3) time.sleep(1) for mid in member_ids: self._click_text("搜索", timeout=2) or self._tap(self._w // 2, 150) self._input(mid, clear=True) time.sleep(1) pos = self._find_text(mid) if pos: self._tap(*pos) time.sleep(0.3) self._back() time.sleep(0.3) self._click_text("完成", timeout=3) or self._click_text("确定", timeout=2) time.sleep(2) if group_name: self._tap(self._w - 60, 80) time.sleep(1) if self._click_text("群聊名称", timeout=3) or self._click_text("群名称", timeout=2): time.sleep(0.5) self._input(group_name, clear=True) self._click_text("完成") or self._click_text("保存") or self._click_text("确定") return self._ok(note=f"群聊已创建: {group_name}") except Exception as e: return self._fail(str(e)) def invite_to_group(self, group_id: str, member_ids: list) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) time.sleep(0.5) self._tap(self._w - 60, 80) time.sleep(1) self._tap(self._w - 100, 350) time.sleep(0.5) for mid in member_ids: self._input(mid, clear=True) time.sleep(1) pos = self._find_text(mid) if pos: self._tap(*pos) time.sleep(0.3) self._click_text("邀请") or self._click_text("完成") or self._click_text("确定") return self._ok(note=f"已邀请 {len(member_ids)} 人入群") except Exception as e: return self._fail(str(e)) def remove_from_group(self, group_id: str, member_ids: list) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) self._tap(self._w - 60, 80) time.sleep(1) if not self._click_text("-", timeout=2): self._scroll_up() self._click_text("删除", timeout=2) time.sleep(0.5) for mid in member_ids: pos = self._find_text(mid) if pos: self._tap(*pos) time.sleep(0.3) self._click_text("删除") or self._click_text("完成") or self._click_text("确定") time.sleep(0.5) self._click_text("确定") or self._click_text("确认") return self._ok(note=f"已移除 {len(member_ids)} 人") except Exception as e: return self._fail(str(e)) def set_group_notice(self, group_id: str, notice: str) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) self._tap(self._w - 60, 80) time.sleep(1) self._scroll_up() time.sleep(0.3) self._click_text("群公告", timeout=3) time.sleep(1) self._input(notice, clear=True) time.sleep(0.3) self._click_text("完成") or self._click_text("发布") time.sleep(0.5) self._click_text("确定") or self._click_text("确认") return self._ok(note="群公告已更新") except Exception as e: return self._fail(str(e)) def set_group_name(self, group_id: str, group_name: str) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) self._tap(self._w - 60, 80) time.sleep(1) if self._click_text("群聊名称", timeout=3) or self._click_text("群名称", timeout=2): time.sleep(0.5) self._input(group_name, clear=True) self._click_text("完成") or self._click_text("保存") return self._ok(note=f"群名已改为: {group_name}") return self._fail("未找到群名称设置入口") except Exception as e: return self._fail(str(e)) def send_group_message(self, group_id: str, content: str, msg_type: str = "text", media_url: str = None, at_all: bool = False, at_list: list = None) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) time.sleep(0.5) msg = content if at_all: msg = "@所有人 " + content elif at_list: msg = " ".join(f"@{a}" for a in at_list) + " " + content self._input(msg) time.sleep(0.3) self._click_text("发送", timeout=3) or self._enter() return self._ok(message_id=f"group_{int(time.time()*1000)}") except Exception as e: return self._fail(str(e)) def set_group_welcome(self, group_id: str, welcome_text: str, welcome_image: str = None) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) self._tap(self._w - 60, 80) time.sleep(1) self._scroll_up() if self._click_text("群管理", timeout=3): time.sleep(0.5) if self._click_text("进群欢迎语", timeout=3) or self._click_text("欢迎语", timeout=2): time.sleep(0.5) self._input(welcome_text, clear=True) self._click_text("完成") or self._click_text("保存") return self._ok(note="群欢迎语已设置") return self._fail("未找到群欢迎语设置") except Exception as e: return self._fail(str(e)) def get_groups(self, limit: int = 100) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(0.5) self._click_text("群聊", timeout=3) time.sleep(1) groups, seen = [], set() for _ in range(min(limit // 8, 15)): xml = self._get_xml() names = re.findall(r'text="([^"]{2,30})"', xml) skip = {"群聊", "通讯录", "搜索", "返回"} for n in names: if n not in skip and n not in seen: seen.add(n) groups.append({"group_name": n}) if len(groups) >= limit: break self._scroll_up() time.sleep(0.5) return self._ok(groups=groups[:limit], count=len(groups[:limit])) except Exception as e: return self._fail(str(e)) def get_group_members(self, group_id: str) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) self._tap(self._w - 60, 80) time.sleep(1) self._click_text("查看全部", timeout=3) or self._click_text("全部群成员", timeout=2) time.sleep(1) members, seen = [], set() for _ in range(10): xml = self._get_xml() names = re.findall(r'text="([^"]{2,20})"', xml) skip = {"群成员", "搜索", "返回", group_id, "邀请"} for n in names: if n not in skip and n not in seen: seen.add(n) members.append({"nickname": n}) self._scroll_up() time.sleep(0.3) return self._ok(members=members, count=len(members)) except Exception as e: return self._fail(str(e)) def quit_group(self, group_id: str) -> dict: try: self._ensure_wechat() self._search_and_enter(group_id) self._tap(self._w - 60, 80) time.sleep(1) for _ in range(5): self._scroll_up() time.sleep(0.3) if self._click_text("删除并退出", timeout=3) or self._click_text("退出群聊", timeout=2): time.sleep(0.5) self._click_text("确定") or self._click_text("确认") return self._ok(note=f"已退出群: {group_id}") return self._fail("未找到退出群聊按钮") except Exception as e: return self._fail(str(e)) # ═══════════════════ 4. 标签管理 ═══════════════════ def create_tag(self, tag_name: str) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(0.5) self._click_text("标签", timeout=3) time.sleep(1) self._click_text("新建标签", timeout=3) or self._tap(self._w // 2, self._h - 100) time.sleep(0.5) self._input(tag_name, clear=True) self._click_text("确定") or self._click_text("保存") return self._ok(note=f"标签已创建: {tag_name}") except Exception as e: return self._fail(str(e)) def delete_tag(self, tag_name: str) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(0.5) self._click_text("标签", timeout=3) time.sleep(1) if self._click_text(tag_name, timeout=3): time.sleep(0.5) self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("删除标签", timeout=3) time.sleep(0.5) self._click_text("确定") or self._click_text("确认") return self._ok(note=f"标签已删除: {tag_name}") return self._fail(f"未找到标签: {tag_name}") except Exception as e: return self._fail(str(e)) def get_tags(self) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(0.5) self._click_text("标签", timeout=3) time.sleep(1) xml = self._get_xml() tags = [] names = re.findall(r'text="([^"]+)"', xml) skip = {"标签", "新建标签", "通讯录", "搜索", "返回"} for n in names: if n not in skip and len(n) > 0: tags.append({"tag_name": n}) return self._ok(tags=tags, count=len(tags)) except Exception as e: return self._fail(str(e)) def add_tag(self, user_id: str, tags: list) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) self._tap(self._w - 60, 80) time.sleep(1) self._tap(self._w // 2, 200) time.sleep(1) if self._click_text("标签", timeout=3) or self._click_text("设置备注和标签", timeout=2): time.sleep(0.5) for tag in tags: if not self._click_text(tag, timeout=2): self._click_text("添加标签", timeout=2) self._input(tag) self._enter() time.sleep(0.3) self._click_text("完成") or self._click_text("保存") or self._click_text("确定") return self._ok(note=f"已为 {user_id} 添加标签: {tags}") return self._fail("未找到标签设置入口") except Exception as e: return self._fail(str(e)) def remove_tag(self, user_id: str, tags: list) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) self._tap(self._w - 60, 80) time.sleep(1) self._tap(self._w // 2, 200) time.sleep(1) if self._click_text("标签", timeout=3): time.sleep(0.5) for tag in tags: self._click_text(tag, timeout=2) time.sleep(0.2) self._click_text("完成") or self._click_text("保存") return self._ok(note=f"已为 {user_id} 移除标签: {tags}") return self._fail("未找到标签设置入口") except Exception as e: return self._fail(str(e)) def get_users_by_tag(self, tag_name: str, limit: int = 100) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("通讯录") time.sleep(0.5) self._click_text("标签", timeout=3) time.sleep(1) if self._click_text(tag_name, timeout=3): time.sleep(1) xml = self._get_xml() users = [] names = re.findall(r'text="([^"]+)"', xml) skip = {tag_name, "标签", "返回", "更多"} for n in names: if n not in skip and len(n) > 1: users.append({"nickname": n}) return self._ok(users=users[:limit], count=len(users[:limit])) return self._fail(f"未找到标签: {tag_name}") except Exception as e: return self._fail(str(e)) # ═══════════════════ 5. 朋友圈管理 ═══════════════════ def post_moments(self, content: str, images: list = None, video_url: str = None, location: str = None, visible_list: list = None, invisible_list: list = None) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("发现") time.sleep(0.5) self._click_text("朋友圈", timeout=3) time.sleep(1.5) if images or video_url: self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("从相册选择", timeout=3) or self._click_text("相册", timeout=2) time.sleep(1) count = min(len(images) if images else 1, 9) for i in range(count): col, row = i % 3, i // 3 x = self._w // 6 + col * (self._w // 3) y = self._h // 4 + row * (self._h // 4) self._tap(x, y) time.sleep(0.3) self._click_text("完成", timeout=3) time.sleep(1) else: self._long_press(self._w - 60, 80, 2000) time.sleep(1) self._input(content) time.sleep(0.3) if location: self._click_text("所在位置", timeout=2) time.sleep(0.5) self._input(location, clear=True) time.sleep(1) pos = self._find_text(location) if pos: self._tap(*pos) time.sleep(0.3) if visible_list or invisible_list: self._click_text("谁可以看", timeout=2) time.sleep(0.5) if visible_list: self._click_text("部分可见", timeout=2) elif invisible_list: self._click_text("不给谁看", timeout=2) time.sleep(0.3) self._click_text("确定") self._click_text("发表", timeout=5) or self._click_text("发送", timeout=3) time.sleep(2) return self._ok(note="朋友圈已发布") except Exception as e: return self._fail(str(e)) def like_moments(self, user_id: str = None, post_index: int = 0) -> dict: try: self._ensure_wechat() self._open_moments(user_id) for _ in range(post_index): self._scroll_up() time.sleep(0.5) xml = self._get_xml() hearts = re.findall( r'content-desc="[^"]*(?:评论|赞)[^"]*"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml ) if hearts: x1, y1, x2, y2 = map(int, hearts[0]) self._tap((x1 + x2) // 2, (y1 + y2) // 2) time.sleep(0.3) self._click_text("赞", timeout=3) return self._ok(note="已点赞") self._tap(self._w - 80, self._h // 2) time.sleep(0.3) self._click_text("赞", timeout=3) return self._ok(note="已点赞") except Exception as e: return self._fail(str(e)) def comment_moments(self, user_id: str = None, post_index: int = 0, comment: str = "", reply_to: str = None) -> dict: try: self._ensure_wechat() self._open_moments(user_id) for _ in range(post_index): self._scroll_up() time.sleep(0.5) self._tap(self._w - 80, self._h // 2) time.sleep(0.3) self._click_text("评论", timeout=3) time.sleep(0.5) if reply_to: self._click_text(reply_to, timeout=2) time.sleep(0.3) self._input(comment) self._click_text("发送", timeout=3) or self._enter() return self._ok(note=f"已评论: {comment[:20]}...") except Exception as e: return self._fail(str(e)) def get_moments(self, user_id: str = None, limit: int = 10) -> dict: try: self._ensure_wechat() self._open_moments(user_id) moments = [] for _ in range(min(limit, 20)): xml = self._get_xml() texts = re.findall(r'text="([^"]{5,})"', xml) skip = {"朋友圈", "发现", "搜索", "返回"} for t in texts: if t not in skip and {"content": t} not in moments: moments.append({"content": t, "timestamp": int(time.time())}) if len(moments) >= limit: break self._scroll_up() time.sleep(0.8) return self._ok(moments=moments[:limit], count=len(moments[:limit])) except Exception as e: return self._fail(str(e)) def delete_moments(self, post_index: int = 0) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("发现") time.sleep(0.5) self._click_text("朋友圈", timeout=3) time.sleep(1.5) self._tap(60, 180) time.sleep(1) for _ in range(post_index): self._scroll_up() time.sleep(0.5) self._tap(self._w // 2, self._h // 2) time.sleep(0.5) self._click_text("删除", timeout=3) time.sleep(0.5) self._click_text("确定") or self._click_text("确认") or self._click_text("删除") return self._ok(note="朋友圈已删除") except Exception as e: return self._fail(str(e)) def _open_moments(self, user_id: str = None): if user_id: self._search_and_enter(user_id) self._tap(self._w - 60, 80) time.sleep(0.5) self._tap(self._w // 2, 200) time.sleep(1) self._click_text("朋友圈", timeout=3) else: self._go_home() self._tap_tab("发现") time.sleep(0.5) self._click_text("朋友圈", timeout=3) time.sleep(1.5) # ═══════════════════ 6. 个人设置 ═══════════════════ def _enter_profile(self): self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._tap(self._w // 2, 200) time.sleep(1) def set_nickname(self, nickname: str) -> dict: try: self._enter_profile() self._click_text("名字", timeout=3) or self._click_text("昵称", timeout=2) time.sleep(0.5) self._input(nickname, clear=True) time.sleep(0.3) self._click_text("完成") or self._click_text("保存") return self._ok(note=f"昵称已改为: {nickname}") except Exception as e: return self._fail(str(e)) def set_signature(self, signature: str) -> dict: try: self._enter_profile() if not self._click_text("个性签名", timeout=3): self._scroll_up() self._click_text("个性签名", timeout=3) time.sleep(0.5) self._input(signature, clear=True) self._click_text("完成") or self._click_text("保存") return self._ok(note=f"签名已改为: {signature}") except Exception as e: return self._fail(str(e)) def set_avatar(self, image_path: str = None) -> dict: try: self._enter_profile() self._click_text("头像", timeout=3) time.sleep(1) self._click_text("从相册选择", timeout=3) or self._click_text("相册", timeout=2) time.sleep(1) self._tap(self._w // 4, self._h // 3) time.sleep(0.5) self._click_text("使用") or self._click_text("完成") or self._click_text("确定") time.sleep(1) return self._ok(note="头像已更换") except Exception as e: return self._fail(str(e)) def set_gender(self, gender: str) -> dict: try: self._enter_profile() if not self._click_text("性别", timeout=3): self._scroll_up() self._click_text("性别", timeout=3) time.sleep(0.5) target = "男" if gender.lower() in ("male", "男") else "女" self._click_text(target, timeout=3) self._click_text("完成") or self._click_text("保存") return self._ok(note=f"性别已设为: {target}") except Exception as e: return self._fail(str(e)) def set_region(self, region: str) -> dict: try: self._enter_profile() if not self._click_text("地区", timeout=3): self._scroll_up() self._click_text("地区", timeout=3) time.sleep(0.5) for part in region.split(): self._click_text(part, timeout=3) time.sleep(0.5) self._click_text("完成") or self._click_text("保存") return self._ok(note=f"地区已设为: {region}") except Exception as e: return self._fail(str(e)) def get_profile(self) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(1) xml = self._get_xml() profile = {} elements = re.findall( r'text="([^"]+)"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml ) skip = {"我", "微信", "通讯录", "发现", "搜索", "状态", "浮窗", "表情", "服务", "收藏", "卡包", "设置", "3个朋友"} for txt, x1, y1, x2, y2 in elements: y = int(y1) if txt.startswith("微信号"): profile["wechat_id"] = txt.replace("微信号：", "").replace("微信号:", "").strip() elif 200 < y < 370 and txt not in skip and len(txt) > 1: profile.setdefault("nickname", txt) return self._ok(profile=profile) except Exception as e: return self._fail(str(e)) # ═══════════════════ 7. 账号安全与解封 ═══════════════════ def check_account_status(self) -> dict: try: if not self._nav_to_account_security(): return self._fail("无法导航到账号与安全") xml = self._get_xml() info = {"account_active": True} texts = re.findall(r'text="([^"]+)"', xml) for t in texts: if "微信号" in t: info["wechat_id"] = t elif "手机号" in t: info["phone"] = t elif "已绑定" in t or "已保护" in t: info["security_status"] = t self._back() self._back() return self._ok(status=info) except Exception as e: return self._fail(str(e)) def unblock_account(self, helper_wxid: str = None) -> dict: try: if self._nav_to_safety_center(): if self._click_text("解封账号", timeout=3) or \ self._click_text("自助解封", timeout=2): time.sleep(1) if helper_wxid: pos = self._find_text("微信号") if pos: self._tap(pos[0], pos[1] + 80) self._input(helper_wxid) self._click_text("下一步", timeout=3) or \ self._click_text("提交", timeout=2) time.sleep(1) return self._ok( note="解封流程已启动，请辅助好友在微信团队中确认") return self._ok(note="账号当前未被封禁", status="normal") return self._fail("无法导航到微信安全中心") except Exception as e: return self._fail(str(e)) def safety_center(self) -> dict: try: if not self._nav_to_safety_center(): return self._fail("无法导航到微信安全中心") xml = self._get_xml() options = [] texts = re.findall(r'text="([^"]+)"', xml) for t in texts: if t not in {"微信安全中心", "返回"} and len(t) > 1: options.append(t) return self._ok(options=options) except Exception as e: return self._fail(str(e)) def change_password(self, old_pwd: str, new_pwd: str) -> dict: try: if not self._nav_to_account_security(): return self._fail("无法导航到账号与安全") if self._click_text("密码", timeout=3) or self._click_text("微信密码", timeout=2): time.sleep(0.5) self._input(old_pwd) self._enter() time.sleep(0.5) self._input(new_pwd) self._enter() time.sleep(0.5) self._input(new_pwd) self._click_text("完成") or self._click_text("确定") or self._enter() return self._ok(note="密码已修改") return self._fail("未找到密码设置入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 8. 收藏管理 ═══════════════════ def add_to_favorites(self, content_desc: str = None) -> dict: try: if content_desc: xml = self._get_xml() pos = self._find_text(content_desc, xml) if pos: self._long_press(*pos) time.sleep(0.5) if self._click_text("收藏", timeout=3): return self._ok(note="已收藏") return self._fail("未找到收藏选项") except Exception as e: return self._fail(str(e)) def get_favorites(self, limit: int = 20) -> dict: try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("收藏", timeout=3) time.sleep(1) favorites = [] for _ in range(min(limit // 5, 10)): xml = self._get_xml() texts = re.findall(r'text="([^"]{2,})"', xml) skip = {"收藏", "返回", "搜索", "全部收藏", "我"} for t in texts: if t not in skip and {"content": t} not in favorites: favorites.append({"content": t}) if len(favorites) >= limit: break self._scroll_up() time.sleep(0.5) return self._ok(favorites=favorites[:limit], count=len(favorites[:limit])) except Exception as e: return self._fail(str(e)) # ═══════════════════ 9. 支付/红包 ═══════════════════ def send_red_packet(self, to_id: str, amount: str, message: str = "恭喜发财") -> dict: try: self._ensure_wechat() self._search_and_enter(to_id) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) self._click_text("红包", timeout=3) time.sleep(1) self._input(amount, clear=True) time.sleep(0.3) if message != "恭喜发财": pos = self._find_text("恭喜发财") if pos: self._tap(*pos) self._input(message, clear=True) self._click_text("塞钱进红包", timeout=3) or self._click_text("支付", timeout=2) time.sleep(1) return self._ok(note=f"红包已准备: {amount}元（需确认支付密码）") except Exception as e: return self._fail(str(e)) def transfer(self, to_id: str, amount: str, message: str = "") -> dict: try: self._ensure_wechat() self._search_and_enter(to_id) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) self._click_text("转账", timeout=3) time.sleep(1) self._input(amount, clear=True) time.sleep(0.3) if message: self._click_text("添加转账说明", timeout=2) self._input(message) self._click_text("转账", timeout=3) time.sleep(1) return self._ok(note=f"转账已准备: {amount}元（需确认支付密码）") except Exception as e: return self._fail(str(e)) # ═══════════════════ 10. 聊天设置 ═══════════════════ def set_chat_top(self, user_id: str, enable: bool = True) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) self._tap(self._w - 60, 80) time.sleep(1) self._click_text("置顶聊天" if enable else "取消置顶", timeout=3) or \ self._click_text("置顶", timeout=2) return self._ok(note=f"{'已置顶' if enable else '已取消置顶'}: {user_id}") except Exception as e: return self._fail(str(e)) def set_mute_chat(self, user_id: str, enable: bool = True) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) self._tap(self._w - 60, 80) time.sleep(1) self._click_text("消息免打扰", timeout=3) return self._ok(note=f"{'已开启' if enable else '已关闭'}免打扰: {user_id}") except Exception as e: return self._fail(str(e)) def clear_chat_history(self, user_id: str) -> dict: try: self._ensure_wechat() self._search_and_enter(user_id) self._tap(self._w - 60, 80) time.sleep(1) self._scroll_up() if self._click_text("清空聊天记录", timeout=3): time.sleep(0.5) self._click_text("确定") or self._click_text("清空") return self._ok(note=f"聊天记录已清空: {user_id}") return self._fail("未找到清空选项") except Exception as e: return self._fail(str(e)) # ═══════════════════ 11. 小程序 ═══════════════════ def open_mini_program(self, name: str) -> dict: try: self._ensure_wechat() self._go_home() self._swipe(self._w // 2, 300, self._w // 2, self._h // 2, 300) time.sleep(1) if self._click_text(name, timeout=3): time.sleep(2) return self._ok(note=f"小程序已打开: {name}") self._click_text("搜索", timeout=2) self._input(name) time.sleep(1) self._click_text(name, timeout=3) time.sleep(2) return self._ok(note=f"小程序: {name}") except Exception as e: return self._fail(str(e)) # ═══════════════════ 12. 公众号 ═══════════════════ def follow_official_account(self, account_name: str) -> dict: try: self._ensure_wechat() self._go_home() self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("添加朋友", timeout=3) time.sleep(0.5) self._click_text("公众号", timeout=3) time.sleep(0.5) self._input(account_name, clear=True) time.sleep(1) self._click_text("搜索", timeout=2) or self._enter() time.sleep(1.5) self._click_text(account_name, timeout=3) time.sleep(1) self._click_text("关注", timeout=3) return self._ok(note=f"已关注公众号: {account_name}") except Exception as e: return self._fail(str(e)) # ═══════════════════ 13. 解封完善 ═══════════════════ def _nav_to_safety_center(self) -> bool: """导航到微信安全中心（缓存路径优先）""" if self._navigate_cached("to_safety_center"): return True self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(1) self._scroll_up() time.sleep(0.5) if not self._click_text("设置", timeout=3): return False time.sleep(1.5) for _ in range(3): self._swipe(self._w // 2, 400, self._w // 2, 1600, 200) time.sleep(0.2) time.sleep(0.5) if not self._click_text("账号与安全", timeout=3): return False time.sleep(1.5) if not self._click_text("微信安全中心", timeout=5): return False time.sleep(2) return True def _nav_to_settings(self) -> bool: """导航到设置页面""" if self._navigate_cached("to_settings"): return True self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(1) self._scroll_up() time.sleep(0.5) if not self._click_text("设置", timeout=3): return False time.sleep(1.5) for _ in range(3): self._swipe(self._w // 2, 400, self._w // 2, 1600, 200) time.sleep(0.2) return True def _nav_to_account_security(self) -> bool: """导航到账号与安全""" if self._navigate_cached("to_account_security"): return True if not self._nav_to_settings(): return False if not self._click_text("账号与安全", timeout=3): return False time.sleep(1.5) return True def unblock_self(self) -> dict: """自助解封完整流程""" try: if not self._nav_to_safety_center(): return self._fail("无法导航到微信安全中心") self._record_page_elements("safety_center") if self._click_text("解封账号", timeout=3) or self._click_text("自助解封", timeout=2): time.sleep(1) self._click_text("下一步", timeout=3) or self._click_text("开始", timeout=2) time.sleep(1) return self._ok(note="自助解封流程已启动，请按页面提示完成验证") return self._ok(note="账号当前未被封禁，无需解封", status="normal") except Exception as e: return self._fail(str(e)) def unblock_appeal(self) -> dict: """申诉解封""" try: if not self._nav_to_safety_center(): return self._fail("无法导航到微信安全中心") if self._click_text("申诉辅助", timeout=3) or self._click_text("申诉", timeout=2): time.sleep(1) self._click_text("开始申诉", timeout=3) or self._click_text("下一步", timeout=2) time.sleep(1) return self._ok(note="申诉解封流程已启动，请按页面提示填写信息") return self._fail("未找到申诉入口") except Exception as e: return self._fail(str(e)) def check_restrictions(self) -> dict: """检查当前限制详情""" try: self._ensure_wechat() self._go_home() self._tap_tab("微信") time.sleep(0.5) xml = self._get_xml() restrictions = [] restriction_keywords = [ "账号暂时无法使用聊天功能", "功能受限", "账号异常", "暂时限制", "封禁", "被限制", ] for kw in restriction_keywords: if self._find_text(kw, xml): restrictions.append(kw) if restrictions: return self._ok(restricted=True, restrictions=restrictions) return self._ok(restricted=False, restrictions=[], note="未检测到功能限制") except Exception as e: return self._fail(str(e)) def appeal_restriction(self) -> dict: """申诉功能限制""" try: self._ensure_wechat() xml = self._get_xml() if self._click_text("了解详情", timeout=3) or self._click_text("查看详情", timeout=2): time.sleep(1) if self._click_text("申诉", timeout=3) or self._click_text("去申诉", timeout=2): time.sleep(1) return self._ok(note="已进入功能限制申诉页面") self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._click_text("账号与安全", timeout=3) time.sleep(0.5) self._scroll_up() self._click_text("微信安全中心", timeout=3) time.sleep(1) if self._click_text("解封账号", timeout=3) or self._click_text("申诉", timeout=2): return self._ok(note="已进入申诉页面") return self._fail("未找到申诉入口") except Exception as e: return self._fail(str(e)) def unblock_with_sms(self, phone: str) -> dict: """通过短信验证解封""" try: if not self._nav_to_safety_center(): return self._fail("无法导航到微信安全中心") self._click_text("解封账号", timeout=3) or self._click_text("自助解封", timeout=2) time.sleep(1) pos = self._find_text("手机号") if pos: self._tap(pos[0], pos[1] + 80) time.sleep(0.3) self._input(phone, clear=True) time.sleep(0.3) self._click_text("下一步", timeout=3) or self._click_text("发送验证码", timeout=2) time.sleep(1) return self._ok(note=f"短信验证解封已启动，验证码将发送至 {phone}") return self._fail("未找到手机号输入框") except Exception as e: return self._fail(str(e)) # ═══════════════════ 14. 视频号 ═══════════════════ def open_video_channel(self) -> dict: """打开视频号""" try: self._ensure_wechat() self._go_home() self._tap_tab("发现") time.sleep(0.5) if self._click_text("视频号", timeout=3): time.sleep(2) return self._ok(note="视频号已打开") return self._fail("未找到视频号入口") except Exception as e: return self._fail(str(e)) def like_video(self, index: int = 0) -> dict: """点赞视频""" try: self._ensure_wechat() result = self.open_video_channel() if not result.get("success"): return result for _ in range(index): self._scroll_up() time.sleep(1.5) xml = self._get_xml() heart_pos = self._find_text("❤", xml) or self._find_text("喜欢", xml) if heart_pos: self._tap(*heart_pos) time.sleep(0.3) return self._ok(note=f"已点赞第 {index+1} 个视频") self._tap(self._w - 80, self._h // 2) time.sleep(0.3) return self._ok(note=f"已尝试点赞第 {index+1} 个视频") except Exception as e: return self._fail(str(e)) def comment_video(self, index: int = 0, comment: str = "") -> dict: """评论视频""" try: self._ensure_wechat() result = self.open_video_channel() if not result.get("success"): return result for _ in range(index): self._scroll_up() time.sleep(1.5) xml = self._get_xml() comment_pos = self._find_text("评论", xml) or self._find_text("💬", xml) if comment_pos: self._tap(*comment_pos) else: self._tap(self._w - 80, self._h * 2 // 3) time.sleep(1) self._input(comment) time.sleep(0.3) self._click_text("发送", timeout=3) or self._enter() return self._ok(note=f"已评论第 {index+1} 个视频: {comment[:20]}") except Exception as e: return self._fail(str(e)) def follow_video_creator(self, index: int = 0) -> dict: """关注视频号创作者""" try: self._ensure_wechat() result = self.open_video_channel() if not result.get("success"): return result for _ in range(index): self._scroll_up() time.sleep(1.5) self._tap(80, self._h // 2) time.sleep(1) if self._click_text("关注", timeout=3): return self._ok(note=f"已关注第 {index+1} 个视频的创作者") return self._fail("未找到关注按钮") except Exception as e: return self._fail(str(e)) def share_video(self, index: int = 0, to_id: str = "") -> dict: """分享视频""" try: self._ensure_wechat() result = self.open_video_channel() if not result.get("success"): return result for _ in range(index): self._scroll_up() time.sleep(1.5) xml = self._get_xml() share_pos = self._find_text("转发", xml) or self._find_text("分享", xml) if share_pos: self._tap(*share_pos) else: self._tap(self._w - 80, self._h * 3 // 4) time.sleep(1) if to_id: self._click_text(to_id, timeout=3) or self._click_text("搜索", timeout=2) if not self._find_text(to_id): self._input(to_id, clear=True) time.sleep(1) self._click_text(to_id, timeout=3) time.sleep(0.5) self._click_text("发送", timeout=3) or self._click_text("分享", timeout=2) return self._ok(note=f"已分享第 {index+1} 个视频给 {to_id}") return self._ok(note="分享面板已打开") except Exception as e: return self._fail(str(e)) def get_video_list(self, limit: int = 10) -> dict: """获取视频列表""" try: self._ensure_wechat() result = self.open_video_channel() if not result.get("success"): return result videos = [] for i in range(min(limit, 20)): xml = self._get_xml() texts = re.findall(r'text="([^"]{3,})"', xml) skip = {"视频号", "返回", "搜索", "发现", "直播"} for t in texts: if t not in skip and {"title": t} not in videos: videos.append({"title": t, "index": i}) break if len(videos) >= limit: break self._scroll_up() time.sleep(1.5) return self._ok(videos=videos[:limit], count=len(videos[:limit])) except Exception as e: return self._fail(str(e)) # ═══════════════════ 15. 扫一扫 ═══════════════════ def scan_qr_code(self) -> dict: """打开扫一扫""" try: self._ensure_wechat() self._go_home() self._tap(self._w - 60, 80) time.sleep(0.5) if self._click_text("扫一扫", timeout=3): time.sleep(2) return self._ok(note="扫一扫已打开，等待扫描") return self._fail("未找到扫一扫入口") except Exception as e: return self._fail(str(e)) def scan_add_friend(self) -> dict: """扫码加好友""" try: result = self.scan_qr_code() if not result.get("success"): return result time.sleep(3) xml = self._get_xml() if self._find_text("添加到通讯录", xml): self._click_text("添加到通讯录", timeout=3) time.sleep(1) self._click_text("发送", timeout=3) return self._ok(note="已扫码并发送好友请求") if self._find_text("关注", xml): self._click_text("关注", timeout=3) return self._ok(note="已扫码关注") return self._ok(note="扫一扫已打开，请对准二维码") except Exception as e: return self._fail(str(e)) def show_my_qr(self) -> dict: """显示我的二维码""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._tap(self._w // 2, 200) time.sleep(1) if self._click_text("二维码", timeout=3) or self._click_text("我的二维码", timeout=2): time.sleep(1) return self._ok(note="我的二维码已显示") return self._fail("未找到二维码入口") except Exception as e: return self._fail(str(e)) def extract_qr_from_image(self) -> dict: """从图片识别二维码""" try: self._ensure_wechat() self._go_home() self._tap(self._w - 60, 80) time.sleep(0.5) self._click_text("扫一扫", timeout=3) time.sleep(1) if self._click_text("相册", timeout=3) or self._click_text("从相册选取", timeout=2): time.sleep(1) self._tap(self._w // 4, self._h // 3) time.sleep(2) xml = self._get_xml() texts = re.findall(r'text="([^"]+)"', xml) return self._ok(note="已从相册选取图片识别二维码", page_texts=texts[:10]) return self._fail("未找到相册入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 16. 支付完善 ═══════════════════ def show_payment_code(self) -> dict: """显示付款码""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("服务", timeout=3) or self._click_text("支付", timeout=2) time.sleep(1) if self._click_text("收付款", timeout=3): time.sleep(1) return self._ok(note="付款码已显示") return self._fail("未找到收付款入口") except Exception as e: return self._fail(str(e)) def receive_payment(self, amount: str = "", desc: str = "") -> dict: """收款码""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("服务", timeout=3) or self._click_text("支付", timeout=2) time.sleep(1) self._click_text("收付款", timeout=3) time.sleep(1) if self._click_text("二维码收款", timeout=3) or self._click_text("我要收款", timeout=2): time.sleep(1) if amount: self._click_text("设置金额", timeout=3) time.sleep(0.5) self._input(amount, clear=True) time.sleep(0.3) if desc: self._click_text("添加收款说明", timeout=2) self._input(desc) self._click_text("确定", timeout=3) time.sleep(0.5) return self._ok(note=f"收款码已生成{f'，金额: {amount}元' if amount else ''}") return self._fail("未找到收款入口") except Exception as e: return self._fail(str(e)) def view_wallet(self) -> dict: """查看钱包余额""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("服务", timeout=3) or self._click_text("支付", timeout=2) time.sleep(1) if self._click_text("钱包", timeout=3): time.sleep(1) xml = self._get_xml() wallet_info = {} texts = re.findall(r'text="([^"]+)"', xml) for t in texts: if "零钱" in t or "余额" in t: wallet_info["balance_label"] = t if re.match(r'^¥?[\d,.]+$', t): wallet_info["amount"] = t return self._ok(wallet=wallet_info, note="钱包信息已获取") return self._fail("未找到钱包入口") except Exception as e: return self._fail(str(e)) def view_transactions(self, limit: int = 20) -> dict: """查看账单""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("服务", timeout=3) or self._click_text("支付", timeout=2) time.sleep(1) self._click_text("钱包", timeout=3) time.sleep(1) if self._click_text("账单", timeout=3): time.sleep(1) transactions = [] for _ in range(min(limit // 5, 10)): xml = self._get_xml() texts = re.findall(r'text="([^"]{2,})"', xml) skip = {"账单", "返回", "搜索", "全部账单", "筛选"} for t in texts: if t not in skip and {"desc": t} not in transactions: transactions.append({"desc": t}) if len(transactions) >= limit: break self._scroll_up() time.sleep(0.5) return self._ok(transactions=transactions[:limit], count=len(transactions[:limit])) return self._fail("未找到账单入口") except Exception as e: return self._fail(str(e)) def receive_red_packet(self, from_id: str = "") -> dict: """领取红包""" try: self._ensure_wechat() if from_id: self._search_and_enter(from_id) time.sleep(0.5) xml = self._get_xml() rp_pos = self._find_text("微信红包", xml) or self._find_text("领取红包", xml) if rp_pos: self._tap(*rp_pos) time.sleep(1) if self._click_text("开", timeout=3) or self._click_text("领取", timeout=2): time.sleep(1) return self._ok(note="红包已领取") xml2 = self._get_xml() if self._find_text("已领取", xml2) or self._find_text("已被领完", xml2): return self._ok(note="红包已被领取过或已领完") return self._fail("红包领取失败") return self._fail("未找到可领取的红包") except Exception as e: return self._fail(str(e)) # ═══════════════════ 17. 通话 ═══════════════════ def voice_call(self, user_id: str) -> dict: """语音通话""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) if self._click_text("语音通话", timeout=3) or self._click_text("音视频通话", timeout=2): time.sleep(0.5) self._click_text("语音通话", timeout=2) time.sleep(2) return self._ok(note=f"正在呼叫 {user_id}（语音通话）") return self._fail("未找到语音通话入口") except Exception as e: return self._fail(str(e)) def video_call(self, user_id: str) -> dict: """视频通话""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) if self._click_text("视频通话", timeout=3) or self._click_text("音视频通话", timeout=2): time.sleep(0.5) self._click_text("视频通话", timeout=2) time.sleep(2) return self._ok(note=f"正在呼叫 {user_id}（视频通话）") return self._fail("未找到视频通话入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 18. 群发助手 ═══════════════════ def mass_send(self, content: str, user_ids: list = None) -> dict: """群发消息""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._click_text("通用", timeout=3) time.sleep(0.5) self._click_text("辅助功能", timeout=3) time.sleep(0.5) self._click_text("群发助手", timeout=3) time.sleep(1) if not self._click_text("启用该功能", timeout=2): pass self._click_text("新建群发", timeout=3) or self._click_text("开始群发", timeout=2) time.sleep(1) if user_ids: for uid in user_ids: pos = self._find_text(uid) if pos: self._tap(*pos) time.sleep(0.3) else: self._click_text("搜索", timeout=2) self._input(uid, clear=True) time.sleep(0.5) pos = self._find_text(uid) if pos: self._tap(*pos) time.sleep(0.3) self._back() time.sleep(0.3) else: self._click_text("全选", timeout=2) time.sleep(0.3) self._click_text("下一步", timeout=3) or self._click_text("确定", timeout=2) time.sleep(0.5) self._input(content) time.sleep(0.3) self._click_text("发送", timeout=3) time.sleep(1) self._click_text("确定", timeout=2) or self._click_text("发送", timeout=2) return self._ok(note=f"群发消息已发送，内容: {content[:30]}") except Exception as e: return self._fail(str(e)) # ═══════════════════ 19. 搜一搜/看一看 ═══════════════════ def wechat_search(self, keyword: str) -> dict: """搜一搜""" try: self._ensure_wechat() self._go_home() self._tap_tab("发现") time.sleep(0.5) if self._click_text("搜一搜", timeout=3): time.sleep(1) self._input(keyword, clear=True) time.sleep(0.5) self._click_text("搜索", timeout=3) or self._enter() time.sleep(2) xml = self._get_xml() results = [] texts = re.findall(r'text="([^"]{3,})"', xml) skip = {"搜一搜", "搜索", "返回", keyword, "全部", "朋友圈", "文章", "公众号", "小程序", "视频", "百科"} for t in texts: if t not in skip and len(t) > 2: results.append({"title": t}) return self._ok(results=results[:20], count=len(results[:20]), note=f"搜一搜 '{keyword}' 结果") return self._fail("未找到搜一搜入口") except Exception as e: return self._fail(str(e)) def top_stories(self) -> dict: """看一看""" try: self._ensure_wechat() self._go_home() self._tap_tab("发现") time.sleep(0.5) if self._click_text("看一看", timeout=3): time.sleep(2) xml = self._get_xml() stories = [] texts = re.findall(r'text="([^"]{5,})"', xml) skip = {"看一看", "返回", "发现", "精选", "朋友在看", "热点"} for t in texts: if t not in skip: stories.append({"title": t}) return self._ok(stories=stories[:20], count=len(stories[:20])) return self._fail("未找到看一看入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 20. 微信运动 ═══════════════════ def get_steps(self) -> dict: """获取步数""" try: self._ensure_wechat() self._go_home() self._search_and_enter("微信运动") time.sleep(1) xml = self._get_xml() steps_info = {} texts = re.findall(r'text="([^"]+)"', xml) for t in texts: if re.match(r'^\d{1,6}$', t) and int(t) > 0: steps_info.setdefault("steps", t) elif "步" in t: steps_info["steps_label"] = t return self._ok(steps=steps_info, note="步数信息已获取") except Exception as e: return self._fail(str(e)) def like_steps(self, user_id: str = "") -> dict: """点赞步数""" try: self._ensure_wechat() self._go_home() self._search_and_enter("微信运动") time.sleep(1) self._click_text("步数排行榜", timeout=3) or self._click_text("排行榜", timeout=2) time.sleep(1) if user_id: pos = self._find_text(user_id) if pos: self._tap(*pos) time.sleep(0.5) xml = self._get_xml() like_pos = self._find_text("❤", xml) or self._find_text("赞", xml) if like_pos: self._tap(*like_pos) return self._ok(note=f"已点赞{f' {user_id} 的' if user_id else ''}步数") return self._fail("未找到点赞按钮") except Exception as e: return self._fail(str(e)) # ═══════════════════ 21. 位置分享 ═══════════════════ def send_location(self, user_id: str) -> dict: """发送位置""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) if self._click_text("位置", timeout=3): time.sleep(0.5) self._click_text("发送位置", timeout=3) time.sleep(2) self._click_text("发送", timeout=3) return self._ok(note=f"已向 {user_id} 发送当前位置") return self._fail("未找到位置入口") except Exception as e: return self._fail(str(e)) def share_real_time_location(self, user_id: str) -> dict: """共享实时位置""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) if self._click_text("位置", timeout=3): time.sleep(0.5) if self._click_text("共享实时位置", timeout=3): time.sleep(2) return self._ok(note=f"已与 {user_id} 开始共享实时位置") return self._fail("未找到共享实时位置选项") return self._fail("未找到位置入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 22. 表情管理 ═══════════════════ def send_emoji(self, user_id: str, emoji_name: str = "") -> dict: """发送表情""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) xml = self._get_xml() emoji_btn = self._find_text("表情", xml) or self._find_text("😊", xml) if emoji_btn: self._tap(*emoji_btn) time.sleep(0.5) if emoji_name: pos = self._find_text(emoji_name) if pos: self._tap(*pos) time.sleep(0.3) self._click_text("发送", timeout=2) return self._ok(note=f"已发送表情 {emoji_name}") self._tap(self._w // 4, self._h * 3 // 4) time.sleep(0.3) self._click_text("发送", timeout=2) return self._ok(note="已发送表情") return self._fail("未找到表情入口") except Exception as e: return self._fail(str(e)) def get_sticker_list(self) -> dict: """获取表情包列表""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) if self._click_text("表情", timeout=3): time.sleep(1) xml = self._get_xml() stickers = [] texts = re.findall(r'text="([^"]{2,})"', xml) skip = {"表情", "返回", "搜索", "我的表情", "我"} for t in texts: if t not in skip: stickers.append({"name": t}) return self._ok(stickers=stickers, count=len(stickers)) return self._fail("未找到表情入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 23. 朋友圈增强 ═══════════════════ def set_moments_cover(self) -> dict: """设置朋友圈封面""" try: self._ensure_wechat() self._go_home() self._tap_tab("发现") time.sleep(0.5) self._click_text("朋友圈", timeout=3) time.sleep(1.5) self._tap(self._w // 2, 300) time.sleep(1) if self._click_text("换封面", timeout=3) or self._click_text("更换相册封面", timeout=2): time.sleep(0.5) self._click_text("从手机相册选择", timeout=3) or self._click_text("相册", timeout=2) time.sleep(1) self._tap(self._w // 4, self._h // 3) time.sleep(0.5) self._click_text("确定", timeout=3) or self._click_text("完成", timeout=2) time.sleep(1) return self._ok(note="朋友圈封面已更换") return self._fail("未找到更换封面入口") except Exception as e: return self._fail(str(e)) def set_moments_privacy(self, days: int = 180) -> dict: """设置朋友圈可见天数""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._click_text("隐私", timeout=3) time.sleep(0.5) if self._click_text("朋友圈", timeout=3) or self._click_text("允许朋友查看朋友圈的范围", timeout=2): time.sleep(0.5) days_map = {3: "最近三天", 30: "最近一个月", 180: "最近半年"} target = days_map.get(days, "全部") if self._click_text(target, timeout=3): time.sleep(0.3) self._click_text("确定", timeout=2) or self._back() return self._ok(note=f"朋友圈可见范围已设为: {target}") return self._fail(f"未找到 '{target}' 选项") return self._fail("未找到朋友圈隐私设置入口") except Exception as e: return self._fail(str(e)) def forward_moments_link(self, url: str, to_id: str = "") -> dict: """分享链接到朋友圈""" try: self._ensure_wechat() if to_id: self._search_and_enter(to_id) time.sleep(0.5) self._input(url) time.sleep(0.3) self._click_text("发送", timeout=3) or self._enter() time.sleep(1) xml = self._get_xml() link_pos = self._find_text(url[:20], xml) if link_pos: self._long_press(*link_pos) time.sleep(0.5) if self._click_text("分享到朋友圈", timeout=3): time.sleep(1) self._click_text("发表", timeout=3) or self._click_text("发送", timeout=2) return self._ok(note="链接已分享到朋友圈") self._go_home() self._tap_tab("发现") time.sleep(0.5) self._click_text("朋友圈", timeout=3) time.sleep(1.5) self._long_press(self._w - 60, 80, 2000) time.sleep(1) self._input(url) self._click_text("发表", timeout=5) or self._click_text("发送", timeout=3) return self._ok(note="链接已发布到朋友圈") except Exception as e: return self._fail(str(e)) # ═══════════════════ 24. 语音消息 ═══════════════════ def send_voice_message(self, user_id: str, duration: int = 3) -> dict: """发送语音消息（长按语音按钮）""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) if self._click_text("按住说话", timeout=2): time.sleep(0.3) else: voice_btn = self._find_text("语音") if voice_btn: self._tap(*voice_btn) time.sleep(0.3) else: self._tap(60, self._h - 120) time.sleep(0.3) hold_ms = max(duration * 1000, 1000) self._long_press(self._w // 2, self._h - 200, hold_ms) time.sleep(0.5) return self._ok(note=f"已向 {user_id} 发送 {duration}s 语音消息") except Exception as e: return self._fail(str(e)) # ═══════════════════ 25. 文件管理 ═══════════════════ def send_file_from_chat(self, user_id: str) -> dict: """在聊天中发送文件""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) self._click_text("+") or self._tap(self._w - 60, self._h - 180) time.sleep(0.5) if not self._click_text("文件", timeout=3): self._scroll_up() self._click_text("文件", timeout=3) time.sleep(1) self._tap(self._w // 2, self._h // 3) time.sleep(0.5) self._click_text("发送", timeout=3) or self._click_text("确定", timeout=2) return self._ok(note=f"已向 {user_id} 发送文件") except Exception as e: return self._fail(str(e)) def download_file(self, user_id: str) -> dict: """下载收到的文件""" try: self._ensure_wechat() self._search_and_enter(user_id) time.sleep(0.5) xml = self._get_xml() file_pos = self._find_text("[文件]", xml) if not file_pos: texts = re.findall( r'text="([^"]*\.(pdf|doc|docx|xls|xlsx|ppt|pptx|zip|rar|txt)[^"]*)"', xml ) if texts: file_pos = self._find_text(texts[0][0], xml) if file_pos: self._tap(*file_pos) time.sleep(2) if self._click_text("下载", timeout=3) or self._click_text("打开", timeout=2): time.sleep(2) return self._ok(note="文件已下载/打开") return self._ok(note="已点击文件，等待加载") return self._fail("未找到可下载的文件") except Exception as e: return self._fail(str(e)) # ═══════════════════ 26. 设置管理 ═══════════════════ def toggle_do_not_disturb(self, enable: bool = True) -> dict: """勿扰模式""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._click_text("新消息通知", timeout=3) or self._click_text("消息通知", timeout=2) time.sleep(0.5) if self._click_text("勿扰模式", timeout=3): time.sleep(0.5) xml = self._get_xml() current_on = self._find_text("开启", xml) is not None if enable != current_on: self._click_text("开启" if not current_on else "关闭", timeout=2) return self._ok(note=f"勿扰模式已{'开启' if enable else '关闭'}") return self._fail("未找到勿扰模式设置") except Exception as e: return self._fail(str(e)) def clear_cache(self) -> dict: """清理缓存""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._click_text("通用", timeout=3) time.sleep(0.5) if self._click_text("存储空间", timeout=3) or self._click_text("微信存储空间", timeout=2): time.sleep(2) if self._click_text("清理", timeout=5) or self._click_text("前往清理", timeout=3): time.sleep(1) self._click_text("全选", timeout=2) time.sleep(0.3) self._click_text("删除", timeout=3) or self._click_text("清理", timeout=2) time.sleep(0.5) self._click_text("确定", timeout=3) or self._click_text("确认", timeout=2) time.sleep(2) return self._ok(note="缓存清理完成") return self._ok(note="存储空间页面已打开") return self._fail("未找到存储空间入口") except Exception as e: return self._fail(str(e)) def check_for_update(self) -> dict: """检查更新""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._click_text("关于微信", timeout=3) time.sleep(0.5) if self._click_text("检查新版本", timeout=3) or self._click_text("版本更新", timeout=2): time.sleep(2) xml = self._get_xml() if self._find_text("已是最新版本", xml): return self._ok(note="当前已是最新版本") if self._find_text("发现新版本", xml) or self._find_text("立即更新", xml): return self._ok(note="发现新版本可更新", has_update=True) return self._ok(note="版本检查完成") return self._fail("未找到检查更新入口") except Exception as e: return self._fail(str(e)) def logout(self) -> dict: """退出登录""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._scroll_up() if self._click_text("退出登录", timeout=3) or self._click_text("退出", timeout=2): time.sleep(0.5) self._click_text("确定", timeout=3) or self._click_text("退出登录", timeout=2) time.sleep(2) return self._ok(note="已退出微信登录") return self._fail("未找到退出登录按钮") except Exception as e: return self._fail(str(e)) def switch_account(self) -> dict: """切换账号""" try: self._ensure_wechat() self._go_home() self._tap_tab("我") time.sleep(0.5) self._click_text("设置", timeout=3) time.sleep(0.5) self._scroll_up() if self._click_text("切换账号", timeout=3): time.sleep(1) return self._ok(note="已进入切换账号页面") if self._click_text("退出登录", timeout=3): time.sleep(0.5) self._click_text("确定", timeout=3) time.sleep(2) return self._ok(note="已退出登录，可切换账号") return self._fail("未找到切换账号入口") except Exception as e: return self._fail(str(e)) # ═══════════════════ 统一调度入口 ═══════════════════ def execute(self, action: str, params: dict) -> dict: """统一入口：根据 action 路由到对应方法""" router = { # 消息 "send_message": lambda p: self.send_message( p.get("to_id", ""), p.get("content", ""), p.get("msg_type", "text"), p.get("media_url"), p.get("at_list")), "get_messages": lambda p: self.get_messages( p.get("conversation_id"), p.get("limit", 20)), "forward_message": lambda p: self.forward_message( p.get("to_id", ""), p.get("content", "")), "recall_message": lambda p: self.recall_message(), "send_card": lambda p: self.send_card( p.get("to_id", ""), p.get("card_wxid", "")), # 好友 "add_friend": lambda p: self.add_friend( p.get("user_id", ""), p.get("message", ""), p.get("source")), "accept_friend": lambda p: self.accept_friend(p.get("user_id")), "set_remark": lambda p: self.set_remark( p.get("user_id", ""), p.get("remark", "")), "delete_friend": lambda p: self.delete_friend(p.get("user_id", "")), "get_contacts": lambda p: self.get_contacts(p.get("limit", 100)), "search_contact": lambda p: self.search_contact(p.get("keyword", "")), "batch_add_friend": lambda p: self.batch_add_friend( p.get("user_ids", []), p.get("message", ""), p.get("interval", 5)), "get_friend_info": lambda p: self.get_friend_info(p.get("user_id", "")), # 群聊 "create_group": lambda p: self.create_group( p.get("group_name", ""), p.get("member_ids", [])), "invite_to_group": lambda p: self.invite_to_group( p.get("group_id", ""), p.get("member_ids", [])), "remove_from_group": lambda p: self.remove_from_group( p.get("group_id", ""), p.get("member_ids", [])), "set_group_notice": lambda p: self.set_group_notice( p.get("group_id", ""), p.get("notice", "")), "set_group_name": lambda p: self.set_group_name( p.get("group_id", ""), p.get("group_name", "")), "send_group_message": lambda p: self.send_group_message( p.get("group_id", ""), p.get("content", ""), p.get("msg_type", "text"), p.get("media_url"), p.get("at_all", False), p.get("at_list")), "set_group_welcome": lambda p: self.set_group_welcome( p.get("group_id", ""), p.get("welcome_text", ""), p.get("welcome_image")), "get_groups": lambda p: self.get_groups(p.get("limit", 100)), "get_group_members": lambda p: self.get_group_members(p.get("group_id", "")), "quit_group": lambda p: self.quit_group(p.get("group_id", "")), # 标签 "create_tag": lambda p: self.create_tag(p.get("tag_name", "")), "delete_tag": lambda p: self.delete_tag(p.get("tag_name", "")), "get_tags": lambda p: self.get_tags(), "add_tag": lambda p: self.add_tag( p.get("user_id", ""), p.get("tags", [])), "remove_tag": lambda p: self.remove_tag( p.get("user_id", ""), p.get("tags", [])), "get_users_by_tag": lambda p: self.get_users_by_tag( p.get("tag_name", ""), p.get("limit", 100)), # 朋友圈 "post_moments": lambda p: self.post_moments( p.get("content", ""), p.get("images"), p.get("video_url"), p.get("location"), p.get("visible_list"), p.get("invisible_list")), "like_moments": lambda p: self.like_moments( p.get("user_id"), p.get("post_index", 0)), "comment_moments": lambda p: self.comment_moments( p.get("user_id"), p.get("post_index", 0), p.get("comment", ""), p.get("reply_to")), "get_moments": lambda p: self.get_moments( p.get("user_id"), p.get("limit", 10)), "delete_moments": lambda p: self.delete_moments(p.get("post_index", 0)), # 个人设置 "set_nickname": lambda p: self.set_nickname(p.get("nickname", "")), "set_signature": lambda p: self.set_signature(p.get("signature", "")), "set_avatar": lambda p: self.set_avatar(p.get("image_path")), "set_gender": lambda p: self.set_gender(p.get("gender", "")), "set_region": lambda p: self.set_region(p.get("region", "")), "get_profile": lambda p: self.get_profile(), # 账号安全 "check_account_status": lambda p: self.check_account_status(), "unblock_account": lambda p: self.unblock_account(p.get("helper_wxid")), "safety_center": lambda p: self.safety_center(), "change_password": lambda p: self.change_password( p.get("old_pwd", ""), p.get("new_pwd", "")), # 收藏 "add_to_favorites": lambda p: self.add_to_favorites(p.get("content_desc")), "get_favorites": lambda p: self.get_favorites(p.get("limit", 20)), # 支付 "send_red_packet": lambda p: self.send_red_packet( p.get("to_id", ""), p.get("amount", ""), p.get("message", "恭喜发财")), "transfer": lambda p: self.transfer( p.get("to_id", ""), p.get("amount", ""), p.get("message", "")), # 聊天设置 "set_chat_top": lambda p: self.set_chat_top( p.get("user_id", ""), p.get("enable", True)), "set_mute_chat": lambda p: self.set_mute_chat( p.get("user_id", ""), p.get("enable", True)), "clear_chat_history": lambda p: self.clear_chat_history(p.get("user_id", "")), # 小程序 "open_mini_program": lambda p: self.open_mini_program(p.get("name", "")), # 公众号 "follow_official_account": lambda p: self.follow_official_account( p.get("account_name", "")), # 解封完善 "unblock_self": lambda p: self.unblock_self(), "unblock_appeal": lambda p: self.unblock_appeal(), "check_restrictions": lambda p: self.check_restrictions(), "appeal_restriction": lambda p: self.appeal_restriction(), "unblock_with_sms": lambda p: self.unblock_with_sms(p.get("phone", "")), # 视频号 "open_video_channel": lambda p: self.open_video_channel(), "like_video": lambda p: self.like_video(p.get("index", 0)), "comment_video": lambda p: self.comment_video( p.get("index", 0), p.get("comment", "")), "follow_video_creator": lambda p: self.follow_video_creator(p.get("index", 0)), "share_video": lambda p: self.share_video( p.get("index", 0), p.get("to_id", "")), "get_video_list": lambda p: self.get_video_list(p.get("limit", 10)), # 扫一扫 "scan_qr_code": lambda p: self.scan_qr_code(), "scan_add_friend": lambda p: self.scan_add_friend(), "show_my_qr": lambda p: self.show_my_qr(), "extract_qr_from_image": lambda p: self.extract_qr_from_image(), # 支付完善 "show_payment_code": lambda p: self.show_payment_code(), "receive_payment": lambda p: self.receive_payment( p.get("amount", ""), p.get("desc", "")), "view_wallet": lambda p: self.view_wallet(), "view_transactions": lambda p: self.view_transactions(p.get("limit", 20)), "receive_red_packet": lambda p: self.receive_red_packet(p.get("from_id", "")), # 通话 "voice_call": lambda p: self.voice_call(p.get("user_id", "")), "video_call": lambda p: self.video_call(p.get("user_id", "")), # 群发助手 "mass_send": lambda p: self.mass_send( p.get("content", ""), p.get("user_ids")), # 搜一搜/看一看 "wechat_search": lambda p: self.wechat_search(p.get("keyword", "")), "top_stories": lambda p: self.top_stories(), # 微信运动 "get_steps": lambda p: self.get_steps(), "like_steps": lambda p: self.like_steps(p.get("user_id", "")), # 位置分享 "send_location": lambda p: self.send_location(p.get("user_id", "")), "share_real_time_location": lambda p: self.share_real_time_location( p.get("user_id", "")), # 表情管理 "send_emoji": lambda p: self.send_emoji( p.get("user_id", ""), p.get("emoji_name", "")), "get_sticker_list": lambda p: self.get_sticker_list(), # 朋友圈增强 "set_moments_cover": lambda p: self.set_moments_cover(), "set_moments_privacy": lambda p: self.set_moments_privacy(p.get("days", 180)), "forward_moments_link": lambda p: self.forward_moments_link( p.get("url", ""), p.get("to_id", "")), # 语音消息 "send_voice_message": lambda p: self.send_voice_message( p.get("user_id", ""), p.get("duration", 3)), # 文件管理 "send_file_from_chat": lambda p: self.send_file_from_chat(p.get("user_id", "")), "download_file": lambda p: self.download_file(p.get("user_id", "")), # 设置管理 "toggle_do_not_disturb": lambda p: self.toggle_do_not_disturb( p.get("enable", True)), "clear_cache": lambda p: self.clear_cache(), "check_for_update": lambda p: self.check_for_update(), "logout": lambda p: self.logout(), "switch_account": lambda p: self.switch_account(), # 截图 "screenshot": lambda p: self.d.screenshot(), } handler = router.get(action) if not handler: logger.warning(f"WeChatADBEngine: 未知 action={action}") return self._fail(f"不支持的操作: {action}") if self.op_log: self.op_log.start_operation(action, params) try: result = handler(params) success = isinstance(result, dict) and result.get("success", False) if self.op_log: self.op_log.end_operation( success, result, result.get("error") if not success else None) return result except Exception as e: if self.op_log: self.op_log.end_operation(False, error=str(e)) return self._fail(str(e))