itgoyo/hermes-skills/engineering-voice-ai-integration-engineer

import subprocess
import json
from pathlib import Path
SUPPORTED_EXTENSIONS = {".wav", ".mp3", ".m4a", ".ogg", ".flac", ".mp4", ".mov", ".webm"}
MAX_DURATION_SECONDS = 14400  # 4 小时
def validate_audio_file(file_path: str) -> dict:
    """
    处理前验证音频文件。
    使用 ffprobe 检测格式、时长、编解码器和声道布局。
    永远不要信任文件扩展名——始终探测实际容器。
    """
    path = Path(file_path)
    if path.suffix.lower() not in SUPPORTED_EXTENSIONS:
        raise ValueError(f"不支持的扩展名: {path.suffix}")
    result = subprocess.run([
        "ffprobe", "-v", "quiet",
        "-print_format", "json",
        "-show_streams", "-show_format",
        str(path)
    ], capture_output=True, text=True, check=True)
    probe = json.loads(result.stdout)
    duration = float(probe["format"]["duration"])
    if duration > MAX_DURATION_SECONDS:
        raise ValueError(f"文件超出最大时长: {duration:.0f}s > {MAX_DURATION_SECONDS}s")
    audio_streams = [s for s in probe["streams"] if s["codec_type"] == "audio"]
    if not audio_streams:
        raise ValueError("文件中未找到音频流")
    stream = audio_streams[0]
    return {
        "duration": duration,
        "codec": stream["codec_name"],
        "sample_rate": int(stream["sample_rate"]),
        "channels": stream["channels"],
        "bit_rate": probe["format"].get("bit_rate"),
        "format": probe["format"]["format_name"]
    }

import subprocess
from pathlib import Path
def preprocess_audio(input_path: str, output_path: str) -> str:
    """
    为 Whisper 系列模型输入规范化音频。
    关键步骤：
    - 重采样为 16kHz（Whisper 的原生采样率）
    - 混音为单声道（防止因声道导致的精度差异）
    - 按 EBU R128 标准规范化响度
    - 剥离视频轨道（减小文件大小，加速处理）
    返回预处理后的 wav 文件路径。
    """
    cmd = [
        "ffmpeg", "-y",
        "-i", input_path,
        "-vn",                        # 剥离视频
        "-acodec", "pcm_s16le",       # 16-bit PCM
        "-ar", "16000",               # 16kHz 采样率
        "-ac", "1",                   # 单声道
        "-af", "loudnorm=I=-16:TP=-1.5:LRA=11",  # EBU R128 响度规范化
        output_path
    ]
    subprocess.run(cmd, check=True, capture_output=True)
    return output_path
def chunk_audio(input_path: str, chunk_dir: str,
                chunk_duration: int = 1800, overlap: int = 30) -> list[str]:
    """
    将长音频拆分为有重叠的分块用于模型处理。
    使用重叠防止分块边界处的单词截断。
    重叠片段在转录组装时会被裁剪。
    chunk_duration: 每块秒数（默认 30 分钟）
    overlap: 重叠窗口秒数（默认 30 秒）
    """
    import math, os
    result = subprocess.run([
        "ffprobe", "-v", "quiet", "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1", input_path
    ], capture_output=True, text=True, check=True)
    total_duration = float(result.stdout.strip())
    chunks = []
    start = 0
    chunk_index = 0
    os.makedirs(chunk_dir, exist_ok=True)
    while start < total_duration:
        end = min(start + chunk_duration + overlap, total_duration)
        out_path = f"{chunk_dir}/chunk_{chunk_index:04d}.wav"
        subprocess.run([
            "ffmpeg", "-y",
            "-i", input_path,
            "-ss", str(start),
            "-to", str(end),
            "-acodec", "copy",
            out_path
        ], check=True, capture_output=True)
        chunks.append({"path": out_path, "start_offset": start, "index": chunk_index})
        start += chunk_duration
        chunk_index += 1
    return chunks

from faster_whisper import WhisperModel
from dataclasses import dataclass
@dataclass
class TranscriptSegment:
    start: float
    end: float
    text: str
    speaker: str | None = None
    confidence: float | None = None
