1528 lines
52 KiB
Python
1528 lines
52 KiB
Python
"""
|
||
微信控制技能 - Agent端实现
|
||
|
||
功能模块：
|
||
1. 消息管理 - 发送/获取消息、批量发送
|
||
2. 好友管理 - 添加/通过好友、设置备注、删除好友
|
||
3. 群聊管理 - 创建群/邀请入群/群发消息/设置群欢迎语
|
||
4. 标签管理 - 添加/删除/查询标签
|
||
5. 朋友圈管理 - 发布/点赞/评论
|
||
|
||
@author 卡若
|
||
@version 3.0.0
|
||
"""
|
||
|
||
import time
|
||
import logging
|
||
import re
|
||
import sys
|
||
import os
|
||
from typing import Dict, Any, List, Optional
|
||
_agent_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
if _agent_dir not in sys.path:
|
||
sys.path.insert(0, _agent_dir)
|
||
from skills.base import BaseSkill
|
||
from error_handler import retry, with_error_handling, ErrorHandler
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 截屏 + AI 视觉（每次操作都先看屏再决定）
|
||
class _VisionFallback(Exception):
|
||
"""视觉流程失败，回退到规则流程"""
|
||
pass
|
||
|
||
|
||
def _vision_step(self, goal: str, step_hint: str = "") -> dict:
|
||
"""截屏 → AI 看屏 → 返回下一步动作"""
|
||
try:
|
||
from vision_helper import ask_vision
|
||
img = self.screenshot()
|
||
return ask_vision(img, goal, step_hint)
|
||
except Exception as e:
|
||
logger.warning(f"视觉调用失败: {e}")
|
||
return {"action": "error", "error": str(e)}
|
||
|
||
|
||
class WechatSkill(BaseSkill):
|
||
"""
|
||
微信控制技能
|
||
|
||
支持的操作：
|
||
- send_message: 发送消息
|
||
- get_messages: 获取消息列表
|
||
- get_contacts: 获取联系人列表
|
||
- add_friend: 添加好友
|
||
- accept_friend: 通过好友请求
|
||
- set_remark: 设置备注
|
||
- delete_friend: 删除好友
|
||
- create_group: 创建群聊
|
||
- invite_to_group: 邀请入群
|
||
- remove_from_group: 移出群聊
|
||
- send_group_message: 发送群消息
|
||
- set_group_notice: 设置群公告
|
||
- set_group_name: 设置群名
|
||
- set_group_welcome: 设置群欢迎语
|
||
- get_groups: 获取群列表
|
||
- get_group_members: 获取群成员
|
||
- add_tag: 添加标签
|
||
- remove_tag: 移除标签
|
||
- create_tag: 创建标签
|
||
- delete_tag: 删除标签
|
||
- get_tags: 获取标签列表
|
||
- get_users_by_tag: 根据标签获取好友
|
||
- post_moments: 发布朋友圈
|
||
- like_moments: 点赞朋友圈
|
||
- comment_moments: 评论朋友圈
|
||
- get_moments: 获取朋友圈
|
||
- batch_send_message: 批量发送消息
|
||
"""
|
||
|
||
PACKAGE = "com.tencent.mm"
|
||
NAME = "微信"
|
||
SEND_MESSAGE_TIMEOUT = 45
|
||
|
||
def send_message_with_vision(self, to_id: str, content: str, msg_type: str = "text") -> Dict[str, Any]:
|
||
"""
|
||
发送消息（以截屏+AI为主，每次操作都先看屏再决定）
|
||
"""
|
||
try:
|
||
start = time.time()
|
||
self.launch()
|
||
self.sleep(0.8)
|
||
goal = f"给 {to_id} 发消息：{content}"
|
||
max_steps = 15
|
||
for step in range(max_steps):
|
||
if time.time() - start > self.SEND_MESSAGE_TIMEOUT:
|
||
return {"success": False, "error": "timeout"}
|
||
hint = f"步骤 {step+1}/{max_steps}。"
|
||
act = _vision_step(self, goal, hint)
|
||
a = act.get("action", "")
|
||
if a == "done":
|
||
logger.info(f"微信消息发送成功(视觉): {to_id}")
|
||
return {"success": True, "message_id": f"wx_{int(time.time() * 1000)}"}
|
||
if a == "error":
|
||
logger.warning(f"视觉返回错误: {act.get('error')}，回退到规则流程")
|
||
raise _VisionFallback()
|
||
if a == "click":
|
||
x, y = int(act.get("x", 540)), int(act.get("y", 1200))
|
||
self.click(x, y)
|
||
self.sleep(0.3)
|
||
elif a == "input":
|
||
t = act.get("text", "")
|
||
if t:
|
||
self.input_text(t, clear=(step < 2))
|
||
self.sleep(0.2)
|
||
elif a == "back":
|
||
self.back()
|
||
self.sleep(0.3)
|
||
raise _VisionFallback()
|
||
except _VisionFallback:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"视觉发送失败: {e}")
|
||
raise _VisionFallback()
|
||
|
||
@with_error_handling("发送微信消息")
|
||
@retry(max_retries=2, delay=0.5)
|
||
def send_message(self, to_id: str, content: str, msg_type: str = "text") -> Dict[str, Any]:
|
||
"""
|
||
发送消息（先尝试视觉流程，失败则用规则流程）
|
||
"""
|
||
start = time.time()
|
||
try:
|
||
out = self.send_message_with_vision(to_id, content, msg_type)
|
||
if not out.get("success") and out.get("error") == "timeout":
|
||
return out
|
||
return out
|
||
except _VisionFallback:
|
||
pass
|
||
try:
|
||
if time.time() - start > self.SEND_MESSAGE_TIMEOUT:
|
||
return {"success": False, "error": "timeout"}
|
||
self.launch()
|
||
self.wait_for_app_ready(timeout=4)
|
||
if time.time() - start > self.SEND_MESSAGE_TIMEOUT:
|
||
return {"success": False, "error": "timeout"}
|
||
if not self.wait_for_ui(["微信", "通讯录", "搜索"], timeout=2):
|
||
self.back()
|
||
self.sleep(0.2)
|
||
if self.click_text("搜索", timeout=2) or self.click_desc("搜索"):
|
||
self.wait_for_ui(["请输入", "搜索"], timeout=2)
|
||
else:
|
||
self.d.xpath('//*[@content-desc="搜索"]').click_exists(timeout=2)
|
||
self.sleep(0.2)
|
||
if time.time() - start > self.SEND_MESSAGE_TIMEOUT:
|
||
return {"success": False, "error": "timeout"}
|
||
self.input_text(to_id)
|
||
self.wait_for_ui([to_id], timeout=3) or self.sleep(0.3)
|
||
if not self.click_text(to_id, timeout=2):
|
||
if not self.click_contains(to_id, timeout=2):
|
||
return {"success": False, "error": f"未找到联系人: {to_id}"}
|
||
if time.time() - start > self.SEND_MESSAGE_TIMEOUT:
|
||
return {"success": False, "error": "timeout"}
|
||
if not self.wait_for_ui(["发送", "按住 说话", "发消息"], timeout=3):
|
||
return {"success": False, "error": "未能进入聊天界面"}
|
||
input_clicked = False
|
||
for selector in ["发消息", "输入", "请输入"]:
|
||
if self.click_contains(selector, timeout=1):
|
||
input_clicked = True
|
||
break
|
||
if not input_clicked:
|
||
info = self.d.info
|
||
self.click(info['displayWidth'] // 2, info['displayHeight'] - 200)
|
||
self.sleep(0.15)
|
||
self.input_text(content, clear=False)
|
||
self.sleep(0.15)
|
||
if not self.click_text("发送", timeout=2):
|
||
self.d.xpath('//*[@text="发送"]').click_exists(timeout=2)
|
||
logger.info(f"微信消息发送成功: {to_id}")
|
||
self.say(f"已发消息给 {to_id}", data={"to_id": to_id, "content_preview": content[:30]})
|
||
return {"success": True, "message_id": f"wx_{int(time.time() * 1000)}"}
|
||
except Exception as e:
|
||
logger.error(f"微信发送失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
def get_messages(self, limit: int = 20, conversation_id: str = None) -> Dict[str, Any]:
|
||
"""
|
||
获取消息列表
|
||
|
||
Args:
|
||
limit: 消息数量限制
|
||
conversation_id: 会话ID（联系人名称），如果指定则获取该会话的消息
|
||
|
||
Returns:
|
||
消息列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
messages = []
|
||
|
||
# 如果指定了会话，先进入
|
||
if conversation_id:
|
||
# 使用搜索进入会话
|
||
if self.click_text("搜索") or self.click_desc("搜索"):
|
||
self.sleep(0.5)
|
||
self.input_text(conversation_id)
|
||
self.sleep(1.5)
|
||
|
||
# 点击搜索结果
|
||
if not self.click_text(conversation_id):
|
||
if not self.click_contains(conversation_id):
|
||
return {"success": False, "error": f"未找到会话: {conversation_id}", "messages": []}
|
||
|
||
self.sleep(1)
|
||
|
||
# 获取UI树分析消息
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
# 解析XML提取消息
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
|
||
# 查找消息元素（通常在聊天界面）
|
||
# 微信消息通常有特定的resource-id或class
|
||
for elem in root.iter():
|
||
# 查找消息文本
|
||
text = elem.get("text", "")
|
||
resource_id = elem.get("resource-id", "")
|
||
bounds = elem.get("bounds", "")
|
||
|
||
if text and len(text) > 0:
|
||
# 判断是否是消息内容（排除按钮、输入框等）
|
||
if "发送" not in text and "输入" not in text:
|
||
# 提取坐标判断位置（消息通常在屏幕中间）
|
||
if bounds:
|
||
try:
|
||
coords = bounds.replace("[", "").replace("]", "").split(",")
|
||
if len(coords) >= 4:
|
||
y = int(coords[1])
|
||
# 消息通常在屏幕上半部分
|
||
info = self.d.info
|
||
if y < info['displayHeight'] * 0.8:
|
||
messages.append({
|
||
"text": text,
|
||
"bounds": bounds,
|
||
"timestamp": int(time.time() * 1000)
|
||
})
|
||
except:
|
||
pass
|
||
|
||
# 限制数量
|
||
messages = messages[:limit]
|
||
|
||
except Exception as e:
|
||
logger.warning(f"解析UI树失败: {e}")
|
||
|
||
return {
|
||
"success": True,
|
||
"messages": messages,
|
||
"count": len(messages)
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取消息失败: {e}")
|
||
return {"success": False, "error": str(e), "messages": []}
|
||
|
||
def get_contacts(self, limit: int = 100) -> Dict[str, Any]:
|
||
"""
|
||
获取联系人列表
|
||
|
||
Args:
|
||
limit: 联系人数量限制
|
||
|
||
Returns:
|
||
联系人列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 点击通讯录
|
||
if not self.click_text("通讯录"):
|
||
return {"success": False, "error": "无法进入通讯录", "contacts": []}
|
||
|
||
self.sleep(1)
|
||
|
||
contacts = []
|
||
|
||
# 获取UI树
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
# 解析联系人
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
|
||
# 查找联系人元素
|
||
for elem in root.iter():
|
||
text = elem.get("text", "")
|
||
resource_id = elem.get("resource-id", "")
|
||
|
||
# 微信联系人通常有特定的resource-id或class
|
||
if text and len(text) > 0:
|
||
# 排除系统元素
|
||
if "通讯录" not in text and "搜索" not in text:
|
||
# 检查是否是联系人（通常有头像或特定布局）
|
||
parent = elem
|
||
for _ in range(3): # 向上查找3层
|
||
if parent is not None:
|
||
parent_id = parent.get("resource-id", "")
|
||
if "contact" in parent_id.lower() or "item" in parent_id.lower():
|
||
contacts.append({
|
||
"name": text,
|
||
"resource_id": resource_id
|
||
})
|
||
break
|
||
parent = parent.getparent() if hasattr(parent, 'getparent') else None
|
||
|
||
# 去重
|
||
seen = set()
|
||
unique_contacts = []
|
||
for contact in contacts:
|
||
name = contact["name"]
|
||
if name not in seen:
|
||
seen.add(name)
|
||
unique_contacts.append(contact)
|
||
|
||
contacts = unique_contacts[:limit]
|
||
|
||
except Exception as e:
|
||
logger.warning(f"解析联系人失败: {e}")
|
||
# 降级方案：使用搜索功能获取常用联系人
|
||
# 可以通过搜索历史记录获取
|
||
|
||
return {
|
||
"success": True,
|
||
"contacts": contacts,
|
||
"count": len(contacts)
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取联系人失败: {e}")
|
||
return {"success": False, "error": str(e), "contacts": []}
|
||
|
||
def add_friend(self, user_id: str, message: str = "") -> Dict[str, Any]:
|
||
"""添加好友"""
|
||
try:
|
||
self.launch()
|
||
self.sleep(2)
|
||
|
||
# 点击+号
|
||
self.click_desc("+") or self.click_text("+")
|
||
self.sleep(0.5)
|
||
|
||
# 添加朋友
|
||
self.click_text("添加朋友")
|
||
self.sleep(0.5)
|
||
|
||
# 输入微信号
|
||
self.input_text(user_id)
|
||
self.sleep(1)
|
||
|
||
# 搜索
|
||
self.click_text("搜索") or self.click_contains("搜索")
|
||
self.sleep(2)
|
||
|
||
# 添加
|
||
if self.click_text("添加到通讯录"):
|
||
if message:
|
||
self.sleep(0.5)
|
||
self.input_text(message)
|
||
self.click_text("发送")
|
||
return {"success": True}
|
||
|
||
return {"success": False, "error": "未找到用户或已是好友"}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def accept_friend(self, user_id: str = None) -> Dict[str, Any]:
|
||
"""通过好友请求"""
|
||
try:
|
||
self.launch()
|
||
self.sleep(2)
|
||
|
||
# 通讯录 -> 新的朋友
|
||
self.click_text("通讯录")
|
||
self.sleep(1)
|
||
self.click_text("新的朋友")
|
||
self.sleep(1)
|
||
|
||
# 找到并通过
|
||
if user_id:
|
||
if self.exists(user_id):
|
||
# 找到特定用户的接受按钮
|
||
pass
|
||
|
||
# 通过第一个
|
||
if self.click_text("接受"):
|
||
return {"success": True}
|
||
|
||
return {"success": False, "error": "无待处理的好友请求"}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# ==========================================================================
|
||
# 三、群聊管理
|
||
# ==========================================================================
|
||
|
||
def create_group(self, group_name: str, member_ids: List[str]) -> Dict[str, Any]:
|
||
"""
|
||
创建群聊
|
||
|
||
Args:
|
||
group_name: 群名称
|
||
member_ids: 成员名称列表（至少2人）
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 点击右上角+号
|
||
self.click_desc("+") or self.click_text("+")
|
||
self.sleep(0.5)
|
||
|
||
# 发起群聊
|
||
self.click_text("发起群聊")
|
||
self.sleep(1)
|
||
|
||
# 选择群成员
|
||
for member_id in member_ids:
|
||
# 搜索成员
|
||
if self.click_text("搜索") or self.click_desc("搜索"):
|
||
self.sleep(0.3)
|
||
self.input_text(member_id)
|
||
self.sleep(1)
|
||
|
||
# 选中成员
|
||
self.click_text(member_id) or self.click_contains(member_id)
|
||
self.sleep(0.5)
|
||
|
||
# 清空搜索框
|
||
self.back()
|
||
self.sleep(0.3)
|
||
|
||
# 确定创建
|
||
self.click_text("完成") or self.click_text("确定")
|
||
self.sleep(2)
|
||
|
||
# 修改群名
|
||
if group_name:
|
||
# 点击群名区域
|
||
self.click_text("群聊") or self.click_contains("的群聊")
|
||
self.sleep(0.5)
|
||
self.input_text(group_name, clear=True)
|
||
self.click_text("完成") or self.click_text("确定")
|
||
|
||
logger.info(f"创建群聊成功: {group_name}")
|
||
return {
|
||
"success": True,
|
||
"group_name": group_name,
|
||
"member_count": len(member_ids) + 1
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"创建群聊失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def invite_to_group(self, group_id: str, member_ids: List[str]) -> Dict[str, Any]:
|
||
"""
|
||
邀请入群
|
||
|
||
Args:
|
||
group_id: 群名称
|
||
member_ids: 要邀请的成员名称列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 搜索进入群聊
|
||
if self.click_text("搜索") or self.click_desc("搜索"):
|
||
self.sleep(0.5)
|
||
self.input_text(group_id)
|
||
self.sleep(1.5)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
# 点击右上角群设置
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3) or \
|
||
self.click_text("...")
|
||
self.sleep(1)
|
||
|
||
# 点击+号添加成员
|
||
self.click_desc("+") or self.click_text("+")
|
||
self.sleep(1)
|
||
|
||
# 选择成员
|
||
invited_count = 0
|
||
for member_id in member_ids:
|
||
if self.click_text("搜索") or self.click_desc("搜索"):
|
||
self.sleep(0.3)
|
||
self.input_text(member_id)
|
||
self.sleep(1)
|
||
|
||
if self.click_text(member_id) or self.click_contains(member_id):
|
||
invited_count += 1
|
||
|
||
self.back()
|
||
self.sleep(0.3)
|
||
|
||
# 确定
|
||
self.click_text("完成") or self.click_text("确定")
|
||
|
||
return {
|
||
"success": True,
|
||
"invited_count": invited_count
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def remove_from_group(self, group_id: str, member_ids: List[str]) -> Dict[str, Any]:
|
||
"""移出群聊"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入群聊
|
||
self.search(group_id)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
# 进入群设置
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 点击-号移除成员
|
||
self.click_desc("-")
|
||
self.sleep(1)
|
||
|
||
removed_count = 0
|
||
for member_id in member_ids:
|
||
if self.click_text(member_id):
|
||
removed_count += 1
|
||
|
||
self.click_text("完成") or self.click_text("删除")
|
||
|
||
return {"success": True, "removed_count": removed_count}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def send_group_message(
|
||
self,
|
||
group_id: str,
|
||
content: str,
|
||
msg_type: str = "text",
|
||
at_all: bool = False,
|
||
at_list: List[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
发送群消息
|
||
|
||
Args:
|
||
group_id: 群名称
|
||
content: 消息内容
|
||
msg_type: 消息类型
|
||
at_all: 是否@所有人
|
||
at_list: @的成员列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 搜索进入群聊
|
||
self.search(group_id)
|
||
self.sleep(1)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
# 处理@
|
||
if at_all:
|
||
content = "@所有人 " + content
|
||
elif at_list:
|
||
at_str = " ".join([f"@{name}" for name in at_list])
|
||
content = at_str + " " + content
|
||
|
||
# 输入消息
|
||
self.input_text(content, clear=False)
|
||
self.sleep(0.3)
|
||
|
||
# 发送
|
||
self.click_text("发送")
|
||
|
||
return {
|
||
"success": True,
|
||
"message_id": f"wx_group_{int(time.time() * 1000)}"
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def set_group_notice(self, group_id: str, notice: str) -> Dict[str, Any]:
|
||
"""设置群公告"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入群聊
|
||
self.search(group_id)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
# 进入群设置
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 群公告
|
||
self.click_text("群公告")
|
||
self.sleep(1)
|
||
|
||
# 输入公告
|
||
self.input_text(notice, clear=True)
|
||
|
||
# 发布
|
||
self.click_text("完成") or self.click_text("发布")
|
||
|
||
return {"success": True}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def set_group_name(self, group_id: str, group_name: str) -> Dict[str, Any]:
|
||
"""设置群名"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入群聊
|
||
self.search(group_id)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
# 进入群设置
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 点击群名
|
||
self.click_text("群聊名称") or self.click_contains("群名")
|
||
self.sleep(0.5)
|
||
|
||
# 输入新群名
|
||
self.input_text(group_name, clear=True)
|
||
|
||
# 确定
|
||
self.click_text("完成") or self.click_text("确定")
|
||
|
||
return {"success": True}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def set_group_welcome(self, group_id: str, welcome_text: str, welcome_image: str = None) -> Dict[str, Any]:
|
||
"""设置群欢迎语（需要群主/管理员权限）"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入群聊设置
|
||
self.search(group_id)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 群管理
|
||
if not self.click_text("群管理"):
|
||
return {"success": False, "error": "未找到群管理选项，可能不是群主/管理员"}
|
||
self.sleep(1)
|
||
|
||
# 入群欢迎语
|
||
self.click_text("入群欢迎语") or self.click_contains("欢迎语")
|
||
self.sleep(1)
|
||
|
||
# 输入欢迎语
|
||
self.input_text(welcome_text, clear=True)
|
||
|
||
# 保存
|
||
self.click_text("完成") or self.click_text("保存")
|
||
|
||
return {"success": True}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_groups(self, limit: int = 100) -> Dict[str, Any]:
|
||
"""获取群聊列表"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 通讯录 -> 群聊
|
||
self.click_text("通讯录")
|
||
self.sleep(1)
|
||
|
||
self.click_text("群聊")
|
||
self.sleep(1)
|
||
|
||
# 获取UI树解析群列表
|
||
groups = []
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
# 解析群名（简化实现）
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
for elem in root.iter():
|
||
text = elem.get("text", "")
|
||
if text and "群聊" not in text and "通讯录" not in text:
|
||
# 可能是群名
|
||
groups.append({"name": text})
|
||
except:
|
||
pass
|
||
|
||
return {
|
||
"success": True,
|
||
"groups": groups[:limit],
|
||
"count": len(groups[:limit])
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e), "groups": []}
|
||
|
||
def get_group_members(self, group_id: str) -> Dict[str, Any]:
|
||
"""获取群成员列表"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入群聊
|
||
self.search(group_id)
|
||
self.click_text(group_id) or self.click_contains(group_id)
|
||
self.sleep(1)
|
||
|
||
# 进入群设置
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 点击查看全部成员
|
||
self.click_contains("查看全部") or self.click_text("全部群成员")
|
||
self.sleep(1)
|
||
|
||
# 获取成员列表
|
||
members = []
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
for elem in root.iter():
|
||
text = elem.get("text", "")
|
||
if text and len(text) > 0:
|
||
if "群成员" not in text and "搜索" not in text:
|
||
members.append({"name": text})
|
||
except:
|
||
pass
|
||
|
||
return {
|
||
"success": True,
|
||
"members": members,
|
||
"count": len(members)
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e), "members": []}
|
||
|
||
# ==========================================================================
|
||
# 四、标签管理
|
||
# ==========================================================================
|
||
|
||
def add_tag(self, user_id: str, tags: List[str]) -> Dict[str, Any]:
|
||
"""
|
||
给好友添加标签
|
||
|
||
Args:
|
||
user_id: 好友名称
|
||
tags: 标签列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 搜索好友
|
||
self.search(user_id)
|
||
self.click_text(user_id) or self.click_contains(user_id)
|
||
self.sleep(1)
|
||
|
||
# 进入聊天界面后点击头像进入详情
|
||
# 或者直接在搜索结果点击头像
|
||
|
||
# 点击更多设置
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3) or \
|
||
self.click_text("...")
|
||
self.sleep(1)
|
||
|
||
# 设置标签
|
||
self.click_text("设置标签") or self.click_contains("标签")
|
||
self.sleep(1)
|
||
|
||
# 选择/创建标签
|
||
for tag in tags:
|
||
if self.exists(tag):
|
||
self.click_text(tag)
|
||
else:
|
||
# 创建新标签
|
||
self.click_text("添加标签") or self.click_text("新建标签")
|
||
self.sleep(0.5)
|
||
self.input_text(tag)
|
||
self.click_text("完成") or self.click_text("确定")
|
||
self.sleep(0.3)
|
||
|
||
# 保存
|
||
self.click_text("完成") or self.click_text("保存")
|
||
|
||
return {"success": True, "tags_added": tags}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def remove_tag(self, user_id: str, tags: List[str]) -> Dict[str, Any]:
|
||
"""移除好友标签"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入好友详情
|
||
self.search(user_id)
|
||
self.click_text(user_id) or self.click_contains(user_id)
|
||
self.sleep(1)
|
||
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 设置标签
|
||
self.click_text("设置标签") or self.click_contains("标签")
|
||
self.sleep(1)
|
||
|
||
# 取消选择标签
|
||
for tag in tags:
|
||
if self.exists(tag):
|
||
self.click_text(tag) # 再次点击取消选择
|
||
self.sleep(0.3)
|
||
|
||
# 保存
|
||
self.click_text("完成") or self.click_text("保存")
|
||
|
||
return {"success": True, "tags_removed": tags}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def create_tag(self, tag_name: str) -> Dict[str, Any]:
|
||
"""创建标签"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 通讯录 -> 标签
|
||
self.click_text("通讯录")
|
||
self.sleep(1)
|
||
|
||
self.click_text("标签")
|
||
self.sleep(1)
|
||
|
||
# 新建标签
|
||
self.click_text("新建") or self.click_text("添加标签")
|
||
self.sleep(0.5)
|
||
|
||
self.input_text(tag_name)
|
||
self.click_text("完成") or self.click_text("保存")
|
||
|
||
return {"success": True, "tag_name": tag_name}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def delete_tag(self, tag_name: str) -> Dict[str, Any]:
|
||
"""删除标签"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 通讯录 -> 标签
|
||
self.click_text("通讯录")
|
||
self.sleep(1)
|
||
|
||
self.click_text("标签")
|
||
self.sleep(1)
|
||
|
||
# 找到并长按标签
|
||
if self.exists(tag_name):
|
||
# 长按删除
|
||
element = self.d(text=tag_name)
|
||
if element.exists:
|
||
element.long_click()
|
||
self.sleep(0.5)
|
||
self.click_text("删除")
|
||
self.click_text("确定")
|
||
return {"success": True}
|
||
|
||
return {"success": False, "error": f"未找到标签: {tag_name}"}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_tags(self) -> Dict[str, Any]:
|
||
"""获取标签列表"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 通讯录 -> 标签
|
||
self.click_text("通讯录")
|
||
self.sleep(1)
|
||
|
||
self.click_text("标签")
|
||
self.sleep(1)
|
||
|
||
# 解析标签列表
|
||
tags = []
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
for elem in root.iter():
|
||
text = elem.get("text", "")
|
||
if text and "标签" not in text and "新建" not in text:
|
||
# 提取标签名（格式可能是"标签名(数量)"）
|
||
match = re.match(r'^(.+?)(?:\(\d+\))?$', text)
|
||
if match:
|
||
tags.append(match.group(1))
|
||
except:
|
||
pass
|
||
|
||
return {"success": True, "tags": tags}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e), "tags": []}
|
||
|
||
def get_users_by_tag(self, tag_name: str, limit: int = 100) -> Dict[str, Any]:
|
||
"""根据标签获取好友列表"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 通讯录 -> 标签
|
||
self.click_text("通讯录")
|
||
self.sleep(1)
|
||
|
||
self.click_text("标签")
|
||
self.sleep(1)
|
||
|
||
# 点击标签
|
||
self.click_text(tag_name) or self.click_contains(tag_name)
|
||
self.sleep(1)
|
||
|
||
# 解析好友列表
|
||
users = []
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
for elem in root.iter():
|
||
text = elem.get("text", "")
|
||
if text and len(text) > 0:
|
||
if tag_name not in text and "人" not in text:
|
||
users.append({"name": text})
|
||
except:
|
||
pass
|
||
|
||
return {
|
||
"success": True,
|
||
"users": users[:limit],
|
||
"count": len(users[:limit])
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e), "users": []}
|
||
|
||
# ==========================================================================
|
||
# 五、朋友圈管理
|
||
# ==========================================================================
|
||
|
||
def post_moments(
|
||
self,
|
||
content: str,
|
||
images: List[str] = None,
|
||
video_url: str = None,
|
||
location: str = None,
|
||
visible_list: List[str] = None,
|
||
invisible_list: List[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
发布朋友圈
|
||
|
||
Args:
|
||
content: 文字内容
|
||
images: 图片URL列表（需要先下载到本地）
|
||
video_url: 视频URL
|
||
location: 位置
|
||
visible_list: 可见名单
|
||
invisible_list: 不可见名单
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 发现 -> 朋友圈
|
||
self.click_text("发现")
|
||
self.sleep(1)
|
||
self.click_text("朋友圈")
|
||
self.sleep(2)
|
||
|
||
# 点击相机图标发布
|
||
camera = self.d.xpath('//*[@content-desc="拍照分享"]')
|
||
if camera.exists:
|
||
if images or video_url:
|
||
# 短按：选择图片/视频
|
||
camera.click()
|
||
else:
|
||
# 长按：纯文字
|
||
camera.long_click()
|
||
self.sleep(1)
|
||
|
||
if images:
|
||
# 从相册选择
|
||
self.click_text("从相册选择")
|
||
self.sleep(1)
|
||
# TODO: 选择图片（需要图片已在相册中）
|
||
# 这里简化处理，假设选择第一张
|
||
self.click_text("完成") or self.click_text("发送")
|
||
|
||
self.sleep(1)
|
||
|
||
# 输入文字内容
|
||
if self.click_contains("这一刻的想法") or self.click_contains("说点什么"):
|
||
self.input_text(content)
|
||
else:
|
||
self.input_text(content)
|
||
|
||
# 设置位置
|
||
if location:
|
||
self.click_text("所在位置")
|
||
self.sleep(1)
|
||
self.input_text(location)
|
||
self.sleep(1)
|
||
self.click_text(location) or self.click_contains(location)
|
||
|
||
# 设置可见范围
|
||
if visible_list or invisible_list:
|
||
self.click_text("谁可以看")
|
||
self.sleep(0.5)
|
||
if visible_list:
|
||
self.click_text("部分可见")
|
||
# TODO: 选择可见的人
|
||
elif invisible_list:
|
||
self.click_text("不给谁看")
|
||
# TODO: 选择不可见的人
|
||
self.click_text("完成")
|
||
|
||
# 发表
|
||
self.click_text("发表")
|
||
self.sleep(2)
|
||
|
||
return {
|
||
"success": True,
|
||
"post_id": f"moments_{int(time.time() * 1000)}"
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"发布朋友圈失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def like_moments(self, user_id: str, post_index: int = 0) -> Dict[str, Any]:
|
||
"""
|
||
点赞朋友圈
|
||
|
||
Args:
|
||
user_id: 好友名称
|
||
post_index: 第几条朋友圈（0表示最新一条）
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入好友朋友圈
|
||
self._enter_user_moments(user_id)
|
||
|
||
# 滚动到指定位置
|
||
for _ in range(post_index):
|
||
self.swipe("up", scale=0.3)
|
||
self.sleep(0.5)
|
||
|
||
# 点击评论按钮
|
||
self.click_desc("评论") or self.d.xpath('//*[@content-desc="评论"]').click()
|
||
self.sleep(0.5)
|
||
|
||
# 点击赞
|
||
self.click_text("赞")
|
||
|
||
return {"success": True}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def comment_moments(
|
||
self,
|
||
user_id: str,
|
||
comment: str,
|
||
post_index: int = 0,
|
||
reply_to: str = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
评论朋友圈
|
||
|
||
Args:
|
||
user_id: 好友名称
|
||
comment: 评论内容
|
||
post_index: 第几条朋友圈
|
||
reply_to: 回复某人
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入好友朋友圈
|
||
self._enter_user_moments(user_id)
|
||
|
||
# 滚动到指定位置
|
||
for _ in range(post_index):
|
||
self.swipe("up", scale=0.3)
|
||
self.sleep(0.5)
|
||
|
||
# 点击评论按钮
|
||
self.click_desc("评论") or self.d.xpath('//*[@content-desc="评论"]').click()
|
||
self.sleep(0.5)
|
||
|
||
# 点击评论
|
||
self.click_text("评论")
|
||
self.sleep(0.3)
|
||
|
||
# 如果要回复某人
|
||
if reply_to:
|
||
# 找到该人的评论并点击
|
||
self.click_text(reply_to)
|
||
self.sleep(0.3)
|
||
|
||
# 输入评论
|
||
self.input_text(comment)
|
||
|
||
# 发送
|
||
self.click_text("发送")
|
||
|
||
return {"success": True}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_moments(self, user_id: str = None, limit: int = 10) -> Dict[str, Any]:
|
||
"""
|
||
获取朋友圈列表
|
||
|
||
Args:
|
||
user_id: 好友名称，不指定则获取自己的朋友圈
|
||
limit: 获取数量
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
if user_id:
|
||
self._enter_user_moments(user_id)
|
||
else:
|
||
# 进入自己的朋友圈
|
||
self.click_text("发现")
|
||
self.sleep(1)
|
||
self.click_text("朋友圈")
|
||
self.sleep(2)
|
||
# 点击自己的头像
|
||
# TODO: 进入个人相册
|
||
|
||
# 获取朋友圈内容
|
||
moments = []
|
||
ui_tree = self.d.dump_hierarchy()
|
||
|
||
# 简化解析
|
||
import xml.etree.ElementTree as ET
|
||
try:
|
||
root = ET.fromstring(ui_tree)
|
||
for elem in root.iter():
|
||
text = elem.get("text", "")
|
||
if text and len(text) > 10: # 可能是朋友圈内容
|
||
moments.append({"content": text[:100]})
|
||
except:
|
||
pass
|
||
|
||
return {
|
||
"success": True,
|
||
"moments": moments[:limit],
|
||
"count": len(moments[:limit])
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e), "moments": []}
|
||
|
||
def _enter_user_moments(self, user_id: str):
|
||
"""进入好友的朋友圈（内部方法）"""
|
||
# 搜索好友
|
||
self.search(user_id)
|
||
self.click_text(user_id) or self.click_contains(user_id)
|
||
self.sleep(1)
|
||
|
||
# 进入好友详情
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 点击头像进入详情页
|
||
# 然后点击朋友圈
|
||
self.click_text("朋友圈") or self.click_contains("个人相册")
|
||
self.sleep(2)
|
||
|
||
# ==========================================================================
|
||
# 六、批量操作
|
||
# ==========================================================================
|
||
|
||
def batch_send_message(
|
||
self,
|
||
to_ids: List[str],
|
||
content: str,
|
||
msg_type: str = "text",
|
||
interval: float = 2.0
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
批量发送消息
|
||
|
||
Args:
|
||
to_ids: 接收者ID列表
|
||
content: 消息内容
|
||
msg_type: 消息类型
|
||
interval: 发送间隔(秒)
|
||
|
||
Returns:
|
||
批量发送结果
|
||
"""
|
||
results = []
|
||
success_count = 0
|
||
failed_count = 0
|
||
|
||
for to_id in to_ids:
|
||
try:
|
||
result = self.send_message(to_id, content, msg_type)
|
||
|
||
if result.get("success"):
|
||
success_count += 1
|
||
else:
|
||
failed_count += 1
|
||
|
||
results.append({
|
||
"to_id": to_id,
|
||
"success": result.get("success", False),
|
||
"error": result.get("error"),
|
||
"message_id": result.get("message_id")
|
||
})
|
||
|
||
# 发送间隔
|
||
time.sleep(interval)
|
||
|
||
except Exception as e:
|
||
failed_count += 1
|
||
results.append({
|
||
"to_id": to_id,
|
||
"success": False,
|
||
"error": str(e)
|
||
})
|
||
|
||
return {
|
||
"success": True,
|
||
"success_count": success_count,
|
||
"failed_count": failed_count,
|
||
"results": results
|
||
}
|
||
|
||
def batch_add_friend(
|
||
self,
|
||
user_ids: List[str],
|
||
message: str = "",
|
||
interval: float = 5.0
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
批量添加好友
|
||
|
||
Args:
|
||
user_ids: 用户ID列表
|
||
message: 验证消息
|
||
interval: 添加间隔(秒)
|
||
"""
|
||
results = []
|
||
success_count = 0
|
||
failed_count = 0
|
||
|
||
for user_id in user_ids:
|
||
try:
|
||
result = self.add_friend(user_id, message)
|
||
|
||
if result.get("success"):
|
||
success_count += 1
|
||
else:
|
||
failed_count += 1
|
||
|
||
results.append({
|
||
"user_id": user_id,
|
||
"success": result.get("success", False),
|
||
"error": result.get("error")
|
||
})
|
||
|
||
time.sleep(interval)
|
||
|
||
except Exception as e:
|
||
failed_count += 1
|
||
results.append({
|
||
"user_id": user_id,
|
||
"success": False,
|
||
"error": str(e)
|
||
})
|
||
|
||
return {
|
||
"success": True,
|
||
"success_count": success_count,
|
||
"failed_count": failed_count,
|
||
"results": results
|
||
}
|
||
|
||
# ==========================================================================
|
||
# 七、好友管理扩展
|
||
# ==========================================================================
|
||
|
||
def set_remark(self, user_id: str, remark: str) -> Dict[str, Any]:
|
||
"""
|
||
设置好友备注
|
||
|
||
Args:
|
||
user_id: 好友名称
|
||
remark: 新备注
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入好友详情
|
||
self.search(user_id)
|
||
self.click_text(user_id) or self.click_contains(user_id)
|
||
self.sleep(1)
|
||
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 设置备注和标签
|
||
self.click_text("设置备注和标签") or self.click_contains("备注")
|
||
self.sleep(0.5)
|
||
|
||
# 输入备注
|
||
self.input_text(remark, clear=True)
|
||
|
||
# 保存
|
||
self.click_text("完成") or self.click_text("保存")
|
||
|
||
return {"success": True, "remark": remark}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def delete_friend(self, user_id: str) -> Dict[str, Any]:
|
||
"""
|
||
删除好友
|
||
|
||
Args:
|
||
user_id: 好友名称
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 进入好友详情
|
||
self.search(user_id)
|
||
self.click_text(user_id) or self.click_contains(user_id)
|
||
self.sleep(1)
|
||
|
||
self.d.xpath('//*[@content-desc="聊天信息"]').click_exists(timeout=3)
|
||
self.sleep(1)
|
||
|
||
# 点击头像进入资料页
|
||
# TODO: 点击头像
|
||
|
||
# 更多 -> 删除
|
||
self.click_text("...") or self.click_desc("更多")
|
||
self.sleep(0.5)
|
||
|
||
self.click_text("删除")
|
||
self.sleep(0.3)
|
||
|
||
# 确认删除
|
||
self.click_text("删除") or self.click_text("确定")
|
||
|
||
return {"success": True}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# ==========================================================================
|
||
# 八、辅助方法
|
||
# ==========================================================================
|
||
|
||
def send_message_with_search(self, to_id: str, content: str) -> Dict[str, Any]:
|
||
"""
|
||
通过搜索发送消息（更可靠）
|
||
|
||
Args:
|
||
to_id: 联系人名称/备注
|
||
content: 消息内容
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready()
|
||
|
||
# 使用搜索功能
|
||
search_result = self.search(to_id)
|
||
if not search_result.get("success"):
|
||
return {"success": False, "error": "搜索联系人失败"}
|
||
|
||
self.sleep(1)
|
||
|
||
# 点击第一个搜索结果
|
||
if not self.click_text(to_id):
|
||
if not self.click_contains(to_id):
|
||
return {"success": False, "error": f"未找到联系人: {to_id}"}
|
||
|
||
self.sleep(1)
|
||
|
||
# 确保在聊天界面
|
||
if not self.exists("发送") and not self.exists("按住 说话"):
|
||
return {"success": False, "error": "未能进入聊天界面"}
|
||
|
||
# 输入并发送消息
|
||
self.input_text(content, clear=False)
|
||
self.sleep(0.3)
|
||
self.click_text("发送")
|
||
|
||
return {
|
||
"success": True,
|
||
"message_id": f"wx_{int(time.time() * 1000)}"
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"微信发送失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def execute_compound_task(self, task_description: str) -> Dict[str, Any]:
|
||
"""
|
||
执行复合任务（如"打开微信给张三发消息：下午开会"）
|
||
|
||
Args:
|
||
task_description: 任务描述
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
try:
|
||
# 解析任务
|
||
# 提取联系人
|
||
contact_match = re.search(r'给(.+?)发', task_description)
|
||
if not contact_match:
|
||
return {"success": False, "error": "无法解析联系人"}
|
||
|
||
contact = contact_match.group(1).strip()
|
||
|
||
# 提取消息内容
|
||
content_match = re.search(r'发消息[：:](.+)', task_description)
|
||
if not content_match:
|
||
content_match = re.search(r'发(.+)', task_description)
|
||
|
||
if not content_match:
|
||
return {"success": False, "error": "无法解析消息内容"}
|
||
|
||
content = content_match.group(1).strip()
|
||
|
||
# 执行发送
|
||
return self.send_message_with_search(contact, content)
|
||
|
||
except Exception as e:
|
||
logger.error(f"执行复合任务失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_current_screen_info(self) -> Dict[str, Any]:
|
||
"""获取当前屏幕信息（调试用）"""
|
||
try:
|
||
ui_tree = self.d.dump_hierarchy()
|
||
info = self.d.info
|
||
|
||
return {
|
||
"success": True,
|
||
"display_width": info.get("displayWidth"),
|
||
"display_height": info.get("displayHeight"),
|
||
"current_package": info.get("currentPackageName"),
|
||
"ui_tree_length": len(ui_tree)
|
||
}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)} |