Delete app.py

app.py

@@ -1,287 +0,0 @@
-import time
-import uuid
-import os
-from dataclasses import dataclass, field
-from typing import Optional
-
-import numpy as np
-import gradio as gr
-from scipy.signal import resample_poly
-from faster_whisper import WhisperModel
-
-try:
-    import webrtcvad
-except ImportError as e:
-    raise ImportError("请先安装：pip install webrtcvad-wheels") from e
-
-# 可选：设置HF_TOKEN以提升下载速度（如果有token的话）
-# os.environ["HF_TOKEN"] = "your_huggingface_token_here"
-
-SAMPLE_RATE = 16000
-FRAME_MS = 30
-FRAME_SAMPLES = SAMPLE_RATE * FRAME_MS // 1000
-SILENCE_END_MS = 900
-SILENCE_END_FRAMES = SILENCE_END_MS // FRAME_MS
-MIN_CLAUSE_SEC = 0.5
-PARTIAL_UPDATE_SEC = 0.8  # 更快的实时更新
-VAD_AGGRESSIVENESS = 3
-ASR_LANGUAGE = "zh"
-
-print("正在加载 whisper tiny 模型...")
-asr_model = WhisperModel("tiny", device="cpu", compute_type="int8")
-print("模型加载完成")
-
-
-def to_16k_mono_float32(sr: int, audio: np.ndarray) -> np.ndarray:
-    """将音频转换为16kHz单声道float32格式"""
-    if audio.ndim > 1:
-        audio = audio.mean(axis=1)
-    audio = audio.astype(np.float32)
-    if np.abs(audio).max() > 1.0:
-        audio = audio / 32768.0
-    if sr != SAMPLE_RATE:
-        audio = resample_poly(audio, SAMPLE_RATE, sr).astype(np.float32)
-    return audio
-
-
-def float32_to_pcm16_bytes(audio_f32: np.ndarray) -> bytes:
-    """将float32音频转换为PCM16字节流"""
-    clipped = np.clip(audio_f32, -1.0, 1.0)
-    return (clipped * 32767).astype(np.int16).tobytes()
-
-
-class VadSegmenter:
-    """VAD语音活动检测器"""
-    def __init__(self, aggressiveness=VAD_AGGRESSIVENESS):
-        self.vad = webrtcvad.Vad(aggressiveness)
-        self.silence_frames = 0
-        self.in_speech = False
-        self.last_is_speech = False
-
-    def feed(self, frame_bytes: bytes) -> str:
-        """处理一帧音频，返回状态：speaking/pause_end/silence"""
-        is_speech = self.vad.is_speech(frame_bytes, SAMPLE_RATE)
-        self.last_is_speech = is_speech
-        if is_speech:
-            self.silence_frames = 0
-            self.in_speech = True
-            return "speaking"
-        if self.in_speech:
-            self.silence_frames += 1
-            if self.silence_frames >= SILENCE_END_FRAMES:
-                self.in_speech = False
-                self.silence_frames = 0
-                return "pause_end"
-            return "speaking"
-        return "silence"
-
-
-@dataclass
-class DebugSession:
-    """会话状态管理"""
-    session_id: str
-    vad: VadSegmenter = field(default_factory=VadSegmenter)
-    buffer: np.ndarray = field(default_factory=lambda: np.zeros(0, dtype=np.float32))
-    leftover: np.ndarray = field(default_factory=lambda: np.zeros(0, dtype=np.float32))
-    confirmed_text: str = ""  # 已确认的文本（最终结果）
-    partial_text: str = ""    # 实时识别的文本
-    last_hyp: str = ""
-    last_partial_ts: float = 0.0
-    log_lines: list = field(default_factory=list)
-    chunk_count: int = 0
-    speech_frame_count: int = 0
-    silence_frame_count: int = 0
-
-
-def _common_prefix(a, b):
-    """计算两个字符串的公共前缀"""
-    n = min(len(a), len(b))
-    i = 0
-    while i < n and a[i] == b[i]:
-        i += 1
-    return a[:i]
-
-
-def ingest(session: Optional[DebugSession], new_chunk, caption_box):
-    """处理音频流的主函数"""
-    if session is None:
-        session = DebugSession(session_id=uuid.uuid4().hex[:8])
-        print(f"[{session.session_id}] 新会话开始")
-
-    if new_chunk is None:
-        return session, caption_box
-
-    session.chunk_count += 1
-    sr, chunk = new_chunk
-
-    # 转换音频格式
-    audio_f32 = to_16k_mono_float32(sr, chunk)
-    session.leftover = np.concatenate([session.leftover, audio_f32])
-
-    # VAD处理
-    while len(session.leftover) >= FRAME_SAMPLES:
-        frame = session.leftover[:FRAME_SAMPLES]
-        session.leftover = session.leftover[FRAME_SAMPLES:]
-        frame_bytes = float32_to_pcm16_bytes(frame)
-        status = session.vad.feed(frame_bytes)
-
-        if session.vad.last_is_speech:
-            session.speech_frame_count += 1
-        else:
-            session.silence_frame_count += 1
-
-        if session.vad.last_is_speech or session.vad.in_speech:
-            session.buffer = np.concatenate([session.buffer, frame])
-
-        # 检测到停顿，进行最终识别
-        if status == "pause_end":
-            if len(session.buffer) < SAMPLE_RATE * MIN_CLAUSE_SEC:
-                print(f"[{session.session_id}] 触发停顿但缓冲区太短，当误判处理")
-                session.vad.in_speech = True
-                session.vad.silence_frames = 0
-            else:
-                dur = len(session.buffer) / SAMPLE_RATE
-                print(f"[{session.session_id}] 触发停顿，缓冲区时长={dur:.2f}s，开始最终识别...")
-                t0 = time.time()
-                segments, info = asr_model.transcribe(
-                    session.buffer, language=ASR_LANGUAGE, task="transcribe",
-                    beam_size=5, vad_filter=True, condition_on_previous_text=False,
-                )
-                text = "".join(s.text for s in segments).strip()
-                print(f"[{session.session_id}] 最终识别耗时={time.time()-t0:.2f}s, 结果='{text}'")
-                
-                # 更新最终文本
-                if text:
-                    session.confirmed_text += text + " "
-                    session.partial_text = ""  # 清空实时文本
-                session.buffer = np.zeros(0, dtype=np.float32)
-                session.last_hyp = ""
-
-    # 实时识别（每0.8秒更新一次）
-    now = time.time()
-    if session.vad.in_speech and now - session.last_partial_ts >= PARTIAL_UPDATE_SEC:
-        session.last_partial_ts = now
-        if len(session.buffer) >= SAMPLE_RATE * 0.3:
-            t0 = time.time()
-            segments, _ = asr_model.transcribe(
-                session.buffer, language=ASR_LANGUAGE, task="transcribe",
-                beam_size=1, vad_filter=False, condition_on_previous_text=False,
-            )
-            hyp = "".join(s.text for s in segments).strip()
-            print(f"[{session.session_id}] 实时识别: '{hyp}'")
-            session.partial_text = hyp  # 更新实时文本
-
-    # 构建字幕显示（最终文本 + 实时文本）
-    subtitle_text = session.confirmed_text
-    if session.partial_text:
-        if subtitle_text:
-            subtitle_text += f"\n[实时] {session.partial_text}"
-        else:
-            subtitle_text = f"[实时] {session.partial_text}"
-    
-    # 如果没有任何文本，显示提示
-    if not subtitle_text:
-        subtitle_text = "等待说话..."
-
-    return session, subtitle_text
-
-
-# 创建Gradio界面
-with gr.Blocks(title="实时中文字幕", css="""
-    .subtitle-box {
-        font-size: 24px !important;
-        line-height: 1.6 !important;
-        color: #2c3e50;
-        background: #f8f9fa;
-        padding: 20px;
-        border-radius: 10px;
-        min-height: 200px;
-        border: 2px solid #3498db;
-    }
-    .subtitle-box:focus {
-        border-color: #e74c3c;
-    }
-    .header-text {
-        color: #2c3e50;
-        margin-bottom: 20px;
-    }
-    .status-badge {
-        display: inline-block;
-        padding: 5px 15px;
-        border-radius: 20px;
-        font-weight: bold;
-    }
-""") as demo:
-    gr.Markdown(
-        """
-        ## 🎙️ 实时中文字幕（类似现享字幕效果）
-        ### 使用说明：
-        1. 点击下方 **🎤 点击麦克风开始说话** 按钮
-        2. **允许浏览器访问麦克风**
-        3. **正常说话**，程序会实时显示识别文本（带 [实时] 标记）
-        4. **停顿1秒**，程序自动确认该句并添加到最终结果
-        5. 支持**连续多句**识别，字幕会持续累积
-        
-        📌 **提示**：识别结果会实时更新，带 `[实时]` 前缀的是未确认的临时识别
-        """
-    )
-    
-    with gr.Row():
-        with gr.Column(scale=1):
-            audio_in = gr.Audio(
-                sources=["microphone"], 
-                type="numpy", 
-                streaming=True, 
-                label="🎤 点击麦克风开始说话",
-                interactive=True
-            )
-            
-            # 添加控制按钮
-            with gr.Row():
-                clear_btn = gr.Button("🗑️ 清空字幕", variant="secondary", size="sm")
-    
-    with gr.Row():
-        subtitle = gr.Textbox(
-            label="📝 字幕显示", 
-            lines=10,
-            elem_classes="subtitle-box",
-            interactive=False,
-            value="🎤 点击麦克风，开始说话..."
-        )
-    
-    # 状态管理
-    state = gr.State(None)
-    
-    # 清空功能
-    def clear_subtitle(session, current_text):
-        if session:
-            session.confirmed_text = ""
-            session.partial_text = ""
-            session.buffer = np.zeros(0, dtype=np.float32)
-            session.last_hyp = ""
-        return session, "字幕已清空，请开始说话..."
-    
-    clear_btn.click(
-        fn=clear_subtitle,
-        inputs=[state, subtitle],
-        outputs=[state, subtitle]
-    )
-
-    # 音频流处理
-    audio_in.stream(
-        fn=ingest,
-        inputs=[state, audio_in, subtitle],
-        outputs=[state, subtitle],
-        stream_every=0.3,  # 每0.3秒更新一次界面
-        concurrency_limit=1  # 限制并发
-    )
-
-
-if __name__ == "__main__":
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=True,  # 生成公网链接
-        debug=False,
-        quiet=False
-    )