一万条视频怎么体检：我用 ffprobe 搭的批量质检流水线，以及真正能查出问题的那一步

ffprobe -v error -show_format -show_streams -of json input.mp4

ffprobe -v error \
  -show_entries format=duration,size,bit_rate,format_name \
  -show_entries stream=codec_type,codec_name,width,height,avg_frame_rate \
  -of default=noprint_wrappers=1 input.mp4

ffmpeg -v error -i input.mp4 -f null - 2>error.log

[mov,mp4,m4a,3gp,3g2,mj2 @ 0x...] moov atom not found
[h264 @ 0x...] error while decoding MB 32 18, bytestream -7
[aac @ 0x...] Number of bands exceeds limit

# 抽样验证：只解最后 30 秒（截断文件通常坏在尾部）
ffmpeg -v error -sseof -30 -i input.mp4 -f null - 2>&1 | head -20

timeout 60 ffprobe -v error -show_format -of json input.mp4

ffmpeg -v error -i bad.mp4 -f null - 2>&1 | head -50

# xargs 并发（-P 4 表示 4 路）
find /data -name '*.mp4' -print0 | xargs -0 -P 4 -I{} sh -c 'ffprobe -v error -show_format -of json "{}" > "{}.json"'

find /data -name '*.mp4' | parallel -j 8 --timeout 60 'ffprobe -v error -show_format -of json {} > {}.json'

总计: 10432 条
─────────────────────────────
P0 致命        :  187 条 (1.8%)
P1 需处理      :  643 条 (6.2%)
P2 仅记录      : 1094 条 (10.5%)
健康           : 8508 条 (81.6%)
─────────────────────────────
P0 细分:
  无法读取        :  92
  解码失败        :  71
  无视频流        :  15
  空文件          :   9
─────────────────────────────
编码分布:
  h264           : 9214
  hevc           :  731
  mpeg4          :  402
  其他/未知      :   85
─────────────────────────────
需要转码(非hevc): 9701 条
预计可节省空间  : ~38%（按 H.265 CRF 26 估算）

path,size_mb,duration_s,vcodec,acodec,width,height,fps,bitrate_kbps,level,issues
/data/a/001.mp4,1024.3,1832.5,h264,aac,1920,1080,29.97,4680,OK,
/data/a/002.mp4,0.0,,,unknown,,0,0,,P0,empty_file
/data/b/003.mp4,512.1,1830.0,h264,,1280,720,25.00,2240,P1,no_audio;duration_mismatch
/data/c/004.mp4,880.2,1200.0,h264,aac,1920,1080,29.97,6000,P0,decode_error:MB_bytestream

# 把 P0 文件移到隔离区
awk -F, '$10=="P0" {print $1}' report.csv | while read f; do mv "$f" /quarantine/; done

#!/usr/bin/env python3
"""视频批量质检：元数据检查 + 解码验证，输出 CSV 明细与汇总。"""
import csv, json, subprocess, sys
from concurrent.futures import ThreadPoolExecutor
from fractions import Fraction
from pathlib import Path

PROBE_TIMEOUT = 60        # 元数据读取超时（秒）
DECODE_TIMEOUT = 900      # 解码验证超时（秒）
WORKERS = 16              # 并发数

VIDEO_EXTS = {'.mp4', '.mkv', '.ts', '.mov', '.avi', '.flv', '.webm', '.m4v'}


def run(cmd, timeout):
    try:
        p = subprocess.run(cmd, capture_output=True, timeout=timeout)
        return p.returncode, p.stdout, p.stderr
    except subprocess.TimeoutExpired:
        return -1, b'', b'TIMEOUT'


def probe(path):
    code, out, err = run(
        ['ffprobe', '-v', 'error', '-show_format', '-show_streams',
         '-of', 'json', str(path)], PROBE_TIMEOUT)
    if code != 0:
        return None
    try:
        return json.loads(out.decode('utf-8', 'replace'))
    except json.JSONDecodeError:
        return None


def pick_streams(info):
    v = next((s for s in info.get('streams', []) if s.get('codec_type') == 'video'), None)
    a = next((s for s in info.get('streams', []) if s.get('codec_type') == 'audio'), None)
    return v, a


