AirtestProject Airtest-Selenium 实操小课②:刷 B 站视频

fishfish-yu · 2024年01月11日 · 2136 次阅读

此文章来源于项目官方公众号:“AirtestProject”\
版权声明:允许转载,但转载必须保留原链接;请勿用作商业或者非法用途

1. 前言

上一课我们讲到用 Airtest-Selenium 爬取网站上我们需要的信息数据,还没看的同学可以戳这里看看~

那么今天的推文,我们就来说说看,怎么实现看 b 站、刷 b 站的日常操作,包括点击暂停,发弹幕,点赞,收藏等操作,仅供大家参考学习~

2.需求分析和准备

整体的需求大致可以分为以下步骤:

  • 打开 chrome 浏览器
  • 打开百度网页
  • 搜索 “哔哩哔哩”
  • 点击进入 “哔哩哔哩” 官网
  • 搜索关键词 “Airtest 酱”
  • 点击进入 “Airtest 酱” 首页,随机点击播放视频
  • 并对视频点击暂停,发弹幕,点赞,收藏

在写脚本之前,我们需要准备好社区版 AirtestIDE(目前最新版为 1.2.16),设置好 chrome.exe 地址和对应的 driver;并且确保我们的 chrome 浏览器版本不是太高以及 selenium 是 4.0 以下即可(这些兼容问题我们都会在后续的版本修复)。

3. 脚本实现与运行效果

3.1 脚本运行效果

我们在编写这次代码的时候,我们主要是使用了页面元素定位的方式去进行操作交互的,除此之外还实现了保存 cookie、读取 cookie 的一个操作。大家在日常使用也会发现,在首次通过脚本开启的 chrome 网页界面是无 cookie 的,那么我们在进行一些任务之前是需要先登录后才能进行下一步操作的,可以通过首次登录时读取 cookie 数据保存到本地,往后每次运行只需要读取本地的 cookie 文件就可以轻松登录啦~

先来看下我们整体的运行效果:

3.2 完整代码分享

这里也附上完整的示例代码给大家参考,有需要的同学可以自取学习哦:

# -- encoding=utf8 --
from airtest.core.api import *
# 引入 selenium 的 webdriver 模块
from airtest_selenium.proxy import WebChrome
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import threading
import time
import random
import json

# 保存以及调用 cookie 的线程
class UtilFunc():
def cookie_is_exist_(self, cook_name='_'): # 检查 txt 文件是否存在
if os.path.exists(f'{cook_name}cookies.txt'):
return True
return False

def cookie_save_(self, driver, cook_name='_'): # 保存 cookie 到 txt 文件中以便下次读取
# 获取当前页面的所有 cookie
cookies = driver.get_cookies()
# 将 cookie 转换为 JSON 字符串
cookies_json = json.dumps(cookies)
# 保存 cookie 到 txt 文件
with open(f'{cook_name}cookies.txt', 'w') as file:
file.write(cookies_json)
print(f"保存 cookies:{cookies}")

def cookie_set_(self, driver, cook_name='_'): # 读取 cookie 文件并给当前网站设置已存 cookie
# 从 txt 文件读取 JSON_cookie 数据
with open(f'{cook_name}cookies.txt', 'r', encoding='gbk') as file:
json_data = file.read()
# 将 JSON 数据转换为列表
data_list = json.loads(json_data)
for cookie in data_list:
driver.add_cookie(cookie)
print("设置 cookie")

# 创建一个实例,代码运行到这里,会打开一个 chrome 浏览器
driver = WebChrome()
isLogin = False # 存储登录状态值,False 为未登录,True 为已登录

# 打开 chrome 浏览器并打开视频播放
def start_selenium():
driver.implicitly_wait(20)
driver.get("https://www.baidu.com/")
# 输入搜索关键词并提交搜索
search_box = driver.find_element_by_name('wd')
search_box.send_keys('哔哩哔哩')
search_box.submit()

try:
# 查找搜索结果中文本为 "哔哩哔哩" 的元素并点击
results = driver.find_elements_by_xpath('//div[@id="content_left"]//span[contains(text(), "哔哩哔哩")]')
if results:
results[0].click()
print("点击了哔哩哔哩搜索结果")
except Exception as e:
element = driver.find_element_by_xpath(
"//div[@id='content_left']/div[@id='1']/div[@class='c-container']/div[1]/h3[@class='c-title t t tts-title']/a")
element.click()
driver.switch_to_new_tab() # 切换界面

