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)