python全自动爬取m3u8网页视频（各类网站都通用）

        当前人工智能，大语言模型的火热，使得python这门编程语言的使用越来越广泛。最近也开始学习了python，发现它在自动化方面的确有得天独厚的优势。python的简单易用，丰富的开源库，完善的生态，使得它有可能成为大语言模型和物理世界连接的桥梁。就像人们使用linux shell操作linux系统，大模型可以使用python来执行它想要执行的能影响物理世界的指令。

        学习了python之后，发现可以做的事情很多，让我们操作计算机，数据的获取和处理变得简单了许多，特别是网络中的数据。我们可以使用python来做许多自动化的操作。为了熟悉使用这门语言，也开始试着写一些简单的爬虫，这些所见即所得的成果比起单纯研究技术和算法有趣的多。当然爬虫也是一门技术。下面就是分享一些视频爬取经验，仅供记录经验，技术分享交流使用。

1、M3U8介绍

        M3U8 是一种基于文本的播放列表文件格式，主要用于 ​​HTTP Live Streaming（HLS）流媒体协议​​，由苹果公司开发并广泛应用于在线视频和音频传输中。

        M3U8它实际上就是用一个文本文件（一般为.m3u8后缀）来定义视频和音频等流媒体的播放行为。我们在浏览器上看视频时，当一个完整的视频文件很大时，比如一个G，如果等网页把这1G的视频文件全部下载下来再播放显然很不现实。那么直接的解决方式就是把这1G的视频文件分成很多个一小段的视频文件（比如1个小视频文件能播放1分钟），边下边播，就不会让用户长时间的等待下载才能播放。M3U8文件就是定义这一个个小视频文件的。M3U8文件的格式类似如下：

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-KEY:METHOD=AES-128,URI="https://aa.bb.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/crypt.key?auth_key=1742691847-72-0-d8787a27fe2eef60b071ac643f875c1b",IV=0xba25433fa8984b5abb5e0f5cc41f1a60
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-TARGETDURATION:5
#EXTINF:5.000,
https://aa.bb.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/0f07ef21ec282c96ac32bb73d221f90c0.ts?auth_key=1742691847-72-0-0b996b02189a9b159cdca1b30a72bc39
#EXTINF:5.000,
https://aa.bb.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/0f07ef21ec282c96ac32bb73d221f90c1.ts?auth_key=1742691847-72-0-bf3bd2faf0a60b57bf33a620c16533db

       比如文件中这样的一行：https://aa.bb.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/0f07ef21ec282c96ac32bb73d221f90c0.ts?auth_key=1742691847-72-0-0b996b02189a9b159cdca1b30a72bc39  

就是一个单独的小视频文件的下载地址。我们一般只要拿到这个下载地址，就可以使用模拟浏览器请求获取到文件数据。当然如果视频数据需要保密，那么返回的数据就是加密后的视频数据，我们要解密后才能播放。

        那我怎么知道视频数据有没有加密？用什么加密方式？怎么解密？答案都在M3U8文件里。文件内容中METHOD=AES-128就是写明了加密方式，加密key的获取通过这个url去获取，URI="https://aa.bb.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/crypt.key?auth_key=1742691847-72-0-d8787a27fe2eef60b071ac643f875c1b"，还有IV向量为IV=0xba25433fa8984b5abb5e0f5cc41f1a60，有了这些信息我们就可以对文件数据进行解密了，AES加密模式一般都是用CBC模式，这里没有明确写出。如果没有加密我们下载下来可以直接用本地视频播放器播放了。

        上面的M3U8文件示例，只是一般形式，还有一些M3U8的文件中的ts文件只有一个ts文件名，并不是完整的URL下载链接，比如只有"12345.ts",此时我们需要把地址拼接完整，前面的地址就是用M3U8文件下载地址来拼接，比如M3U8文件的下载URL为：https://aa.b.com/video91/m3u8/2024/07/02/a6ea9246/index.m3u8，那么ts的下载地址拼接完后就是：https://aa.b.com/12345.ts

