字幕做完别急着交付：时间轴、重叠、超长行与乱码的自动检查

1
00:00:01,000 --> 00:00:04,000
这是第一行字幕
可以有第二行

2
00:00:05,000 --> 00:00:08,500
第二条

[Events]
Dialogue: 0,0:00:01.00,0:00:04.00,Default,,0,0,0,,这是字幕

import re
import csv
from dataclasses import dataclass
from pathlib import Path


@dataclass
class Cue:
    index: int
    start: float
    end: float
    lines: list


def parse_time(t: str) -> float:
    """'00:01:02,500' -> 62.5"""
    t = t.strip().replace(',', '.')
    h, m, rest = t.split(':')
    return int(h) * 3600 + int(m) * 60 + float(rest)


def parse_srt(path: str) -> list:
    text = Path(path).read_text(encoding='utf-8-sig', errors='replace')
    blocks = re.split(r'\n\s*\n', text.strip())
    cues = []
    for b in blocks:
        lines = [l for l in b.split('\n') if l.strip()]
        if len(lines) < 2:
            continue
        # 第一行可能是序号
        m = re.match(r'^(\d+)$', lines[0].strip())
        idx = int(m.group(1)) if m else None
        time_line = lines[1] if m else lines[0]
        tm = re.match(r'([\d:,.]+)\s*-->\s*([\d:,.]+)', time_line)
        if not tm:
            continue
        start, end = parse_time(tm.group(1)), parse_time(tm.group(2))
        text_lines = lines[(2 if m else 1):]
        cues.append(Cue(idx or len(cues) + 1, start, end, text_lines))
    return cues


def check_srt(path: str, cfg: dict) -> list:
    """返回问题列表。"""
    cues = parse_srt(path)
    issues = []

    for i, c in enumerate(cues):
        dur = c.end - c.start
        text = ''.join(c.lines)
        n_chars = len(text.replace('\n', ''))
        cps = n_chars / dur if dur > 0 else 999

        # 时长
        if dur < cfg['min_duration']:
            issues.append({'file': path, 'cue': c.index, 'type': '过短',
                           'detail': f'{dur:.2f}s < {cfg["min_duration"]}s',
                           'level': 'fail'})
        if dur > cfg['max_duration']:
            issues.append({'file': path, 'cue': c.index, 'type': '过长',
                           'detail': f'{dur:.2f}s > {cfg["max_duration"]}s',
                           'level': 'warn'})

        # 阅读速度
        if cps > cfg['max_cps_cjk']:
            issues.append({'file': path, 'cue': c.index, 'type': '阅读过快',
                           'detail': f'{cps:.1f} 字/秒', 'level': 'fail'})

        # 行数
        if len(c.lines) > cfg['max_lines']:
            issues.append({'file': path, 'cue': c.index, 'type': '行数过多',
                           'detail': f'{len(c.lines)} 行', 'level': 'warn'})

        # 每行长度
        for ln in c.lines:
            if len(ln) > cfg['max_chars_per_line']:
                issues.append({'file': path, 'cue': c.index, 'type': '单行过长',
                               'detail': f'{len(ln)} 字', 'level': 'fail'})

        # 与下一行的间隔 / 重叠
        if i + 1 < len(cues):
            nxt = cues[i + 1]
            gap = nxt.start - c.end
            if gap < 0:
                issues.append({'file': path, 'cue': c.index, 'type': '重叠',
                               'detail': f'与下一条重叠 {abs(gap):.2f}s', 'level': 'fail'})
            elif gap < cfg['min_gap']:
                issues.append({'file': path, 'cue': c.index, 'type': '间隔过小',
                               'detail': f'{gap:.3f}s', 'level': 'warn'})

    return issues


DEFAULT_CFG = {
    'min_duration': 0.7,
    'max_duration': 7.0,
    'min_gap': 0.08,
    'max_cps_cjk': 9.0,
    'max_lines': 2,
    'max_chars_per_line': 20,
}


if __name__ == '__main__':
    import sys
    files = sys.argv[1:] or sorted(Path('.').glob('*.srt'))
    all_issues = []
    for f in files:
        all_issues += check_srt(str(f), DEFAULT_CFG)

    with open('subtitle_qc.csv', 'w', newline='', encoding='utf-8') as fp:
        w = csv.DictWriter(fp, fieldnames=['file', 'cue', 'type', 'detail', 'level'])
        w.writeheader()
        w.writerows(all_issues)

    fails = [i for i in all_issues if i['level'] == 'fail']
    print(f'共 {len(all_issues)} 个问题，其中必须修 {len(fails)} 个')
    for i in fails[:20]:
        print(f"  {Path(i['file']).name} #{i['cue']} {i['type']}: {i['detail']}")
    sys.exit(1 if fails else 0)

