Python打造视频解析下载器：m3u8与断点续传实战

简介这是一份面向 TypeScript 开发者的视频解析库源码包专为需要解析 MP4、FLV、MKV 等容器格式并提取元数据、编码信息、时间轴与字幕轨道的应用场景而设计适合正在搭建视频处理工具或希望学习媒体解析原理的中级开发者。压缩包共 13 个文件以 7 个 TypeScript 源码文件为核心辅以 package.json、tsconfig.json、vercel.json 等配置以及 yarn.lock、Procfile 等部署辅助文件整体仅 61KB结构紧凑。源码按 services、controllers、interfaces、entity 等模块划分职责清晰同时包含单元测试与构建脚本可快速接入 npm 工作流方便二次开发与调试包内测试用例还展示了不同容器格式的预期输出便于对照学习。已有 197 人学习下载。通过研读该库开发者可掌握视频容器识别、编码参数读取和元数据提取的工程化实现思路也可以将其作为视频处理功能模块的基础骨架节省从零开发的时间。 做视频解析工具这事说实话一开始我心里是有点抗拒的。市面上的下载器、解析网站一抓一大把随便搜一下都是现成的重新造轮子好像没必要。但真到自己动手处理批量视频资源时就会发现那些现成工具要么限制多要么广告满天飞要么格式转得乱七八糟。后来我花了一周时间用 Python 写了个叫video_parser的小工具专门解决从网页里把视频真实地址挖出来、按需求下载、顺手做初步处理这件事。这篇文章就把整个项目的思路、代码细节、踩过的坑一次说清楚。如果你也在做资源采集、课程存档、视频离线备份或者只是经常碰到网页能看但就是没法下载的尴尬情况这个项目能给你一套完整可复用的方案。它不依赖重型框架理解起来不费劲而且你完全可以在我的代码基础上扩展成自己的工具。1. 项目背景与整体思路拆解1.1 为什么需要自研 video_parser视频解析这个需求说白了就是三个问题第一个网页上的视频地址藏得很深普通用户根本拿不到真实播放地址第二个就算拿到了地址很多是分片格式不合并根本没法直接播放第三个需要批量处理的时候手动操作逐个下载完全不现实。市面上的工具为什么不好用我总结过三类痛点。一类是在线解析站每天换域名不说解析出来的画质经常被压缩还夹带私货。一类是通用下载器功能确实全但配置复杂而且很多只支持特定浏览器或特定网站遇到改版就失效。还有一类是浏览器插件下载单个视频没问题但批量处理、定时任务、命令行调用这些场景完全无能为力。所以自研工具的核心诉求是纯命令行、可脚本化调用、解析逻辑自己可控、不依赖某个特定网站的接口。这样一来无论你是想跑批量任务还是想集成到自己的自动化流程里都能自由操作。video_parser的设计目标就是做这样一个轻量但足够灵活的解析工具。1.2 技术选型与架构设计整个工具的核心依赖就三样requests负责网络请求BeautifulSoup和正则表达式负责从 HTML 里挖视频地址ffmpeg负责处理分片视频的合并和格式转换。听起来简单但组合起来能处理绝大多数解析场景。为什么要用requests而不是httpx或aiohttp说实话requests虽然同步阻塞但胜在生态成熟、API 稳定出错信息直观。视频解析这个场景本身是 IO 密集型的如果追求极致性能可以用异步但作为通用工具稳定可靠比速度更重要。我把所有请求都放到requests.Session里管理这样能自动维持 Cookie处理需要登录态的资源时后续请求会顺畅很多。# 会话管理示例 import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept-Language: zh-CN,zh;q0.9,en;q0.8 })架构上分成了四层请求层、解析层、下载层、后处理层。请求层负责抓页面、带请求头、处理重试解析层用不同的解析器从不同格式的页面里提取视频地址下载层负责拿真实地址下载处理分片和断点续传后处理层调用 ffmpeg 做格式转换。这种分层设计的好处是各模块独立日后增加新的站点支持只需新增一个解析器不用动其他代码。2. 解析流程详解与核心代码实现2.1 页面抓取与请求头伪装解析的第一步是拿到含视频信息的页面源码。这个环节做得好不好直接决定了后面能不能顺利解析。实际开发中我最常遇到的不是解析逻辑出错而是请求发出去就被对方服务器拒了。请求头伪装的核心是理解服务器如何识别非浏览器请求。通常看三点User-Agent是不是常见浏览器的、Referer是否符合站内跳转逻辑、Accept和Sec-Fetch-*这类头是不是正常浏览器会带的。所以我的请求头会配得比较完整而不是只填一个 UA 就完事。def build_headers(referer_url: str ) - dict: headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/avif, image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1, Sec-Fetch-Dest: document, Sec-Fetch-Mode: navigate, Sec-Fetch-Site: none, Sec-Fetch-User: ?1, } if referer_url: headers[Referer] referer_url return headers请求时设置超时是必须的我习惯将timeout设为(3, 10)即连接超时 3 秒、读取超时 10 秒。同时配合Retry策略自动处理临时性的网络抖动重试之间的退避时间递增避免对目标服务器造成压力。2.2 HTML5 视频标签解析最常见的视频嵌入方式是页面里直接放video标签视频地址就在src或source子标签里。这种解析最简单但有个坑不是所有video标签的src都是真实地址有些是经过一层跳转的中间链接需要二次解析。我的解析逻辑分两步走。第一步用 BeautifulSoup 找出所有video和source标签把src属性收集起来。第二步逐个判断这些地址是否是直链。直链的判断标准是URL 以常见的视频扩展名结尾或者响应头里的Content-Type是video/*。from bs4 import BeautifulSoup def parse_html5_video(html: str, page_url: str) - list: soup BeautifulSoup(html, html.parser) video_urls [] for video_tag in soup.find_all(video): src video_tag.get(src) if src: video_urls.append(urljoin(page_url, src)) for source_tag in video_tag.find_all(source): src source_tag.get(src) if src: video_urls.append(urljoin(page_url, src)) return list(set(video_urls))用urljoin处理相对路径是很容易忽略的细节。页面里的src有时是/media/video.mp4这种不补全成绝对地址后续请求必挂。补全之后还要做一次去重一个页面反复出现同一个资源的情况非常常见。2.3 流媒体地址提取与 m3u8 处理现在很多平台不再直接给 mp4而是用 HLS 协议把视频切成一个个 ts 分片通过 m3u8 索引文件管理。遇到这种情况解析工作的重点就从找地址变成了找 m3u8 地址再下载所有分片并合并。m3u8 地址通常藏在script标签里的 JavaScript 变量中或者藏在iframe的嵌套页面里。我用正则加 BeautifulSoup 双重手段来挖。最常见的格式是var videoUrl https://example.com/path/index.m3u8这种用正则rhttps?://[^\s]?\.m3u8[^\s]*就能直接命中。复杂一些的情况是地址经过 base64 编码或拼接处理这时需要看具体代码做解码。拿到 m3u8 文件后下一步是解析其中的 ts 分片地址。m3u8 本身是纯文本格式每行一个条目#EXTINF后面跟的下一行就是分片地址。需要注意分片地址可能是相对路径要基于 m3u8 的 URL 做拼接。另外有些 m3u8 是嵌套的指向下一级 m3u8多码率适配这种情况要递归解析。def parse_m3u8_playlist(m3u8_url: str) - list: content session.get(m3u8_url, headersbuild_headers()).text base_url m3u8_url.rsplit(/, 1)[0] segments [] for line in content.splitlines(): line line.strip() if not line or line.startswith(#): continue segments.append(urljoin(base_url /, line)) return segments分片下载和合并是我强烈建议交给 ffmpeg 处理的自己写并发下载 ts 再合并理论上可行但处理不好容易出现音画不同步。ffmpeg 一条命令就能完成ffmpeg -i input.m3u8 -c copy output.mp4。-c copy是流复制模式不重新编码速度快画质无损。2.4 真实地址二次跳转处理很多视频播放页面里的地址是经过防盗链验证的临时链接直接下载会拿到 403。这种链接有几个共同特征带大量 query 参数、域名是 CDN 子域、有时效性。我的处理方案是先用请求头带上Referer播放页地址去探测一次如果返回 200就用这个地址下载如果返回 403就尝试去掉参数重新拼一次或者从响应体里找真正的跳转目标。关于临时链接的时效性实际测试中不少链接只有几分钟到几小时的有效期。因此解析和下载尽量放在同一个流程里完成如果任务量太大需要分批次执行就要重新解析获取新地址不能沿用旧链接。3. 关键参数配置与实操细节3.1 下载策略与断点续传下载环节的稳定性直接决定工具好不好用。初期版本我直接拿到地址就用requests.get一把梭结果遇到网络抖动就是一个Connection reset前面的进度全白费。后来改用 Range 请求手动实现断点续传稳定很多。断点续传的原理是 HTTP 头里的Range字段。服务器支持的话会在响应头返回Accept-Ranges: bytes请求时指定Range: bytes1048576-就能从指定位置继续下载。实现逻辑不复杂下载前先看本地文件已存在的大小然后从这个位置继续请求。def download_with_resume(url: str, filepath: str, headers: dict) - bool: resume_pos 0 if os.path.exists(filepath): resume_pos os.path.getsize(filepath) current_headers headers.copy() if resume_pos 0: current_headers[Range] fbytes{resume_pos}- with session.get(url, headerscurrent_headers, streamTrue, timeout(5, 30)) as r: if r.status_code 416: return True # 文件已完整 mode ab if resume_pos 0 else wb with open(filepath, mode) as f: for chunk in r.iter_content(chunk_size1024 * 512): f.write(chunk) return True这里有个细节要注意不是所有服务器都支持断点续传。如果响应码是 200 而不是 206说明服务器忽略了 Range 头这时你要么放弃续传改为覆盖写要么就换一个支持 Range 的下载方式。判断逻辑里要处理这两种情况。3.2 重试机制与超时参数调优网络请求没有重试机制等于裸奔。但重试不是盲目多试几次就行重点是什么情况该重试、间隔多久、最多几次。我总结的经验是连接超时类错误值得重试HTTP 4xx 类错误不要重试5xx 可以有限重试。requests库自带HTTPAdapter可以配置重试策略不需要自己写循环。我配置了最多 3 次重试状态码 500、502、503、504 和连接相关错误触发每次退避时间按 0.5 秒、1 秒、2 秒递增避免快速重试触发对方风控。from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry retry_strategy Retry( total3, backoff_factor0.5, status_forcelist[500, 502, 503, 504], allowed_methods[GET, HEAD], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter)超时参数的设置同样讲究。timeout我习惯拆成连接超时和读取超时两个值。连接超时设 3-5 秒超过这个时间连不上就直接判定失败读取超时设 10-30 秒具体看视频文件大小和网络情况。下载大文件时如果读取超时设太短容易在网速波动时误判失败。3.3 并发下载控制处理多个视频来源时串行下载效率太低但盲目开线程池又会给服务器造成压力甚至触发封禁。我的做法是用ThreadPoolExecutor配合信号量控制并发数一般控制在 3-5 个并发既不会太慢也不会被服务器拉黑。from concurrent.futures import ThreadPoolExecutor import threading semaphore threading.Semaphore(3) def bounded_download(url, save_path): with semaphore: return download_with_resume(url, save_path, build_headers()) with ThreadPoolExecutor(max_workers5) as executor: futures [executor.submit(bounded_download, url, path) for url, path in task_list]并发这块有个容易踩的坑requests.Session不是线程安全的多线程共用同一个 Session 时偶尔会出现 Cookie 错乱或连接池异常。稳妥的做法是每个线程独立建 Session或者用requests官方推荐的thread-localSession 方案。我自己测试下来简单场景下开 3-5 个线程、每线程一个 Session最稳定。3.4 文件命名与去重策略批量下载时文件名规则不合理会造成后患。我见过太多工具把文件名直接按视频标题处理结果碰到特殊字符就报错。推荐的做法是标题清洗掉非法字符、长度截断到 80 字符以内、同一页面提取多个视频时加序号前缀。import re def safe_filename(name: str, max_len: int 80) - str: name re.sub(r[\\/:*?|], _, name) name name.strip().strip(.) if len(name) max_len: name name[:max_len] return name or untitled去重逻辑则看文件大小和 MD5简单场景下对比文件大小就够了。真正麻烦的是同一个视频地址在不同时间解析出不同临时链接导致重复下载。我在任务列表里维护一个 URL 集合基于原始页面 URL 视频索引生成任务 ID确保同一个任务不会重复执行。4. 常见问题与排查技巧实录4.1 HTTP 403 与防盗链应对403 是解析下载工具遇到最多的错误也是排查起来最折腾的。第一次写工具时我天真地以为加了 UA 就能通行结果被现实教育得很惨。403 通常意味着服务器已经确认你是非正常访问应对思路有两个方向。第一个方向是把请求头伪装得更像浏览器。尤其是Referer这个字段很多 CDN 就是靠它做防盗链验证的下载时必须带上播放页地址。其次是Origin头部分服务器会校验跨域请求。第二个方向是降低请求频次有些 403 不是规则拦截而是频率触发了阈值。把并发数降下来请求间加一点随机延时基本能解决。import time import random def gentle_delay(): time.sleep(random.uniform(0.5, 1.5))4.2 分片下载失败与残缺视频处理合并后的视频播放到某一段时间卡住或者直接跳秒多半是分片下载漏了或下载不完整。原因是分片请求失败后没有可靠的重试机制或者某些分片被服务器临时限流。排查方法是先看 ffmpeg 合并时输出的 warning如果有Packet corrupted或Invalid data字样基本就是某个分片文件不完整。定位到具体哪个分片缺失后单独重新请求该分片。预防措施是在分片下载循环里增加下载后校验文件大小大于 0的逻辑大小不对直接重试。4.3 视频地址提取为空页面结构复杂时解析器可能什么都拿不到。这时不要急着改代码先做三件事第一把页面源码存到本地看结构第二确认视频地址是否在 iframe 嵌套的页面里第三检查是否有额外的 JavaScript 加密逻辑。iframe 是常见的大坑。视频播放页的实际视频地址经常在嵌了另一层的播放器页面里直接解析外层的页面当然什么都拿不到。我的做法是在解析器里加一步 iframe 地址收集依次请求子页面再解析最多递归嵌套两层避免死循环。4.4 内存占用过高下载超大视频文件时如果用r.content一次性读入内存几百 MB 的视频直接内存爆掉。这个问题在新手代码里太常见了。正确处理是流式写入用iter_content分块迭代每次只保留一个 chunk 在内存里。同时建议对超大文件的下载任务做特殊标记下载进度打印频率调低避免频繁刷新终端信息影响性能。我一般是以 5 秒或 10 MB 为间隔打印一次进度。4.5 常见问题速查表问题现象常见原因推荐排查方案HTTP 403缺少 Referer 或触发频率限制补全请求头降低并发增加随机延时分片合并后有花屏某分片下载不完整分片校验大小并重试必要时重新下载解析结果为空iframe 嵌套或 JS 加密保存源码分析结构递归解析 iframe 子页面下载到一半连接断读取超时过短或网络波动调整超时参数启用断点续传重新执行文件名乱码或保存失败特殊字符未清洗使用 safe_filename 统一处理SSL 证书报错目标服务器证书链异常尝试verifyFalse并确认无安全风险注意verifyFalse等于跳过证书校验可能面临中间人攻击风险。只建议在明确目标来源可信且没有敏感信息传输的情况下使用不存在通杀所有网站的万能方案。5. 扩展方向与实践总结5.1 新增站点解析器video_parser目前的核心解析能力集中在通用页面结构上如果你需要支持特定站点最好的方式是写独立的解析器类。我建议按这个接口设计class BaseParser: def __init__(self, session): self.session session def parse(self, page_url: str) - dict: raise NotImplementedError def extract(self, html: str, page_url: str) - list: raise NotImplementedError每个站点写一个子类重写parse和extract方法。主程序里按站点域名路由到对应解析器没有匹配的则用通用解析器兜底。这样扩展新站点完全不影响已有功能每个解析器之间的边界也很清晰。5.2 命令行封装与批量任务工具核心跑通后我把它封装成了命令行入口支持三个参数--url指定页面地址--output指定输出目录--format指定输出格式。配合一个简单的 JSON 文件作为任务清单就能实现批量下载。python video_parser.py --url https://example.com/course/lesson-1 --output ./videos --format mp4批量任务的 JSON 格式我设计成数组结构每个元素包含页面地址、输出文件名、格式选项。主程序读取后按顺序执行执行结果写入日志文件。这套逻辑做下来我可以把它直接挂到 cron 定时任务里每天早上自动同步一次更新视频完全不需人工干预。5.3 个人体会写video_parser这段时间最大的体会是解析工具的核心不在代码多花哨而在对网络协议的理解和细节处理。请求头伪装、断点续传、分片合并、错误重试每个环节单独看都不难但组合在一起做到稳定可靠就需要大量真实场景的测试和打磨。如果让我重写一遍我会在一开始就做好日志系统把每个请求的地址、响应码、耗时都记录下来排查问题时会省掉很多猜测的时间。这也是我对所有做类似工具的朋友的第一条建议。本文还有配套的精品资源点击获取