zikai
00e2a95fb7
- 音频/视频转双语(英/中)SRT 字幕，Docker 容器化，CPU 开发/GPU 生产同一份代码 - faster-whisper ASR(词级时间戳) + 断句时间戳重算 + NLLB 翻译(模型不共驻) - 分片上传(断点续传) + SQLite 持久化 + 主页/历史/日志页面 - 历史页文件名搜索；缓存定时清理(默认保留7天，可配置) - 双 Dockerfile(cpu/gpu) + setup/start/stop 脚本
234 lines
7.9 KiB
Python
234 lines
7.9 KiB
Python
"""断句 + 时间戳重算（纯算法，零模型开销）。
|
||
|
||
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])
|