提示：文章寫完后，目錄可以自動生成，如何生成可參考右邊的幫助文檔

---

前言

提示：這里可以添加本文要記錄的大概內容：

爬蟲案例七協程爬取視頻

---

提示：以下是本篇文章正文內容，下面案例可供參考

一、Python協程爬取視頻

"""
網址:https://www.tpua.vip/play/84942-1-1.html
目標:爬取視頻
"""
import requests
import re
from urllib.parse import urljoin
import aiohttp
import aiofiles
import asyncio
import os
import subprocess
head = {"":""#自行添加自己的頭
}
def get_m3u8_url():url = "https://www.tpua.vip/play/84942-1-1.html"session = requests.session()session.headers = {"":""#自行添加自己的頭}resp = session.get(url)obj = re.compile(r'"player":"\\/public\\/","url":"(?P<url>.*?)"')m3u8_url = obj.search(resp.text).group("url").replace("\\","")print(m3u8_url)return m3u8_urldef download_m3u8(url):session = requests.session()session.headers =  {"":""#自行添加自己的頭}m3u8_resp = session.get(url)# # 保存m3u8with open('index.m3u8',mode="w",encoding='utf-8') as f:f.write(m3u8_resp.text)
def has_next_m3u8():with open("index.m3u8",mode="r",encoding="utf-8") as f:for line in f:if line.startswith("#EXT-X-STREAM-INF"):return f.readline().strip()return Falseasync def download_one(ts_url,file_name,sem):print(f"{file_name},開始下載")# 設置并發量async with sem:# 下載# 設置超時時間async with aiohttp.ClientSession(headers=head) as sess:async with sess.get(ts_url) as resp:content = await resp.content.read()async with aiofiles.open(f"./source/{file_name}", mode="wb") as f:await f.write(content)print(f"{file_name},下載完畢")
async def download_all_ts(m3u8_url_2):# 信號量,控制并發量sem = asyncio.Semaphore(10)tasks = []i = 1with open("index.m3u8",mode='r',encoding='utf-') as f:for line in f:if i == 10:breakline = line.strip()if line.startswith("#"):continueif not line.startswith("https"):line = urljoin(m3u8_url_2, line)print(line)# 去下載一個tst = asyncio.create_task(download_one(line,f"{i}.ts",sem))tasks.append(t)i += 1await asyncio.wait(tasks)def create_list():input_folder = './source'  # 替換為你的 .ts 文件文件夾路徑# merge_ts_files(input_folder, output_file)# 獲取所有 .ts 文件并排序ts_files = sorted([os.path.join(input_folder, f) for f in os.listdir(input_folder) if f.endswith('.ts')])if not ts_files:print("未找到 .ts 文件！")# 創建一個臨時文件，列出所有 .ts 文件的路徑file_list_path = os.path.join(input_folder, 'file_list.txt')with open(file_list_path, 'w') as f:for ts_file in ts_files:ts_name = ts_file.split("\\")[-1]# print(ts_name)f.write("file " + ts_name+"\n")
def main():# 提取m3u8的urlm3u8_url = get_m3u8_url()print("提取的m3u8_url",m3u8_url)# 下載m3u8文件download_m3u8(m3u8_url)print("下載m3u8_url文件",m3u8_url)# 是否有下一層m3u8m3u8_url_2 = has_next_m3u8()print("是否有下一層next_m3u8_url",m3u8_url_2)while m3u8_url_2:# 有下一層拼接urlm3u8_url = urljoin(m3u8_url,m3u8_url_2)# 下載文件download_m3u8(m3u8_url)print("下載m3u8",m3u8_url)# 判斷是否還要下一層m3u8_url_2 = has_next_m3u8()print("是否還有下一層",m3u8_url_2)# 異步下載tsloop = asyncio.get_event_loop()loop.run_until_complete(download_all_ts(m3u8_url))if __name__ == '__main__':main()create_list()# cmd 上輸入命令合并視頻 ffmpeg -f concat -i file_list.txt -c copy output.mp4 # ffmpeg需要自行去下載，并將其bin路徑添加到path的環境變量中

流程是在源代碼里要找到并用re提取出m3u8的url，并異步去.ts文件，當然有的可能不是.ts，可能是其他的結尾形式如:.jpeg，最后使用ffmpeg合并.ts文件,我并沒有全部對.ts文件全部爬取，只是爬取了前10個.ts文件，并進行了視頻合并。

---