370 lines
12 KiB
Python
370 lines
12 KiB
Python
"""
|
||
Soul 控制技能 - Agent 端实现
|
||
|
||
功能模块：
|
||
1. 瞬间管理 - 发布瞬间（文字/图片/视频）
|
||
2. 私聊管理 - 发送消息、获取消息列表
|
||
3. 好友/关注 - 获取联系人列表
|
||
|
||
安卓包名：cn.soulapp.android
|
||
UI 文案与 resourceId 需根据真机 Soul 版本微调。
|
||
|
||
@author 卡若
|
||
@version 3.0.0
|
||
"""
|
||
|
||
import time
|
||
import logging
|
||
from typing import Dict, Any, List, Optional
|
||
import sys
|
||
import os
|
||
|
||
_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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SoulSkill(BaseSkill):
|
||
"""
|
||
Soul 控制技能
|
||
|
||
支持的操作：
|
||
- post_moments: 发布瞬间（文字/图片/视频）
|
||
- comment_moments: 评论某条瞬间（如回复 123）
|
||
- send_message: 发送私聊消息
|
||
- get_messages: 获取私聊消息列表
|
||
- get_contacts: 获取好友/关注列表
|
||
"""
|
||
|
||
PACKAGE = "cn.soulapp.android"
|
||
NAME = "Soul"
|
||
|
||
# Soul 常见 UI 文案（随版本可能变化，真机可调）
|
||
TEXT_PUBLISH = "发布"
|
||
TEXT_MOMENT = "瞬间"
|
||
TEXT_MESSAGE = "消息"
|
||
TEXT_ME = "我的"
|
||
TEXT_SEND = "发送"
|
||
TEXT_SEARCH = "搜索"
|
||
TEXT_VIDEO = "视频"
|
||
TEXT_PHOTO = "图片"
|
||
TEXT_TEXT = "文字"
|
||
|
||
# ========== 发瞬间 ==========
|
||
|
||
def post_moments(
|
||
self,
|
||
content: str,
|
||
images: Optional[List[str]] = None,
|
||
video_url: Optional[str] = None,
|
||
location: Optional[str] = None,
|
||
visible_list: Optional[List[str]] = None,
|
||
invisible_list: Optional[List[str]] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
发布 Soul 瞬间（文字/图片/视频）
|
||
|
||
Args:
|
||
content: 文案
|
||
images: 图片 URL 或本地路径列表（可选）
|
||
video_url: 视频 URL 或本地路径（可选）
|
||
location: 位置（可选）
|
||
visible_list / invisible_list: 可见/不可见名单（可选，Soul 若支持）
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready(timeout=8)
|
||
self.sleep(1.5)
|
||
|
||
# 点击发布入口：常见为底部「+」或「发布」
|
||
published = False
|
||
for hint in ["+", "发布", "发瞬间", "写瞬间", "加号"]:
|
||
if self.click_text(hint) or self.click_contains(hint):
|
||
self.sleep(1)
|
||
published = True
|
||
break
|
||
if not published:
|
||
# 尝试点击屏幕中央偏下（常见发布按钮位置）
|
||
info = self.d.info
|
||
self.d.click(info["displayWidth"] // 2, info["displayHeight"] * 4 // 5)
|
||
self.sleep(1)
|
||
|
||
# 选择类型：优先视频 > 图片 > 文字
|
||
if video_url:
|
||
if self.click_text(self.TEXT_VIDEO) or self.click_contains("视频"):
|
||
self.sleep(1)
|
||
# TODO: 将 video_url 推到手机并选择文件
|
||
logger.warning("Soul post_moments: 视频选择需真机适配路径")
|
||
if images:
|
||
if self.click_text(self.TEXT_PHOTO) or self.click_contains("图片") or self.click_contains("相册"):
|
||
self.sleep(1)
|
||
# TODO: 将 images 推到手机并多选
|
||
logger.warning("Soul post_moments: 多图选择需真机适配路径")
|
||
|
||
# 输入文案
|
||
if content:
|
||
self.sleep(0.5)
|
||
self.input_text(content, clear=True)
|
||
self.sleep(0.3)
|
||
|
||
# 点击发布/发送
|
||
for send_btn in ["发布", "发送", "发瞬间", "完成", "下一步"]:
|
||
if self.click_text(send_btn) or self.click_contains(send_btn):
|
||
self.sleep(1.5)
|
||
break
|
||
|
||
logger.info("Soul 瞬间发布流程已执行")
|
||
return {
|
||
"success": True,
|
||
"message_id": f"soul_m_{int(time.time() * 1000)}",
|
||
"message": "发布流程已执行，请以真机实际界面为准核对",
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Soul 发布瞬间失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# ========== 评论瞬间 ==========
|
||
|
||
def comment_moments(
|
||
self,
|
||
user_id: str = "",
|
||
comment: str = "",
|
||
post_index: int = 0,
|
||
reply_to: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
在 Soul 广场找一条瞬间并评论（如回复 123）
|
||
|
||
Args:
|
||
user_id: 可选，发瞬间的用户标识（Soul 上可不传，按 post_index 选条数）
|
||
comment: 评论内容，如 "123"
|
||
post_index: 第几条瞬间，0 表示第一条
|
||
reply_to: 可选，回复某人
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready(timeout=8)
|
||
self.sleep(1.5)
|
||
|
||
# 进入广场/发现（首页往往是广场流）
|
||
for tab in ["广场", "发现", "首页", "推荐"]:
|
||
if self.click_text(tab) or self.click_contains(tab):
|
||
self.sleep(2)
|
||
break
|
||
else:
|
||
# 默认已在首页，稍等加载
|
||
self.sleep(2)
|
||
|
||
# 点第 post_index 条瞬间（0=第一条）：通过多次下滑再点，或直接点第一条
|
||
for _ in range(post_index):
|
||
self.swipe_up(scale=0.4)
|
||
self.sleep(0.8)
|
||
self.sleep(0.5)
|
||
# 点击第一条瞬间卡片进入详情（常见：点击卡片区域或「评论」入口）
|
||
info = self.d.info
|
||
self.d.click(info["displayWidth"] // 2, info["displayHeight"] // 3)
|
||
self.sleep(1.5)
|
||
|
||
# 点评论按钮
|
||
commented = False
|
||
for btn in ["评论", "写评论", "说点什么", "留言"]:
|
||
if self.click_text(btn) or self.click_contains(btn):
|
||
self.sleep(1)
|
||
commented = True
|
||
break
|
||
if not commented:
|
||
# 尝试点击屏幕下方评论图标区域
|
||
self.d.click(info["displayWidth"] // 2, info["displayHeight"] * 4 // 5)
|
||
self.sleep(1)
|
||
|
||
# 输入评论内容
|
||
if comment:
|
||
self.input_text(comment, clear=True)
|
||
self.sleep(0.3)
|
||
# 发送
|
||
for send_btn in ["发送", "发布", "评论", "完成"]:
|
||
if self.click_text(send_btn) or self.click_contains(send_btn):
|
||
self.sleep(1)
|
||
break
|
||
|
||
logger.info(f"Soul 评论瞬间已执行: {comment}")
|
||
return {
|
||
"success": True,
|
||
"message": f"已对第 {post_index + 1} 条瞬间评论: {comment}",
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Soul 评论瞬间失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# ========== 私聊 ==========
|
||
|
||
def send_message(
|
||
self,
|
||
to_id: str,
|
||
content: str,
|
||
msg_type: str = "text",
|
||
media_url: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
发送 Soul 私聊消息
|
||
|
||
Args:
|
||
to_id: 对方昵称或 Soul ID
|
||
content: 消息内容
|
||
msg_type: text / image / video
|
||
media_url: 图片/视频 URL（可选）
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready(timeout=8)
|
||
self.sleep(1)
|
||
|
||
# 进入消息
|
||
if not (self.click_text(self.TEXT_MESSAGE) or self.click_contains("消息")):
|
||
self.sleep(0.5)
|
||
self.click_text(self.TEXT_MESSAGE) or self.click_contains("消息")
|
||
self.sleep(1)
|
||
|
||
# 搜索或找到对方
|
||
if self.click_text(self.TEXT_SEARCH) or self.click_desc(self.TEXT_SEARCH):
|
||
self.sleep(0.5)
|
||
self.input_text(to_id, clear=True)
|
||
self.sleep(1.5)
|
||
if self.click_contains(to_id) or self.click_text(to_id):
|
||
self.sleep(1)
|
||
else:
|
||
return {"success": False, "error": f"未找到用户: {to_id}"}
|
||
|
||
# 输入并发送
|
||
self.input_text(content, clear=False)
|
||
self.sleep(0.3)
|
||
if self.click_text(self.TEXT_SEND) or self.click_contains("发送"):
|
||
self.sleep(0.5)
|
||
|
||
logger.info(f"Soul 私聊已发送: {to_id}")
|
||
return {
|
||
"success": True,
|
||
"message_id": f"soul_{int(time.time() * 1000)}",
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Soul 发送消息失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_messages(
|
||
self,
|
||
limit: int = 20,
|
||
conversation_id: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取 Soul 私聊消息列表（当前会话或列表）
|
||
|
||
Args:
|
||
limit: 条数
|
||
conversation_id: 会话对象（对方昵称或 ID）
|
||
|
||
Returns:
|
||
消息列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready(timeout=8)
|
||
self.sleep(1)
|
||
|
||
self.click_text(self.TEXT_MESSAGE) or self.click_contains("消息")
|
||
self.sleep(1)
|
||
|
||
if conversation_id:
|
||
if self.click_text(self.TEXT_SEARCH) or self.click_desc(self.TEXT_SEARCH):
|
||
self.sleep(0.5)
|
||
self.input_text(conversation_id, clear=True)
|
||
self.sleep(1.5)
|
||
self.click_contains(conversation_id) or self.click_text(conversation_id)
|
||
self.sleep(1)
|
||
|
||
messages = []
|
||
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:
|
||
messages.append({"text": text, "timestamp": int(time.time() * 1000)})
|
||
except Exception:
|
||
pass
|
||
|
||
return {
|
||
"success": True,
|
||
"messages": messages[:limit],
|
||
"count": len(messages[:limit]),
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Soul 获取消息失败: {e}")
|
||
return {"success": False, "error": str(e), "messages": []}
|
||
|
||
# ========== 好友/联系人 ==========
|
||
|
||
def get_contacts(self, limit: int = 200) -> Dict[str, Any]:
|
||
"""
|
||
获取 Soul 好友/关注列表（从消息列表或「我的」-「关注」解析）
|
||
|
||
Args:
|
||
limit: 数量上限
|
||
|
||
Returns:
|
||
联系人列表
|
||
"""
|
||
try:
|
||
self.launch()
|
||
self.wait_for_app_ready(timeout=8)
|
||
self.sleep(1)
|
||
|
||
# 从消息页获取会话列表作为「联系人」近似
|
||
self.click_text(self.TEXT_MESSAGE) or self.click_contains("消息")
|
||
self.sleep(1.5)
|
||
|
||
contacts = []
|
||
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 and "发送" not in text:
|
||
contacts.append({"name": text, "user_id": text})
|
||
except Exception:
|
||
pass
|
||
|
||
# 去重并限制数量
|
||
seen = set()
|
||
unique = []
|
||
for c in contacts:
|
||
k = c.get("name") or c.get("user_id", "")
|
||
if k and k not in seen:
|
||
seen.add(k)
|
||
unique.append(c)
|
||
if len(unique) >= limit:
|
||
break
|
||
|
||
return {
|
||
"success": True,
|
||
"contacts": unique,
|
||
"count": len(unique),
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Soul 获取联系人失败: {e}")
|
||
return {"success": False, "error": str(e), "contacts": []}
|