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

news2025/3/10 9:06:15

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

文章目录

前言
一、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_url

def download_m3u8(url):
    session = requests.session()
    session.headers =  {
		"":""#自行添加自己的头
    }

    m3u8_resp = session.get(url)
    # # 保存m3u8
    with 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 False

async 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 = 1
    with open("index.m3u8",mode='r',encoding='utf-') as f:
        for line in f:
            if i == 10:
                break
            line = line.strip()
            if line.startswith("#"):
                continue
            if not line.startswith("https"):
                line = urljoin(m3u8_url_2, line)
            print(line)
            # 去下载一个ts
            t = asyncio.create_task(download_one(line,f"{i}.ts",sem))
            tasks.append(t)
            i += 1

        await 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的url
    m3u8_url = get_m3u8_url()
    print("提取的m3u8_url",m3u8_url)
    # 下载m3u8文件
    download_m3u8(m3u8_url)
    print("下载m3u8_url文件",m3u8_url)
    # 是否有下一层m3u8
    m3u8_url_2 = has_next_m3u8()
    print("是否有下一层next_m3u8_url",m3u8_url_2)
    while m3u8_url_2:
        # 有下一层拼接url
        m3u8_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)

    # 异步下载ts
    loop = 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的环境变量中