交付前的最后一道关：视频自动验收清单与脚本

{
  "name": "客户A-课程交付-v2",
  "container": {
    "format": "mp4",
    "require_faststart": true,
    "max_size_mb": 2048
  },
  "video": {
    "codec": "h264",
    "width": 1920,
    "height": 1080,
    "fps": [24, 25, 30],
    "pix_fmt": ["yuv420p"],
    "profile": ["high", "main"],
    "max_level": 41,
    "bitrate_kbps": {"min": 1500, "max": 8000},
    "color_space": "bt709",
    "color_range": "tv"
  },
  "audio": {
    "codec": "aac",
    "sample_rate": [44100, 48000],
    "channels": 2,
    "min_bitrate_kbps": 128
  },
  "quality": {
    "require_clean_decode": true,
    "max_black_seconds": 3.0,
    "max_frozen_seconds": 2.0,
    "duration_tolerance_seconds": 1.0
  },
  "naming": {
    "pattern": "^[A-Z]{2}-\\d{4}-.+\\.mp4$"
  }
}

#!/usr/bin/env python3
"""视频交付验收。用法：python accept.py file.mp4 --spec specs/customer_a.json"""
import json
import re
import subprocess
import sys
from pathlib import Path


def probe(path: str) -> dict:
    """ffprobe 拿全部信息。"""
    cmd = ['ffprobe', '-v', 'error', '-show_format', '-show_streams',
           '-of', 'json', path]
    out = subprocess.run(cmd, capture_output=True, text=True, check=True).stdout
    return json.loads(out)


class Result:
    def __init__(self):
        self.items = []

    def check(self, name, ok, detail='', level='fail'):
        self.items.append({'name': name, 'ok': ok, 'detail': detail, 'level': level})

    @property
    def failed(self):
        return [i for i in self.items if not i['ok'] and i['level'] == 'fail']

    @property
    def warned(self):
        return [i for i in self.items if not i['ok'] and i['level'] == 'warn']


def accept(path: str, spec: dict) -> Result:
    r = Result()
    info = probe(path)
    fmt = info['format']
    v = next((s for s in info['streams'] if s['codec_type'] == 'video'), None)
    a = next((s for s in info['streams'] if s['codec_type'] == 'audio'), None)

    # ---- 容器层 ----
    fmt_name = fmt.get('format_name', '')
    r.check('容器格式', spec['container']['format'] in fmt_name,
            f'实际: {fmt_name}')

    if spec['container'].get('require_faststart'):
        r.check('faststart', moov_at_front(path), 'moov 不在文件开头')

    size_mb = int(fmt.get('size', 0)) / 1024 / 1024
    r.check('文件大小', size_mb <= spec['container']['max_size_mb'],
            f'{size_mb:.1f} MB')

    # ---- 视频流 ----
    if v is None:
        r.check('视频流存在', False, '没有视频流')
        return r

    r.check('视频编码', v.get('codec_name') == spec['video']['codec'],
            f"实际: {v.get('codec_name')}")
    r.check('分辨率', v.get('width') == spec['video']['width'] and
            v.get('height') == spec['video']['height'],
            f"{v.get('width')}x{v.get('height')}")

    fps = eval_fps(v.get('avg_frame_rate', '0/0'))
    r.check('帧率', any(abs(fps - f) < 0.1 for f in spec['video']['fps']),
            f'{fps:.2f}')

    r.check('像素格式', v.get('pix_fmt') in spec['video']['pix_fmt'],
            f"实际: {v.get('pix_fmt')}")
    r.check('profile', v.get('profile') in spec['video']['profile'],
            f"实际: {v.get('profile')}")

    vb = int(v.get('bit_rate') or fmt.get('bit_rate') or 0) / 1000
    rng = spec['video']['bitrate_kbps']
    r.check('视频码率', rng['min'] <= vb <= rng['max'], f'{vb:.0f} kbps',
            level='warn')      # 码率超标一般只警告

    # ---- 音频流 ----
    if a is None:
        r.check('音频流存在', False, '没有音轨')
    else:
        r.check('音频编码', a.get('codec_name') == spec['audio']['codec'],
                f"实际: {a.get('codec_name')}")
        sr = int(a.get('sample_rate', 0))
        r.check('采样率', sr in spec['audio']['sample_rate'], f'{sr} Hz')
        r.check('声道数', a.get('channels') == spec['audio']['channels'],
                f"{a.get('channels')}")
        ab = int(a.get('bit_rate') or 0) / 1000
        r.check('音频码率', ab >= spec['audio']['min_bitrate_kbps'],
                f'{ab:.0f} kbps', level='warn')

    # ---- 质量层 ----
    if spec['quality'].get('require_clean_decode'):
        errs = decode_check(path)
        r.check('解码无错误', not errs, errs[:500])

    dur = float(fmt.get('duration', 0))
    nb = int(v.get('nb_frames') or 0)
    if nb and fps:
        calc = nb / fps
        r.check('时长与帧数一致', abs(calc - dur) <= spec['quality']['duration_tolerance_seconds'],
                f'容器 {dur:.1f}s vs 帧数推算 {calc:.1f}s')

    # ---- 命名 ----
    pat = spec.get('naming', {}).get('pattern')
    if pat:
        r.check('文件名规范', re.match(pat, Path(path).name) is not None,
                Path(path).name, level='warn')

    return r