2、获取M3U8文件

        从上面的介绍中，我们知道只要获取到M3U8文件，基本上就能下载到视频文件。怎么获取呢？一般手动获取就是用我们浏览器的F12打开调试控制台，搜索网络请求，直接查找请求URL中包含关键字".m3u8"的请求，然后查看响应，复制出来就可以。如果手动查找，可以参考:Python爬取下载m3u8视频，原来这么简单！-CSDN博客

        但是不是每个小白都能那么顺利的找到M3U8文件，我们想要的就是全自动下载，python就是干自动化的工作的。我们能不能做成一个通用的视频下载脚本，自动获取分析网页，自动获取M3U8文件，自动下载ts视频文件，自动合并视频文件呢。当前可以的，python的开源库playwright，Selenium​等都是自动化模拟浏览器的。我们为了能做一个通用的m3u8视频抓取脚本，就不用一个个网站的去分析请求和逆向js。我们使用最简单的方式，就是启动一个浏览器去模拟浏览器访问包含m3u8视频的网页，然后监控网络请求，捕获到返回的m3u8文件就可以了。

3、解析M3U8文件

        只要了解了M3U8文件内容的构成，很容易解析出加密方式和获取ts文件下载链接

4、下载ts文件

        ts文件的下载，可以使用requests库直接请求，有的网站有一些反爬措施，需要加Referer和User-Agent请求头。可以使用requests+多线程池并行下载多个ts，下载的很快，取决于你的网速，当然要考虑网站的承受能力也不用开太多线程。也可以使用aiohttp+协程并行下载

5、ts文件数据解密

        前面提到ts文件数据可能被加密了，我们可以直接根据M3U8的内容获取到所有解密需要用的参数，一般目前都是用AES-128，CBC模式进行加密，还有的都是不加密的，不加密可以省略这个步骤

6、ts视频流合并

        ts视频文件是一个个小的视频文件，能单独播放，但是我们要合并成一整个视频文件才好用我们的本地播放器连续播放，此时就需要借助ffmpeg这个工具进行合并，这是一个独立的可执行程序，通过命令行合并。如果没有安装需要安装

7、python m3u8视频爬虫代码

        以下爬虫代码可以处理不加密的和AES-128 CBC加密的ts文件，目前我发现的视频网站都是这两种加密方式，如果有其他的加密方式可以很容易自行扩展。

        脚本支持下载当前页面所有的m3u8视频，视频文件名称从网页标题中提取，多个视频按xxx_1.mp4,xxx_2.mp4命名

（1）需要安装的python库

playwright（模拟浏览器） requests（http请求） pycryptodome（AES解密）

python版本要python3

（2）需要安装的工具

ffmpeg（用于ts视频合并）

mac安装：

brew install ffmpeg

Ubuntu/Debian安装：

sudo apt update  
sudo apt install ffmpeg

安装完测试一下： 

mypc$ ffmpeg -h
ffmpeg version 7.1.1 Copyright (c) 2000-2025 the FFmpeg developersbuilt with Apple clang version 16.0.0 (clang-1600.0.26.6)
...

（3） 执行

需要给main.py函数加可执行权限:chmod +x main.py

./main.py https://aa.bb.com/1.html

或者

python main.py https://aa.bb.com/1.html

  （4）代码

第一个文件是并发下载ts文件的模块parallel_download_m3u8.py