def check_encoding(path: str) -> dict:
    raw = Path(path).read_bytes()
    has_bom = raw.startswith(b'\xef\xbb\xbf')
    try:
        raw.decode('utf-8')
        valid_utf8 = True
    except UnicodeDecodeError:
        valid_utf8 = False

    result = {'has_bom': has_bom, 'valid_utf8': valid_utf8}
    if not valid_utf8:
        # 尝试判断是不是 GBK
        try:
            raw.decode('gbk')
            result['maybe'] = 'gbk'
        except UnicodeDecodeError:
            result['maybe'] = 'unknown'
    return result

def convert_to_utf8(src, dst):
    raw = Path(src).read_bytes()
    for enc in ('utf-8-sig', 'gbk', 'gb18030', 'big5'):
        try:
            text = raw.decode(enc)
            break
        except UnicodeDecodeError:
            continue
    else:
        raise ValueError('无法识别编码')
    Path(dst).write_text(text, encoding='utf-8-sig')

import numpy as np
import subprocess


def extract_audio(video: str, sr: int = 16000) -> np.ndarray:
    p = subprocess.run(
        ['ffmpeg', '-v', 'error', '-i', video, '-vn', '-ac', '1',
         '-ar', str(sr), '-f', 's16le', '-'], capture_output=True, check=True)
    return np.frombuffer(p.stdout, dtype=np.int16).astype(np.float32) / 32768.0


def speech_segments(audio, sr=16000, win=0.02, thresh=0.02, min_len=0.3):
    """用短时能量检测语音段（粗糙，用于检测系统性偏移）。"""
    w = int(sr * win)
    energy = np.array([np.sqrt((audio[i:i + w] ** 2).mean())
                       for i in range(0, len(audio) - w, w)])
    loud = energy > thresh
    segs = []
    start = None
    for i, v in enumerate(loud):
        t = i * win
        if v and start is None:
            start = t
        elif not v and start is not None:
            if t - start >= min_len:
                segs.append((start, t))
            start = None
    return segs


def detect_offset(sub_cues, speech_segs):
    """计算字幕起点与最近语音起点的偏移中位数。"""
    offsets = []
    for c in sub_cues:
        # 找最近的语音段起点
        best = None
        for s, e in speech_segs:
            d = s - c.start
            if best is None or abs(d) < abs(best):
                best = d
            if s > c.start + 5:
                break
        if best is not None and abs(best) < 3.0:      # 只考虑 3 秒内的
            offsets.append(best)
    return float(np.median(offsets)) if offsets else None

def check_ass(path: str) -> list:
    text = Path(path).read_text(encoding='utf-8-sig')
    issues = []

    # PlayRes
    m = re.search(r'PlayResX:\s*(\d+)', text)
    m2 = re.search(r'PlayResY:\s*(\d+)', text)
    playres = (int(m.group(1)), int(m2.group(1))) if m and m2 else None

    # 样式定义
    styles = set(re.findall(r'^Style:\s*([^,]+),', text, re.M))
    if styles and 'Default' not in styles:
        issues.append({'type': '缺少 Default 样式', 'level': 'warn'})

    # Dialogue 引用的样式
    used = set(re.findall(r'^Dialogue:\s*[^,]+,([^,]+),', text, re.M))
    for s in used - styles:
        issues.append({'type': f'引用了未定义的样式: {s}', 'level': 'fail'})

    # 字体
    fonts = set(re.findall(r'^Style:.*?,([^,]+),\d+', text, re.M))
    for f in fonts:
        issues.append({'type': f'字体（需人工确认可用）: {f}', 'level': 'info'})
    return issues

def accept_delivery(video, subtitle, video_spec):
    issues = accept(video, video_spec)              # 视频检查
    if subtitle:
        issues += check_srt(subtitle, SUBTITLE_CFG) # 字幕检查
        enc = check_encoding(subtitle)
        if not enc['valid_utf8']:
            issues.append({'file': subtitle, 'cue': '-', 'type': '非 UTF-8',
                           'detail': enc.get('maybe', ''), 'level': 'fail'})
    return issues