def eval_fps(rate: str) -> float:
    """'30000/1001' -> 29.97"""
    try:
        n, d = rate.split('/')
        return float(n) / float(d) if float(d) else 0.0
    except Exception:
        return 0.0


def decode_check(path: str, timeout: int = 600) -> str:
    """完整解码一遍，返回错误信息（空字符串表示没问题）。"""
    cmd = ['ffmpeg', '-v', 'error', '-i', path, '-f', 'null', '-']
    p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return (p.stderr or '').strip()

ffmpeg -v error -i file.mp4 -f null - 2>err.log
# err.log 为空 = 能完整解码

dur_container = float(fmt['duration'])       # 容器声明的时长
nb_frames = int(v['nb_frames'])              # 实际帧数
fps = 帧率
dur_calc = nb_frames / fps                   # 推算的时长

ffprobe -v error -count_frames -select_streams v:0 \
  -show_entries stream=nb_read_frames -of csv=p=0 file.mp4

def moov_at_front(path: str, scan_bytes: int = 2 * 1024 * 1024) -> bool:
    """粗略判断 moov 是否在文件开头（读前 2MB 找 'moov'）。"""
    with open(path, 'rb') as f:
        head = f.read(scan_bytes)
    # ftyp 之后应该很快出现 moov
    idx = head.find(b'moov')
    if idx == -1:
        return False          # 前 2MB 里没有 → 大概率在末尾
    return idx < 1024 * 1024  # 在前 1MB 内认为是在开头

ffmpeg -i in.mp4 -c copy -movflags +faststart out.mp4

from concurrent.futures import ProcessPoolExecutor


def main(paths, spec_path):
    spec = json.load(open(spec_path))
    with ProcessPoolExecutor(max_workers=8) as ex:
        results = list(ex.map(lambda p: (p, accept(p, spec)), paths))

    rows = []
    for path, r in results:
        rows.append({
            'file': Path(path).name,
            'status': 'FAIL' if r.failed else ('WARN' if r.warned else 'PASS'),
            'failed': '; '.join(i['name'] for i in r.failed),
            'warned': '; '.join(i['name'] for i in r.warned),
        })

    # 输出 CSV + 控制台摘要
    import csv
    with open('acceptance_report.csv', 'w', newline='', encoding='utf-8') as f:
        w = csv.DictWriter(f, fieldnames=['file', 'status', 'failed', 'warned'])
        w.writeheader()
        w.writerows(rows)

    n_fail = sum(1 for x in rows if x['status'] == 'FAIL')
    print(f'共 {len(rows)} 个文件：通过 {len(rows) - n_fail}，失败 {n_fail}')
    for x in rows:
        if x['status'] == 'FAIL':
            print(f"  FAIL {x['file']}: {x['failed']}")

    sys.exit(1 if n_fail else 0)        # 退出码给 CI 用

# 简单的 HTML 表格 + 颜色标记
import html
rows_html = ''.join(
    f"<tr class='{r['status'].lower()}'><td>{html.escape(r['file'])}</td>"
    f"<td>{r['status']}</td><td>{html.escape(r['failed'])}</td></tr>"
    for r in rows)

1. 客户提供验收标准（文档/邮件）
      ↓
2. 我们翻译成 spec JSON
      ↓
3. 发给客户确认："我们理解的标准是这样，请确认"
      ↓
4. 客户确认后，作为交付依据
      ↓
5. 每次交付跑脚本，报告作为交付附件