zw/iOSAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

合并代码后提交

Browse Source
This commit is contained in:
zw
2025-08-13 20:20:13 +08:00
parent cd61c2138a
commit 33610b27b0
6 changed files with 418 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

299
script/ScriptManager.py
View File

@@ -1,4 +1,5 @@
import random
import re
import threading
import time
from enum import Enum
@@ -7,7 +8,7 @@ import os
from Utils.AiUtils import AiUtils
from Utils.ControlUtils import ControlUtils
from Utils.LogManager import LogManager
from Entity.Variables import anchorList, removeModelFromAnchorList, prologueList
from Entity.Variables import anchorList, removeModelFromAnchorList, prologueList, anchorWithSession
from Utils.Requester import Requester
@@ -199,24 +200,12 @@ class ScriptManager():
# 返回上一步
def goBack(count):
# 如需回复消息
if needReply:
count += 1
for i in range(count):
ControlUtils.clickBack(session)
time.sleep(2)
# 循环条件。1、 循环关闭 2、 数据处理完毕
while not event.is_set() or len(anchorList) > 0:
if needReply:
print("如果需要回复主播消息。走此逻辑")
if AiUtils.getUnReadMsgCount(session) > 0:
# 此处调用另外一个方法处理
# 设置查找深度
session.appium_settings({"snapshotMaxDepth": 15})
# 点击搜索按钮
ControlUtils.clickSearch(session)
# 查找输入框
input = session.xpath('//XCUIElementTypeSearchField')
@@ -237,15 +226,18 @@ class ScriptManager():
# 切换UI查找深度
session.appium_settings({"snapshotMaxDepth": 25})
# 定位 "关注" 按钮 通过关注按钮的位置点击主播首页
follow_button = session.xpath('//XCUIElementTypeOther[@name="zhang1231511"]//XCUIElementTypeButton[@enabled="true" and @visible="true"]')
follow_button = session.xpath("//XCUIElementTypeButton[@traits='Button' and @index='1']")
time.sleep(2)
if follow_button.exists:
print(follow_button.bounds)
# session.appium_settings({"snapshotMaxDepth": 10})
print("找到关注按钮！")
x = follow_button.bounds.x - 200
x = follow_button.bounds.x - 100
y = follow_button.bounds.y
print(x, y)
client.click(x, y)
print("进入主播首页啦")
else:
goBack(1)
removeModelFromAnchorList(anchor)
@@ -253,98 +245,295 @@ class ScriptManager():
continue
time.sleep(3)
session.appium_settings({"snapshotMaxDepth": 25})
time.sleep(2)
# 找到并点击第一个视频
cellClickResult = ControlUtils.clickFirstVideoFromDetailPage(session)
time.sleep(2)
# 观看主播视频
def viewAnchorVideo():
print("开始查看视频")
count = 2
print("开始查看视频，并且重新调整查询深度")
session.appium_settings({"snapshotMaxDepth": 5})
count = 3
while count != 0:
time.sleep(5)
print("条件满足，继续查看")
img = client.screenshot()
time.sleep(1)
filePath = f"resources/{udid}/bgv.png"
img.save(filePath)
LogManager.info("保存屏幕图像成功", udid)
print("保存了背景图")
time.sleep(2)
# 查找add图标
addX, addY = AiUtils.findImageInScreen("add", udid)
if addX != -1:
r = ControlUtils.clickLike(session, udid)
# 点赞成功。总数量减1，否则就滑到下一个视频。原因太多。不好判断。不如直接下一个视频
if r == True:
count -= 1
else:
client.swipe_up()
else:
print("没找有效视频")
client.swipe_up()
break
r = ControlUtils.clickLike(session, udid)
# 点赞成功。
if r == True:
count -= 1
# 假装看几秒视频
time.sleep(5)
client.swipe_up()
if count == 0:
ControlUtils.clickBack(session)
break
# 右滑返回
client.swipe_right()
# 如果打开视频失败。说明该主播没有视频
if cellClickResult == True:
print("点击了视频")
session.appium_settings({"snapshotMaxDepth": 15})
time.sleep(2)
print("重新设置了匹配深度")
# 观看主播视频
LogManager.info("去查看主播视频",udid)
viewAnchorVideo()
time.sleep(2)
print("视频看完了")
time.sleep(3)
LogManager.info("视频看完了，重置试图查询深度", udid)
session.appium_settings({"snapshotMaxDepth": 25})
# 点击关注按钮
followButton = AiUtils.getFollowButton(session)
if followButton.exists:
if followButton is not None:
LogManager.info("找到关注按钮了", udid)
followButton.click()
else:
goBack(2)
return
LogManager.info("没找到关注按钮", udid)
removeModelFromAnchorList(anchor)
goBack(3)
continue
time.sleep(2)
AiUtils.getSendMesageButton(session)
msgButton = AiUtils.getSendMesageButton(session)
time.sleep(2)
if msgButton is not None:
print("找到发消息按钮了")
# 进入聊天页面
msgButton.click()
else:
print("没有识别出发消息按钮")
removeModelFromAnchorList(anchor)
goBack(3)
continue
time.sleep(3)
# 查找聊天界面中的输入框节点
chatInput = session.xpath("//*[className='XCUIElementTypeTextView']")
chatInput = session.xpath("//TextView")
if chatInput.exists:
print("找到了")
print("找到输入框了, 准备发送一条打招呼消息")
# 准备打招呼的文案
# text = random.choice(prologueList)
text = "你好"
# 翻译成主播国家的语言
msg = Requester.translation(text, "法国")
# 准备发送一条信息
chatInput.click()
time.sleep(2)
text = random.choice(prologueList)
msg = Requester.translation(text, anchorCountry)
# 发送消息
chatInput.set_text(msg + "\n")
time.sleep(1)
else:
print("无法发送信息，换个主播")
# 接着下一个主播
goBack(3)
print("无法发送信息")
# 接着下一个主播
removeModelFromAnchorList(anchor)
goBack(4)
else:
print("没找到主播的第一个视频")
print(f"{anchor.anchorId}：该主播没有视频")
# 删除当前数据
removeModelFromAnchorList(anchor)
goBack(2)
goBack(3)
continue
# 设置查找深度
session.appium_settings({"snapshotMaxDepth": 15})
time.sleep(2)
if needReply:
print("如果需要回复主播消息。走此逻辑")
if AiUtils.getUnReadMsgCount(session) > 0:
# 执行回复消息逻辑
self.monitorMessages(session, udid)
homeButton = AiUtils.findHomeButton(udid)
if homeButton.exists:
homeButton.click()
else:
ControlUtils.closeTikTok(session, udid)
time.sleep(2)
ControlUtils.openTikTok(session, udid)
time.sleep(3)
# 执行完成之后。继续点击搜索
session.appium_settings({"snapshotMaxDepth": 15})
# 点击搜索按钮
ControlUtils.clickSearch(session)
else:
session.appium_settings({"snapshotMaxDepth": 15})
# 点击搜索按钮
ControlUtils.clickSearch(session)
# 检查未读消息并回复
def monitorMessages(self, session, udid):
ControlUtils.closeTikTok(session, udid)
time.sleep(2)
ControlUtils.closeTikTok(session, udid)
time.sleep(3)
session.appium_settings({"snapshotMaxDepth": 7})
el = session(xpath='//XCUIElementTypeButton[@name="a11y_vo_inbox"]')
# 如果收件箱有消息 则进行点击
if el.exists:
m = re.search(r'(\d+)', el.label) # 抓到的第一个数字串
count = int(m.group(1)) if m else 0
if count:
el.click()
time.sleep(3)
session.appium_settings({"snapshotMaxDepth": 22})
while True:
el = session(xpath='//XCUIElementTypeButton[@name="a11y_vo_inbox"]')
m = re.search(r'(\d+)', el.label) # 抓到的第一个数字串
count = int(m.group(1)) if m else 0
print("count", count)
if not count:
break
# 新粉丝
xp_new_fan_badge = (
"//XCUIElementTypeCell[.//XCUIElementTypeLink[@name='新粉丝']]"
"//XCUIElementTypeStaticText[string-length(@value)>0 and translate(@value,'0123456789','')='']"
)
# 活动
xp_activity_badge = (
"//XCUIElementTypeCell[.//XCUIElementTypeLink[@name='活动']]"
"//XCUIElementTypeStaticText[string-length(@value)>0 and translate(@value,'0123456789','')='']"
)
# 系统通知
xp_system_badge = (
"//XCUIElementTypeCell[.//XCUIElementTypeLink[@name='系统通知']]"
"//XCUIElementTypeStaticText[string-length(@value)>0 and translate(@value,'0123456789','')='']"
)
# 用户消息
xp_badge_numeric = (
'//XCUIElementTypeOther[@name="AWEIMChatListCellUnreadCountViewComponent"]'
'//XCUIElementTypeStaticText[@value and translate(@value,"0123456789","")=""]'
)
try:
# 如果 2 秒内找不到，会抛异常
badge_text = session.xpath(xp_new_fan_badge).get(timeout=2.0)
val = (badge_text.info.get("value") or
badge_text.info.get("label") or
badge_text.info.get("name"))
print("新粉丝未读数量:", val)
if badge_text:
badge_text.tap()
time.sleep(1)
ControlUtils.clickBack(session)
time.sleep(1)
except Exception:
print("当前屏幕没有找到 新粉丝 未读徽标数字")
badge_text = None
try:
# 如果 2 秒内找不到，会抛异常
badge_text = session.xpath(xp_activity_badge).get(timeout=2.0)
val = (badge_text.info.get("value") or
badge_text.info.get("label") or
badge_text.info.get("name"))
print("活动未读数量:", val)
if badge_text:
badge_text.tap()
time.sleep(1)
ControlUtils.clickBack(session)
time.sleep(1)
except Exception:
print("当前屏幕没有找到 活动 未读徽标数字")
badge_text = None
try:
# 如果 2 秒内找不到，会抛异常
badge_text = session.xpath(xp_system_badge).get(timeout=2.0)
val = (badge_text.info.get("value") or
badge_text.info.get("label") or
badge_text.info.get("name"))
print("系统通知未读数量:", val)
if badge_text:
badge_text.tap()
time.sleep(1)
ControlUtils.clickBack(session)
time.sleep(1)
except Exception:
print("当前屏幕没有找到 系统通知 未读徽标数字")
badge_text = None
try:
# 如果 2 秒内找不到，会抛异常
badge_text = session.xpath(xp_badge_numeric).get(timeout=2.0)
val = (badge_text.info.get("value") or
badge_text.info.get("label") or
badge_text.info.get("name"))
print("用户未读数量:", val)
if badge_text:
badge_text.tap()
time.sleep(3)
xml = session.source()
msgs = AiUtils.extract_messages_from_xml(xml)
# 检测出对方发的最后一条信息
last_msg_text = next(item['text'] for item in reversed(msgs) if item['type'] == 'msg')
# 向ai发送信息
# 获取主播的名称
anchor_name = AiUtils.get_navbar_anchor_name(session)
# 找到输入框
sel = session.xpath(
"//XCUIElementTypeTextView[@name='消息...' or @label='消息...' or @value='消息...']")
if anchor_name not in anchorWithSession:
# 如果是第一次发消息(没有sessionId的情况)
response = Requester.chatToAi({"msg": last_msg_text})
aiResult = response['result']
sessionId = response['session_id']
anchorWithSession[anchor_name] = sessionId
# 找到输入框，输入ai返回出来的消息
if sel.exists:
sel.click() # 聚焦
time.sleep(1)
sel.clear_text()
sel.set_text(aiResult + "\n")
else:
# 如果不是第一次发消息（证明存储的有sessionId）
sessionId = anchorWithSession[anchor_name]
response = Requester.chatToAi({"msg": last_msg_text, "sid": sessionId})
aiResult = response['result']
if sel.exists:
sel.click() # 聚焦
time.sleep(1)
sel.clear_text()
sel.set_text(aiResult + "\n")
time.sleep(1)
# 返回
ControlUtils.clickBack(session)
except Exception:
print("当前屏幕没有找到 用户 未读徽标数字")
badge_text = None
def test(self, udid):
client = wda.USBClient(udid)
session = client.session()
session.appium_settings({"snapshotMaxDepth": 10})
print(client.source())
# manager = ScriptManager()
# manager.test("eca000fcb6f55d7ed9b4c524055214c26a7de7aa")
# manager.growAccount("eca000fcb6f55d7ed9b4c524055214c26a7de7aa")