263 lines
11 KiB
Plaintext
263 lines
11 KiB
Plaintext
import json
|
||
import subprocess
|
||
import os
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
from ahserver.filestorage import FileStorage
|
||
from ahserver.serverenv import ServerEnv
|
||
|
||
result = {}
|
||
|
||
try:
|
||
# 1. Get parameters
|
||
audio_rel = params_kw.get("audio_file", "")
|
||
lyrics = params_kw.get("lyrics", "")
|
||
|
||
if not audio_rel or not lyrics:
|
||
result = {"error": "missing audio_file or lyrics parameter", "status": "error"}
|
||
else:
|
||
# 2. Resolve audio file path
|
||
fs = FileStorage()
|
||
audio_path = fs.realPath(audio_rel)
|
||
|
||
if not os.path.exists(audio_path):
|
||
result = {"error": f"file not found: {audio_path}", "status": "error"}
|
||
else:
|
||
# 3. Call fastwhisper ASR
|
||
asr_url = "http://127.0.0.1:9925/api/asr"
|
||
|
||
# Build multipart form data
|
||
boundary = f"----Boundary{int(time.time()*1000)}"
|
||
body_parts = []
|
||
|
||
# audio_file field
|
||
fname = os.path.basename(audio_path)
|
||
body_parts.append(f"--{boundary}\r\n")
|
||
body_parts.append(f'Content-Disposition: form-data; name="audio_file"; filename="{fname}"\r\n')
|
||
body_parts.append("Content-Type: audio/wav\r\n\r\n")
|
||
with open(audio_path, "rb") as f:
|
||
audio_data = f.read()
|
||
body_parts.append(audio_data)
|
||
body_parts.append(f"\r\n--{boundary}--\r\n")
|
||
|
||
# Build body
|
||
body = b""
|
||
for part in body_parts:
|
||
if isinstance(part, str):
|
||
body += part.encode("utf-8")
|
||
else:
|
||
body += part
|
||
|
||
req = urllib.request.Request(
|
||
asr_url,
|
||
data=body,
|
||
headers={
|
||
"Content-Type": f"multipart/form-data; boundary={boundary}"
|
||
},
|
||
method="POST"
|
||
)
|
||
|
||
with urllib.request.urlopen(req, timeout=300) as resp:
|
||
asr_data = json.loads(resp.read().decode("utf-8"))
|
||
|
||
if asr_data.get("status") != "SUCCEEDED":
|
||
result = {"error": "ASR failed", "asr_response": asr_data, "status": "error"}
|
||
else:
|
||
# 4. Parse ASR segments - extract word-level timings
|
||
segments = asr_data["result"].get("segments", [])
|
||
|
||
# Filter out hallucination segments
|
||
# Hallucinations typically: very short duration, or known patterns
|
||
hallucination_patterns = [
|
||
"优优独播", "YoYo Television", "Thank you",
|
||
"subtitle", "字幕", "by ", "翻译"
|
||
]
|
||
|
||
clean_segments = []
|
||
for seg in segments:
|
||
text = seg[2] if len(seg) > 2 else ""
|
||
is_hallucination = any(p.lower() in text.lower() for p in hallucination_patterns)
|
||
# Also skip very short segments at end
|
||
if not is_hallucination and (seg[1] - seg[0]) > 0.5:
|
||
word_timings = seg[3] if len(seg) > 3 else []
|
||
clean_segments.append({
|
||
"start": seg[0],
|
||
"end": seg[1],
|
||
"text": text,
|
||
"words": [[w[0], w[1], w[2]] for w in word_timings] if word_timings else []
|
||
})
|
||
|
||
# 5. Build LLM prompt
|
||
asr_summary = []
|
||
for i, seg in enumerate(clean_segments):
|
||
asr_summary.append(f"Segment {i+1} [{seg['start']:.2f}s - {seg['end']:.2f}s]: {seg['text']}")
|
||
|
||
# Format word timings for context
|
||
word_detail = []
|
||
for seg in clean_segments:
|
||
if seg["words"]:
|
||
words_str = ", ".join([f"'{w[2]}'[{w[0]:.2f}-{w[1]:.2f}]" for w in seg["words"][:15]])
|
||
word_detail.append(f" [{seg['start']:.2f}-{seg['end']:.2f}]: {words_str}")
|
||
|
||
env = ServerEnv()
|
||
api_key = "0V4xNbIsR061JaYGt1f1L"
|
||
|
||
prompt = f"""你是一个歌词时间轴校准专家。我需要你将ASR语音识别的结果与原始歌词进行对齐校准。
|
||
|
||
## 原始歌词（逐行）：
|
||
{lyrics}
|
||
|
||
## ASR识别结果（带时间戳的分段）：
|
||
{chr(10).join(asr_summary)}
|
||
|
||
## ASR逐词时间戳：
|
||
{chr(10).join(word_detail)}
|
||
|
||
## 任务：
|
||
1. 忽略ASR中的幻觉内容（如"优优独播剧场"等无关文字）
|
||
2. 根据语音相似性，将每行歌词与对应的ASR段落匹配
|
||
3. 为每个歌词字符分配正确的时间戳（基于ASR的词级时间戳）
|
||
4. 如果某行歌词没有对应的ASR段落，根据前后行的时间进行插值
|
||
|
||
## 输出格式（严格JSON）：
|
||
```json
|
||
[
|
||
{{
|
||
"line": "歌词文本",
|
||
"start": 起始秒数,
|
||
"end": 结束秒数,
|
||
"chars": [
|
||
{{"char": "字", "start": 起始秒, "end": 结束秒}},
|
||
...
|
||
]
|
||
}},
|
||
...
|
||
]
|
||
```
|
||
|
||
只输出JSON，不要其他内容。跳过空行。"""
|
||
|
||
# 6. Call LLM
|
||
llm_url = "https://token.opencomputing.cn/llmage/v1/chat/completions"
|
||
llm_payload = {
|
||
"model": "qwen3.7-max",
|
||
"messages": [
|
||
{"role": "system", "content": "你是歌词时间轴校准专家，精通中文歌词与ASR识别结果的音韵匹配。你只输出JSON格式的结果。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
"temperature": 0.1,
|
||
"max_tokens": 8000
|
||
}
|
||
|
||
llm_req = urllib.request.Request(
|
||
llm_url,
|
||
data=json.dumps(llm_payload).encode("utf-8"),
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {api_key}"
|
||
},
|
||
method="POST"
|
||
)
|
||
|
||
with urllib.request.urlopen(llm_req, timeout=300) as resp:
|
||
llm_resp = json.loads(resp.read().decode("utf-8"))
|
||
|
||
llm_text = llm_resp["choices"][0]["message"]["content"]
|
||
|
||
# 7. Parse LLM response
|
||
# Strip markdown code blocks if present
|
||
if "```json" in llm_text:
|
||
llm_text = llm_text.split("```json")[1].split("```")[0]
|
||
elif "```" in llm_text:
|
||
llm_text = llm_text.split("```")[1].split("```")[0]
|
||
|
||
calibrated = json.loads(llm_text.strip())
|
||
|
||
# 8. Generate ASS subtitle
|
||
ass_header = """[Script Info]
|
||
Title: KTV Lyrics (Calibrated)
|
||
ScriptType: v4.00+
|
||
PlayResX: 1920
|
||
PlayResY: 1080
|
||
WrapStyle: 0
|
||
|
||
[V4+ Styles]
|
||
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
|
||
Style: Default,WenQuanYi Zen Hei,72,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,1,0,0,0,100,100,0,0,1,3,1,2,30,30,60,1
|
||
Style: Highlight,WenQuanYi Zen Hei,72,&H0000FFFF,&H0000FFFF,&H00000000,&H80000000,1,0,0,0,100,100,0,0,1,3,1,2,30,30,60,1
|
||
|
||
[Events]
|
||
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
|
||
"""
|
||
|
||
def seconds_to_ass_time(s):
|
||
h = int(s // 3600)
|
||
m = int((s % 3600) // 60)
|
||
sec = s % 60
|
||
return f"{h}:{m:02d}:{sec:05.2f}"
|
||
|
||
events = []
|
||
for line_data in calibrated:
|
||
line_text = line_data["line"]
|
||
start = line_data["start"]
|
||
end = line_data["end"]
|
||
|
||
# Skip empty lines or lines with no valid timing
|
||
if not line_text.strip() or start < 0:
|
||
continue
|
||
|
||
start_str = seconds_to_ass_time(start)
|
||
end_str = seconds_to_ass_time(end)
|
||
|
||
# Create karaoke-style line with word highlighting
|
||
# Simple approach: one Dialogue event per line
|
||
events.append(f"Dialogue: 0,{start_str},{end_str},Default,,0,0,0,,{line_text}")
|
||
|
||
# If we have char-level timings, create highlight overlay
|
||
if "chars" in line_data:
|
||
chars = line_data["chars"]
|
||
for j, ch in enumerate(chars):
|
||
if ch["char"].strip() and ch["start"] >= 0 and ch["end"] > ch["start"]:
|
||
cs = seconds_to_ass_time(ch["start"])
|
||
ce = seconds_to_ass_time(ch["end"])
|
||
# Highlight current char with color change
|
||
before = line_text[:j]
|
||
current = ch["char"]
|
||
after = line_text[j+1:]
|
||
highlighted = f"{{\\c&H00FFFF&}}{before}{{\\c&H00FFFFFF&}}{current}{after}"
|
||
events.append(f"Dialogue: 1,{cs},{ce},Highlight,,0,0,0,,{highlighted}")
|
||
|
||
ass_content = ass_header + "\n".join(events) + "\n"
|
||
|
||
# Save ASS file
|
||
output_dir = f"/tmp/lyric_calibrate/{int(time.time())}"
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
ass_path = os.path.join(output_dir, "lyrics.ass")
|
||
with open(ass_path, "w", encoding="utf-8") as f:
|
||
f.write(ass_content)
|
||
|
||
# Also save calibrated data as JSON
|
||
json_path = os.path.join(output_dir, "calibrated.json")
|
||
with open(json_path, "w", encoding="utf-8") as f:
|
||
json.dump(calibrated, f, ensure_ascii=False, indent=2)
|
||
|
||
ass_rel = ass_path.replace("/tmp/", "")
|
||
json_rel = json_path.replace("/tmp/", "")
|
||
|
||
result = {
|
||
"status": "ok",
|
||
"calibrated_lines": len(calibrated),
|
||
"ass_file": f"/idfile?path={ass_rel}",
|
||
"json_data": f"/idfile?path={json_rel}",
|
||
"segments_used": len(clean_segments),
|
||
"segments_total": len(segments),
|
||
"llm_response_preview": llm_text[:500]
|
||
}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
result = {"error": str(e), "traceback": traceback.format_exc(), "status": "error"}
|
||
|
||
return json.dumps(result, ensure_ascii=False)
|