Initial commit: audio2text 双语字幕生成服务

- 音频/视频转双语(英/中)SRT 字幕，Docker 容器化，CPU 开发/GPU 生产同一份代码 - faster-whisper ASR(词级时间戳) + 断句时间戳重算 + NLLB 翻译(模型不共驻) - 分片上传(断点续传) + SQLite 持久化 + 主页/历史/日志页面 - 历史页文件名搜索；缓存定时清理(默认保留7天，可配置) - 双 Dockerfile(cpu/gpu) + setup/start/stop 脚本
2026-07-06 06:54:19 +00:00
commit 00e2a95fb7
44 changed files with 4110 additions and 0 deletions
--- a/app/services/segmenter.py
+++ b/app/services/segmenter.py
@@ -0,0 +1,233 @@
+"""断句 + 时间戳重算（纯算法，零模型开销）。
+
+Whisper 原始 segment 的断句通常很混乱：每段不一定是完整句子，时间戳也不对齐句界。
+本模块基于词级时间戳重新断句，得到规范的字幕条目。
+
+两路策略：
+1. 精确路（有 word_timestamps）：按句末标点（. ! ? ;）切句，超长句再按逗号拆，
+   每条字幕的时间戳直接取首词.start ~ 末词.end，精确无误。
+2. 匀速估算路（无 word_timestamps）：段内按字符数比例分配时间——
+   句start = 段start + (前缀字符数 / 段总字符数) × 段时长。
+   即「短时匀速」假设，无需大模型。
+
+最后做 SRT 规范化：单条 1–7 秒、≤2 行、每行 ≤42 字符。
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+
+from ..config import get_settings
+from .types import Segment, Subtitle, Word
+
+logger = logging.getLogger("audio2text.segmenter")
+
+# 句末标点：句号、感叹号、问号、分号
+_SENTENCE_END = re.compile(r"[.!?;]+")
+# 句内停顿：逗号、冒号、破折号
+_CLAUSE_BREAK = re.compile(r"[,:\-—]+")
+
+
+def resegment(segments: list[Segment]) -> list[Subtitle]:
+    """把 ASR segments 重组为规范字幕条目。
+
+    Args:
+        segments: asr_service.transcribe 的输出（可能含词级时间戳）
+
+    Returns:
+        list[Subtitle]，已做时长/字数规范化。
+    """
+    cfg = get_settings().segmentation
+    max_words = cfg.max_words_per_line
+    max_dur = cfg.max_duration_seconds
+    min_dur = cfg.min_duration_seconds
+    max_chars = cfg.max_chars_per_line
+
+    # 第一步：把所有词串成大列表（精确路）或退化为段级（估算路）
+    all_words: list[Word] = []
+    has_word_ts = True
+    for seg in segments:
+        if not seg.words:
+            has_word_ts = False
+            break
+        all_words.extend(seg.words)
+
+    if has_word_ts and all_words:
+        subs = _resegment_by_words(all_words, max_words, max_dur)
+    else:
+        logger.warning("无词级时间戳，退化为匀速估算路。")
+        subs = _resegment_by_estimate(segments, max_words, max_dur)
+
+    # 规范化：合并过短条目、拆分行宽
+    subs = _normalize(subs, min_dur, max_chars)
+    logger.debug("断句完成：%d 条字幕。", len(subs))
+    return subs
+
+
+# ---------------- 精确路：按词级时间戳 ----------------
+
+def _resegment_by_words(
+    words: list[Word], max_words: int, max_dur: float,
+) -> list[Subtitle]:
+    """按句末标点切句，超长句按逗号拆，时间戳取首末词。"""
+    subs: list[Subtitle] = []
+    # 当前句的词
+    current: list[Word] = []
+
+    def flush(wlist: list[Word]) -> None:
+        if not wlist:
+            return
+        text = " ".join(w.text for w in wlist).strip()
+        if not text:
+            return
+        subs.append(Subtitle(
+            text=text,
+            start=wlist[0].start,
+            end=wlist[-1].end,
+        ))
+
+    for w in words:
+        current.append(w)
+        # 句末标点 → 收尾
+        if _SENTENCE_END.search(w.text):
+            _maybe_split_and_flush(current, max_words, max_dur, flush)
+            current = []
+            continue
+        # 超长（词数或时长）→ 优先在最近的逗号处断
+        cur_dur = (current[-1].end - current[0].start) if len(current) > 1 else 0
+        if len(current) >= max_words or cur_dur >= max_dur:
+            _maybe_split_and_flush(current, max_words, max_dur, flush)
+            current = []
+
+    flush(current)
+    return subs
+
+
+def _maybe_split_and_flush(
+    wlist: list[Word], max_words: int, max_dur: float, flush,
+) -> None:
+    """若 wlist 过长，在逗号处再拆；否则整条 flush。"""
+    if len(wlist) <= max_words and (len(wlist) <= 1 or
+                                    wlist[-1].end - wlist[0].start < max_dur):
+        flush(wlist)
+        return
+    # 找逗号断点
+    parts: list[list[Word]] = []
+    cur: list[Word] = []
+    for w in wlist:
+        cur.append(w)
+        if _CLAUSE_BREAK.search(w.text) and len(cur) >= max_words // 2:
+            parts.append(cur)
+            cur = []
+    if cur:
+        parts.append(cur)
+    # 若逗号拆不开（无逗号），强制按 max_words 等分
+    if len(parts) == 1 and len(parts[0]) > max_words:
+        parts = [parts[0][i:i + max_words] for i in range(0, len(parts[0]), max_words)]
+    for p in parts:
+        flush(p)
+
+
+# ---------------- 匀速估算路：段内按字符比例 ----------------
+
+def _resegment_by_estimate(
+    segments: list[Segment], max_words: int, max_dur: float,
+) -> list[Subtitle]:
+    """无词级时间戳时：先按文本断句，再按字符数比例估算时间戳。"""
+    subs: list[Subtitle] = []
+    for seg in segments:
+        text = seg.text.strip()
+        if not text:
+            continue
+        dur = seg.end - seg.start
+        # 按句末标点切
+        sentences = _split_sentences(text)
+        if not sentences:
+            sentences = [text]
+        # 段内按字符数比例分配时间
+        total_chars = sum(len(s) for s in sentences) or 1
+        cursor = seg.start
+        for sent in sentences:
+            frac = len(sent) / total_chars
+            est_end = cursor + dur * frac
+            # 超长句再按逗号拆（时间按字符比例再分）
+            if len(sent.split()) > max_words or (est_end - cursor) > max_dur:
+                for clause in _split_clauses(sent):
+                    cfrac = len(clause) / len(sent) if len(sent) else 1
+                    c_end = cursor + (est_end - cursor) * cfrac
+                    subs.append(Subtitle(text=clause.strip(), start=cursor, end=c_end))
+                    cursor = c_end
+            else:
+                subs.append(Subtitle(text=sent.strip(), start=cursor, end=est_end))
+                cursor = est_end
+    return subs
+
+
+def _split_sentences(text: str) -> list[str]:
+    """按句末标点切句，保留标点。"""
+    parts = _SENTENCE_END.split(text)
+    marks = _SENTENCE_END.findall(text)
+    out = []
+    for i, p in enumerate(parts):
+        p = p.strip()
+        if not p:
+            continue
+        out.append(p + (marks[i] if i < len(marks) else ""))
+    return out
+
+
+def _split_clauses(sentence: str) -> list[str]:
+    """按逗号/冒号拆子句，保留标点。"""
+    parts = _CLAUSE_BREAK.split(sentence)
+    marks = _CLAUSE_BREAK.findall(sentence)
+    out = []
+    for i, p in enumerate(parts):
+        p = p.strip()
+        if not p:
+            continue
+        out.append(p + (marks[i - 1] if 0 < i <= len(marks) else ""))
+    return out
+
+
+# ---------------- 规范化 ----------------
+
+def _normalize(subs: list[Subtitle], min_dur: float, max_chars: int) -> list[Subtitle]:
+    """合并过短条目、拆分过宽行。"""
+    # 1. 合并过短（< min_dur 且非末尾）
+    merged: list[Subtitle] = []
+    for s in subs:
+        if merged and (s.end - s.start) < min_dur:
+            prev = merged[-1]
+            prev.text = (prev.text + " " + s.text).strip()
+            prev.end = s.end
+        else:
+            merged.append(Subtitle(text=s.text, start=s.start, end=s.end))
+
+    # 2. 拆分超过 max_chars 的行（按词折行，不改时间戳）
+    out: list[Subtitle] = []
+    for s in merged:
+        if len(s.text) <= max_chars:
+            out.append(s)
+            continue
+        lines = _wrap_text(s.text, max_chars)
+        out.append(Subtitle(text=lines, start=s.start, end=s.end))
+    return out
+
+
+def _wrap_text(text: str, max_chars: int) -> str:
+    """按词折成 ≤2 行，每行 ≤max_chars 字符（SRT 规范）。"""
+    words = text.split()
+    lines: list[str] = []
+    cur = ""
+    for idx, w in enumerate(words):
+        if cur and len(cur) + 1 + len(w) > max_chars:
+            lines.append(cur)
+            # 已有一行 + 当前行，合并剩余为一行
+            cur = " ".join([w] + words[idx + 1:])
+            break
+        else:
+            cur = (cur + " " + w).strip() if cur else w
+    if cur:
+        lines.append(cur)
+    return "\n".join(lines[:2])