def transcribe_chunk(audio_path: str, model: WhisperModel,
                     language: str | None = None) -> list[TranscriptSegment]:
    """
    使用 faster-whisper 转录单个音频分块。
    返回带时间戳的片段。启用词级时间戳
    以确保字幕生成精度。
    模型大小指南：
    - tiny/base：本地实时使用，精度较低
    - small/medium：大多数场景的精度/速度平衡点
    - large-v3：最高精度，需要 GPU，在 A10G 上约 2-3 倍实时
    """
    segments, info = model.transcribe(
        audio_path,
        language=language,
        word_timestamps=True,
        beam_size=5,
        vad_filter=True,           # 语音活动检测——跳过静音
        vad_parameters={"min_silence_duration_ms": 500}
    )
    result = []
    for seg in segments:
        result.append(TranscriptSegment(
            start=seg.start,
            end=seg.end,
            text=seg.text.strip(),
            confidence=getattr(seg, "avg_logprob", None)
        ))
    return result
def assemble_chunks(chunk_results: list[dict],
                    overlap_seconds: int = 30) -> list[TranscriptSegment]:
    """
    将分块转录结果合并为统一时间线。
    裁剪除第一块外所有分块的重叠区域，
    以防止分块边界处的重复片段。
    """
    merged = []
    for chunk in sorted(chunk_results, key=lambda c: c["start_offset"]):
        offset = chunk["start_offset"]
        trim_start = overlap_seconds if chunk["index"] > 0 else 0
        for seg in chunk["segments"]:
            adjusted_start = seg.start + offset
            if adjusted_start < offset + trim_start:
                continue  # 跳过前一分块的重叠区域
            merged.append(TranscriptSegment(
                start=adjusted_start,
                end=seg.end + offset,
                text=seg.text,
                confidence=seg.confidence
            ))
    return merged

from pyannote.audio import Pipeline
import torch
def run_diarization(audio_path: str, hf_token: str,
                    num_speakers: int | None = None) -> list[dict]:
    """
    使用 pyannote.audio 运行说话人分离。
    返回说话人片段 [{start, end, speaker}]。
    在下一步与转录片段合并。
    num_speakers: 如果已知，传入——可显著提高精度。
    如果未知，pyannote 将自动估计（精度较低）。
    """
    pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token
    )
    pipeline.to(torch.device("cuda" if torch.cuda.is_available() else "cpu"))
    diarization = pipeline(audio_path, num_speakers=num_speakers)
    segments = []
    for turn, _, speaker in diarization.itertracks(yield_label=True):
        segments.append({
            "start": turn.start,
            "end": turn.end,
            "speaker": speaker
        })
    return segments
def assign_speakers(transcript_segments: list[TranscriptSegment],
                    diarization_segments: list[dict]) -> list[TranscriptSegment]:
    """
    使用时间重叠为转录片段分配说话人标签。
    对于每个转录片段，找到重叠最大的说话人分离片段
    并分配该说话人标签。
    """
    def overlap(seg, dia):
        return max(0, min(seg.end, dia["end"]) - max(seg.start, dia["start"]))
    for seg in transcript_segments:
        best_match = max(diarization_segments,
                         key=lambda d: overlap(seg, d),
                         default=None)
        if best_match and overlap(seg, best_match) > 0:
            seg.speaker = best_match["speaker"]
    return transcript_segments

import json
import re
def normalize_transcript(segments: list[TranscriptSegment]) -> list[TranscriptSegment]:
    """
    模型输出后清理转录文本。
    处理 Whisper 系列模型的常见伪影：
    - 音乐/噪声导致的全大写转录片段
    - 双空格、前后空白
    - 填充词规范化（可配置）
    - 跨片段拆分的句子边界修复
    """
    for seg in segments:
        text = seg.text
        text = re.sub(r"\s+", " ", text).strip()
        # 标记可能的噪声片段——不要静默丢弃它们
        if text.isupper() and len(text) > 20:
            seg.text = f"[NOISE: {text}]"
        else:
            seg.text = text
    return segments
def export_srt(segments: list[TranscriptSegment], output_path: str) -> str:
    """
    将转录文本导出为 SRT 字幕文件。
    验证阅读速度（按广播标准每秒最多 20 个字符）。
    将过长片段拆分以符合行长度限制。
    """
    def format_timestamp(seconds: float) -> str:
        h = int(seconds // 3600)
        m = int((seconds % 3600) // 60)
        s = int(seconds % 60)
        ms = int((seconds % 1) * 1000)
        return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
    lines = []
    for i, seg in enumerate(segments, 1):
        lines.append(str(i))
        lines.append(f"{format_timestamp(seg.start)} --> {format_timestamp(seg.end)}")
        speaker_prefix = f"[{seg.speaker}] " if seg.speaker else ""
        lines.append(f"{speaker_prefix}{seg.text}")
        lines.append("")
    content = "\n".join(lines)
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(content)
    return output_path
def export_structured_json(segments: list[TranscriptSegment],
                            metadata: dict) -> dict:
    """
    将完整转录文本导出为结构化 JSON 供下游消费者使用。