from collections import namedtuple
import os
from enum import Enum
import threading
import requests
import logging
from concurrent.futures import ThreadPoolExecutor
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
import os
import time
import common
# 定义m3u8的加密类型
class M3u8EncryptType(Enum):UNKNOWN = 0PLAIN = 1AES128 = 2
# 定义多线下载函数
def download_thread_fun(url:str, referrer_url:str, ts_name:str, save_dir:str, status_dict: dict[str, bool], lock: threading.Lock)->None:r"""多线程下载函数:param url: ts文件下载链接:param referrer_url: 请求头部中的referer:param ts_name: ts文件名:param save_dir: 保存目录:param status_dict: 下载状态字典:param lock: 线程锁:return: None"""req_times = 1while req_times <= 10:try:# 开始下载logging.info(f"request({req_times}):{url}")req_times += 1# 访问链接，获取ts文件headers = {'Referer': referrer_url, "User-Agent": common.USER_AGENT}rsp = requests.get(url, headers=headers, timeout=10)if rsp.status_code != 200:logging.error(f"获取ts文件失败,ts_name:{ts_name} url:{url} status_code:{rsp.status_code} text:{rsp.text}")continue#把响应内容写入文件with open(os.path.join(save_dir, ts_name), 'wb') as f:f.write(rsp.content)# 下载完成，修改状态字典with lock:status_dict[ts_name] = Truelogging.info(f"download finish, ts_name: {ts_name}")# 处理完成，跳出循环    breakexcept requests.Timeout as timeout_error:logging.error(f"连接超时:{timeout_error}")# 其他异常except Exception as e:logging.error(f"othererror:{e}")def download_m3u8_ts(ts_url_list: list[str], referer_url: str, output_dir: str, ts_name_list: list[str])->bool:r"""多线程下载m3u8的分片文件:param ts_url_list: ts文件下载链接列表:param referer_url: 请求头部中的referer:param output_dir: 保存目录:param ts_name_list: ts文件名列表,输出参数，后续按这个顺序合并ts文件:return: bool,True: 下载成功, False: 下载失败"""curr_ts_file_list = []# 创建保存目录if not os.path.exists(output_dir):os.makedirs(output_dir)else:# 获取目录下的所有ts文件名称，只保留文件名称，去掉路径curr_ts_file_list = os.listdir(output_dir)# list转换成setcurr_ts_file_set = set(curr_ts_file_list)# 下载状态字典status_dict = {}lock = threading.Lock()# 使用线程池下载ts文件with ThreadPoolExecutor(max_workers=100) as executor:for ts_url in ts_url_list:# url类似：https://aa.b.com/videos/638c59a97f4443188948172f17656210/si/c_qtQ_AoVUPdMt8Ok9AyhzH9IvyNbpsCrFBX8Hq-PohdL8k_jBQ.ts?mm=OWFkMWZkOTM0NjYxZGZmZmYzOTJkMzg3MmNmMTUyZWM4OTE1MDcyMDM5ZGIxZjcz&t=1743165232620&d=rfjyb3bu5j0r.com&e=1743186832620&ip=240e:47e:3460:347b:878:2364:a97c:49e&gap=c9b8c4a5827b11e7fea3a276f8d3e70f&ic=CN&slot=2cd0dcc81c53a97a3aaebf2583296138#提取出ts文件名ts_name = ts_url.split('/')[-1]ts_name = ts_name.split('?')[0]logging.info(f"parse ts_name:{ts_name}")if ts_name == '':logging.error(f'ts_name is empty, url:{ts_url}')continuets_name_list.append(ts_name)# 下载状态字典中添加ts文件名if ts_name in curr_ts_file_set:logging.info(f"ts_file: {ts_name} already exists, skip download")status_dict[ts_name] = Truecontinueelse:status_dict[ts_name] = False# 任务提交到线程池executor.submit(download_thread_fun, ts_url, referer_url, ts_name, output_dir, status_dict, lock)# 检查下载状态字典all_download_finish = True# logging.info(f"download status_dict:{status_dict}")   for ts_name, is_download in status_dict.items():if not is_download:logging.error(f"ts_name: {ts_name} download failed")all_download_finish = Falsereturn all_download_finish# 定义获取IV向量的函数
def parse_iv(txt: str)->str:"""解析IV向量:param txt: m3u8文件内容:return: IV向量字符串"""# 先判断是不是文本类型if not isinstance(txt, str):logging.error("输入参数不是字符串")return ""index = txt.find("IV=")if -1 == index:logging.error("未找到IV向量")return ""# 查找\n位置end_index = txt.find("\n", index)if -1 == end_index:logging.error("未找到IV向量结束位置")return ""# 取出IV向量iv = txt[index+3:end_index]# 去掉前面的0xiv = iv.removeprefix("0x")logging.info(f"iv:{iv}")return iv# 定义获取加密秘钥URI的函数
def parse_key_uri(txt):# 先判断是不是文本类型if not isinstance(txt, str):logging.error("输入参数不是字符串")return ""'''value的格式如下：#EXTM3U#EXT-X-VERSION:3#EXT-X-KEY:METHOD=AES-128,URI="https://aa.b.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/crypt.key?auth_key=1742912656-80-0-e70e0b290d6dba2ddde376dc9dbdb22c",IV=0xba25433fa8984b5abb5e0f5cc41f1a60#EXT-X-MEDIA-SEQUENCE:0#EXT-X-TARGETDURATION:5#EXTINF:5.000,# '''# 先从value中获取到加密密钥的URI：https://aa.b.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/crypt.key?auth_key=1742912656-80-0-e70e0b290d6dba2ddde376dc9dbdb22curi_start = "URI=\""index = txt.find(uri_start)if -1 == index:logging.error("未找到加密密钥的URI")return ""# 继续找下一个双引号的位置index += len(uri_start)end_index = txt.find("\"", index)if -1 == end_index:logging.error("未找到加密密钥的URI结束位置")return ""# 取出加密密钥的URIuri = txt[index:end_index]return uri#定义返回结构
ProcResult = namedtuple("ProcResult", "enc_method key_url iv ts_url_list", defaults=[M3u8EncryptType.UNKNOWN, None, None, None])def parse_m3u8_file(m3u8_req_url: str, m3u8_file_content: str):r"""解析m3u8文件内容:param m3u8_file_content: m3u8文件内容:return: 返回一个命名元祖，包括加密方式method, 秘钥获取url key_url, ts文件下载链接列表 ts_url_list"""# 截取m3u8请求Url的资源路径，比如：https://aa.b.com/video91/m3u8/2024/07/02/a6ea9246/index.m3u8index = m3u8_req_url.rfind("/")url_prefix = m3u8_req_url[:index+1]logging.info(f"url_prefix:{url_prefix}")# 初始化变量，避免未定义key_url = Noneiv = None# 如果是AES128加密，会有这行：#EXT-X-KEY:METHOD=AES-128,URI="https://aa.b.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/crypt.key?auth_key=1742691847-72-0-d8787a27fe2eef60b071ac643f875c1b",IV=0xba25433fa8984b5abb5e0f5cc41f1a60index = m3u8_file_content.find("#EXT-X-KEY:METHOD=")enc_method = M3u8EncryptType.UNKNOWNif -1 == index:enc_method = M3u8EncryptType.PLAINelse:# 截取出加密方式start_index = index + len("#EXT-X-KEY:METHOD=")newline_index = m3u8_file_content.find("\n", start_index)end_index = m3u8_file_content.find(",", start_index)if newline_index < end_index:end_index = newline_indexenc_method_str = m3u8_file_content[start_index:end_index]enc_method_str = enc_method_str.strip()if enc_method_str == "AES-128":# 截取出秘钥获取urlkey_url = parse_key_uri(m3u8_file_content)# 截取出IV向量iv = parse_iv(m3u8_file_content)enc_method = M3u8EncryptType.AES128if key_url == "" or iv == "":logging.error(f"AES128加密，但未找到秘钥获取url或IV向量, key_url:{key_url}, iv:{iv}")# 抛出异常raise Exception("AES128加密，但未找到秘钥获取url或IV向量")if not key_url.startswith("http"):key_url = url_prefix + key_urllogging.info(f"AES128加密, key_url:{key_url}, iv:{iv}")elif enc_method_str == "NONE":enc_method = M3u8EncryptType.PLAINelse :logging.error(f"不支持的加密方式:{enc_method_str}")# 抛出异常raise Exception(f"不支持的加密方式:{enc_method_str}")lines = m3u8_file_content.split("\n")def change_ts_url(line):if line.startswith("http"):return lineelse:return url_prefix + linets_url_list = [change_ts_url(item) for item in lines if not item.startswith("#") and not item.strip() == ""]return ProcResult(enc_method, key_url, iv, ts_url_list)def judge_m3u8_encript_type(m3u8_content: str)->M3u8EncryptType:"""判断m3u8 ts文件类型，比如是否使用明文，哪种加密方式等:param txt: m3u8文件内容:return: M3u8EncryptType枚举类型"""# 如果是AES128加密，会有这行：#EXT-X-KEY:METHOD=AES-128,URI="https://aa.b.com/videos4/0f07ef21ec282c96ac32bb73d221f90c/crypt.key?auth_key=1742691847-72-0-d8787a27fe2eef60b071ac643f875c1b",IV=0xba25433fa8984b5abb5e0f5cc41f1a60index = m3u8_content.find("#EXT-X-KEY:METHOD=")if -1 == index:return M3u8EncryptType.PLAIN# 截取出加密方式start_index = index + len("#EXT-X-KEY:METHOD=")end_index = m3u8_content.find(",", start_index)enc_method = m3u8_content[start_index:end_index]enc_method = enc_method.strip()if enc_method == "AES-128":return M3u8EncryptType.AES128else :return M3u8EncryptType.UNKNOWNdef deal_ts_file_plain(ts_file_dir: str, ts_name_list: list[str], ts_merge_file_name: str)->bool:"""处理未加密的ts文件，直接合并:param ts_file_dir: ts文件目录:param ts_name_list: ts文件名列表:param ts_merge_file_name: 合并后的ts文件名:return: 合并是否成功"""try:with open(os.path.join(ts_file_dir, ts_merge_file_name), "wb") as merge_file:# 合并ts文件for ts_name in ts_name_list:with open(os.path.join(ts_file_dir, ts_name), "rb") as ts_file:ts_file_data = ts_file.read()merge_file.write(ts_file_data)except Exception as e:logging.error("合并ts文件失败:{}".format(e))return Falsereturn Truedef request_encrypt_key(key_uri: str, referer_url: str)->bytes:"""请求加密密钥:param key_url: 加密密钥获取url:param referer_url: 请求头部中的referer:return: 加密密钥"""try:# 访问uri，获取加密密钥# 设置请求头headers = {'referer': referer_url,'User-Agent': common.USER_AGENT}response = requests.get(key_uri, headers=headers, timeout=10)if response.status_code != 200:logging.error("获取加密密钥失败, req_url:{}, status_code:{}".format(key_uri, response.status_code))return Noneexcept Exception as e:logging.error("访问加密密钥URI失败, req_url:{}, error:{}".format(key_uri, e))return Noneaes_key = response.contentreturn aes_keydef deal_ts_file_aes128_new(aes_key: bytes, iv: str, ts_file_dir: str, ts_name_list: list[str], ts_merge_file_name: str)->bool:"""处理AES128加密的ts文件，进行解密，并需要把所有解密后的文件数据，都追加到同一个文件中，ts的拼接顺序按m3u8文件中的顺序，后续使用ffmpeg等工具进行视频流的合并:param aes_key: 加密密钥:param iv: iv向量:param ts_file_dir: ts文件目录:param ts_name_list: ts文件名列表:param ts_merge_file_name: 合并后的ts文件名:return: 合并是否成功"""# iv十六进制字符串转换成二进制byteiv_bytes = bytes.fromhex(iv)assert len(iv_bytes) == 16, "IV 必须是 16 字节"# ase_key是密钥，iv是向量（小写的16进制字符串）cipher = AES.new(aes_key, AES.MODE_CBC, iv_bytes)# 解密后就是m3u8文件，可以直接保存try:merge_file_path = os.path.abspath(os.path.join(ts_file_dir, ts_merge_file_name))with open(merge_file_path, "wb") as f:# 遍历ts文件列表，逐个解密并写入文件for ts_name in ts_name_list:ts_path = os.path.abspath(os.path.join(ts_file_dir, ts_name))with open(ts_path, "rb") as ts_file:ts_file_data = ts_file.read()decrypted_data = cipher.decrypt(ts_file_data)plaintext = unpad(decrypted_data, AES.block_size)  # 移除PKCS7填充f.write(plaintext)logging.info(f"ts合并文件已保存到: {merge_file_path}")except Exception as e:logging.error("error:{}".format(e))return Falsereturn Truedef title_to_filename(title: str)->str:"""将标题转换为文件名,去除非法路径字符，空格替换成"-":param title: 标题:return: 文件名"""# 修复条件检查if not title:curr_ts = int(time.time())name = f"video_{curr_ts}"return name# 去除非法路径字符name = "".join([c for c in title if c.isalpha() or c.isdigit() or c in "._- "])# 空格替换成"-"name = name.replace(" ", "-")return name