util_cookie = UtilFunc()
if util_cookie.cookie_is_exist_("Airtest 酱登录"): # 存在 cookie 文件,设置 cookie
util_cookie.cookie_set_(driver, "Airtest 酱登录")
# 输入搜索关键词并提交搜索
search_box = driver.find_element_by_class_name('nav-search-input')
search_box.send_keys('Airtest 酱')
# 模拟发送 Enter 键
search_box.send_keys(Keys.ENTER)
sleep(5)
driver.switch_to_new_tab() # 切换界面

results_ = driver.find_elements_by_xpath(
'//div[@class="bili-video-card_info--right"]//span[contains(text(),"Airtest 酱")]')
if results
:
results_[0].click()
driver.switch_to_new_tab() # 切换界面

driver.refresh()
sleep(2)
video_ele = driver.find_element_by_xpath("//div[@title='14 天 Airtest 自动化测试小白课程']")
# 滚动到指定元素处
driver.execute_script("arguments[0].scrollIntoView(true);", video_ele)
sleep(5)
video_ele.click()
driver.switch_to_new_tab() # 切换界面

# 获取所有视频
video_list = driver.find_elements_by_xpath("//ul[@class='row video-list clearfix']//a[@class='title']")
random_element = random.choice(video_list)
random_element.click() # 随机播放一个
driver.switch_to_new_tab() # 切换界面

# 登录
def is_login():
"""线程检测登录弹窗"""

def is_no_login(*args):
global isLogin # 在线程内修改外部常量的值
no_login_tip = True
while True:
element = driver.find_elements_by_css_selector('.bili-mini-content-wp')
if len(element) > 0:
if no_login_tip:
print("未登录 请在五分钟内扫码")
no_login_tip = False
else:
print("未检测到登录弹窗")
check_login_ele = driver.find_elements_by_css_selector('.bpx-player-dm-wrap')
if not check_login_ele:
isLogin = True
UtilFunc().cookie_save_(driver, "Airtest 酱登录")
print("保存 cookie")
break
log_text_array = [element.text for element in check_login_ele] # 使用列表推导式简化代码
if "请先登录或注册" in log_text_array:
loginbtn = driver.find_elements_by_xpath(
"//div[@class='bili-header fixed-header']//div[@class='header-login-entry']")
if loginbtn:
loginbtn[0].click()
isLogin = False
print("判断 cookie 文件是否存在,方便下次调用,设置后刷新页面")
else:
isLogin = True
UtilFunc().cookie_save_(driver, "Airtest 酱登录")
print("保存 cookie")
break

thread = threading.Thread(target=is_no_login, args=("args",))
thread.start()

