Delete app.py
Browse files
app.py
DELETED
|
@@ -1,215 +0,0 @@
|
|
| 1 |
-
from fastapi import FastAPI, File, UploadFile, HTTPException, Form
|
| 2 |
-
from fastapi.responses import PlainTextResponse
|
| 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 |
-
import re
|
| 24 |
-
|
| 25 |
-
# 每种语言的默认分段字数(用户未显式传 max_chars 时生效)
|
| 26 |
-
DEFAULT_MAX_CHARS = {"zh": 12, "en": 55}
|
| 27 |
-
FALLBACK_MAX_CHARS = 40 # 其他语言兜底值
|
| 28 |
-
|
| 29 |
-
# 需要从最终字幕文本中清除的标点符号(中英文标点、引号、括号等)
|
| 30 |
-
# 特意不包含直引号撇号 ' ,避免破坏英文缩写(don't / it's)
|
| 31 |
-
_PUNCT_CHARS = ",。!?、;:“”‘’「」『』()()《》【】[]<>—–…‥·,.!?;:\""
|
| 32 |
-
_PUNCT_PATTERN = re.compile("[" + re.escape(_PUNCT_CHARS) + "]")
|
| 33 |
-
|
| 34 |
-
|
| 35 |
-
def strip_punctuation(text: str) -> str:
|
| 36 |
-
"""去除字幕文本中的标点符号,仅做视觉清洗,不影响时间戳/断句逻辑"""
|
| 37 |
-
if not text:
|
| 38 |
-
return text
|
| 39 |
-
cleaned = _PUNCT_PATTERN.sub("", text)
|
| 40 |
-
cleaned = re.sub(r"[ \t]+", " ", cleaned).strip()
|
| 41 |
-
return cleaned
|
| 42 |
-
|
| 43 |
-
|
| 44 |
-
def resolve_max_chars(user_max_chars: int, lang_code: str) -> int:
|
| 45 |
-
"""决定实际使用的 max_chars:用户显式指定则用用户的,否则按语言给默认值"""
|
| 46 |
-
if user_max_chars and user_max_chars > 0:
|
| 47 |
-
return user_max_chars
|
| 48 |
-
base_lang = (lang_code or "").split("-")[0].lower()
|
| 49 |
-
return DEFAULT_MAX_CHARS.get(base_lang, FALLBACK_MAX_CHARS)
|
| 50 |
-
|
| 51 |
-
|
| 52 |
-
def format_timestamp(seconds: float) -> str:
|
| 53 |
-
hours = int(seconds // 3600)
|
| 54 |
-
minutes = int((seconds % 3600) // 60)
|
| 55 |
-
secs = int(seconds % 60)
|
| 56 |
-
millisecs = int((seconds - int(seconds)) * 1000)
|
| 57 |
-
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millisecs:03d}"
|
| 58 |
-
|
| 59 |
-
|
| 60 |
-
def group_by_chars(result, max_chars: int):
|
| 61 |
-
"""用词级时间戳按字数重新组合字幕段,时间戳精确来自 Whisper"""
|
| 62 |
-
words = []
|
| 63 |
-
for seg in result["segments"]:
|
| 64 |
-
for w in seg.get("words", []):
|
| 65 |
-
word = w["word"].strip()
|
| 66 |
-
if word:
|
| 67 |
-
words.append({"word": word, "start": w["start"], "end": w["end"]})
|
| 68 |
-
|
| 69 |
-
if not words:
|
| 70 |
-
print("[Warning] 没有词级时间戳,回退到原始分段", flush=True)
|
| 71 |
-
return result["segments"]
|
| 72 |
-
|
| 73 |
-
segments = []
|
| 74 |
-
buf_text = ""
|
| 75 |
-
buf_start = None
|
| 76 |
-
buf_end = None
|
| 77 |
-
|
| 78 |
-
for w in words:
|
| 79 |
-
if buf_start is None:
|
| 80 |
-
buf_start = w["start"]
|
| 81 |
-
buf_text += w["word"]
|
| 82 |
-
buf_end = w["end"]
|
| 83 |
-
|
| 84 |
-
if len(buf_text) >= max_chars or buf_text[-1] in ",。!?、,.!?":
|
| 85 |
-
segments.append({"start": buf_start, "end": buf_end, "text": buf_text})
|
| 86 |
-
buf_text = ""
|
| 87 |
-
buf_start = None
|
| 88 |
-
|
| 89 |
-
if buf_text:
|
| 90 |
-
segments.append({"start": buf_start, "end": buf_end, "text": buf_text})
|
| 91 |
-
|
| 92 |
-
# 断句依赖标点判断已经用完了,这里统一清洗掉标点,不影响时间戳
|
| 93 |
-
for seg in segments:
|
| 94 |
-
seg["text"] = strip_punctuation(seg["text"])
|
| 95 |
-
|
| 96 |
-
return segments
|
| 97 |
-
|
| 98 |
-
|
| 99 |
-
def generate_srt(segments) -> str:
|
| 100 |
-
srt_content = []
|
| 101 |
-
for i, segment in enumerate(segments, 1):
|
| 102 |
-
start_time = format_timestamp(segment['start'])
|
| 103 |
-
end_time = format_timestamp(segment['end'])
|
| 104 |
-
text = segment['text'].strip()
|
| 105 |
-
srt_content.append(f"{i}\n{start_time} --> {end_time}\n{text}\n")
|
| 106 |
-
return "\n".join(srt_content)
|
| 107 |
-
|
| 108 |
-
|
| 109 |
-
def generate_vtt(segments) -> str:
|
| 110 |
-
vtt_content = ["WEBVTT\n"]
|
| 111 |
-
for i, segment in enumerate(segments):
|
| 112 |
-
start_time = format_timestamp(segment['start']).replace(',', '.')
|
| 113 |
-
end_time = format_timestamp(segment['end']).replace(',', '.')
|
| 114 |
-
text = segment['text'].strip()
|
| 115 |
-
vtt_content.append(f"{start_time} --> {end_time}\n{text}\n")
|
| 116 |
-
return "\n".join(vtt_content)
|
| 117 |
-
|
| 118 |
-
|
| 119 |
-
@app.get("/")
|
| 120 |
-
async def root():
|
| 121 |
-
return {
|
| 122 |
-
"message": "Subtitle Generator API",
|
| 123 |
-
"model": model_size,
|
| 124 |
-
"device": "cpu"
|
| 125 |
-
}
|
| 126 |
-
|
| 127 |
-
|
| 128 |
-
@app.get("/health")
|
| 129 |
-
async def health_check():
|
| 130 |
-
return {"status": "healthy", "model": model_size}
|
| 131 |
-
|
| 132 |
-
|
| 133 |
-
@app.post("/generate_subtitle")
|
| 134 |
-
async def generate_subtitle(
|
| 135 |
-
file: UploadFile = File(...),
|
| 136 |
-
output_format: str = Form("srt"),
|
| 137 |
-
language: Optional[str] = Form(None),
|
| 138 |
-
max_chars: int = Form(0) # 0/未传 = 按语言自动使用默认值(zh=12, en=55, 其他=40)
|
| 139 |
-
):
|
| 140 |
-
valid_formats = ["srt", "vtt", "txt", "json"]
|
| 141 |
-
if output_format not in valid_formats:
|
| 142 |
-
raise HTTPException(status_code=400, detail=f"Invalid format. Choose from: {valid_formats}")
|
| 143 |
-
|
| 144 |
-
suffix = Path(file.filename).suffix if file.filename else ".tmp"
|
| 145 |
-
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file:
|
| 146 |
-
content = await file.read()
|
| 147 |
-
tmp_file.write(content)
|
| 148 |
-
tmp_path = tmp_file.name
|
| 149 |
-
|
| 150 |
-
file_mb = len(content) / 1024 / 1024
|
| 151 |
-
print(f"[Transcribe] 文件: {file.filename} ({file_mb:.1f} MB)", flush=True)
|
| 152 |
-
print(f"[Transcribe] 格式: {output_format} | 语言: {language or '自动检测'} | max_chars: {max_chars if max_chars > 0 else '未指定(将按语言自动选择)'}", flush=True)
|
| 153 |
-
|
| 154 |
-
try:
|
| 155 |
-
# word_timestamps 永远开启:分组逻辑现在始终执行,不再有"完全不分组"的原始模式
|
| 156 |
-
transcribe_options = {"verbose": True, "word_timestamps": True}
|
| 157 |
-
if language:
|
| 158 |
-
transcribe_options["language"] = language
|
| 159 |
-
|
| 160 |
-
print(f"[Transcribe] 开始转写,请稍候...", flush=True)
|
| 161 |
-
t0 = time.time()
|
| 162 |
-
result = model.transcribe(tmp_path, **transcribe_options)
|
| 163 |
-
elapsed = time.time() - t0
|
| 164 |
-
|
| 165 |
-
seg_count = len(result["segments"])
|
| 166 |
-
detected_lang = result.get("language", "unknown")
|
| 167 |
-
print(f"[Transcribe] 完成 ✅ 耗时 {elapsed:.1f}s | 语言: {detected_lang} | 原始段数: {seg_count}", flush=True)
|
| 168 |
-
|
| 169 |
-
effective_max_chars = resolve_max_chars(max_chars, language or detected_lang)
|
| 170 |
-
source = "用户指定" if (max_chars and max_chars > 0) else f"按语言({language or detected_lang})自动默认"
|
| 171 |
-
segments = group_by_chars(result, effective_max_chars)
|
| 172 |
-
print(f"[Transcribe] 按字数拆分完成,共 {len(segments)} 段(max_chars={effective_max_chars},{source})", flush=True)
|
| 173 |
-
|
| 174 |
-
full_text = strip_punctuation(result["text"])
|
| 175 |
-
|
| 176 |
-
if output_format == "srt":
|
| 177 |
-
subtitle_text = generate_srt(segments)
|
| 178 |
-
media_type = "text/plain"
|
| 179 |
-
filename = "subtitle.srt"
|
| 180 |
-
elif output_format == "vtt":
|
| 181 |
-
subtitle_text = generate_vtt(segments)
|
| 182 |
-
media_type = "text/plain"
|
| 183 |
-
filename = "subtitle.vtt"
|
| 184 |
-
elif output_format == "txt":
|
| 185 |
-
subtitle_text = full_text
|
| 186 |
-
media_type = "text/plain"
|
| 187 |
-
filename = "subtitle.txt"
|
| 188 |
-
elif output_format == "json":
|
| 189 |
-
subtitle_text = json.dumps({
|
| 190 |
-
"text": full_text,
|
| 191 |
-
"segments": segments,
|
| 192 |
-
"language": result["language"],
|
| 193 |
-
"max_chars_used": effective_max_chars
|
| 194 |
-
}, ensure_ascii=False, indent=2)
|
| 195 |
-
media_type = "application/json"
|
| 196 |
-
filename = "subtitle.json"
|
| 197 |
-
|
| 198 |
-
print(f"[Transcribe] 返回结果: {filename}", flush=True)
|
| 199 |
-
return PlainTextResponse(
|
| 200 |
-
content=subtitle_text,
|
| 201 |
-
media_type=media_type,
|
| 202 |
-
headers={"Content-Disposition": f"attachment; filename={filename}"}
|
| 203 |
-
)
|
| 204 |
-
|
| 205 |
-
except Exception as e:
|
| 206 |
-
print(f"[Error] 转写失败: {e}", flush=True)
|
| 207 |
-
raise HTTPException(status_code=500, detail=f"Transcription failed: {str(e)}")
|
| 208 |
-
finally:
|
| 209 |
-
if os.path.exists(tmp_path):
|
| 210 |
-
os.unlink(tmp_path)
|
| 211 |
-
|
| 212 |
-
|
| 213 |
-
if __name__ == "__main__":
|
| 214 |
-
import uvicorn
|
| 215 |
-
uvicorn.run(app, host="0.0.0.0", port=7860)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|