第二个是main函数入口模块main.py

#! /opt/homebrew/bin/python3
# 此处需要把路径修改为你自己本地的python可执行文件的路径，mac或者linux下使用which python3（或者which python）命令查看路径
import loggingfrom playwright.sync_api import sync_playwright
import os
import shutil
import sys
import common# 导入parallel_download_m3u8模块
import parallel_download_m3u8 as pdm# 定义main函数
def main(url:str):logging.info(f"开始下载视频:{url}")page_url = ""with sync_playwright() as p:# 启动浏览器,headless=True使用无头模式，如果要查看网页加载过程，可以设置为False，会打开一个浏览器窗口，能直观的看到网页加载过程browser = p.webkit.launch(headless=True)# 以下是模拟手机端的上下文# context = browser.new_context(#     user_agent='Mozilla/5.0 (Linux; Android 11; Pixel 5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.91 Mobile Safari/537.36',#     viewport={'width': 360, 'height': 640},#     device_scale_factor=2.625,#     has_touch=True,#     is_mobile=True,#     extra_http_headers={#         'Accept-Language': 'en-US,en;q=0.9'#     }# )# 以下是模拟PC端的上下文context = browser.new_context(user_agent=common.USER_AGENT,viewport={'width': 1920, 'height': 1080},extra_http_headers={'sec-ch-ua-platform':'"macOS"','sec-ch-ua':'"Microsoft Edge";v="135", "Not-A.Brand";v="8", "Chromium";v="135"','Accept-Language': 'en-US,en;q=0.9'})page = context.new_page()# 监听响应page_has_been_closed = Falsedict_m3u8_content = {}title = ""def capture_response(response):nonlocal m3u8_content, title, page_has_been_closed, page_url#logging.info("url:{}".format(response.url))if "m3u8" in response.url and "#EXTM3U" in response.text():logging.info("捕获到m3u8文件URL:{}".format(response.url))logging.info("m3u8文件内容:{}".format(response.body()))dict_m3u8_content[response.url] = response.text()title = page.title()page_url = page.urlpage.on("response", capture_response)try:page.goto(url, timeout=60000, wait_until="load")# 滚动到底部page.evaluate("window.scrollTo(0, 3);")# 等待页面加载完成page.wait_for_timeout(6000)except Exception as e:if "been closed" in str(e) and not page_has_been_closed:logging.info("页面已关闭")else:logging.error("加载页面失败:{}".format(e))finally:# 关闭页面if not page_has_been_closed:logging.info("页面加载结束，关闭页面")page.close()logging.info(f"当前页面标题: {title}")context.close()browser.close()file_name = pdm.title_to_filename(title)logging.info(f"共捕获到{len(dict_m3u8_content)}个m3u8视频链接")# 获取网页浏览器中的url，比如https://a.com/movie/?viewKey=3b942de4147b413a 截取为https://a.com/index = page_url.rfind("://")if index == -1:logging.error(f"无法获取网址, url:{page_url}")returnindex = page_url.find("/", index+3)if index != -1:referer_url = page_url[:index+1]else:referer_url = page_urllogging.info(f"referer_url:{referer_url}")vidio_count = 1for m3u8_req_url, m3u8_content in dict_m3u8_content.items():logging.info("m3u8 content:{}".format(m3u8_content))# 解析m3u8文件try:proc_result = pdm.parse_m3u8_file(m3u8_req_url, m3u8_content)except Exception as e:logging.error("解析m3u8文件失败:{}, url:{}".format(e, m3u8_req_url))continueaes_key = Noneenc_type = proc_result.enc_methodlogging.info("m3u8 encrypt type:{}".format(enc_type))if enc_type == pdm.M3u8EncryptType.UNKNOWN :logging.error("未知的加密类型, content:{}".format(m3u8_content))continueelif enc_type == pdm.M3u8EncryptType.AES128:# 先下载key文件aes_key = pdm.request_encrypt_key(proc_result.key_url, referer_url)if not aes_key:logging.error("下载key文件失败, m3u8 url:{}".format(m3u8_req_url))continuets_url_list = proc_result.ts_url_list# 调用线程池下载ts文件ts_name_list = []download_result = Falsets_file_dir = "./ts_files"try:download_result = pdm.download_m3u8_ts(ts_url_list, referer_url, ts_file_dir, ts_name_list)except Exception as e:logging.error("下载ts文件异常:{}, m3u8 url:{}".format(e, m3u8_req_url))continueif not download_result:logging.error("下载ts文件失败, m3u8 url:{}".format(m3u8_req_url))# 删除临时目录和所有ts文件if os.path.exists(ts_file_dir):shutil.rmtree(ts_file_dir)# 继续下一个视频continuets_merge_file_name = "tsfile.tmp"# 根据加密类型做相应处理match enc_type:case pdm.M3u8EncryptType.PLAIN:deal_result = pdm.deal_ts_file_plain(ts_file_dir, ts_name_list, ts_merge_file_name)if not deal_result:logging.error("处理未加密的ts文件失败")continue# 无需解密，直接合并ts文件case pdm.M3u8EncryptType.AES128:deal_result = pdm.deal_ts_file_aes128_new(aes_key, proc_result.iv, ts_file_dir, ts_name_list, ts_merge_file_name)if not deal_result:logging.error("处理加密的ts文件失败")continuecase _:logging.error("未知的加密类型")continueif len(dict_m3u8_content) == 1:mp4_file = f"{file_name}.mp4"else:mp4_file = f"{file_name}_{vidio_count}.mp4"# 用ffmpeg合并ts文件merge_file_path = os.path.abspath(os.path.join(ts_file_dir, ts_merge_file_name))mp4_path = os.path.abspath(mp4_file)cmd = f"ffmpeg -i \"{merge_file_path}\" -c copy \"{mp4_path}\""logging.info(f"执行ffmpeg命令: {cmd}")ret = os.system(cmd)# 判断命令是否执行成功if ret != 0:logging.error("合并ts文件失败,name: {}".format(mp4_file))else:logging.info("视频文件合并完成,name: {}".format(mp4_file))vidio_count += 1# 删除临时文件if os.path.exists(ts_merge_file_name):os.remove(ts_merge_file_name)if os.path.exists(ts_file_dir):shutil.rmtree(ts_file_dir)def setup_logger():# 创建 Loggerlogger = logging.getLogger()logger.setLevel(logging.DEBUG)  # 设置全局日志级别# 定义日志格式formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d - %(funcName)s] - %(message)s')# 1. 控制台 Handler（输出到屏幕）console_handler = logging.StreamHandler()console_handler.setLevel(logging.INFO)  # 控制台日志级别console_handler.setFormatter(formatter)# 2. 文件 Handler（保存到文件）file_handler = logging.FileHandler('app.log', encoding='utf-8')file_handler.setLevel(logging.DEBUG)  # 文件日志级别file_handler.setFormatter(formatter)# 将 Handler 添加到 Loggerlogger.addHandler(console_handler)logger.addHandler(file_handler)# 调用main函数
if __name__ == '__main__':if len(sys.argv) > 1:url = sys.argv[1]else:print("usage: python main.py <url>")exit(1)setup_logger()main(url)