本帖最后由 Yin07 于 2025-8-6 20:33 编辑
目标网站：
[HTML] 纯文本查看 复制代码 https://www.xdf.cn/bbc/yingyu/list_4851_1.html
用到的工具：webdriver
任意浏览器驱动皆可
Edge：[HTML] 纯文本查看 复制代码 https://developer.microsoft.com/en-us/microsoft-edge/tools/webdriver/?form=MA13LH#installation
Firefox ： [HTML] 纯文本查看 复制代码 https://github.com/mozilla/geckodriver/releases/
Chrome：[HTML] 纯文本查看 复制代码 http://chromedriver.storage.googleapis.com/index.html
IE：[HTML] 纯文本查看 复制代码 http://selenium-release.storage.googleapis.com/index.html
Opera：[HTML] 纯文本查看 复制代码 https://github.com/operasoftware/operachromiumdriver/releases
模块安装 selenium
[Python] 纯文本查看 复制代码 pip install selenium
思路分析:
首页数据排列整齐，直接爬取
视频链接地址：
值得注意是视频的加载方式为异步加载，直接使用requests是无法得到视频链接的，
在阻止getsource请求后，视频就不会加载了，并且可以看到与jquery.js有关
这里直接用selenium模拟浏览器操作。
由于需要爬取的视频数量多，可以采用多线程
为了实时观察代码的运行情况，可以在特殊位置打印一下
以下使用Edge进行演示：
模块导入：
[Python] 纯文本查看 复制代码 import time
import requests
import re
from selenium import webdriver
from selenium.webdriver.edge.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as ec
from pyquery import PyQuery as pq
from selenium.webdriver import Edge
from selenium.webdriver.edge.service import Service
from selenium.webdriver.common.action_chains import ActionChains
from bs4 import BeautifulSoup
import os
from threading import Thread
import threading
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
from queue import Empty
selenium的设置：
[Python] 纯文本查看 复制代码 options = Options()
options.add_argument('--headless')
#options.add_experimental_option('detach',True) #保持网页
service = Service(executable_path=r"D:\python\new\msedgedriver.exe")
driver = webdriver.Edge(service=service, options=options)
为了不影响视觉效果，这里使用浏览器的无头模式
驱动地址最好直接复制，再替换
数据的获取：
[Python] 纯文本查看 复制代码 driver.get(urls)
time.sleep(5)
先访问链接，通过5秒的等待让视频能够加载
[Python] 纯文本查看 复制代码 data= driver.page_source
#print(data)
#id=driver.find_element(By.CSS_SELECTOR,'div[id^ ="player-con_component_"]')
link=re.findall('<video webkit-playsinline="" playsinline="" x-webkit-airplay="" x5-playsinline="" preload="preload" autoplay="autoplay" src="(.*?)" style="width: 100%; height: 100%;"></video>',data)
#print(link)
#time.sleep(1)
driver.quit()
使用drive.page_source获取加载5秒后的页面
再用正则表达式获取到视频链接
考虑到有些页面是没有视频的或者是等待的时间过短
写一个判断有助于发现问题所在
[Python] 纯文本查看 复制代码 if not link:
print(title+'链接获取失败，请检查延时是否合理')
with open(r'D:\python\new\errorlist.txt','a',encoding='utf-8') as txt:
txt.write(title +':'+ urls +'\n')
else:
数据保存：
[Python] 纯文本查看 复制代码 source = requests.get(url =link[0])
with open('H:/English/'+title+'.mp4','wb') as f:
f.write(source.content)
写到这里已经是可以运行的
为了代码的美观舒适可以进行一点小小的修饰
先创建几个全局变量用来存储返回的链接与文件名
[Python] 纯文本查看 复制代码
task_queue1 =Queue() #to getlink
task_queue2 =Queue()
task_queue3=Queue() # to download as link
task_queue4=Queue()
将整个代码分为三段
定义获取所有视频链接所在网页链接：
[Python] 纯文本查看 复制代码 def geturllist():
将返回的链接与文件名传输至全局变量中
[Python] 纯文本查看 复制代码 for u , n in zip(links,titles):
task_queue1.put(u)
task_queue2.put(n)
打印一下获取到底链接数量，便于查看
[Python] 纯文本查看 复制代码 print('页面链接收集到',task_queue1.qsize(),'条')
定义获取所有视频链接所属链接：
[Python] 纯文本查看 复制代码 def getlink():
获取全局变量中的链接和文件名并发送请求
[Python] 纯文本查看 复制代码 pageurl = task_queue1.get_nowait()
name = task_queue2.get_nowait()
driver.get(pageurl)
将链接传输至全局变量中
[Python] 纯文本查看 复制代码 task_queue3.put(link)
task_queue4.put(title)
定义视频下载：
[Python] 纯文本查看 复制代码 def download():
遍历视频下载链接和文件名
[Python] 纯文本查看 复制代码 while True:
url = task_queue3.get_nowait()
caption = task_queue4.get_nowait()
主程序
[Python] 纯文本查看 复制代码
if __name__ == '__main__':
thread1 = []
thread2 = []
thread3=[]
getlist = Thread(target=geturllist)
getlist.start()
thread1.append(getlist)
for get in thread1:
getlist.join()
time.sleep(2)
for pool in range(5):
geturl = Thread(target=getlink)
geturl.start()
thread2.append(geturl)
for i in thread2:
geturl.join()
downloaddata = Thread(target=download)
downloaddata.start()
thread3.append(downloaddata)
for down in thread3:
downloaddata.join()
print(f'第{x}页下载完成')
运行效果：
  
源代码：
自动化下载测试.zip
(2.82 KB, 下载次数: 52)
| 
[复制链接]
发表于 2025-8-6 20:21
