feat: 微信引擎v2全功能(96actions/98API) + 导航缓存系统 + 操作日志
- 引擎v1→v2: 54→96个actions, 新增14模块(解封/视频号/扫一扫/支付/通话/群发/搜一搜/运动/位置/表情/朋友圈增强/语音/文件/设置) - API端点: 55→98个, 23个功能分类 - 新建 wechat_nav_cache.py: 导航路径缓存+页面元素录制+操作日志+自动纠错 - 真机录制Tab精确坐标(y=2148), 我/设置/账号与安全页面元素 - 全局容错: _retry/_recover_state/_verify_action - 导航方法重构: 缓存路径优先, 消除重复UI探索 - 微信全功能矩阵文档 v8.0.56 Made-with: Cursor
This commit is contained in:
卡若
@@ -406,12 +406,18 @@ def _parse_contacts_from_xml(xml: str, limit: int) -> list:
|
||||
async def _anti_ban_guard(device_id: str, platform: str, action: str, content: Optional[str] = None) -> dict:
|
||||
"""
|
||||
防封守卫：在执行任何操作前进行防封检查。
|
||||
返回 {"pass": True/False, "error": str, "content": str(处理后的内容)}
|
||||
返回 {"pass": True/False, "reason": str, "content": str(处理后的内容)}
|
||||
检查链：时段 → 生命周期 → L1全局 → L2设备 → L3动作 → 敏感词 → 内容差异化
|
||||
"""
|
||||
from services.anti_ban_alert import (
|
||||
alert_daily_limit, alert_outside_hours, alert_sensitive_content,
|
||||
)
|
||||
from services.rate_limiter import OutsideOperationHours
|
||||
|
||||
try:
|
||||
lifecycle_check = await account_lifecycle.check_allowed(device_id, platform, action)
|
||||
if not lifecycle_check["allowed"]:
|
||||
return {"pass": False, "error": lifecycle_check["reason"], "content": content}
|
||||
return {"pass": False, "reason": lifecycle_check["reason"], "content": content}
|
||||
except Exception as e:
|
||||
logger.warning(f"[anti_ban] 生命周期检查异常(跳过): {e}")
|
||||
|
||||
@@ -420,8 +426,12 @@ async def _anti_ban_guard(device_id: str, platform: str, action: str, content: O
|
||||
wait = await rate_limiter.check_and_wait(device_id, platform, action, is_new_account=is_new)
|
||||
if wait > 0:
|
||||
logger.info(f"[anti_ban] {device_id}/{platform}.{action} 等待了 {wait:.1f}s")
|
||||
except OutsideOperationHours as e:
|
||||
asyncio.ensure_future(alert_outside_hours(device_id, action))
|
||||
return {"pass": False, "reason": str(e), "content": content}
|
||||
except DailyLimitExceeded as e:
|
||||
return {"pass": False, "error": str(e), "content": content}
|
||||
asyncio.ensure_future(alert_daily_limit(device_id, platform, action, -1, -1))
|
||||
return {"pass": False, "reason": str(e), "content": content}
|
||||
except Exception as e:
|
||||
logger.warning(f"[anti_ban] 限流检查异常(跳过): {e}")
|
||||
|
||||
@@ -430,9 +440,10 @@ async def _anti_ban_guard(device_id: str, platform: str, action: str, content: O
|
||||
sensitive = has_sensitive_words(content)
|
||||
if sensitive:
|
||||
logger.warning(f"[anti_ban] 检测到敏感词: {sensitive}")
|
||||
asyncio.ensure_future(alert_sensitive_content(device_id, platform, sensitive))
|
||||
safe_content = diversify_content(content, add_invisible=True, add_emoji=False)
|
||||
|
||||
return {"pass": True, "error": "", "content": safe_content}
|
||||
return {"pass": True, "reason": "", "content": safe_content}
|
||||
|
||||
|
||||
async def _execute_skill(device_id: str, platform: str, action: str, params: dict, timeout: int = 30) -> dict:
|
||||
@@ -529,7 +540,7 @@ async def send_message(req: SendMessageRequest):
|
||||
# ---- 防封守卫 ----
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "send_message", req.content)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "data": {"success": False, "error": guard["error"], "error_code": "anti_ban_blocked"}, "channel_used": "none"}
|
||||
return {"code": 200, "data": {"success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}, "channel_used": "none"}
|
||||
req.content = guard["content"]
|
||||
|
||||
mode = _get_device_mode(req.device_id)
|
||||
@@ -614,7 +625,7 @@ async def batch_send_message(req: BatchSendMessageRequest):
|
||||
"""
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "batch_send", req.content)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "data": {"success": False, "error": guard["error"], "error_code": "anti_ban_blocked"}, "channel_used": "none"}
|
||||
return {"code": 200, "data": {"success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}, "channel_used": "none"}
|
||||
req.content = guard["content"]
|
||||
|
||||
_check_device_online(req.device_id)
|
||||
@@ -644,9 +655,12 @@ async def batch_send_message(req: BatchSendMessageRequest):
|
||||
async def reply_comment(req: ReplyCommentRequest):
|
||||
"""回复评论（抖音/小红书等）"""
|
||||
_check_device_online(req.device_id)
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "comment", req.content)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}
|
||||
result = await _execute_skill(
|
||||
req.device_id, req.platform.value, "reply_comment",
|
||||
{"video_id": req.video_id, "comment_id": req.comment_id, "content": req.content}
|
||||
{"video_id": req.video_id, "comment_id": req.comment_id, "content": guard.get("content", req.content)}
|
||||
)
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {
|
||||
@@ -665,7 +679,7 @@ async def add_friend(req: AddFriendRequest):
|
||||
"""添加好友"""
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "add_friend", req.message)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "data": {"success": False, "error": guard["error"], "error_code": "anti_ban_blocked"}, "channel_used": "none"}
|
||||
return {"code": 200, "data": {"success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}, "channel_used": "none"}
|
||||
req.message = guard["content"] or req.message
|
||||
|
||||
_check_device_online(req.device_id)
|
||||
@@ -739,14 +753,15 @@ async def delete_friend(req: DeleteFriendRequest):
|
||||
@router.post("/friend/batch-add", response_model=dict, tags=["好友管理"])
|
||||
async def batch_add_friend(req: BatchAddFriendRequest):
|
||||
"""批量添加好友"""
|
||||
|
||||
_check_device_online(req.device_id)
|
||||
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "add_friend", req.message)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}
|
||||
result = await _execute_skill(
|
||||
req.device_id, req.platform.value, "batch_add_friend",
|
||||
{
|
||||
"user_ids": req.user_ids,
|
||||
"message": req.message,
|
||||
"message": guard.get("content", req.message),
|
||||
"interval": req.interval
|
||||
},
|
||||
timeout=len(req.user_ids) * 15
|
||||
@@ -873,14 +888,15 @@ async def set_group_name(req: SetGroupNameRequest):
|
||||
@router.post("/group/send-message", response_model=dict, tags=["群聊管理"])
|
||||
async def send_group_message(req: GroupMessageRequest):
|
||||
"""发送群消息"""
|
||||
|
||||
_check_device_online(req.device_id)
|
||||
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "send_group_message", req.content)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}
|
||||
result = await _execute_skill(
|
||||
req.device_id, req.platform.value, "send_group_message",
|
||||
{
|
||||
"group_id": req.group_id,
|
||||
"content": req.content,
|
||||
"content": guard.get("content", req.content),
|
||||
"msg_type": req.msg_type.value,
|
||||
"media_url": req.media_url,
|
||||
"at_all": req.at_all,
|
||||
@@ -1073,7 +1089,7 @@ async def post_moments(req: PostMomentsRequest):
|
||||
"""发布朋友圈/瞬间（统一返回 200，业务失败用 success=false 表示，避免 502）"""
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "post_moments", req.content)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["error"], "data": {}, "channel_used": "none"}
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "data": {}, "channel_used": "none"}
|
||||
req.content = guard["content"]
|
||||
|
||||
try:
|
||||
@@ -1122,14 +1138,14 @@ async def post_moments(req: PostMomentsRequest):
|
||||
@router.post("/moments/like", response_model=dict, tags=["朋友圈管理"])
|
||||
async def like_moments(req: LikeMomentsRequest):
|
||||
"""点赞朋友圈"""
|
||||
|
||||
_check_device_online(req.device_id)
|
||||
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "like_moments")
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}
|
||||
result = await _execute_skill(
|
||||
req.device_id, req.platform.value, "like_moments",
|
||||
{"user_id": req.user_id, "post_index": req.post_index}
|
||||
)
|
||||
|
||||
return {
|
||||
"code": 200,
|
||||
"data": result.get("data", {}),
|
||||
@@ -1140,15 +1156,16 @@ async def like_moments(req: LikeMomentsRequest):
|
||||
@router.post("/moments/comment", response_model=dict, tags=["朋友圈管理"])
|
||||
async def comment_moments(req: CommentMomentsRequest):
|
||||
"""评论朋友圈"""
|
||||
|
||||
_check_device_online(req.device_id)
|
||||
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "comment_moments", req.comment)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}
|
||||
result = await _execute_skill(
|
||||
req.device_id, req.platform.value, "comment_moments",
|
||||
{
|
||||
"user_id": req.user_id,
|
||||
"post_index": req.post_index,
|
||||
"comment": req.comment,
|
||||
"comment": guard.get("content", req.comment),
|
||||
"reply_to": req.reply_to
|
||||
}
|
||||
)
|
||||
@@ -1571,6 +1588,416 @@ async def follow_official_account(device_id: str, platform: Platform, account_na
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 十六、解封与限制管理
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/account/unblock-self", response_model=dict, tags=["账号安全"])
|
||||
async def unblock_self(device_id: str, platform: Platform):
|
||||
"""自助解封"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "unblock_self", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/account/unblock-appeal", response_model=dict, tags=["账号安全"])
|
||||
async def unblock_appeal(device_id: str, platform: Platform):
|
||||
"""申诉解封"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "unblock_appeal", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.get("/account/restrictions", response_model=dict, tags=["账号安全"])
|
||||
async def check_restrictions(device_id: str, platform: Platform):
|
||||
"""检查当前功能限制"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "check_restrictions", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/account/appeal-restriction", response_model=dict, tags=["账号安全"])
|
||||
async def appeal_restriction(device_id: str, platform: Platform):
|
||||
"""申诉功能限制"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "appeal_restriction", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/account/unblock-sms", response_model=dict, tags=["账号安全"])
|
||||
async def unblock_with_sms(device_id: str, platform: Platform, phone: str = ""):
|
||||
"""短信验证解封"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "unblock_with_sms", {"phone": phone})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 十七、视频号
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/video-channel/list", response_model=dict, tags=["视频号"])
|
||||
async def get_video_list(device_id: str, platform: Platform, limit: int = 10):
|
||||
"""获取视频号列表"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "get_video_list", {"limit": limit})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/video-channel/like", response_model=dict, tags=["视频号"])
|
||||
async def like_video(device_id: str, platform: Platform, index: int = 0):
|
||||
"""点赞视频"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "like_video", {"index": index})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/video-channel/comment", response_model=dict, tags=["视频号"])
|
||||
async def comment_video(device_id: str, platform: Platform, index: int = 0, comment: str = ""):
|
||||
"""评论视频"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "comment_video", {"index": index, "comment": comment})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/video-channel/follow", response_model=dict, tags=["视频号"])
|
||||
async def follow_video_creator(device_id: str, platform: Platform, index: int = 0):
|
||||
"""关注视频号创作者"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "follow_video_creator", {"index": index})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/video-channel/share", response_model=dict, tags=["视频号"])
|
||||
async def share_video(device_id: str, platform: Platform, index: int = 0, to_id: str = ""):
|
||||
"""分享视频"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "share_video", {"index": index, "to_id": to_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 十八、扫一扫
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/scan/qr-code", response_model=dict, tags=["扫一扫"])
|
||||
async def scan_qr_code(device_id: str, platform: Platform):
|
||||
"""扫描二维码"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "scan_qr_code", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/scan/add-friend", response_model=dict, tags=["扫一扫"])
|
||||
async def scan_add_friend(device_id: str, platform: Platform):
|
||||
"""扫码加好友"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "scan_add_friend", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.get("/scan/my-qr", response_model=dict, tags=["扫一扫"])
|
||||
async def show_my_qr(device_id: str, platform: Platform):
|
||||
"""显示我的二维码"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "show_my_qr", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/scan/extract-qr", response_model=dict, tags=["扫一扫"])
|
||||
async def extract_qr_from_image(device_id: str, platform: Platform):
|
||||
"""从图片识别二维码"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "extract_qr_from_image", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 十九、支付增强
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/payment/code", response_model=dict, tags=["支付"])
|
||||
async def show_payment_code(device_id: str, platform: Platform):
|
||||
"""显示付款码"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "show_payment_code", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/payment/receive", response_model=dict, tags=["支付"])
|
||||
async def receive_payment(device_id: str, platform: Platform, amount: str = "", desc: str = ""):
|
||||
"""收款"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "receive_payment", {"amount": amount, "desc": desc})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.get("/payment/wallet", response_model=dict, tags=["支付"])
|
||||
async def view_wallet(device_id: str, platform: Platform):
|
||||
"""查看钱包"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "view_wallet", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.get("/payment/transactions", response_model=dict, tags=["支付"])
|
||||
async def view_transactions(device_id: str, platform: Platform, limit: int = 20):
|
||||
"""查看账单"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "view_transactions", {"limit": limit})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/payment/receive-red-packet", response_model=dict, tags=["支付"])
|
||||
async def receive_red_packet(device_id: str, platform: Platform, from_id: str = ""):
|
||||
"""领取红包"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "receive_red_packet", {"from_id": from_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十、通话
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/call/voice", response_model=dict, tags=["通话"])
|
||||
async def voice_call(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""语音通话"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "voice_call", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/call/video", response_model=dict, tags=["通话"])
|
||||
async def video_call(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""视频通话"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "video_call", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十一、群发助手
|
||||
# =============================================================================
|
||||
|
||||
class MassSendRequest(BaseModel):
|
||||
device_id: str
|
||||
platform: Platform
|
||||
content: str
|
||||
user_ids: List[str]
|
||||
|
||||
@router.post("/mass-send", response_model=dict, tags=["群发助手"])
|
||||
async def mass_send(req: MassSendRequest):
|
||||
"""群发消息"""
|
||||
_check_device_online(req.device_id)
|
||||
guard = await _anti_ban_guard(req.device_id, req.platform.value, "batch_send", req.content)
|
||||
if not guard["pass"]:
|
||||
return {"code": 200, "success": False, "error": guard["reason"], "error_code": "anti_ban_blocked"}
|
||||
result = await _execute_skill(
|
||||
req.device_id, req.platform.value, "mass_send",
|
||||
{"content": guard.get("content", req.content), "user_ids": req.user_ids}
|
||||
)
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十二、搜一搜/看一看
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/search/wechat", response_model=dict, tags=["搜一搜"])
|
||||
async def wechat_search(device_id: str, platform: Platform, keyword: str = ""):
|
||||
"""搜一搜"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "wechat_search", {"keyword": keyword})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.get("/discover/top-stories", response_model=dict, tags=["看一看"])
|
||||
async def top_stories(device_id: str, platform: Platform):
|
||||
"""看一看"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "top_stories", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十三、微信运动
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/wechat-sport/steps", response_model=dict, tags=["微信运动"])
|
||||
async def get_steps(device_id: str, platform: Platform):
|
||||
"""获取步数"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "get_steps", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/wechat-sport/like-steps", response_model=dict, tags=["微信运动"])
|
||||
async def like_steps(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""点赞步数"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "like_steps", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十四、位置分享
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/location/send", response_model=dict, tags=["位置"])
|
||||
async def send_location(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""发送位置"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "send_location", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/location/share-realtime", response_model=dict, tags=["位置"])
|
||||
async def share_real_time_location(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""共享实时位置"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "share_real_time_location", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十五、表情管理
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/emoji/send", response_model=dict, tags=["表情"])
|
||||
async def send_emoji(device_id: str, platform: Platform, user_id: str = "", emoji_name: str = ""):
|
||||
"""发送表情"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "send_emoji", {"user_id": user_id, "emoji_name": emoji_name})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.get("/emoji/stickers", response_model=dict, tags=["表情"])
|
||||
async def get_sticker_list(device_id: str, platform: Platform):
|
||||
"""获取表情包列表"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "get_sticker_list", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十六、朋友圈增强
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/moments/set-cover", response_model=dict, tags=["朋友圈管理"])
|
||||
async def set_moments_cover(device_id: str, platform: Platform):
|
||||
"""设置朋友圈封面"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "set_moments_cover", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/moments/set-privacy", response_model=dict, tags=["朋友圈管理"])
|
||||
async def set_moments_privacy(device_id: str, platform: Platform, days: int = 180):
|
||||
"""设置朋友圈可见天数"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "set_moments_privacy", {"days": days})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/moments/share-link", response_model=dict, tags=["朋友圈管理"])
|
||||
async def share_link_to_moments(device_id: str, platform: Platform, url: str = "", title: str = ""):
|
||||
"""分享链接到朋友圈"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "forward_moments_link", {"url": url, "title": title})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十七、语音消息
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/message/voice", response_model=dict, tags=["消息管理"])
|
||||
async def send_voice_message(device_id: str, platform: Platform, user_id: str = "", duration: int = 3):
|
||||
"""发送语音消息"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "send_voice_message", {"user_id": user_id, "duration": duration})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十八、文件管理
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/file/send", response_model=dict, tags=["文件管理"])
|
||||
async def send_file(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""发送文件"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "send_file_from_chat", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/file/download", response_model=dict, tags=["文件管理"])
|
||||
async def download_file(device_id: str, platform: Platform, user_id: str = ""):
|
||||
"""下载文件"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "download_file", {"user_id": user_id})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 二十九、设置管理
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/settings/do-not-disturb", response_model=dict, tags=["设置"])
|
||||
async def toggle_dnd(device_id: str, platform: Platform, enable: bool = True):
|
||||
"""勿扰模式"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "toggle_do_not_disturb", {"enable": enable})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/settings/clear-cache", response_model=dict, tags=["设置"])
|
||||
async def clear_cache(device_id: str, platform: Platform):
|
||||
"""清理缓存"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "clear_cache", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/settings/check-update", response_model=dict, tags=["设置"])
|
||||
async def check_update(device_id: str, platform: Platform):
|
||||
"""检查更新"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "check_for_update", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/settings/logout", response_model=dict, tags=["设置"])
|
||||
async def logout(device_id: str, platform: Platform):
|
||||
"""退出登录"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "logout", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
@router.post("/settings/switch-account", response_model=dict, tags=["设置"])
|
||||
async def switch_account(device_id: str, platform: Platform):
|
||||
"""切换账号"""
|
||||
_check_device_online(device_id)
|
||||
result = await _execute_skill(device_id, platform.value, "switch_account", {})
|
||||
payload = result.get("data") if result.get("data") is not None else result
|
||||
return {"code": 200, "data": payload, "channel_used": Channel.SDK_CONTROL.value}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 内部实现方法
|
||||
# =============================================================================
|
||||
@@ -1813,3 +2240,83 @@ async def _send_via_agent(req: SendMessageRequest) -> dict:
|
||||
return {"success": False, "error": err or "ADB 执行失败"}
|
||||
|
||||
return {"success": False, "error": "设备离线"}
|
||||
|
||||
|
||||
# ========== AF12: 风控中心监控 ==========
|
||||
|
||||
|
||||
@router.get("/anti-ban/dashboard")
|
||||
async def anti_ban_dashboard():
|
||||
"""防封风控中心：汇总全局限流/账号生命周期/指纹碰撞状态"""
|
||||
import time as _time
|
||||
from services.rate_limiter import rate_limiter as _rl
|
||||
|
||||
devices = await device_manager.get_all_devices(limit=500)
|
||||
|
||||
total_devices = len(devices)
|
||||
online = sum(1 for d in devices if d.get("status") == "online")
|
||||
|
||||
seen_fp: dict = {}
|
||||
fp_collision_devices = []
|
||||
for d in devices:
|
||||
fp = d.get("fingerprint_hash", "")
|
||||
if not fp:
|
||||
continue
|
||||
did = d.get("device_id", "")
|
||||
seen_fp.setdefault(fp, []).append(did)
|
||||
for fp, dids in seen_fp.items():
|
||||
if len(dids) > 1:
|
||||
for did in dids:
|
||||
fp_collision_devices.append({
|
||||
"device_id": did,
|
||||
"collided_with": [x for x in dids if x != did],
|
||||
})
|
||||
|
||||
lifecycle_summary = []
|
||||
for d in devices[:50]:
|
||||
did = d.get("device_id", "")
|
||||
plat = d.get("platform", "wechat")
|
||||
phase_info = await account_lifecycle.get_phase(did, plat)
|
||||
rules = await account_lifecycle.get_rules(did, plat)
|
||||
lifecycle_summary.append({
|
||||
"device_id": did,
|
||||
"phase": phase_info.value if hasattr(phase_info, "value") else str(phase_info),
|
||||
"max_daily_add_friend": rules.get("max_daily_add_friend", "N/A"),
|
||||
"max_daily_message": rules.get("max_daily_send_message", "N/A"),
|
||||
})
|
||||
|
||||
return {
|
||||
"timestamp": int(_time.time()),
|
||||
"devices": {"total": total_devices, "online": online},
|
||||
"fingerprint_collisions": fp_collision_devices,
|
||||
"account_lifecycle": lifecycle_summary,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/anti-ban/device/{device_id}")
|
||||
async def anti_ban_device_detail(device_id: str):
|
||||
"""单设备防封详情"""
|
||||
device = await device_manager.get_device(device_id)
|
||||
if not device:
|
||||
raise HTTPException(status_code=404, detail="设备不存在")
|
||||
|
||||
plat = device.get("platform", "wechat")
|
||||
phase = await account_lifecycle.get_phase(device_id, plat)
|
||||
rules = await account_lifecycle.get_rules(device_id, plat)
|
||||
fp = device.get("fingerprint_hash", "")
|
||||
|
||||
collision_result = {"collision": False, "collided_with": []}
|
||||
if fp and device_manager.db:
|
||||
collision_result = await device_manager.check_fingerprint_collision(
|
||||
device_id, device.get("fingerprint_info", {})
|
||||
)
|
||||
|
||||
return {
|
||||
"device_id": device_id,
|
||||
"status": device.get("status", "unknown"),
|
||||
"fingerprint_hash": fp,
|
||||
"fingerprint_collision": collision_result,
|
||||
"account_phase": phase.value if hasattr(phase, "value") else str(phase),
|
||||
"phase_rules": rules,
|
||||
"last_heartbeat": str(device.get("last_heartbeat", "")),
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
438
sdk/app/services/wechat_nav_cache.py
Normal file
438
sdk/app/services/wechat_nav_cache.py
Normal file
@@ -0,0 +1,438 @@
|
||||
"""
|
||||
微信 UI 导航缓存系统 v1.0
|
||||
—— 记录所有页面路径、元素坐标，避免重复探索，支持自动纠错
|
||||
|
||||
核心理念:
|
||||
1. 首次探索时录制页面结构到 JSON（持久化）
|
||||
2. 后续操作直接查缓存，跳过 uiautomator dump
|
||||
3. 缓存失效（点击后页面不符预期）自动触发重新探索
|
||||
4. 每个 action 的操作步骤作为 "路径模板" 存储
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import os
|
||||
import re
|
||||
import logging
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CACHE_DIR = Path(__file__).parent.parent.parent / "data" / "nav_cache"
|
||||
LOG_DIR = Path(__file__).parent.parent.parent / "data" / "operation_logs"
|
||||
|
||||
|
||||
class PageSignature:
|
||||
"""页面特征签名——用于快速验证当前在哪个页面"""
|
||||
|
||||
def __init__(self, name: str, must_have: List[str] = None,
|
||||
must_not_have: List[str] = None, title: str = None):
|
||||
self.name = name
|
||||
self.must_have = must_have or []
|
||||
self.must_not_have = must_not_have or []
|
||||
self.title = title
|
||||
|
||||
def match(self, texts: List[str]) -> bool:
|
||||
text_set = set(texts)
|
||||
for t in self.must_have:
|
||||
if t not in text_set:
|
||||
return False
|
||||
for t in self.must_not_have:
|
||||
if t in text_set:
|
||||
return False
|
||||
if self.title and self.title not in text_set:
|
||||
return False
|
||||
return True
|
||||
|
||||
def to_dict(self):
|
||||
return {
|
||||
"name": self.name,
|
||||
"must_have": self.must_have,
|
||||
"must_not_have": self.must_not_have,
|
||||
"title": self.title,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
return cls(d["name"], d.get("must_have", []),
|
||||
d.get("must_not_have", []), d.get("title"))
|
||||
|
||||
|
||||
class NavStep:
|
||||
"""单步导航操作"""
|
||||
|
||||
def __init__(self, action: str, target: str = None,
|
||||
coords: List[int] = None, wait: float = 1.0,
|
||||
verify_page: str = None, fallback: str = None):
|
||||
self.action = action # tap_text / tap_coord / tap_tab / scroll_up / scroll_down / back / swipe
|
||||
self.target = target # 文本目标 or tab名
|
||||
self.coords = coords # [x, y] 精确坐标（缓存命中时用）
|
||||
self.wait = wait
|
||||
self.verify_page = verify_page # 期望到达的页面签名名
|
||||
self.fallback = fallback # 失败后备用动作
|
||||
|
||||
def to_dict(self):
|
||||
return {k: v for k, v in {
|
||||
"action": self.action, "target": self.target,
|
||||
"coords": self.coords, "wait": self.wait,
|
||||
"verify_page": self.verify_page, "fallback": self.fallback,
|
||||
}.items() if v is not None}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
return cls(**d)
|
||||
|
||||
|
||||
class NavPath:
|
||||
"""一条完整的导航路径（从微信首页到目标页面）"""
|
||||
|
||||
def __init__(self, name: str, steps: List[NavStep] = None,
|
||||
target_page: str = None):
|
||||
self.name = name
|
||||
self.steps = steps or []
|
||||
self.target_page = target_page
|
||||
self.success_count = 0
|
||||
self.fail_count = 0
|
||||
self.last_success = None
|
||||
|
||||
def to_dict(self):
|
||||
return {
|
||||
"name": self.name,
|
||||
"steps": [s.to_dict() for s in self.steps],
|
||||
"target_page": self.target_page,
|
||||
"success_count": self.success_count,
|
||||
"fail_count": self.fail_count,
|
||||
"last_success": self.last_success,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
p = cls(d["name"], [NavStep.from_dict(s) for s in d.get("steps", [])],
|
||||
d.get("target_page"))
|
||||
p.success_count = d.get("success_count", 0)
|
||||
p.fail_count = d.get("fail_count", 0)
|
||||
p.last_success = d.get("last_success")
|
||||
return p
|
||||
|
||||
|
||||
class WeChatNavCache:
|
||||
"""导航缓存管理器"""
|
||||
|
||||
def __init__(self, wechat_version: str = "8.0.56",
|
||||
screen_w: int = 1080, screen_h: int = 2400):
|
||||
self.version = wechat_version
|
||||
self.screen_w = screen_w
|
||||
self.screen_h = screen_h
|
||||
|
||||
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
self.cache_file = CACHE_DIR / f"nav_v{wechat_version}_{screen_w}x{screen_h}.json"
|
||||
|
||||
self.pages: Dict[str, Dict] = {}
|
||||
self.paths: Dict[str, NavPath] = {}
|
||||
self.signatures: Dict[str, PageSignature] = {}
|
||||
self.element_cache: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
self._load()
|
||||
if not self.pages:
|
||||
self._init_builtin()
|
||||
|
||||
def _load(self):
|
||||
if self.cache_file.exists():
|
||||
try:
|
||||
data = json.loads(self.cache_file.read_text(encoding="utf-8"))
|
||||
for name, page in data.get("pages", {}).items():
|
||||
self.pages[name] = page
|
||||
for name, path in data.get("paths", {}).items():
|
||||
self.paths[name] = NavPath.from_dict(path)
|
||||
for name, sig in data.get("signatures", {}).items():
|
||||
self.signatures[name] = PageSignature.from_dict(sig)
|
||||
self.element_cache = data.get("element_cache", {})
|
||||
logger.info(f"导航缓存已加载: {len(self.pages)} 页面, "
|
||||
f"{len(self.paths)} 路径, {len(self.element_cache)} 元素")
|
||||
except Exception as e:
|
||||
logger.warning(f"加载缓存失败: {e}")
|
||||
|
||||
def save(self):
|
||||
data = {
|
||||
"version": self.version,
|
||||
"screen": f"{self.screen_w}x{self.screen_h}",
|
||||
"updated": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"pages": self.pages,
|
||||
"paths": {k: v.to_dict() for k, v in self.paths.items()},
|
||||
"signatures": {k: v.to_dict() for k, v in self.signatures.items()},
|
||||
"element_cache": self.element_cache,
|
||||
}
|
||||
self.cache_file.write_text(
|
||||
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
|
||||
def _init_builtin(self):
|
||||
"""初始化内置的已知页面数据（基于真机录制 2026-03-14）"""
|
||||
|
||||
# ─── 底部 Tab 坐标（精确录制） ───
|
||||
tab_y = 2148
|
||||
self.element_cache["tabs"] = {
|
||||
"微信": {"center": [135, tab_y], "bounds": [0, 2119, 270, 2176]},
|
||||
"通讯录": {"center": [405, tab_y], "bounds": [270, 2119, 540, 2176]},
|
||||
"发现": {"center": [675, tab_y], "bounds": [540, 2119, 810, 2176]},
|
||||
"我": {"center": [945, tab_y], "bounds": [810, 2119, 1080, 2176]},
|
||||
}
|
||||
|
||||
# ─── 页面特征签名 ───
|
||||
sigs = {
|
||||
"home": PageSignature("home", must_have=[], title=None),
|
||||
"me": PageSignature("me", must_have=["服务", "收藏", "设置"]),
|
||||
"settings": PageSignature("settings", must_have=["账号与安全"],
|
||||
title="设置"),
|
||||
"account_security": PageSignature(
|
||||
"account_security", must_have=["微信号", "手机号", "微信密码"],
|
||||
title="账号与安全"),
|
||||
"safety_center": PageSignature(
|
||||
"safety_center", must_have=["微信安全中心"]),
|
||||
"contacts": PageSignature("contacts", must_have=["新的朋友"]),
|
||||
"discover": PageSignature("discover", must_have=["朋友圈", "视频号"]),
|
||||
}
|
||||
for name, sig in sigs.items():
|
||||
self.signatures[name] = sig
|
||||
|
||||
# ─── "我"页面元素坐标（精确录制） ───
|
||||
self.pages["me"] = {
|
||||
"recorded_at": "2026-03-14 16:32",
|
||||
"elements": {
|
||||
"头像区": {"center": [608, 298]},
|
||||
"微信号": {"center": [643, 398], "text": "微信号：Lytiao1"},
|
||||
"状态": {"center": [379, 507]},
|
||||
"3个朋友": {"center": [676, 507]},
|
||||
"服务": {"center": [197, 706]},
|
||||
"收藏": {"center": [197, 883]},
|
||||
"朋友圈": {"center": [220, 1037]},
|
||||
"视频号": {"center": [220, 1191]},
|
||||
"订单与卡包": {"center": [266, 1345]},
|
||||
"表情": {"center": [197, 1499]},
|
||||
"设置": {"center": [197, 1676]},
|
||||
},
|
||||
}
|
||||
|
||||
# ─── 设置页面元素坐标 ───
|
||||
self.pages["settings"] = {
|
||||
"recorded_at": "2026-03-14 16:32",
|
||||
"needs_scroll_to_top": True,
|
||||
"elements": {
|
||||
"账号与安全": {"center": [250, 300], "note": "顶部滑后第一项"},
|
||||
"青少年模式": {"center": [250, 479]},
|
||||
"关怀模式": {"center": [158, 631]},
|
||||
"新消息通知": {"center": [204, 800]},
|
||||
"聊天": {"center": [112, 958]},
|
||||
"通用": {"center": [158, 1112]},
|
||||
"隐私": {"center": [89, 1257], "section_header": True},
|
||||
"朋友权限": {"center": [112, 1423]},
|
||||
"个人信息与权限": {"center": [135, 1574]},
|
||||
"个人信息收集清单": {"center": [158, 1725]},
|
||||
"第三方信息共享清单": {"center": [181, 1876]},
|
||||
"插件": {"center": [89, 2018]},
|
||||
"关于微信": {"center": [89, 2168]},
|
||||
},
|
||||
}
|
||||
|
||||
# ─── 账号与安全页面 ───
|
||||
self.pages["account_security"] = {
|
||||
"recorded_at": "2026-03-14 16:33",
|
||||
"elements": {
|
||||
"微信号": {"center": [112, 300], "value_center": [603, 300]},
|
||||
"手机号": {"center": [112, 451], "value_center": [603, 451]},
|
||||
"微信密码": {"center": [135, 625]},
|
||||
"声音锁": {"center": [112, 778]},
|
||||
"应急联系人": {"center": [158, 953]},
|
||||
"登录过的设备": {"center": [181, 1104]},
|
||||
"更多安全设置": {"center": [181, 1257]},
|
||||
"微信安全中心": {"center": [513, 1423]},
|
||||
},
|
||||
}
|
||||
|
||||
# ─── 常用导航路径 ───
|
||||
self.paths["to_me"] = NavPath("to_me", [
|
||||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||||
], "me")
|
||||
|
||||
self.paths["to_settings"] = NavPath("to_settings", [
|
||||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||||
NavStep("scroll_up", None, None, 0.5),
|
||||
NavStep("tap_text", "设置", [197, 1676], 2.0, "settings"),
|
||||
NavStep("scroll_to_top", None, None, 0.5),
|
||||
], "settings")
|
||||
|
||||
self.paths["to_account_security"] = NavPath("to_account_security", [
|
||||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||||
NavStep("scroll_up", None, None, 0.5),
|
||||
NavStep("tap_text", "设置", [197, 1676], 2.0, "settings"),
|
||||
NavStep("scroll_to_top", None, None, 0.5),
|
||||
NavStep("tap_text", "账号与安全", [250, 300], 2.0, "account_security"),
|
||||
], "account_security")
|
||||
|
||||
self.paths["to_safety_center"] = NavPath("to_safety_center", [
|
||||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||||
NavStep("scroll_up", None, None, 0.5),
|
||||
NavStep("tap_text", "设置", [197, 1676], 2.0, "settings"),
|
||||
NavStep("scroll_to_top", None, None, 0.5),
|
||||
NavStep("tap_text", "账号与安全", [250, 300], 2.0, "account_security"),
|
||||
NavStep("tap_text", "微信安全中心", [513, 1423], 3.0, "safety_center"),
|
||||
], "safety_center")
|
||||
|
||||
self.paths["to_discover"] = NavPath("to_discover", [
|
||||
NavStep("tap_tab", "发现", [675, tab_y], 1.5, "discover"),
|
||||
], "discover")
|
||||
|
||||
self.paths["to_contacts"] = NavPath("to_contacts", [
|
||||
NavStep("tap_tab", "通讯录", [405, tab_y], 1.5, "contacts"),
|
||||
], "contacts")
|
||||
|
||||
self.save()
|
||||
logger.info("内置导航缓存已初始化并保存")
|
||||
|
||||
def get_tab_coord(self, tab_name: str) -> Optional[List[int]]:
|
||||
tabs = self.element_cache.get("tabs", {})
|
||||
if tab_name in tabs:
|
||||
return tabs[tab_name]["center"]
|
||||
return None
|
||||
|
||||
def get_element_coord(self, page: str, element: str) -> Optional[List[int]]:
|
||||
page_data = self.pages.get(page, {})
|
||||
elements = page_data.get("elements", {})
|
||||
if element in elements:
|
||||
return elements[element].get("center")
|
||||
return None
|
||||
|
||||
def get_path(self, path_name: str) -> Optional[NavPath]:
|
||||
return self.paths.get(path_name)
|
||||
|
||||
def record_page(self, page_name: str, elements: Dict[str, Dict]):
|
||||
"""录制/更新一个页面的元素数据"""
|
||||
self.pages[page_name] = {
|
||||
"recorded_at": time.strftime("%Y-%m-%d %H:%M"),
|
||||
"elements": elements,
|
||||
}
|
||||
self.save()
|
||||
logger.info(f"页面 [{page_name}] 已录制: {len(elements)} 个元素")
|
||||
|
||||
def record_path_result(self, path_name: str, success: bool):
|
||||
"""记录路径执行结果"""
|
||||
path = self.paths.get(path_name)
|
||||
if path:
|
||||
if success:
|
||||
path.success_count += 1
|
||||
path.last_success = time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
else:
|
||||
path.fail_count += 1
|
||||
self.save()
|
||||
|
||||
def update_element_coord(self, page: str, element: str,
|
||||
new_center: List[int]):
|
||||
"""自动纠错：更新元素坐标"""
|
||||
if page not in self.pages:
|
||||
self.pages[page] = {"elements": {}}
|
||||
if "elements" not in self.pages[page]:
|
||||
self.pages[page]["elements"] = {}
|
||||
if element not in self.pages[page]["elements"]:
|
||||
self.pages[page]["elements"][element] = {}
|
||||
old = self.pages[page]["elements"][element].get("center")
|
||||
self.pages[page]["elements"][element]["center"] = new_center
|
||||
self.pages[page]["elements"][element]["updated_at"] = \
|
||||
time.strftime("%Y-%m-%d %H:%M")
|
||||
self.save()
|
||||
logger.info(f"坐标纠错: [{page}].{element} {old} -> {new_center}")
|
||||
|
||||
|
||||
class OperationLogger:
|
||||
"""操作日志记录器——每次操作自动记录步骤/耗时/结果"""
|
||||
|
||||
def __init__(self, device_id: str = ""):
|
||||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
self.device_id = device_id
|
||||
self.log_file = LOG_DIR / f"ops_{time.strftime('%Y%m%d')}.jsonl"
|
||||
self._current_op = None
|
||||
|
||||
def start_operation(self, action: str, params: dict):
|
||||
self._current_op = {
|
||||
"action": action,
|
||||
"params": {k: v for k, v in params.items()
|
||||
if k not in ("password", "pwd", "new_pwd", "old_pwd")},
|
||||
"device": self.device_id,
|
||||
"started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"start_ts": time.time(),
|
||||
"steps": [],
|
||||
}
|
||||
|
||||
def log_step(self, step_type: str, detail: str = "",
|
||||
coords: List[int] = None, success: bool = True):
|
||||
if not self._current_op:
|
||||
return
|
||||
self._current_op["steps"].append({
|
||||
"type": step_type,
|
||||
"detail": detail,
|
||||
"coords": coords,
|
||||
"success": success,
|
||||
"ts": round(time.time() - self._current_op["start_ts"], 2),
|
||||
})
|
||||
|
||||
def end_operation(self, success: bool, result: dict = None,
|
||||
error: str = None):
|
||||
if not self._current_op:
|
||||
return
|
||||
self._current_op["success"] = success
|
||||
self._current_op["duration_ms"] = round(
|
||||
(time.time() - self._current_op["start_ts"]) * 1000)
|
||||
self._current_op["ended_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
if result:
|
||||
self._current_op["result_summary"] = str(result)[:200]
|
||||
if error:
|
||||
self._current_op["error"] = error
|
||||
|
||||
del self._current_op["start_ts"]
|
||||
|
||||
try:
|
||||
with open(self.log_file, "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(self._current_op, ensure_ascii=False) + "\n")
|
||||
except Exception as e:
|
||||
logger.warning(f"写入操作日志失败: {e}")
|
||||
|
||||
op = self._current_op
|
||||
self._current_op = None
|
||||
return op
|
||||
|
||||
def get_recent_ops(self, limit: int = 20) -> List[dict]:
|
||||
"""读取最近的操作记录"""
|
||||
if not self.log_file.exists():
|
||||
return []
|
||||
ops = []
|
||||
try:
|
||||
with open(self.log_file, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
ops.append(json.loads(line))
|
||||
except Exception:
|
||||
pass
|
||||
return ops[-limit:]
|
||||
|
||||
def get_action_stats(self) -> Dict[str, Dict]:
|
||||
"""统计每个 action 的成功率和平均耗时"""
|
||||
ops = self.get_recent_ops(1000)
|
||||
stats: Dict[str, Dict] = {}
|
||||
for op in ops:
|
||||
action = op.get("action", "unknown")
|
||||
if action not in stats:
|
||||
stats[action] = {"total": 0, "success": 0,
|
||||
"total_ms": 0, "errors": []}
|
||||
stats[action]["total"] += 1
|
||||
if op.get("success"):
|
||||
stats[action]["success"] += 1
|
||||
stats[action]["total_ms"] += op.get("duration_ms", 0)
|
||||
if op.get("error"):
|
||||
stats[action]["errors"].append(op["error"][:50])
|
||||
|
||||
for action, s in stats.items():
|
||||
s["success_rate"] = (
|
||||
f"{s['success']/s['total']*100:.0f}%" if s["total"] else "N/A")
|
||||
s["avg_ms"] = (
|
||||
s["total_ms"] // s["total"] if s["total"] else 0)
|
||||
return stats
|
||||
Reference in New Issue
Block a user