def fps_of(stream):
    """帧率在 ffprobe 里是分数字符串，要转成浮点数。"""
    raw = (stream or {}).get('avg_frame_rate', '0/0')
    try:
        return round(float(Fraction(raw)), 3)
    except (ValueError, ZeroDivisionError):
        return 0.0


def check(path):
    rec = {'path': str(path), 'size_mb': round(path.stat().st_size / 1048576, 2)}
    issues, level = [], 'OK'

    if path.stat().st_size < 1024:
        return {**rec, 'level': 'P0', 'issues': 'empty_file'}

    info = probe(path)
    if info is None:
        return {**rec, 'level': 'P0', 'issues': 'probe_failed'}

    fmt = info.get('format', {})
    v, a = pick_streams(info)

    if not v:
        return {**rec, 'level': 'P0', 'issues': 'no_video_stream'}

    # 元数据字段
    rec.update({
        'duration_s': round(float(fmt.get('duration') or 0), 2),
        'vcodec': v.get('codec_name', ''),
        'acodec': (a or {}).get('codec_name', ''),
        'width': v.get('width', 0),
        'height': v.get('height', 0),
        'fps': fps_of(v),
        'bitrate_kbps': round(int(fmt.get('bit_rate') or 0) / 1000),
    })

    # P1 规则
    if not a:
        issues.append('no_audio')
    if rec['duration_s'] < 1:
        issues.append('duration_zero')
    if rec['width'] <= 0 or rec['height'] <= 0:
        issues.append('bad_resolution')
    if rec['bitrate_kbps'] and rec['bitrate_kbps'] < 100:
        issues.append('bitrate_too_low')
    if rec['vcodec'] == 'h264' and rec['bitrate_kbps'] > 80000:
        issues.append('bitrate_abnormal_high')

    # 解码验证（只对通过元数据检查的做，省时间）
    if not issues:
        code, _, err = run(['ffmpeg', '-v', 'error', '-i', str(path),
                            '-f', 'null', '-'], DECODE_TIMEOUT)
        errtxt = err.decode('utf-8', 'replace').strip()
        if code == -1:
            issues.append('decode_timeout'); level = 'P0'
        elif errtxt:
            issues.append('decode_error:' + errtxt.splitlines()[0][:80])
            level = 'P0'

    if issues and level == 'OK':
        level = 'P1'
    return {**rec, 'level': level, 'issues': ';'.join(issues)}


def main(root, out_csv):
    files = [p for p in Path(root).rglob('*') if p.suffix.lower() in VIDEO_EXTS]
    print(f'发现 {len(files)} 个文件，开始检查...', file=sys.stderr)

    with ThreadPoolExecutor(max_workers=WORKERS) as ex, \
            open(out_csv, 'w', newline='', encoding='utf-8-sig') as f:
        fields = ['path', 'size_mb', 'duration_s', 'vcodec', 'acodec',
                  'width', 'height', 'fps', 'bitrate_kbps', 'level', 'issues']
        w = csv.DictWriter(f, fieldnames=fields, extrasaction='ignore')
        w.writeheader()
        for i, rec in enumerate(ex.map(check, files), 1):
            w.writerow(rec)
            if i % 200 == 0:
                print(f'  已处理 {i}/{len(files)}', file=sys.stderr)


if __name__ == '__main__':
    main(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else 'report.csv')

python3 qa_check.py /data/videos report.csv

ffmpeg -i input.mp4 -vf blackdetect=d=2:pix_th=0.10 -f null - 2>&1 | grep blackdetect

[blackdetect @ 0x...] black_start:0 black_end:15.2 black_duration:15.2

ffmpeg -i input.mp4 -af silencedetect=n=-50dB:d=2 -f null - 2>&1 | grep silence

# 每 10% 抽一帧，拼成一张 5x2 的图
ffmpeg -i input.mp4 -vf "select='not(mod(n\,300))',scale=320:-1,tile=5x2" \
  -frames:v 1 contact_sheet.jpg

ffmpeg -i encoded.mp4 -i source.mp4 -lavfi libvmaf -f null -