直播结束 20 分钟出点播：录制、转码、切片的自动流水线

直播结束（录制进程退出 / 收到结束回调）
        ↓
   [事件] recording.finished
        ↓
   ① REMUX      修复录制文件（重建索引、重整时间戳）
        ↓
   ② PROBE      提取元信息（时长、分辨率、编码、是否有音轨）
        ↓
   ③ TRANSCODE  转码多档（1080p / 720p / 480p）
        ↓
   ④ PACKAGE    生成 HLS（fMP4 切片 + master playlist）
        ↓
   ⑤ THUMBNAIL  抽封面（自动选帧）
        ↓
   ⑥ UPLOAD     上传到对象存储 / CDN
        ↓
   ⑦ PUBLISH    更新数据库状态、发布
        ↓
   ⑧ NOTIFY     回调 / 通知
        ↓
     DONE

ffmpeg -i rtmp://live.example.com/stream \
  -c copy \
  -f segment -segment_time 600 -segment_format mpegts \
  -strftime 1 "records/%Y%m%d_%H%M%S.ts"

[mov,mp4,m4a,3gp,3g2,mj2 @ 0x...] moov atom not found

# 1. 尝试忽略错误重新封装
ffmpeg -err_detect ignore_err -i broken.mp4 -c copy -movflags +faststart fixed.mp4

# 2. 如果上面失败，强制重新生成时间戳
ffmpeg -fflags +genpts+igndts -i broken.mp4 -c copy -avoid_negative_ts make_zero fixed.mp4

# 3. 极端情况：只救数据（可能丢一点结尾）
ffmpeg -i broken.mp4 -c copy -fflags +igndts -movflags +frag_keyframe+empty_moov fixed.mp4

def validate_media(path, min_duration=60):
    """验证文件可用：能 probe、时长合理、有视频流、能解码。"""
    info = probe(path)
    if not info.get('streams'):
        return False, '没有流'
    v = next((s for s in info['streams'] if s['codec_type'] == 'video'), None)
    if v is None:
        return False, '没有视频流'
    dur = float(info['format'].get('duration', 0))
    if dur < min_duration:
        return False, f'时长过短: {dur}'
    # 解码验证（抽样，不整片）
    err = decode_check(path, duration=30)
    if err:
        return False, f'解码错误: {err[:200]}'
    return True, 'ok'

ffmpeg -i fixed.mp4 \
  -map 0:v -map 0:a -map 0:v -map 0:a -map 0:v -map 0:a \
  -c:v libx264 -profile:v high -preset slow \
  -c:a aac -ar 48000 -b:a 128k \
  -b:v:0 5000k -maxrate:0 6000k -bufsize:0 10000k -s:v:0 1920x1080 \
  -b:v:1 2500k -maxrate:1 3000k -bufsize:1 5000k  -s:v:1 1280x720 \
  -b:v:2 1000k -maxrate:2 1200k -bufsize:2 2000k  -s:v:2 854x480 \
  -g 120 -keyint_min 120 -sc_threshold 0 \          # 关键：固定 GOP，保证对齐
  -f hls -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2" \
  -hls_time 4 -hls_segment_type fmp4 \
  -hls_playlist_type vod \
  -master_pl_name master.m3u8 \
  -hls_segment_filename 'v%v/seg_%06d.m4s' 'v%v/index.m3u8'

def fix_bandwidth(master_path, measured):
    """把 ffmpeg 生成的 BANDWIDTH 替换成实测的峰值码率。"""
    text = Path(master_path).read_text()
    for idx, kbps in measured.items():
        text = re.sub(rf'(name:v{idx}.*?BANDWIDTH=)(\d+)',
                      rf'\g<1>{int(kbps * 1000)}', text, flags=re.S)
    Path(master_path).write_text(text)

def validate_hls(master_path):
    """校验 HLS：分片存在、数量与 EXTINF 一致。"""
    master = Path(master_path).read_text()
    subs = re.findall(r'^(?!#)(.+m3u8)$', master, re.M)
    ok = True
    for s in subs:
        p = Path(master_path).parent / s
        if not p.exists():
            ok = False
            continue
        content = p.read_text()
        segs = re.findall(r'^(?!#)(.+)$', content, re.M)
        for seg in segs:
            if not (p.parent / seg).exists():
                ok = False
    return ok

from thumbnail import pick_best_frame
frame, stamp = pick_best_frame(fixed_path)
save_as_jpeg(frame, 'cover.jpg')

def find_chapter_points(video_path, min_gap=120, silence_thresh=-40, min_silence=1.5):
    """用静音检测找章节点（演讲中的长停顿通常对应换主题）。"""
    cmd = ['ffmpeg', '-v', 'error', '-i', video_path, '-af',
           f'silencedetect=noise={silence_thresh}dB:d={min_silence}', '-f', 'null', '-']
    p = subprocess.run(cmd, capture_output=True, text=True)
    starts = [float(m) for m in re.findall(r'silence_start: ([\d.]+)', p.stderr)]
    # 过滤：间隔太近的合并
    points = []
    for s in starts:
        if not points or s - points[-1] > min_gap:
            points.append(s)
    return points

import boto3


def upload_directory(local_dir, bucket, prefix):
    s3 = boto3.client('s3', endpoint_url=S3_ENDPOINT, ...)
    for p in Path(local_dir).rglob('*'):
        if p.is_file():
            key = f'{prefix}/{p.relative_to(local_dir).as_posix()}'
            ct = guess_content_type(p)
            # HLS 的 m3u8 要设正确的 Content-Type 和缓存策略
            s3.upload_file(str(p), bucket, key,
                           ExtraArgs={'ContentType': ct,
                                      'CacheControl': cache_policy_for(p)})

def notify(vod_id, status, extra=None):
    payload = {'vod_id': vod_id, 'status': status, 'ts': int(time.time())}
    if extra:
        payload.update(extra)
    # 回调（要有重试）
    for attempt in range(3):
        try:
            requests.post(WEBHOOK_URL, json=payload, timeout=5)
            break
        except Exception:
            time.sleep(2 ** attempt)
    # 内部通知
    send_group_message(f'点播 {vod_id} 已{status}')

STATES = ['PENDING', 'REMUX', 'PROBE', 'TRANSCODE', 'PACKAGE',
          'THUMBNAIL', 'UPLOAD', 'PUBLISH', 'NOTIFY', 'DONE', 'FAILED']


def advance(vod_id, next_state):
    with transaction.atomic():
        v = VodJob.objects.select_for_update().get(id=vod_id)
        if v.state == 'FAILED':
            return
        v.state = next_state
        v.state_started_at = timezone.now()
        v.save(update_fields=['state', 'state_started_at'])