爬虫案例七Python协程爬取视频

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

前言
一、Python协程爬取视频

前言

提示：这里可以添加本文要记录的大概内容：

爬虫案例七协程爬取视频

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

一、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的环境变量中