# 暂停播放
def video_pause_and_play(check_btn=False):
if isLogin:
try:
paus_btn = driver.find_elements_by_xpath(
"//[@id=\"bilibili-player\"]//div[@class='bpx-player-ctrl-btn bpx-player-ctrl-play']")
if paus_btn[0]:
detection_time1 = driver.find_elements_by_xpath(
'//
[@class="bpx-player-control-bottom-left"]//div[@class="bpx-player-ctrl-time-label"]')
start_time = detection_time1[0].text
sleep(5)
# 时间戳检测是否在播放
detection_time2 = driver.find_elements_by_xpath(
'//*[@class="bpx-player-control-bottom-left"]//div[@class="bpx-player-ctrl-time-label"]')
end_time = detection_time2[0].text
if start_time == end_time or check_btn:
print("点击播放 (暂停) 按钮")
paus_btn[0].click()
except Exception as e:
print(f"点击播放 (暂停) 出错{e}")

# 发送弹幕
def video_sms(sms_body="不错"):
if isLogin:
try:
sms_input_edit = driver.find_element_by_xpath("//input[@class='bpx-player-dm-input']")
sms_input_edit.send_keys(sms_body)
# 模拟发送 Enter 键
sms_input_edit.send_keys(Keys.ENTER)
except Exception as e:
print(f"发弹幕出错{e}")
print(f"发送弹幕:{sms_body}")

# 点赞
def video_love():
if isLogin:
print("点赞")
try:
sms_input_edit = driver.find_elements_by_xpath(
"//div[@class='toolbar-left-item-wrap']//div[@class='video-like video-toolbar-left-item']")
if not sms_input_edit:
print("已经点赞")
return
sms_input_edit[0].click()
except Exception as e:
print(f"点赞出错{e}")

# 收藏
def video_collect():
if isLogin:
print("收藏")
try:
colle_btn = driver.find_elements_by_xpath(
"//div[@class='toolbar-left-item-wrap']//div[@class='video-fav video-toolbar-left-item']")
if not colle_btn:
print("已经收藏")
return
colle_btn[0].click()
sleep(2)
list_coll = driver.find_elements_by_xpath("//div[@class='group-list']//ul/li/label")
random_element = random.choice(list_coll) # 随机收藏
# 滚动到指定元素处
driver.execute_script("arguments[0].scrollIntoView(true);", random_element)
sleep(2)
random_element.click() # 随机收藏一个
sleep(2)
driver.find_element_by_xpath("//div/button[@class='btn submit-move']").click() # 确认收藏
except Exception as e:
print(f"收藏出错{e}")

# 等待元素出现
def wait_for_element(driver, selector, timeout=60 * 5):
try:
element = WebDriverWait(driver, timeout).until(
EC.presence_of_element_located((By.XPATH, selector))
)
return element
except Exception:
print("元素未出现")
return None

# 头像元素初始化
selem = "//div[@class='bili-header fixed-header']//*[contains(@class, 'header-avatar-wrap--container mini-avatar--init')]"

if name == "main":
start_selenium() # 开启浏览器找到视频播放
is_login() # 检测是否出现登录弹窗
# 等待元素出现
element = wait_for_element(driver, selem)
if element:
print("检测到已经登录")
# 暂停和播放视频
for _ in range(2):
video_pause_and_play()
sleep(3)
driver.refresh()
# 发送弹幕
sms_list = ["感觉不错,收藏了", "666,这么强", "自动化还得看 airtest", "干货呀", "麦克阿瑟直呼内行"]
for item in sms_list:
wait_time = random.randint(5, 10) # 随机生成等待时间,单位为秒
time.sleep(wait_time) # 等待随机的时间
video_sms(item) # 评论

# 点赞和收藏视频
for action in [video_love, video_collect]:
action()
sleep(3)
else:
print("登录超时")

3.2 重要知识点

1)切换新页面并打开新的标签页
driver.switch_to_new_tab()
**2)将随机的元素 random_element 对象的 “顶端” 移动到与当前窗口的 “顶部”** 对齐。
driver.execute_script("arguments[0].scrollIntoView(true);", random_element)

3) 从非空序列中随机选取一个数据并返回,该序列可以是list、tuple、str、set**。**

random.choice()

4) 通过实例化 threading.Thread 类创建线程,target:在线程中调用的对象,可以为函数或者方法;args 为 target 对象的参数。

start():开启线程,如果线程是通过继承 threading.Thread 子类的方法定义的,则调用该类中的 run() 方法;start() 只能调用一次,否则报 RuntimeError。

threading.Thread(target=is_no_login, args=("args",))
thread.start()

5) 使用 expected_conditions 模块(在使用时通常重命名为 EC 模块)去判断特定元素是否存在于页面 DOM 树中,如果是,则返回该元素(单个元素),否则就报错。

EC.presence_of_element_located((By.XPATH, selector))

4. 注意事项与小结

4.1 相关教程

4.2 课程小结

在本周的课程中,我们介绍了如何使用 Airtest-selenium 进行自动化刷 B 站视频的操作流程,也分享了 Airtest-selenium 比较常见的用法。但是,请大家注意,我们的分享仅供学习参考哦!我们分享的代码并不是永远适用的,因为网页的页面元素可能会不断更新。

同时,我们也非常欢迎同学们能够提供自己常用场景的代码,我们会积极分享相关的使用技巧。让我们一起努力,共同进步~


AirtestIDE 下载:airtest.netease.com/\
Airtest 教程官网:airtest.doc.io.netease.com/\
搭建企业私有云服务:airlab.163.com/b2b

官方答疑 Q 群:117973773

暂无回复。
需要 登录 后方可回复, 如果你还没有账号请点击这里 注册