Update app.py
Browse files
app.py
CHANGED
|
@@ -3,15 +3,22 @@ from fastapi.responses import PlainTextResponse
|
|
| 3 |
import whisper
|
| 4 |
import tempfile
|
| 5 |
import os
|
|
|
|
| 6 |
import json
|
|
|
|
| 7 |
from pathlib import Path
|
| 8 |
from typing import Optional
|
| 9 |
|
|
|
|
|
|
|
|
|
|
| 10 |
app = FastAPI(title="Subtitle Generator API")
|
| 11 |
|
| 12 |
-
# ↓↓↓ 就改了这里:base → medium
|
| 13 |
model_size = os.getenv("WHISPER_MODEL", "medium")
|
|
|
|
| 14 |
model = whisper.load_model(model_size)
|
|
|
|
|
|
|
| 15 |
|
| 16 |
def format_timestamp(seconds: float) -> str:
|
| 17 |
hours = int(seconds // 3600)
|
|
@@ -20,6 +27,7 @@ def format_timestamp(seconds: float) -> str:
|
|
| 20 |
millisecs = int((seconds - int(seconds)) * 1000)
|
| 21 |
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millisecs:03d}"
|
| 22 |
|
|
|
|
| 23 |
def group_by_chars(result, max_chars: int):
|
| 24 |
"""用词级时间戳按字数重新组合字幕段,时间戳精确来自 Whisper"""
|
| 25 |
words = []
|
|
@@ -30,7 +38,7 @@ def group_by_chars(result, max_chars: int):
|
|
| 30 |
words.append({"word": word, "start": w["start"], "end": w["end"]})
|
| 31 |
|
| 32 |
if not words:
|
| 33 |
-
|
| 34 |
return result["segments"]
|
| 35 |
|
| 36 |
segments = []
|
|
@@ -44,7 +52,6 @@ def group_by_chars(result, max_chars: int):
|
|
| 44 |
buf_text += w["word"]
|
| 45 |
buf_end = w["end"]
|
| 46 |
|
| 47 |
-
# 达到字数上限,或遇到句末标点就切断
|
| 48 |
if len(buf_text) >= max_chars or buf_text[-1] in ",。!?、,.!?":
|
| 49 |
segments.append({"start": buf_start, "end": buf_end, "text": buf_text})
|
| 50 |
buf_text = ""
|
|
@@ -55,6 +62,7 @@ def group_by_chars(result, max_chars: int):
|
|
| 55 |
|
| 56 |
return segments
|
| 57 |
|
|
|
|
| 58 |
def generate_srt(segments) -> str:
|
| 59 |
srt_content = []
|
| 60 |
for i, segment in enumerate(segments, 1):
|
|
@@ -64,6 +72,7 @@ def generate_srt(segments) -> str:
|
|
| 64 |
srt_content.append(f"{i}\n{start_time} --> {end_time}\n{text}\n")
|
| 65 |
return "\n".join(srt_content)
|
| 66 |
|
|
|
|
| 67 |
def generate_vtt(segments) -> str:
|
| 68 |
vtt_content = ["WEBVTT\n"]
|
| 69 |
for i, segment in enumerate(segments):
|
|
@@ -73,6 +82,7 @@ def generate_vtt(segments) -> str:
|
|
| 73 |
vtt_content.append(f"{start_time} --> {end_time}\n{text}\n")
|
| 74 |
return "\n".join(vtt_content)
|
| 75 |
|
|
|
|
| 76 |
@app.get("/")
|
| 77 |
async def root():
|
| 78 |
return {
|
|
@@ -81,10 +91,12 @@ async def root():
|
|
| 81 |
"device": "cpu"
|
| 82 |
}
|
| 83 |
|
|
|
|
| 84 |
@app.get("/health")
|
| 85 |
async def health_check():
|
| 86 |
return {"status": "healthy", "model": model_size}
|
| 87 |
|
|
|
|
| 88 |
@app.post("/generate_subtitle")
|
| 89 |
async def generate_subtitle(
|
| 90 |
file: UploadFile = File(...),
|
|
@@ -102,14 +114,29 @@ async def generate_subtitle(
|
|
| 102 |
tmp_file.write(content)
|
| 103 |
tmp_path = tmp_file.name
|
| 104 |
|
|
|
|
|
|
|
|
|
|
|
|
|
| 105 |
try:
|
| 106 |
-
transcribe_options = {"word_timestamps": True}
|
| 107 |
if language:
|
| 108 |
transcribe_options["language"] = language
|
| 109 |
|
|
|
|
|
|
|
| 110 |
result = model.transcribe(tmp_path, **transcribe_options)
|
|
|
|
| 111 |
|
| 112 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 113 |
|
| 114 |
if output_format == "srt":
|
| 115 |
subtitle_text = generate_srt(segments)
|
|
@@ -132,6 +159,7 @@ async def generate_subtitle(
|
|
| 132 |
media_type = "application/json"
|
| 133 |
filename = "subtitle.json"
|
| 134 |
|
|
|
|
| 135 |
return PlainTextResponse(
|
| 136 |
content=subtitle_text,
|
| 137 |
media_type=media_type,
|
|
@@ -139,11 +167,13 @@ async def generate_subtitle(
|
|
| 139 |
)
|
| 140 |
|
| 141 |
except Exception as e:
|
|
|
|
| 142 |
raise HTTPException(status_code=500, detail=f"Transcription failed: {str(e)}")
|
| 143 |
finally:
|
| 144 |
if os.path.exists(tmp_path):
|
| 145 |
os.unlink(tmp_path)
|
| 146 |
|
|
|
|
| 147 |
if __name__ == "__main__":
|
| 148 |
import uvicorn
|
| 149 |
uvicorn.run(app, host="0.0.0.0", port=7860)
|
|
|
|
| 3 |
import whisper
|
| 4 |
import tempfile
|
| 5 |
import os
|
| 6 |
+
import sys
|
| 7 |
import json
|
| 8 |
+
import time
|
| 9 |
from pathlib import Path
|
| 10 |
from typing import Optional
|
| 11 |
|
| 12 |
+
# 把 stderr 合并到 stdout,HuggingFace Spaces log 才能显示 Whisper 进度条
|
| 13 |
+
sys.stderr = sys.stdout
|
| 14 |
+
|
| 15 |
app = FastAPI(title="Subtitle Generator API")
|
| 16 |
|
|
|
|
| 17 |
model_size = os.getenv("WHISPER_MODEL", "medium")
|
| 18 |
+
print(f"[Startup] 加载 Whisper 模型: {model_size} ...", flush=True)
|
| 19 |
model = whisper.load_model(model_size)
|
| 20 |
+
print(f"[Startup] 模型加载完成 ✅", flush=True)
|
| 21 |
+
|
| 22 |
|
| 23 |
def format_timestamp(seconds: float) -> str:
|
| 24 |
hours = int(seconds // 3600)
|
|
|
|
| 27 |
millisecs = int((seconds - int(seconds)) * 1000)
|
| 28 |
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millisecs:03d}"
|
| 29 |
|
| 30 |
+
|
| 31 |
def group_by_chars(result, max_chars: int):
|
| 32 |
"""用词级时间戳按字数重新组合字幕段,时间戳精确来自 Whisper"""
|
| 33 |
words = []
|
|
|
|
| 38 |
words.append({"word": word, "start": w["start"], "end": w["end"]})
|
| 39 |
|
| 40 |
if not words:
|
| 41 |
+
print("[Warning] 没有词级时间戳,回退到原始分段", flush=True)
|
| 42 |
return result["segments"]
|
| 43 |
|
| 44 |
segments = []
|
|
|
|
| 52 |
buf_text += w["word"]
|
| 53 |
buf_end = w["end"]
|
| 54 |
|
|
|
|
| 55 |
if len(buf_text) >= max_chars or buf_text[-1] in ",。!?、,.!?":
|
| 56 |
segments.append({"start": buf_start, "end": buf_end, "text": buf_text})
|
| 57 |
buf_text = ""
|
|
|
|
| 62 |
|
| 63 |
return segments
|
| 64 |
|
| 65 |
+
|
| 66 |
def generate_srt(segments) -> str:
|
| 67 |
srt_content = []
|
| 68 |
for i, segment in enumerate(segments, 1):
|
|
|
|
| 72 |
srt_content.append(f"{i}\n{start_time} --> {end_time}\n{text}\n")
|
| 73 |
return "\n".join(srt_content)
|
| 74 |
|
| 75 |
+
|
| 76 |
def generate_vtt(segments) -> str:
|
| 77 |
vtt_content = ["WEBVTT\n"]
|
| 78 |
for i, segment in enumerate(segments):
|
|
|
|
| 82 |
vtt_content.append(f"{start_time} --> {end_time}\n{text}\n")
|
| 83 |
return "\n".join(vtt_content)
|
| 84 |
|
| 85 |
+
|
| 86 |
@app.get("/")
|
| 87 |
async def root():
|
| 88 |
return {
|
|
|
|
| 91 |
"device": "cpu"
|
| 92 |
}
|
| 93 |
|
| 94 |
+
|
| 95 |
@app.get("/health")
|
| 96 |
async def health_check():
|
| 97 |
return {"status": "healthy", "model": model_size}
|
| 98 |
|
| 99 |
+
|
| 100 |
@app.post("/generate_subtitle")
|
| 101 |
async def generate_subtitle(
|
| 102 |
file: UploadFile = File(...),
|
|
|
|
| 114 |
tmp_file.write(content)
|
| 115 |
tmp_path = tmp_file.name
|
| 116 |
|
| 117 |
+
file_mb = len(content) / 1024 / 1024
|
| 118 |
+
print(f"[Transcribe] 文件: {file.filename} ({file_mb:.1f} MB)", flush=True)
|
| 119 |
+
print(f"[Transcribe] 格式: {output_format} | 语言: {language or '自动检测'} | max_chars: {max_chars or '不拆分'}", flush=True)
|
| 120 |
+
|
| 121 |
try:
|
| 122 |
+
transcribe_options = {"word_timestamps": True, "verbose": True}
|
| 123 |
if language:
|
| 124 |
transcribe_options["language"] = language
|
| 125 |
|
| 126 |
+
print(f"[Transcribe] 开始转写,请稍候...", flush=True)
|
| 127 |
+
t0 = time.time()
|
| 128 |
result = model.transcribe(tmp_path, **transcribe_options)
|
| 129 |
+
elapsed = time.time() - t0
|
| 130 |
|
| 131 |
+
seg_count = len(result["segments"])
|
| 132 |
+
detected_lang = result.get("language", "unknown")
|
| 133 |
+
print(f"[Transcribe] 完成 ✅ 耗时 {elapsed:.1f}s | 语言: {detected_lang} | 原始段数: {seg_count}", flush=True)
|
| 134 |
+
|
| 135 |
+
if max_chars > 0:
|
| 136 |
+
segments = group_by_chars(result, max_chars)
|
| 137 |
+
print(f"[Transcribe] 按字数拆分完成,共 {len(segments)} 段(max_chars={max_chars})", flush=True)
|
| 138 |
+
else:
|
| 139 |
+
segments = result["segments"]
|
| 140 |
|
| 141 |
if output_format == "srt":
|
| 142 |
subtitle_text = generate_srt(segments)
|
|
|
|
| 159 |
media_type = "application/json"
|
| 160 |
filename = "subtitle.json"
|
| 161 |
|
| 162 |
+
print(f"[Transcribe] 返回结果: {filename}", flush=True)
|
| 163 |
return PlainTextResponse(
|
| 164 |
content=subtitle_text,
|
| 165 |
media_type=media_type,
|
|
|
|
| 167 |
)
|
| 168 |
|
| 169 |
except Exception as e:
|
| 170 |
+
print(f"[Error] 转写失败: {e}", flush=True)
|
| 171 |
raise HTTPException(status_code=500, detail=f"Transcription failed: {str(e)}")
|
| 172 |
finally:
|
| 173 |
if os.path.exists(tmp_path):
|
| 174 |
os.unlink(tmp_path)
|
| 175 |
|
| 176 |
+
|
| 177 |
if __name__ == "__main__":
|
| 178 |
import uvicorn
|
| 179 |
uvicorn.run(app, host="0.0.0.0", port=7860)
|