网上请人做软件的网站,铝合金型材外发加工网,我的ip地址查询,公司注册地址变更需要多长时间提示#xff1a;文章写完后#xff0c;目录可以自动生成#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、Python协程爬取视频 前言
提示#xff1a;这里可以添加本文要记录的大概内容#xff1a;
爬虫案例七协程爬取视频 提示#xff1a;以下是本篇文章正文… 提示文章写完后目录可以自动生成如何生成可参考右边的帮助文档 文章目录 前言一、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.htmlsession requests.session()session.headers {:#自行添加自己的头}resp session.get(url)obj re.compile(rplayer:\\/public\\/,url:(?Purl.*?))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,modew,encodingutf-8) as f:f.write(m3u8_resp.text)
def has_next_m3u8():with open(index.m3u8,moder,encodingutf-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(headershead) as sess:async with sess.get(ts_url) as resp:content await resp.content.read()async with aiofiles.open(f./source/{file_name}, modewb) 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,moder,encodingutf-) 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文件并进行了视频合并。
