selenium爬取某音的最新视频（多博主）
仅限学习交流
1.需求
隔断时间是获取自己关注列表中，发布时间小于5分钟的视频
2.系统环境
2.1 linux系统
-
服务器配置信息：
-
系统：CentOs 8.2 64bit
-
配置：2核|1Gib（实际内存800MB左右）
-
系统：CentOs 7.9 64.bit
-
配置：2核4Gib
错误现象1：
原因：
-
不同的WebDriver对运行内存需求时不一样的
-
对于简单的网页访问和抓取任务，一个Webdriver实例可能需要几百MBh和1GB左右的内存
-
对于复杂的网页或需要长时间运行的任务，内存的需求可能会增加到几GB
-
如果并发运行多个WebDriver实例，内存需求将按照实例数量线性增长。
尝试：优化内存的使用
-
定期清理内存(效果:无效)
- 在任务结束后，使用Webdriver的quit()方法关闭浏览器并释放资源
如果不行就手动进行释放
top #查询linux中的进程。
top -o +%MEM # 按内存使用率的排序
ps aux --sort=-%mem | head -n 10 # 列出内存使用做多的前10个进程
kill PID (进程号)
kill -9 PID(进程号) # 强制结束进程（慎用，可能会导致数据丢失）
# 清除页缓存
sudo sh -c 'echo 1 > /proc/sys/vm/drop_caches '
# 上面的命令的平替：
sudo sysctl -w vm.drop_caches=1
# 清除目录缓存
sudo sh -c 'echo 2 > /proc/sys/vm/drop_caches'
# 上面指令的平替
sudo sysctl -w vm.drop_caches=2
# 同时清理页缓存和目录缓存
sudo sh -c 'echo 3 > /proc/sys/vm/drop_caches'
sudo sysctl -w vm.drop_caches=3
-
优化浏览器设置（尝试：无效）
--headless # 使用无头模式
--incognito # 使用无痕模式：不保存浏览历史和缓存和cookie等
--disable-gpu # 禁用GPU ，有时候可以解决一些兼容性或性能问题
blink-settings=imagesEnabled=false # 禁止图片和视频的加载
--no-sandbox #适用于Chrome ，以最高权限运行浏览器
-
控制并发的数量（有效，但是运行后一段时间，自动崩溃）
总结：自己代码时功能相对复杂一点，并且时运行的多个WebDriver实例的，所以云服务器的资源是无法满足的
如果你配置相对好一点，可以试试接下来的流程。
a.资源包准备
b.部署流程
低版本的谷歌浏览器
第2步可能会遇到的问题：libgtk-3.so.0: cannot open shared object file: No such file or directory Couldn't load XPCOM.（火狐浏览器的是gtk3）谷歌的好像是2.忘记了
yum install gtk3 # (可变动，谷歌的好像不是这个版本)
# libgtk-3.so.0 是 GTK+ 3 库的一部分，它是许多 Linux 桌面应用程序（包括 Firefox）用于创建图形用户界面的工具包
-
安装Chromedriver
# 1.下载Chromedriver
wget https://storage.googleapis.com/chrome-for-testing-public/125.0.6422.141/linux64/chromedriver-linux64.zip # 一般是下载到/root目录,或者你令改其他的位置
# 2.解压
unzip chromedriver-linux64.zip
# 3.进入解压后的目录，并将chromedriver 移动到/usr/bin中
cd chromedriver-linux64
mv chromedriver /usr/bin
# 4.检验
chromedriver --version
-
安装anaconda3
# 1.下载anaconda3的安装包
wget https://repo.anaconda.com/archive/Anaconda3-2024.10-1-Linux-x86_64.sh
# 2.执行安装
sh Anaconda3-2024.10-1-Linux-x86_64.sh
# 3.设置环境变量
vim ~/.bashrc # 进入改文件中，在最后一行添加 export PATH=/root/anaconda3/bin:$PATH
source ~/.bashrc #更新可以进行使用了
可能出现的问题：source ：not found 说明正在使用的 shell (/bin/sh) 不支持 source 命令
# 执行bash 指令，再运行source命令
bash # 切换
-
conda创建爬虫脚本的运行环境
# 1.创建python环境，其中dy,是名字自己可以更改
conda create -n dy -c main python=3.12
# 2.进行环境的激活
conda activate dy
可能出现的问题：我们无法执行conda activate face ，让我们去执行conda init，执行之后还是不能够执行
bash # 输入进行shell切换，在执行conda activate dy
2.2 window系统
-
配置信息
- 系统：windows11
- 处理器：12th Gen Intel(R) Core(TM) i5-12600KF 3.70 GHz (10核16线程)
- 内存：32GB
a.资源包准备：
-
资源目录（以谷歌浏览器为例）
-
谷歌浏览器安装包
-
与谷歌浏览器对应版本的chromedriver
-
pycharm
b.部署流程
省略...，简单的一批
3.代码
a.需求：
隔断时间是获取自己关注列表中，发布时间小于5分钟的视频
b.分析：
- 基础流程：
c.代码中的函数
c1.获取本地excel中的url数据
- 【博主主页url的数据】：我存放在本地的excel中，我们需要进行读取。第一列是编号，第二列是名字，第三列是博主主页的url
- 其中我们主要的就是要编号和博主主页的url，编号的目的：方便我们进行区分和过筛。博主主页的url,是必须存在的。
- 为什么存放在字典当中，方便我们对其过筛的一种思路
- 【代码】：
from openpyxl import load_workbook # 操作excel表中的导入
def convert_excel_data_to_dict_data(path,i)
"""
利用openpyxl获取本地excel表中A列和C列
利用dict函数和zip函数，分别将两个数据进行组合成为字典类型的数据
返回：dict_data 该数据，方便接下来的调用 """
wb = load_workbook(path) # 加载数据
ws = wb.worksheets[i] # 选择工作表 传入i的参数，方便我们多博主的并行爬取。
print(f"已经获取表格{i}的数据")
for A in ws.iter_cols(min_row=1, max_row=ws.max_row, min_col=1, max_col=1, values_only=True):
print("读取Excel中A列数据成功.....")
for C in ws.iter_cols(min_row=1, max_row=ws.max_row, min_col=3, max_col=3, values_only=True):
print("读取Excel中C列数据成功.....")
# 方式一：使用字典推导式
dict_data = {key: value for key, value in zip(A, C)}
return dict_data
-
【知识补充】：
-
工作表中的迭代方法iter_cols() ,返回的是一个元组。
-
ws的属性max_row，最大的行数。
-
zip函数：内置函数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。如果长度对照，则返回列表长度与最短的对象相同。利用 * 号操作符，可以将元组解压为列表
c2.获取博主主页的最近20条左右的视频(代码不全，只展示函数)
-
由于抖音web页面采用的异步加载，所以我们使用selenium进行爬取
-
进入博主的主页(我将其封装成了一个类了。好进行统一调用)
def enter_homepage(self,url):
self.driver.get(url)
time.sleep(3) # 常规休息3秒。模拟真实用户，防止被反爬检测。
-
存在登录窗口（我们需要取消登录窗口）
def log_out(self,timeout=5)
"""
定位取消按钮元素，进行取消。
特定情况分析：会存在不一样的登录窗口，注意补充。"""
try:
# 尝试第一个取消按钮
login_frame=WebDriverWait(self.driver,timeout).until(EC.presence_of_element_located((By.XPATH, '//div[@class="douyin-login__close dy-account-close"]')))
login_frame.click()
# 尝试第二个取消按钮
except TimeoutException:
login_frame = WebDriverWait(self.driver,timeout).until(EC.presence_of_element_located((By.XPATH, '//div[@class="tPz8yDXt"]')))[0]
login_frame.click()
# 尝试最后一个备选按钮
except Exception as e:
login_frame = WebDriverWait(self.driver,timeout).until(EC.presence_of_element_located((By.XPATH, '//div[@class="lEGqAh8b"]')))
login_frame.click()
print(f"{e}")
该代码可以优化为一下代码，但是没有办事尝试封装成一个了。
def click_element_by_xpath(driver, xpath, timeout=5):
try:
element = WebDriverWait(driver, timeout).until(EC.presence_of_element_located((By.XPATH, xpath)))
element.click()
except TimeoutException:
print(f"Timed out waiting for element with XPath: {xpath}")
# 可以选择抛出异常或进行其他处理
try:
# 尝试点击第一个关闭按钮
click_element_by_xpath(self.driver, '//div[@class="douyin-login__close dy-account-close"]')
except TimeoutException:
# 如果第一个失败，尝试点击第二个关闭按钮
click_element_by_xpath(self.driver, '//div[@class="tPz8yDXt"]')
except Exception as e:
# 如果所有尝试都失败，尝试点击最后的备选按钮，并打印异常信息
click_element_by_xpath(self.driver, '//div[@class="lEGqAh8b"]')
print(f"An unexpected error occurred: {e}")
-
获取主页中第一个视频（优化跳过置顶视频）
def get_frist_video(self):
""" 1.解析该博主主页中显示的所有视频链接
返回的是一个元素对象list,我们通过for遍历其属性，也就是其中的url，用一个list存储url
2.跳过置顶视频 目的是进行赛选，较少时间
通过获取 《置顶 标签》，然后进行跳过
3.返回目标视频的url，并点击进入（模拟真人）"""
# 1.解析该博主的所有视频连接
a_element_list = WebDriverWait(self.driver, 10).until(
EC.presence_of_all_elements_located((By.XPATH, "//div[@class='pCVdP6Bb']/ul/li/div/a")))
complete_url = []
for a_url_element in a_element_list:
url = a_url_element.get_attribute("href")
complete_url.append(f"{url}")
# 2.跳过置顶的视频 (如果含有”置顶的标签“，则不获取该url)
# top_tag_list = content.xpath("//div[@class='semi-tag-content semi-tag-content-ellipsis']") # 获取置顶的标签的视频
try:
top_tag_list = WebDriverWait(self.driver, 5).until(
EC.presence_of_all_elements_located(
(By.XPATH, "//div[@class='semi-tag-content semi-tag-content-ellipsis']")))
a = len(top_tag_list) - 1
frist_video_url = complete_url[a]
a_element_list[a].click()
except:
frist_video_url = complete_url[0]
a_element_list[0].click()
return frist_video_url
c3.获取目标视频的无水印url
- 我在主页进入的视频是没办法获取目标视频的（只存在第4个视频中的url）。只能通过主页中获取的详情页的list中url,才能获取。
def get_video_url(self, frist_video_url):
"""
由于点击之后无法进行获取，只能进入通过获取主页list中url,进入详情页获取
可能出现的情况：
1.会出现找不到相关指定的视频，所以我们尽量少获取（在时间符合要求的情况下进行获取，并进行一定反馈（包括异常反馈））"""
self.driver.get(frist_video_url)
try:
complete_video_url = WebDriverWait(self.driver, 5).until(
EC.presence_of_all_elements_located((By.XPATH, r"//xg-video-container/video/source")))[0].get_attribute(
"src")
return complete_video_url
except Exception as e:
print(f"出现图文,不执行获取,或者出现异常{e}")
return None
c4.获取目标发布时间的视频
def get_local_video_url(self, frist_video_url, dict_data, key):
"""
1.获取发布时间文本
真实情况下:通过点击进入视频页面中总共存在两个视频或者三个视频的数据
2.利用re进行数据清洗，保留发布的时间数据
3.根据发布时间置顶规则。发布时间小于5分钟的进行本地url的获取(无水印下载)
4.返回print_str 可能是local_video_url(无水印的url)，也可能返回不符合的字符串。"""
try:
video_create_time = WebDriverWait(self.driver, 5).until(EC.presence_of_all_elements_located((By.XPATH, "//div[@class='video-create-time']/span")))
except Exception as e:
print(f"{e}")
finally:
"""一般是三个"""
"""或者是两个"""
if len(video_create_time) == 2:
scrapy_time = video_create_time[0].text
if len(video_create_time) == 3:
scrapy_time = video_create_time[1].text
# 赛选时间
if scrapy_time.startswith('钟前', -2):
print(scrapy_time)
a = re.search(pattern=r"\d+", string=scrapy_time).group()
if int(a) <= 5:
local_video_url = self.get_local_video_url(frist_video_url)
# print(f"执行爬取,视频url:{local_video_url}")
print_str = local_video_url
return print_str
else:
# 大于5小时的视频，如果不加返回值，可能会返回NoneType
print_str = "不符合要求"
return print_str
else:
print_str = "不符合要求"
return print_str
c5.筛选功能
-
当我们获取一个符合我们要求的视频后，不再进行获取。（一般博主一天或者多天才发布一个视频，所以一直爬取是没有必要的。）
def screening(self, keys_list, key):
"""
如果获取该播主的视频之后，不在获取该博主的视频。
由于博主大约一天之获取一个视频，所以我们进行过滤一下
我们将博主url进行字典进行存储，进行了编号。
如果获取了博主的视频，我们就减少了list对应的值，映射出字典中的不获取对应的值。
字典中的key值，与list中的值进行相关联。如果获取了博主的，也就是获取其中的key值，然后讲list中的key进行删除
:return:"""
"""所以我们需要给出一个指定的keys_list"""
# 执行移除的操作
if key in keys_list: # 【执行视频获取的等替】
keys_list.remove(key)
return keys_list
c6.主函数
def processing_main(i)
# 将本地的数据写入字典当中
helper = SeleniumHelper()
dict_data = helper.convert_Excel_data_to_dict_data(i)
# 获取字典中的所有建，并形成了一个list,方便我们进行赛选，
# 如果封装到一个screening的函数中，就是失去了过滤的效果。每次都是重新输入dict_data的数据。
keys = dict_data.keys()
keys_list = list(keys)
while True:
for key in keys_list:
url = dict_data.get(key) # 获取博主主页的url
helper.enter_homepage(url) # 通过webdriver进入博主主页
helper.log_out() # 取消登录窗口
frist_video_url = helper.get_frist_video() # 获取最新发布的视频
print_str = helper.get_video_time(frist_video_url, dict_data, key)
if len(print_str) > 5:
# 根据获取的时间创建文件夹并进行保存。
current_time = datetime.now()
hour_data = current_time.strftime('%H')
date_data = current_time.strftime('%m-%d')
time_data = current_time.strftime('%H:%M:%S')
if os.path.exists(f"./file/{date_data}") is False:
os.makedirs(f"./file/{date_data}")
with open(f"./file/{date_data}/{hour_data}.txt", 'a', encoding="utf-8") as f:
f.write(f"{time_data}-{print_str}\n")
keys_list = helper.screening(keys_list, key) # 筛选，自动删除已经获取过的博主编号
print(f"第{i}工作表中的，第{key}个博主视频:{print_str}")
print("休息五分钟后，继续爬取")
time.sleep(300) # 休息五分钟
c7.开启多进程
"""
我尝试过开启3个进程，但是会被抖音检测到异常，所以我开启2个进程同时进行爬取。但是爬取的excel表中博主数据是不一致的。
未加入Ip代{过}{滤}理
未加入浏览器的user_agent信息"""
if __name__ == "__main__":
processes = []
for i in range(0, 2):
p = multiprocessing.Process(target=processing_main, args=(i,))
processes.append(p)
p.start()
for p in processes:
p.join()
print("all worker finished")
4.优化过程
a.优化1：
添加功能【判断发布的时间】：判断是否当前时间相差不到5分钟的博主，或者是与当下时间相同的发布时间对应上的。
**思路***
-
先判断是否当前的视频时间与当下时间是否相同，如果相同就执行获取，并将设置一个状态，当前已经爬取成功。1天之内不在进行博主视频的爬取。
正常情况，一个博主，一天都是一个视频，该博主爬取成功后，不在进行该博主的爬取。
-
怎么实现设置状态【等替的方法】
0，1开关，在excel文档中，第三列设置0，1开关。默认为0
如果爬取成功，则将其值设置为1。
代码逻辑：首先获取该播主，也就是单行中的数值是否为0，如果为0，就进行获取。
如果发现了发布时间与当下时间相对应，则将数值变为1.
缺点：我需要去修改excel的文件
-
【通过一个变量设置一个状态与博主的url，进行绑定。】
通过字典：进行编号。 利用key的值
如果获取，就将key值放入到一个list中。
然后，其中list的值，key不在list中，则进行获取。 如果在不进行获取。
# 获取字典中所有的key
keys = url_dist.keys()
keys_value = list(keys)
# 通过key获取value
url_dist.get(key)
"值从上一步来，以list为媒介进行删减。"
# 假设已经获取了1，就删除.
new_key_list = list.remove(key)
for key_ in new_key_list:
-
现在缺的是讲excel数据，写入字典当中。(已经完成)
b.优化2：
c.优化3（未设置）
访问抖音页面过多的情况下，会被抖音平台检测到，会进入不到目标视频的详情页中，也就是提取不到目标视频的url。
解决方案：添加ip代{过}{滤}理(付费代{过}{滤}理或者免费代{过}{滤}理)和不同的user-agent
# 不使用ip代{过}{滤}理的方法，相对于上面更加麻烦
将项目部署到linux服务器中，只保存指定博主的信息和获取的时间
本地，根据获取博主的信息，手动或者自动提取信息。
[复制链接]
发表于 2025-1-10 16:04
|
发表于 2025-1-10 16:02
|楼主
分享到朋友圈
