★ FEATURED ARTICLE 简介video_parser 是一套基于 TypeScript 编写的视频解析库面向需要在播放器、转码服务或短视频处理流程中读取媒体信息的开发者适合有一定 TypeScript 基础和多媒体处理经验的工程人员。该库能够从 MP4、FLV、MKV 等常见容器中提取元数据、帧率、编码格式、字幕轨道与时间轴信息并输出相对规范的结构化结果使用者不必自行处理底层容器与编码细节。压缩包共 13 个文件、约 61KB其中 7 个 ts 文件是解析核心按 services、interfaces、entity、controllers 等模块拆开职责清楚便于替换和扩展json 配置与 lock 文件用于还原依赖环境Procfile 支持部署到云平台。已有 203 人浏览学习。研读这份代码可以理解视频容器解析、编码识别、元数据抽取和时间轴同步的实现思路同时能直接复用其模块化设计快速集成到自己的项目中减少从零搭建解析流程的工作量。1. video_parser 到底拆的是视频的什么给素材库做一次全面体检当你接手一个几万条视频的素材库客户问“库里有多少条 4K、多少条 H.265、哪些片子时长超过 10 分钟”靠人工打开播放器一个个看属性显然不现实靠文件名去猜更是玄学——有人叫“final_修2_最终版.mp4”有人叫“IMG_20240312_093021.mp4”命名规律根本不存在。video_parser 解决的就是这个场景把视频文件的元数据自动读出来输出成程序能直接用的 JSON 或字典让你用几行代码把整个素材库盘完。video_parser 本质上是元数据解析器不是播放器也不是转码器。它能告诉你时长、分辨率、编码格式、码率、帧率、音轨信息甚至旋转角度和 HDR 标记但不会去解码画面、抽帧或转码。适合它的人很明确做内容管理的后端工程师、自动化测试的脚本维护者、剪辑工具链的开发者以及任何需要批量处理视频素材的从业者。顺着这个标题搜进来的人多半是想自己动手做一套解析工具或者正在评估现成方案够不够用。这类工具最常见的实现是包一层 ffprobe 调用底层不做任何魔法解析系统里已装好的 ffprobe 输出返回结构化结果。下面从最小的命令开始把 video_parser 的落地方案从头到尾拆开。2. 先拆开 video_parser 的底层解析的是“声明信息”不是画面内容2.1 元数据从哪来容器层与编码层的分工ffprobe 怎么读要理解 video_parser 能干什么、不能干什么先要知道视频文件的分层结构。一个 MP4 文件由两层组成外层是容器container负责把视频轨、音频轨、字幕、章节组织在同一份文件里内层是编码数据视频轨是 H.264 或 H.265 编码的帧序列音频轨是 AAC 或 MP3 的采样数据。容器层有 moov、mdat 这样的 box 结构里面记录了时长、总码率、轨道列表每一轨的开头又带着编码参数比如分辨率、帧率、profile、像素格式。ffprobe 的工作方式是只读这两层的声明信息不去解码任何一帧的像素。它能把容器层的时长、码率、轨道类型以及编码层的 codec_name、width、height、r_frame_rate 等字段完整捞出来。这个特性带来两件好事第一解析速度极快一个几十 GB 的 4K 文件通常几百毫秒内就能返回结果因为它根本不碰画面数据第二解析不依赖 GPU 或高 CPU普通的批量任务用多线程也能扛住。但同时也带来一个边界它给出的时长和码率是文件声明的不是实际计算出来的。如果文件损坏或者录制设备写入时长不准拿到的 duration 可能和真实播放长度对不上。这条边界也划清了 video_parser 和帧级分析的界线。想要精确帧数、关键帧位置、场景切换点或者想判断画面内容黑场、人脸、字幕那要上 ffprobe 的 show_frames或者干脆走完整的解码链路代价是完全不同的量级。把这条界线画清楚后面排查时会省非常多时间。拿这类封装和 OpenCV 的 VideoCapture 对比会更直观OpenCV 打开视频后拿到的元数据同样来自容器层但它的定位是解码和帧处理打开一个文件会初始化解码器对恶劣文件的容忍度和 ffprobe 不在一个层级而 video_parser 这类工具只做探测拿完声明信息立即退出遇到损坏文件的失败方式是“拒绝”而不是“挂住”。这就是为什么素材盘点类任务的第一选择永远是 ffprobe 系的解析器而不是 OpenCV。2.2 最小可用命令一条 ffprobe 拿全所有元数据先说结论不管 video_parser 包装得多漂亮核心一定落在这条命令上ffprobe -v quiet -print_format json -show_format -show_streams sample.mp4逐项拆解-v quiet把日志级别压到最低避免 ffprobe 自身的 warning 混进标准输出否则 JSON 解析会直接报错-print_format json指定输出格式为 JSON这是后续程序解析的前提-show_format输出容器层信息包括时长、总码率、封装格式名-show_streams输出每一条轨道的详情视频轨、音频轨、字幕轨都在里面。两个开关一起给一次调用就能拿全所有关键字段。运行后你会得到一个顶层含streams数组和format对象的 JSON。format.format_name是容器格式常见值是 mp4、mov、matroskaformat.duration是声明时长单位秒format.bit_rate是总码率。streams数组里每个元素是一条轨道codec_typevideo的那条里才有 width、height、avg_frame_rate、r_frame_rate、profile、pix_fmtcodec_typeaudio的轨道带 sample_rate、channels、channel_layout。多音轨、多字幕轨会各自占一个元素所以写解析逻辑时永远不要假设 streams 只有一条。常见字段可以整理成一张对照表写 SQL 或做字段映射时直接照着抄字段来源层级类型含义format.format_nameformatstring容器封装格式format.durationformatstring声明时长秒format.bit_rateformatstring总码率bpsstreams[].codec_typestreamstringvideo / audio / subtitlestreams[].codec_namestreamstringh264 / hevc / aac / mp3streams[].width / heightvideo streamint编码层宽高streams[].r_frame_ratevideo streamstring分数形式帧率如 30000/1001streams[].pix_fmtvideo streamstring像素格式如 yuv420p注意这个表里有三个字段的值是字符串duration、bit_rate、r_frame_rate。ffprobe 的 JSON 输出里它们都不是数字。r_frame_rate 更特殊它是一个分数表达式30000/1001才是真正的 29.97fps。太多解析器在这三个字段上直接 int() 或 float() 翻车稍后封装代码里会用专门的函数处理。提示r_frame_rate 在 JSON 里是字符串形式的分数比如 30000/1001转 float 前要先拆开计算。这条命令本身也可以作为手工排坑的利器。当上层封装返回的字段和预期不符先手动跑一遍 ffprobe 看原始 JSON能快速定位是容器数据的问题还是中间解析逻辑的问题。我一般会把原始 JSON 落一份到临时文件再谈上层处理排查成本会低很多。2.3 封装成 video_parser 核心函数subprocess 调用与 JSON 落地命令行能跑通封装就是水到渠成的事。最朴素的 video_parser 核心函数用 Python 的 subprocess 调 ffprobe再把 stdout 交给 json.loadsimport subprocess import json def parse_video(path: str, timeout: int 30) - dict: cmd [ ffprobe, -v, quiet, -print_format, json, -show_format, -show_streams, path ] try: proc subprocess.run( cmd, capture_outputTrue, textTrue, timeouttimeout ) proc.check_returncode() return json.loads(proc.stdout) except subprocess.TimeoutExpired: raise TimeoutError(f解析超时: {path}) except subprocess.CalledProcessError as e: raise RuntimeError(fffprobe 非零退出: {path}, stderr{e.stderr}) except json.JSONDecodeError: raise ValueError(fffprobe 输出不是合法 JSON: {path})这里几个参数必须讲清楚。textTrue让 stdout 以字符串返回否则拿到的是 bytes后面 json.loads 会报类型错误timeout30是给解析上的保险正常情况下 ffprobe 秒回但遇到网络流或损坏文件可能挂住没有 timeout 整个任务会卡死check_returncode()把 ffprobe 的非零退出码转成异常文件不存在、权限不足时 ffprobe 会往 stderr 写错误并退出这一步把环境问题集中拦在异常分支里。异常要分开处理不要三种失败混成一个。超时多半是文件在远端或 IO 卡住非零退出多半是文件不存在或根本不是视频JSON 解析失败通常意味着 ffprobe 版本太老、输出被干扰、或者有人改了 -v 级别导致 warning 混入。分开抛异常上层才能按类型决定重试、跳过还是告警。如果你想要返回值更符合业务习惯可以在核心函数里顺手做两层整理。第一层是把 streams 按 codec_type 分组抽出 video、audio 两个子列表第二层是把字符串字段转成真正的数字把 r_frame_rate 算成 float。下面是一段常用的整理代码def normalize(raw: dict) - dict: streams raw.get(streams, []) video next((s for s in streams if s.get(codec_type) video), None) audio next((s for s in streams if s.get(codec_type) audio), None) def to_float(v): try: return float(v) if v is not None else None except (TypeError, ValueError): return None def to_fps(s): if not s: return None try: num, den s.get(avg_frame_rate, 0/0).split(/) return round(float(num) / float(den), 3) if float(den) else None except (ValueError, AttributeError): return None fmt raw.get(format, {}) return { container: fmt.get(format_name), duration: to_float(fmt.get(duration)), bit_rate: to_float(fmt.get(bit_rate)), video_codec: video.get(codec_name) if video else None, width: video.get(width) if video else None, height: video.get(height) if video else None, fps: to_fps(video), audio_codec: audio.get(codec_name) if audio else None, }注意这里帧率取的是avg_frame_rate而不是r_frame_rate。两者的区别是r_frame_rate 是容器里记录的基础帧率avg_frame_rate 是按时长和帧数算出的平均帧率对可变帧率VFR视频avg_frame_rate 更接近真实观感。对于固定帧率视频两者几乎一致。normalize 之后业务侧拿到的就是一个干净、完全是基本类型的 dict可以直接进 JSON 落盘或 SQLite 入库不需要再操心字段名和字符串转换。核心函数到这一步video_parser 的最小可用版本就成立了。3. 把 video_parser 做成批量工具目录扫描、多线程与结果入库3.1 从单文件到目录先解决“哪些文件要解析”单文件函数写好后第一个真实需求必然是“把整个目录下的视频全部解析掉”。这一步的关键不是调用解析函数而是想清楚文件从哪来。常见的做法是用 pathlib 的 rglob 做递归遍历按扩展名过滤from pathlib import Path VIDEO_EXTS {.mp4, .mov, .mkv, .avi, .flv, .ts, .m4v} def find_videos(root: str): root_path Path(root) for p in root_path.rglob(*): if p.is_file() and p.suffix.lower() in VIDEO_EXTS: yield p用生成器而不是直接返回列表是为了避免大目录下先把所有路径一次性装进内存。p.suffix.lower()这步不能省相机直出、Windows 导出的文件后缀大小写不统一是常态。扩展名过滤只是粗筛它不能识别“改了后缀但根本不是视频”的文件正确判断靠 ffprobe 返回的format.format_name扫描阶段不要试图做过多验证那是解析阶段的事。这里有两个容易被忽略的工程细节。第一rglob 会遍历整个目录树包括.thumbnails、eaDir这类隐藏目录里面可能有一堆缩略图和 .mp4 后缀的残留文件。扫描时先过滤掉隐藏目录批量任务会干净很多。第二符号链接和硬链接会导致同一个文件被扫到多次如果素材库里有链接结构解析前先用resolve()做一次路径去重。原理很简单解析一个 4K 文件要几百毫秒重复解析十次就是几秒的浪费累积下来很可观。还有一种更稳的扫描策略如果目录结构本身有规律比如按日期或项目分了子目录直接让调用方传入要扫的具体目录列表而不是扫根目录。这样既能跳过无关区域又能让扫描范围成为可审计的配置项。批量工具的输入越可控输出就越可信。全量重扫一遍一万个文件单线程大约要五十分钟开了 8 线程能压到七八分钟但前提是扫描范围本身没有浪费否则多线程也救不回来。3.2 多线程解析与异常隔离别让一个坏文件拖垮全批ffprobe 是外部进程Python 调 subprocess 时不持有 GIL所以批量解析天然适合多线程。用 concurrent.futures.ThreadPoolExecutor 是最省事的写法from concurrent.futures import ThreadPoolExecutor, as_completed def parse_batch(paths, max_workers8): results {} with ThreadPoolExecutor(max_workersmax_workers) as pool: future_map {pool.submit(parse_video, p): p for p in paths} for future in as_completed(future_map): p future_map[future] try: results[str(p)] future.result() except Exception as exc: results[str(p)] {error: str(exc)} return resultsworkers 数量怎么定ffprobe 是 IO 密集型进程瓶颈通常在磁盘和进程启动开销不在 CPU。机械盘建议 4普通 SSD 可以上 8NVMe 上 16 也不会有压力。但要注意如果同一台机器同时还在做转码或剪辑workers 要主动让路否则 ffprobe 会和转码进程抢 IO导致两边都变慢。不要用 Python 的普通 thread 去调 ffprobe 以外的重活也不要把 max_workers 调到和文件数一样多进程启动成本会反噬总耗时。异常隔离是批量任务的生命线。一个坏文件直接抛异常整个批次都会中断。上面的代码把每个文件的异常捕获后塞回结果字典用{error: ...}标记失败项最后统一看哪些文件需要人工复核这是批量解析任务最基本的容错姿态。更进一步的做法是给失败文件单独维护一个列表重试一次后再判死刑很多 IO 抖动造成的失败重试一次就过了。批量任务跑久了能跑出多少条“错误”是确定的真正重要的问题是你能不能从结果里一眼看出哪些是环境问题、哪些是文件问题。我习惯在结果里把 error 信息原样保留而不是只存一个布尔值这样筛日志的时候能按 stderr 内容批量归类。比如同样是失败Permission denied和Invalid data found的处理路径完全不同把原始错误丢掉了排查就得重新解析一遍。3.3 输出去向JSON 落盘、JSONL 流式写、SQLite 入库解析结果如果只是存在内存里等于白跑。批量任务通常有两个出口调试阶段落 JSON 文件生产阶段写 SQLite。先看最简单的 JSON 落盘def dump_results(results: dict, out_path: str): with open(out_path, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2)ensure_asciiFalse必须写否则中文路径会被转成\uXXXX转义序列排查时眼睛直接废掉。indent2只是让文件可读如果结果要交给下游程序消费可以去掉 indent 减小体积。文件数上万时一次性 dump 整个 dict 有两个风险内存峰值高、写盘时万一中断丢全部。更稳妥的是 JSONL每解析完一个文件立刻 append 一行天然流式不怕中断def parse_to_jsonl(paths, out_path): with open(out_path, w, encodingutf-8) as f: for p in paths: try: data parse_video(p) except Exception as exc: data {path: str(p), error: str(exc)} f.write(json.dumps(data, ensure_asciiFalse) \n)生产环境我一般直接写 SQLite。先建表再用 INSERT OR REPLACE 做幂等写入重复跑脚本不会产生重复记录import sqlite3 import datetime SCHEMA CREATE TABLE IF NOT EXISTS videos ( path TEXT PRIMARY KEY, duration REAL, width INTEGER, height INTEGER, codec_name TEXT, bit_rate INTEGER, fps REAL, audio_codec TEXT, parsed_at TEXT ); def init_db(db_path: str) - sqlite3.Connection: conn sqlite3.connect(db_path) conn.execute(SCHEMA) return conn def upsert_video(conn, path: str, meta: dict): conn.execute( INSERT OR REPLACE INTO videos (path, duration, width, height, codec_name, bit_rate, fps, audio_codec, parsed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , ( path, meta.get(duration), meta.get(width), meta.get(height), meta.get(video_codec), meta.get(bit_rate), meta.get(fps), meta.get(audio_codec), datetime.datetime.now().isoformat() ))表结构的设计原则是“按查询需求建字段”不要一上来就把所有可能用到的字段都塞进去。上面这个表对应的是素材盘点最常见的四类查询按时长、分辨率、编码、码率筛选。后续要加字段ALTER TABLE 加一列即可不要推翻重建。主键用文件路径做自然键对本地素材库够用如果文件会被移动或重命名再考虑换成内容 hash。批量解析到入库的完整链路建议分两步跑第一步解析并写 JSONL 或 SQLite第二步从库里做校验和报表。不要把“解析”和“分析”混在一个脚本里两步分开才能让中间结果可复现。毕竟批量任务的定位是产生可信数据不是产生一份只跑一次的报告。4. video_parser 高频翻车现场四个必踩的坑与排查路径4.1 现象解析秒回但 duration 是 0.0视频明明有内容这个坑在两类文件上最经常出现监控摄像头导出的 MP4以及某些手机厂商的录屏文件。原因在容器层文件的 moov box 里没有写入时长或者时长字段本身就是 0。ffprobe 忠实返回容器里声明的 0.0这不是它算错了而是文件没声明。如果直接拿这个 duration 去做时长筛选这些文件会全部被误判成无效视频。解决路径分两步。第一步看视频轨里有没有 nb_frames如果有用帧数除以帧率自己算真实时长。第二步如果 nb_frames 也没有只能让 ffprobe 实际数一遍帧ffprobe -v error -count_frames -select_streams v:0 \ -show_entries streamnb_read_frames,r_frame_rate input.mp4-count_frames会逐帧读取视频轨代价是解析时间从毫秒级变成秒级甚至更久所以不要把这条命令写进常规批量流程只作为异常分支的兜底。判断逻辑放在“duration 为 0 或缺失时才走兜底”的位置而不是每个文件都强制数帧。批量任务里遇到 duration0 的文件先单独收集到一个列表最后统一走一遍兜底效率最高。注意duration 为 0 的文件要标记而不是删除等兜底数据出来后决定去留。素材库清理时最怕的就是依据错误元数据做删除操作删了就找不回来了。4.2 现象中文路径、带空格的文件一拍就崩很多人第一次批量解析翻车就翻在路径上然后怀疑 subprocess 的转义有问题。实际上 Python 的 subprocess 传参是列表形式不会走 shell 转义空格和中文本身不会导致 ffprobe 失败。真正会崩的是三种情况。第一种路径以-开头ffprobe 把它当成命令行选项而不是文件。解法是传参时在路径前补-iffprobe 的所有工具都支持用-i显式声明输入。第二种路径里带了换行符或不可见字符多从网盘、FTP 下载后出现这类文件在扫描阶段就应该单独捞出来不要混进批量任务。第三种Windows 下路径过长或含特殊字符导致进程创建失败这时要先把路径转成短路径属于系统层面的限制。一个容易被误伤的案例是开发者在 mac 上解析/tmp/测试 视频.mp4没事到 Windows 上同样的代码就崩于是归咎于中文。实际上多半是分隔符和权限问题和编码无关。排查这类问题最快的方式是打印传给 ffprobe 的完整参数列表亲眼确认路径没有被动过手脚。血泪经验是永远不要在上层对路径做 replace 或所谓“清洗”除非你知道自己在干什么很多 bug 反而是清理逻辑引入的。def safe_probe(path: str): # 绝对路径可以避免 ffprobe 把以 - 开头的文件名当成选项 abs_path os.path.abspath(path) cmd [ffprobe, -v, quiet, -print_format, json, -show_format, -show_streams, -i, abs_path] ...注意os.path.abspath会把相对路径变成绝对路径依赖当前工作目录。如果你的程序最终会在守护进程里跑工作目录可能不是你以为的那个建议入口处统一把工作目录固定下来再调用。4.3 现象批量解析几百个文件后内存暴涨如果结果字典里保留每个文件的完整 streams 数组一个文件包含几十条轨道时解析结果就有好几 KB万一某条流的 tags 里有大段嵌入字幕或录制参数单文件结果能到几十 KB。一万个文件累加内存占用直接以 GB 计。这不是 ffprobe 的问题是上层把“全量原始数据”和“业务需要的数据”混为一谈了。解法有两个方向。第一入库前只抽需要的字段不保留原始 JSON。上一章的 normalize 函数就是为了这个而存在的正式批量任务里不要保留完整原始结果内存立刻降下来。第二如果确实要保留原始 JSON 做审计就不能一次性攒完再写盘要改成流式写 JSONL每个文件解析完立刻写一行。判断内存是否健康的简单标准一万条素材解析结果字典占用的内存不应超过 200 MB。如果明显超标第一件事就是查结果里有没有塞了非业务字段。还有个小技巧用tracemalloc看看峰值内存到底被谁吃了不要凭感觉优化凭感觉优化出来的代码往往比原来更难维护。4.4 现象竖屏视频的 width/height 和预期相反手机拍摄的竖屏视频文件里记录的画面是横向编码的真正的显示方向靠容器里的 rotation 元数据标定。部分 ffprobe 版本、部分封装下拿到的 width 和 height 就是编码层原始值手机上拍的竖屏视频会返回 width1920, height1080和预期完全相反。rotation 以 90 度为步进常见取值为 0、90、180、270存放在视频轨的 side_data 里。判断竖屏不能直接比较 width 和 height 大小要先读 rotation。下面是常用的辅助函数def effective_size(video_stream: dict) - tuple: rotation 0 for side in video_stream.get(side_data_list, []): if side.get(side_data_type) Display Matrix: rotation int(float(side.get(rotation, 0))) width video_stream.get(width) or 0 height video_stream.get(height) or 0 if rotation in (90, 270): return height, width return width, height注意 rotation 字段在 JSON 里可能是字符串 90 而不是数字 90int(float(...)) 是为了兼容两种输出。这个坑影响最严重的是封面生成和播放器截图方向如果在解析阶段不处理 rotation后面每一环节的宽高判断都会跟着错。素材盘点如果只关心分辨率档位对 90/270 这种旋转视频要避免把竖屏素材统计进横屏分类。这里还衍生出另一个提醒ffprobe 的版本对 side_data 的输出格式不统一老版本可能不会输出 rotation 字段。量产环境建议固定 ffprobe 版本至少在同一批任务里版本一致否则同一个文件在不同机器上解析结果不同排查起来极其痛苦。5. 把 video_parser 接进真实业务转码预判、素材盘点与兜底策略5.1 用解析结果做转码预判先算账再决定动不动文件转码是所有视频业务里最贵的操作CPU 占用和耗时都是真金白银。用 video_parser 的结果可以在转码前把账算清楚如果源文件本身就是 H.264、分辨率不超过目标值、码率在预期区间直接复制或硬链接完全不需要转码。反过来H.265 要转 H.264或分辨率远高于目标才进转码队列。这个判断逻辑可以收敛成一张决策表场景元数据条件动作直接复用codec_nameh264 且 width≤目标 且 bit_rate≤阈值复制文件不转码需要重编码codec_namehevc 或 width目标 或 bit_rate阈值进转码队列需要人工确认duration 为 0 或 format_name 未知走兜底策略直接拒绝无视频轨或解析异常记录告警跳过这张表的价值在于把“要不要转码”从拍脑袋变成可审计的规则。实现上就是几行 if 分支但要特别注意判断顺序先处理异常和缺失字段再处理直接复用和重编码否则一个 duration0 的坏文件可能被错误地送进转码队列。转码队列吃掉的资源可比解析大得多判断上宁可保守。有人会问码率阈值怎么定这没有通用答案要按你的播放端和带宽来定。常见的经验是1080p 的 H.264 视频码率超过 20 Mbps 属于高码率值得重编码4K 超过 60 Mbps 同理。但更靠谱的做法是先统计素材库的码率分布取 P90 作为阈值避免少数电影级素材把阈值拉高导致大量文件误入重编码队列。元数据先行收益其实是在转码调度上省出来的电费和工时。代码实现上把决策逻辑单独放一个函数接收 normalize 后的 dict返回一个动作枚举而不是散落在调用方里def decide_action(meta: dict, max_width: int, max_bitrate: float) - str: if meta[width] is None or meta[video_codec] is None: return review if meta[bit_rate] is not None and meta[bit_rate] max_bitrate \ and meta[width] max_width and meta[video_codec] h264: return copy if meta[video_codec] in (hevc, vp9) or meta[width] max_width: return transcode return review这里把 h264 直接复用作为最优先判断因为低分辨率的 H.265 文件虽然可以直接复用但多数播放端兼容性差宁可转一道。这是业务取舍不是纯技术判断video_parser 的角色是提供准确输入决策权留在业务层。5.2 素材盘点报表把解析结果变成业务能读的数字盘点需求通常是“按分辨率分组看数量”“按编码格式看占比”“找出码率异常的片子”。这些可以直接在 SQLite 上跑聚合查询SELECT codec_name, COUNT(*) AS cnt FROM videos WHERE codec_name IS NOT NULL GROUP BY codec_name ORDER BY cnt DESC;分辨率分组统计要稍微讲究一点。直接按 width 和 height 精确分组会把 1920×1080 和 1920×816 分成两个桶业务上通常没法接受。常见做法是先按长边分档SELECT CASE WHEN MAX(width, height) 1280 THEN 720p以下 WHEN MAX(width, height) 1920 THEN 1080p以下 WHEN MAX(width, height) 3840 THEN 4K以下 ELSE 4K以上 END AS tier, COUNT(*) AS cnt FROM videos WHERE width IS NOT NULL AND height IS NOT NULL GROUP BY tier ORDER BY cnt DESC;这里用MAX(a, b)标量函数取长边SQLite 较老的版本对多参数标量函数的支持有差异稳妥做法是先执行一条SELECT MAX(1920, 1080)验证一下环境。我见过一次因为服务器上的 SQLite 是古董版本函数本身能用但查询计划异常慢最后是换了新版本才解决排查了半天。报表想要更有业务价值还可以加上按时间维度的增量统计今天新增多少文件、新增文件的编码分布如何这需要表里有一个代表入库时间的字段比如 parsed_at。字段在首次建表时就规划好比后面再补要省事。盘点报表的最佳实践是“一次生成多处使用”跑一个脚本把聚合结果写进统计表业务侧只读统计表不要在报表每次打开时全表扫一遍视频表。几万条记录全表扫描其实也不慢但这是习惯问题等视频到了百万级这个习惯能帮你避免一次大的性能返工。5.3 解析失败的兜底策略宁可漏报不可崩批批量解析里总会有几个失败文件这是写在剧本里的确定性事件不是意外。兜底策略分等级不能一刀切。文件不存在或权限不足的记录告警这属于环境问题重试也可能失败不该反复跑文件存在但 ffprobe 返回非零的可能是伪装成视频的文件或文件头损坏单独列到 review 队列留待人工检查唯一值得重试的是超时因为网络存储或 IO 抖动导致的超时重试两次后成功率极高闭环如下def parse_with_retry(path, retries2, timeout30): for attempt in range(retries 1): try: return parse_video(path, timeouttimeout) except TimeoutError: continue except RuntimeError: raise raise TimeoutError(f重试 {retries} 次仍超时: {path})重点在于 RuntimeError 不重试文件不存在这种确定性错误重试多少次结果都一样只会拖慢队列。把“可重试”和“不可重试”分开是批量任务能不能跑完的分水岭。我习惯在任务里维护 ok、retry、dead 三档结果retry 档自动重试一次dead 档写进日志并汇总到每日告警。还有一个提醒解析失败的记录也要落库不能只在日志里。失败本身就是数据后续可以用它反查哪些目录容易出问题、哪些设备产出的文件格式不规范。如果没有失败记录统计永远只能看到成功样本问题会被掩盖。兜底策略的最终目标不是消灭失败而是让失败可见、可分类、不拖垮主流程。6. 最后一招只解析变化的文件——用增量缓存把重复劳动砍掉八成批量解析跑第二遍的时候绝大多数文件的元数据根本没变全量重跑就是在浪费磁盘 IO 和机器时间。给 video_parser 加一层增量缓存是最划算的最后一公里优化。做法并不复杂。解析前先取文件的mtime_ns和size两个值和缓存表里的记录比对没变就直接读缓存结果变了才重新调 ffprobe。为什么不用内容 hash因为内容 hash 得把整个文件读一遍对几十 GB 的视频来说代价比解析本身还大得不偿失。mtime 加 size 的组合在本地素材库上足够可靠。def cached_parse(conn, path: str, timeout: int 30) - dict: st os.stat(path) row conn.execute( SELECT mtime, size, result FROM parse_cache WHERE path ?, (path,) ).fetchone() if row and row[0] st.st_mtime_ns and row[1] st.st_size: return json.loads(row[2]) result parse_video(path, timeouttimeout) conn.execute( INSERT OR REPLACE INTO parse_cache (path, mtime, size, result) VALUES (?, ?, ?, ?), (path, st.st_mtime_ns, st.st_size, json.dumps(result, ensure_asciiFalse)) ) return result缓存表独立建一个 db 文件和生产数据表分开。这样缓存随时可以删掉重建不影响业务表。每当你改了解析字段映射或者换了 ffprobe 版本直接删掉缓存表全量重跑一次避免新旧结构混在同一个缓存里。注意mtime 加 size 的缓存能挡住绝大多数重复解析但它不是内容级的完整性校验缓存必须设计失效出口。增量缓存最大的坑是“元数据变了但 mtime 没变”比如某些工具原地改写文件头却不更新时间戳。对这种边界我的态度是容忍因为这类工具极少而且通常后续操作会更新 mtime如果你真的被咬过一次就退一步改成每周一次全量无缓存解析来校正。没有失效出口的缓存迟早变成脏数据的温床。我维护这类解析工具时还有个习惯把批量任务拆成 scan、parse、report 三步每步的中间产物都落盘。scan 产出待解析清单parse 消费清单写缓存和业务表report 读业务表出统计。哪一步挂了就从哪一步重跑不需要从头再来。这比一个脚本从头跑到尾可控得多。希望这个增量缓存和分步任务的设计能帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
