不只是.ts后缀:用Python批量处理m3u8下载的‘变种’视频分片(附完整脚本)

张开发
2026/4/20 14:31:01 15 分钟阅读
不只是.ts后缀:用Python批量处理m3u8下载的‘变种’视频分片(附完整脚本)
破解伪装视频分片Python自动化修复与合并实战指南最近在抓取某些视频资源时发现一个有趣的现象——明明应该是标准的TS视频分片却伪装成了PNG图片格式。这种变种分片不仅让常规下载工具束手无策还会导致FFmpeg这样的专业工具误判格式而无法合并。本文将分享一套完整的Python自动化解决方案从识别、修复到最终合并帮你轻松搞定这些狡猾的视频分片。1. 理解非常规m3u8分片的本质当你在Chrome中下载这些分片时浏览器会默认将它们识别为PNG图片。但如果你仔细观察会发现这些图片的大小明显异常——通常远大于普通缩略图。用十六进制编辑器打开后真相大白文件开头确实是PNG的魔数签名89 50 4E 47但后面跟着的却是标准的TS视频内容。为什么会出现这种情况这其实是某些平台为防止自动化抓取而设计的混淆手段。他们知道大多数爬虫会直接过滤非视频后缀的链接于是故意去掉.ts后缀甚至伪装成图片格式。但只要我们理解其本质就能轻松破解文件头欺骗前4个字节被设置为PNG签名真实内容从第5个字节开始是正常的TS视频数据关键特征文件大小通常在几百KB到几MB之间远大于普通PNG注意直接删除PNG文件头会导致文件损坏必须用特定字节填充替换2. 自动化处理流程设计完整的解决方案需要实现以下关键步骤智能下载自动解析m3u8文件并下载所有分片格式检测快速识别被伪装的分片内容修复安全地修复文件头而不损坏视频数据无缝合并使用FFmpeg将修复后的分片合并为完整视频2.1 核心组件与技术选型功能模块技术方案关键工具/库m3u8解析正则表达式Python re模块文件下载异步请求aiohttp格式检测魔数分析文件头读取内容修复二进制操作Python文件IO视频合并子进程调用subprocess FFmpeg3. 实战代码从下载到修复的全流程实现下面是一个完整的Python脚本实现了上述所有功能import os import re import aiohttp import asyncio from pathlib import Path async def download_file(session, url, save_path): async with session.get(url) as response: with open(save_path, wb) as f: while True: chunk await response.content.read(1024) if not chunk: break f.write(chunk) def is_disguised_ts(file_path): 检查文件是否被伪装成PNG的TS分片 with open(file_path, rb) as f: header f.read(4) return header b\x89PNG def repair_ts_file(input_path, output_path): 修复被伪装的TS文件 with open(input_path, rb) as infile, open(output_path, wb) as outfile: # 读取全部内容 data infile.read() # 替换前4个字节为0xFF repaired_data b\xFF\xFF\xFF\xFF data[4:] outfile.write(repaired_data) async def process_m3u8(m3u8_url, output_dir): 处理整个m3u8流程 os.makedirs(output_dir, exist_okTrue) raw_dir os.path.join(output_dir, raw) repaired_dir os.path.join(output_dir, repaired) os.makedirs(raw_dir, exist_okTrue) os.makedirs(repaired_dir, exist_okTrue) async with aiohttp.ClientSession() as session: # 下载m3u8文件 m3u8_content await (await session.get(m3u8_url)).text() # 提取所有分片URL ts_urls re.findall(r^[^#].*\.(?:ts|m4s|vtt)?$, m3u8_content, re.MULTILINE) # 下载所有分片 download_tasks [] for i, ts_url in enumerate(ts_urls): save_path os.path.join(raw_dir, f{i:04d}.ts) download_tasks.append(download_file(session, ts_url, save_path)) await asyncio.gather(*download_tasks) # 修复所有分片 for i in range(len(ts_urls)): input_path os.path.join(raw_dir, f{i:04d}.ts) output_path os.path.join(repaired_dir, f{i:04d}.ts) if is_disguised_ts(input_path): repair_ts_file(input_path, output_path) else: # 如果不是伪装文件直接复制 with open(input_path, rb) as infile, open(output_path, wb) as outfile: outfile.write(infile.read()) # 生成文件列表供FFmpeg合并 with open(os.path.join(output_dir, filelist.txt), w) as f: for i in range(len(ts_urls)): f.write(ffile {os.path.join(repaired_dir, f{i:04d}.ts)}\n) # 使用FFmpeg合并 cmd [ ffmpeg, -f, concat, -safe, 0, -i, os.path.join(output_dir, filelist.txt), -c, copy, os.path.join(output_dir, output.mp4) ] subprocess.run(cmd, checkTrue) if __name__ __main__: import sys m3u8_url sys.argv[1] if len(sys.argv) 1 else input(请输入m3u8 URL: ) output_dir sys.argv[2] if len(sys.argv) 2 else output asyncio.run(process_m3u8(m3u8_url, output_dir))4. 高级技巧与性能优化4.1 并行下载加速上述代码已经使用了异步IO来加速下载但对于大量分片我们还可以进一步优化# 在process_m3u8函数中添加semaphore控制并发数 semaphore asyncio.Semaphore(16) # 限制并发数为16 async def limited_download(session, url, save_path): async with semaphore: await download_file(session, url, save_path) # 然后替换原来的download_tasks创建方式 download_tasks [limited_download(session, ts_url, os.path.join(raw_dir, f{i:04d}.ts)) for i, ts_url in enumerate(ts_urls)]4.2 智能格式检测增强除了PNG伪装我们还可能遇到其他类型的混淆。增强的检测函数可以识别更多变种def is_disguised_ts(file_path): 增强版格式检测 with open(file_path, rb) as f: header f.read(8) # 检测常见图片格式 image_headers { b\x89PNG: PNG, b\xFF\xD8\xFF: JPEG, bGIF87a: GIF, bGIF89a: GIF, bBM: BMP } for sig, fmt in image_headers.items(): if header.startswith(sig): return True return False4.3 内存高效处理大文件对于特别大的分片文件我们可以采用流式处理来减少内存占用def repair_large_ts(input_path, output_path, chunk_size1024*1024): 流式修复大文件 with open(input_path, rb) as infile, open(output_path, wb) as outfile: # 处理前4个字节 outfile.write(b\xFF\xFF\xFF\xFF) infile.seek(4) # 流式复制剩余内容 while True: chunk infile.read(chunk_size) if not chunk: break outfile.write(chunk)5. 常见问题与解决方案在实际使用中你可能会遇到以下问题FFmpeg合并失败确保所有分片都已正确修复尝试使用-fflags genpts参数重建时间戳检查分片顺序是否正确下载速度慢调整并发数不要过高以免被封禁考虑使用代理轮换检查网络连接是否稳定部分分片无法修复检查这些分片是否使用了不同的混淆方式尝试手动分析文件结构考虑是否为加密分片需要额外解密步骤输出视频音画不同步确保m3u8中的EXTINF时长准确尝试使用-avoid_negative_ts make_zero参数检查原始分片的帧率和时间基这套解决方案已经在多个实际项目中验证有效能够处理绝大多数变种分片。根据我的经验最关键的是准确识别伪装模式并采用正确的修复方法——简单的删除会导致文件损坏而精确的填充则能完美保留视频内容。

更多文章