xiaoxia
f4b4f1fc4f
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production Runtime Images (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
feat: ASR自动字幕能力
114 lines
3.6 KiB
Python
Executable File
114 lines
3.6 KiB
Python
Executable File
"""Mock ASR 服务 — 用于测试和开发环境。
|
|
|
|
生成模拟的字幕时间轴，不依赖真实ASR服务。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from packages.domain.subtitle import (
|
|
SubtitleSegment,
|
|
SubtitleTimeline,
|
|
SubtitleWord,
|
|
)
|
|
from packages.ports.asr_service import ASRService, ASRServiceError
|
|
|
|
|
|
class MockASRService(ASRService):
|
|
"""Mock ASR 服务，生成模拟字幕数据。
|
|
|
|
如果 audio_path 对应的目录下有同名 .txt 文件，
|
|
就读取该文件内容作为字幕文本，按时间均匀分段。
|
|
否则生成默认的测试字幕。
|
|
"""
|
|
|
|
def __init__(self, mock_text: Optional[str] = None):
|
|
self._mock_text = mock_text
|
|
|
|
def transcribe(
|
|
self,
|
|
audio_path: Path,
|
|
language: Optional[str] = None,
|
|
with_word_timestamps: bool = True,
|
|
) -> SubtitleTimeline:
|
|
if not audio_path.exists():
|
|
raise ASRServiceError(f"音频文件不存在: {audio_path}", provider="mock")
|
|
|
|
# 尝试读取同名 txt 文件作为字幕文本
|
|
text = self._mock_text
|
|
if text is None:
|
|
txt_path = audio_path.with_suffix(".txt")
|
|
if txt_path.exists():
|
|
text = txt_path.read_text(encoding="utf-8").strip()
|
|
else:
|
|
text = "这是一段测试字幕。它用于验证ASR自动字幕功能是否正常工作。每一句话都会被正确地分段并显示在视频底部。字幕的样式可以根据用户的喜好进行自定义调整。"
|
|
|
|
# 估算音频时长（用ffmpeg probe或者直接假设）
|
|
# mock模式下按字数估算，每秒4个字
|
|
total_duration = max(5.0, len(text) / 4.0)
|
|
|
|
segments = self._text_to_segments(text, total_duration, with_word_timestamps)
|
|
|
|
return SubtitleTimeline(
|
|
segments=segments,
|
|
language=language or "zh",
|
|
total_duration=total_duration,
|
|
)
|
|
|
|
def _text_to_segments(
|
|
self,
|
|
text: str,
|
|
total_duration: float,
|
|
with_word_timestamps: bool,
|
|
) -> list[SubtitleSegment]:
|
|
"""将文本按句切分成带时间轴的字幕片段。"""
|
|
# 按句末标点拆分
|
|
sentences = re.split(r"(?<=[。！？!?])", text)
|
|
sentences = [s.strip() for s in sentences if s.strip()]
|
|
|
|
if not sentences:
|
|
sentences = [text]
|
|
|
|
total_chars = sum(len(s) for s in sentences)
|
|
if total_chars == 0:
|
|
return []
|
|
|
|
segments = []
|
|
current_time = 0.0
|
|
|
|
for sentence in sentences:
|
|
char_count = len(sentence)
|
|
duration = total_duration * (char_count / total_chars)
|
|
end_time = current_time + duration
|
|
|
|
words: list[SubtitleWord] = []
|
|
if with_word_timestamps:
|
|
# 每个字作为一个词级单元（中文按字，英文按词）
|
|
word_time = current_time
|
|
word_duration = duration / char_count
|
|
|
|
for char in sentence:
|
|
words.append(
|
|
SubtitleWord(
|
|
text=char,
|
|
start=word_time,
|
|
end=word_time + word_duration,
|
|
)
|
|
)
|
|
word_time += word_duration
|
|
|
|
segments.append(
|
|
SubtitleSegment(
|
|
text=sentence,
|
|
start=current_time,
|
|
end=end_time,
|
|
words=words,
|
|
)
|
|
)
|
|
current_time = end_time
|
|
|
|
return segments
|