From 6eab0979d038d7a3de196c80d3d5496c3b593aef Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 4 Jul 2026 20:11:17 +0800 Subject: [PATCH] feat: initial commit - media-server --- .gitignore | 10 + ah.py | 91 ++++ app/api/calibrate/index.dspy | 110 ++++ app/api/demucs/index.dspy | 96 ++++ app/api/gpu_lock/index.dspy | 14 + app/api/ktv/index.dspy | 63 +++ app/api/lyric_calibrate/index.dspy | 262 ++++++++++ app/api/merge/index.dspy | 109 ++++ app/api/pipeline_status/index.dspy | 24 + app/api/pipeline_submit/index.dspy | 85 +++ app/api/status/index.dspy | 41 ++ app/api/submit/index.dspy | 18 + app/api/subtitle/index.dspy | 41 ++ app/api/task/index.dspy | 12 + conf/config.json | 42 ++ i18n/en/msg.txt | 3 + i18n/zh-cn/msg.txt | 3 + start.sh | 14 + stop.sh | 3 + workers/__init__.py | 0 workers/ktv_pipeline.py | 803 +++++++++++++++++++++++++++++ workers/ktv_pipeline.py.bak | 539 +++++++++++++++++++ 22 files changed, 2383 insertions(+) create mode 100644 .gitignore create mode 100644 ah.py create mode 100644 app/api/calibrate/index.dspy create mode 100644 app/api/demucs/index.dspy create mode 100644 app/api/gpu_lock/index.dspy create mode 100644 app/api/ktv/index.dspy create mode 100644 app/api/lyric_calibrate/index.dspy create mode 100644 app/api/merge/index.dspy create mode 100644 app/api/pipeline_status/index.dspy create mode 100644 app/api/pipeline_submit/index.dspy create mode 100644 app/api/status/index.dspy create mode 100644 app/api/submit/index.dspy create mode 100644 app/api/subtitle/index.dspy create mode 100644 app/api/task/index.dspy create mode 100644 conf/config.json create mode 100644 i18n/en/msg.txt create mode 100644 i18n/zh-cn/msg.txt create mode 100755 start.sh create mode 100755 stop.sh create mode 100644 workers/__init__.py create mode 100644 workers/ktv_pipeline.py create mode 100644 workers/ktv_pipeline.py.bak diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5cbdaeb --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +__pycache__/ +*.pyc +*.pyo +logs/ +*.log +nohup.out +nohup_gpu*.out +py3/ +*.egg-info/ +*.pid diff --git a/ah.py b/ah.py new file mode 100644 index 0000000..f5e6fc1 --- /dev/null +++ b/ah.py @@ -0,0 +1,91 @@ +# -*- coding:utf-8 -*- +import asyncio +import json +from ahserver import filedownload +from ahserver.webapp import webapp +from ahserver.serverenv import ServerEnv +from ahserver.configuredServer import add_startup +from longtasks.longtasks import LongTasks, schedule_once +from appPublic.log import debug, exception + + +class MediaTasks(LongTasks): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._handlers = {} + + def register(self, task_type: str, handler): + self._handlers[task_type] = handler + debug(f'MediaTasks: registered handler for {task_type}') + + async def process_task(self, payload: dict, workid: int = None): + if isinstance(payload, str): + payload = json.loads(payload) + task_type = payload.get('task_type', '') + debug(f'MediaTasks processing: type={task_type}') + handler = self._handlers.get(task_type) + if handler is None: + raise ValueError(f'Unknown task_type: {task_type}') + return await handler(payload, workid) + + +class GPULock: + def __init__(self, redis_client, total_gpus=8): + self.redis = redis_client + self.total_gpus = total_gpus + self.lock_prefix = 'gpu:lock:' + + async def acquire(self, task_id: str, timeout=600): + for gpu_id in range(self.total_gpus): + key = f'{self.lock_prefix}{gpu_id}' + acquired = await self.redis.set(key, task_id, nx=True, ex=timeout) + if acquired: + return gpu_id + return None + + async def release(self, gpu_id: int): + key = f'{self.lock_prefix}{gpu_id}' + await self.redis.delete(key) + + async def status(self): + result = {} + for gpu_id in range(self.total_gpus): + key = f'{self.lock_prefix}{gpu_id}' + owner = await self.redis.get(key) + result[gpu_id] = {'busy': owner is not None, 'owner': owner} + return result + + +async def handle_ktv_pipeline(payload, workid=None): + from workers.ktv_pipeline import run_pipeline + pipeline_id = payload.get('pipeline_id', '') + debug(f'KTV pipeline handler: {pipeline_id}') + await run_pipeline(pipeline_id) + return {'pipeline_id': pipeline_id, 'status': 'completed'} + + +async def on_app_built(app): + env = ServerEnv() + longtasks = env.longtasks + if longtasks: + schedule_once(0.1, longtasks.run) + debug('longtasks worker started') + + +def init(): + env = ServerEnv() + longtasks = MediaTasks( + 'redis://127.0.0.1:6379', + 'media', + worker_cnt=4, + stuck_seconds=1800, + max_age_hours=24 + ) + longtasks.register('ktv_pipeline', handle_ktv_pipeline) + env.longtasks = longtasks + + add_startup(on_app_built) + + +if __name__ == '__main__': + webapp(init) diff --git a/app/api/calibrate/index.dspy b/app/api/calibrate/index.dspy new file mode 100644 index 0000000..effe663 --- /dev/null +++ b/app/api/calibrate/index.dspy @@ -0,0 +1,110 @@ +""" +POST /api/calibrate +字幕校准服务 - 用LLM将WhisperX识别的歌词时间戳与原始歌词对齐 + +Parameters: + original_lyrics: 原始歌词（准确文字） + whisperx_json: WhisperX输出的JSON（时间戳准，文字不准） + +Returns: + JSON with calibrated subtitles (accurate text + precise timestamps) +""" +import json +import os +import sys + +CALIBRATE_PROMPT = """你是一个专业的歌词字幕校准专家。 + +任务：将WhisperX语音识别输出的时间戳与原始歌词文字进行精确对齐。 + +规则： +1. 保留WhisperX输出的所有时间戳（start/end），这些时间是准确的 +2. 将WhisperX识别的文字替换为原始歌词中对应的文字 +3. 按段落顺序匹配：WhisperX的第N段对应原歌词的第N段 +4. 处理副歌重复：如果WhisperX识别出重复段落，映射到同一歌词段落 +5. 忽略纯音乐段落（无歌词的时间段） + +输出格式（严格JSON）： +{ + "segments": [ + { + "text": "校准后的歌词文字", + "start": 1.234, + "end": 4.567, + "chars": [ + {"char": "爱", "start": 1.234, "end": 1.500}, + {"char": "上", "start": 1.500, "end": 1.800} + ] + } + ] +} + +原始歌词： +{original_lyrics} + +WhisperX输出JSON： +{whisperx_json} + +请输出校准后的JSON（不要markdown代码块，直接输出JSON）： +""" + +try: + original_lyrics = params_kw.get('original_lyrics', '') + whisperx_json = params_kw.get('whisperx_json', '') + + if not original_lyrics or not whisperx_json: + result = json.dumps({"status": "error", "error": "missing original_lyrics or whisperx_json"}, ensure_ascii=False) + else: + # Build the prompt + prompt = CALIBRATE_PROMPT.format( + original_lyrics=original_lyrics, + whisperx_json=whisperx_json + ) + + # Call LLM via Sage llmage API + import aiohttp + LLM_API_BASE = os.environ.get('LLM_API_BASE', 'https://token.opencomputing.cn/llmage/v1') + LLM_API_KEY = os.environ.get('LLM_API_KEY', '') + + if not LLM_API_KEY: + # Try to get from config + from ahserver.serverenv import ServerEnv + env = ServerEnv() + LLM_API_KEY = getattr(env, 'llm_api_key', '') or '' + + async with aiohttp.ClientSession() as session: + payload = { + "model": "qwen3-235b-a22b", + "catelogid": "t2t", + "messages": [{"role": "user", "content": prompt}], + "temperature": 0.1, + "max_tokens": 4096 + } + headers = { + "Authorization": f"Bearer {LLM_API_KEY}", + "Content-Type": "application/json" + } + + async with session.post(f"{LLM_API_BASE}/chat/completions", json=payload, headers=headers, timeout=120) as resp: + if resp.status == 200: + data = await resp.json() + content = data.get('choices', [{}])[0].get('message', {}).get('content', '') + # Parse the LLM response + content = content.strip() + if content.startswith('```'): + content = content.split('```')[1] + if content.startswith('json'): + content = content[4:] + content = content.strip() + try: + calibrated = json.loads(content) + result = json.dumps({"status": "success", "data": calibrated}, ensure_ascii=False) + except json.JSONDecodeError: + result = json.dumps({"status": "error", "error": "LLM response not valid JSON", "raw": content[:500]}, ensure_ascii=False) + else: + text = await resp.text() + result = json.dumps({"status": "error", "error": f"LLM API returned {resp.status}", "detail": text[:300]}, ensure_ascii=False) + +except Exception as e: + import traceback + result = json.dumps({"status": "error", "error": str(e), "traceback": traceback.format_exc()}, ensure_ascii=False) diff --git a/app/api/demucs/index.dspy b/app/api/demucs/index.dspy new file mode 100644 index 0000000..f509907 --- /dev/null +++ b/app/api/demucs/index.dspy @@ -0,0 +1,96 @@ +import json +import os +import asyncio +import time +import traceback +from ahserver.filestorage import FileStorage +from urllib.parse import quote + +try: + DEMUCS_WRAPPER = '/tmp/demucs_wrapper.py' + PYTHON_BIN = '/share/vllm-0.8.5/bin/python' + OUTPUT_BASE_DIR = '/tmp/demucs_output' + + os.makedirs(OUTPUT_BASE_DIR, exist_ok=True) + + fs = FileStorage() + filesroot = os.path.abspath(fs.root) # e.g. /tmp + + input_file = None + web_path = None + + # 方式1: 文件上传 (multipart/form-data) + if 'audio_file' in params_kw: + web_path = params_kw['audio_file'] + if isinstance(web_path, str) and len(web_path) > 0: + input_file = fs.realPath(web_path) + + # 方式2: 文件路径 + elif 'filepath' in params_kw or 'audio_path' in params_kw: + input_file = params_kw.get('filepath') or params_kw.get('audio_path') + + if not input_file: + return json.dumps({ + 'status': 'error', + 'message': 'Missing parameter: audio_file (upload) or filepath (path)' + }, ensure_ascii=False) + + if not os.path.isfile(input_file): + return json.dumps({ + 'status': 'error', + 'message': f'File not found: {input_file}', + 'web_path': web_path + }, ensure_ascii=False) + + run_id = f'demucs_{int(time.time() * 1000)}' + output_dir = os.path.join(OUTPUT_BASE_DIR, run_id) + os.makedirs(output_dir, exist_ok=True) + + cmd = [PYTHON_BIN, DEMUCS_WRAPPER, '--two-stems=vocals', '-o', output_dir, input_file] + + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode != 0: + error_msg = stderr.decode('utf-8', errors='replace').strip() + return json.dumps({ + 'status': 'error', + 'message': f'Demucs failed: {error_msg}' + }, ensure_ascii=False) + + filename_no_ext = os.path.splitext(os.path.basename(input_file))[0] + stems_dir = os.path.join(output_dir, 'htdemucs', filename_no_ext) + vocals_path = os.path.join(stems_dir, 'vocals.wav') + no_vocals_path = os.path.join(stems_dir, 'no_vocals.wav') + + if not os.path.isfile(vocals_path): + return json.dumps({ + 'status': 'error', + 'message': f'Output not found: {vocals_path}' + }, ensure_ascii=False) + + # 计算相对于 filesroot 的 web path + vocals_webpath = fs.webpath(vocals_path) + no_vocals_webpath = fs.webpath(no_vocals_path) + + # 构建下载 URL (通过 idfile 端点) + vocals_url = f'/idfile?path={quote(vocals_webpath)}' + no_vocals_url = f'/idfile?path={quote(no_vocals_webpath)}' + + return json.dumps({ + 'status': 'success', + 'vocals_url': vocals_url, + 'no_vocals_url': no_vocals_url, + 'input_file': input_file + }, ensure_ascii=False) + +except Exception as e: + return json.dumps({ + 'status': 'error', + 'message': str(e), + 'traceback': traceback.format_exc() + }, ensure_ascii=False) diff --git a/app/api/gpu_lock/index.dspy b/app/api/gpu_lock/index.dspy new file mode 100644 index 0000000..2ad05cf --- /dev/null +++ b/app/api/gpu_lock/index.dspy @@ -0,0 +1,14 @@ +# -*- coding:utf-8 -*- +# GET /api/gpu_lock - GPU锁状态 + +import aioredis + +r = await aioredis.from_url('redis://127.0.0.1:6379', decode_responses=True) +result = {} +for gpu_id in range(8): + key = f'gpu:lock:{gpu_id}' + owner = await r.get(key) + result[gpu_id] = {'busy': owner is not None, 'owner': owner} +await r.close() + +return json.dumps(result) diff --git a/app/api/ktv/index.dspy b/app/api/ktv/index.dspy new file mode 100644 index 0000000..5488ae0 --- /dev/null +++ b/app/api/ktv/index.dspy @@ -0,0 +1,63 @@ +import json +import os +import asyncio + +video_path = params_kw.get('video_path', '') +vocals_path = params_kw.get('vocals_path', '') +accompaniment_path = params_kw.get('accompaniment_path', '') +subtitle_path = params_kw.get('subtitle_path', '') +output_name = params_kw.get('output_name', 'ktv_output') + +output_dir = os.path.join('/tmp/ffmpeg_output', output_name) +os.makedirs(output_dir, exist_ok=True) + +output_path = os.path.join(output_dir, 'output.mkv') + +try: + missing = [] + if not video_path: + missing.append('video_path') + if not vocals_path: + missing.append('vocals_path') + if not accompaniment_path: + missing.append('accompaniment_path') + if not subtitle_path: + missing.append('subtitle_path') + + if missing: + result = json.dumps({"status": "error", "error": f"Missing required params: {', '.join(missing)}"}) + elif not os.path.isfile(video_path): + result = json.dumps({"status": "error", "error": f"Video file not found: {video_path}"}) + elif not os.path.isfile(vocals_path): + result = json.dumps({"status": "error", "error": f"Vocals file not found: {vocals_path}"}) + elif not os.path.isfile(accompaniment_path): + result = json.dumps({"status": "error", "error": f"Accompaniment file not found: {accompaniment_path}"}) + elif not os.path.isfile(subtitle_path): + result = json.dumps({"status": "error", "error": f"Subtitle file not found: {subtitle_path}"}) + else: + proc = await asyncio.create_subprocess_exec( + 'ffmpeg', '-y', + '-i', video_path, + '-i', vocals_path, + '-i', accompaniment_path, + '-vf', f"ass={subtitle_path}", + '-map', '0:v', + '-map', '1:a', + '-map', '2:a', + '-c:v', 'libx264', + '-c:a', 'aac', + output_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode == 0: + result = json.dumps({"status": "success", "output_path": output_path}) + else: + result = json.dumps({"status": "error", "error": stderr.decode()}) + +except Exception as e: + result = json.dumps({"status": "error", "error": str(e)}) + +return result diff --git a/app/api/lyric_calibrate/index.dspy b/app/api/lyric_calibrate/index.dspy new file mode 100644 index 0000000..3384fc5 --- /dev/null +++ b/app/api/lyric_calibrate/index.dspy @@ -0,0 +1,262 @@ +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) diff --git a/app/api/merge/index.dspy b/app/api/merge/index.dspy new file mode 100644 index 0000000..3375399 --- /dev/null +++ b/app/api/merge/index.dspy @@ -0,0 +1,109 @@ +import json +import os +import asyncio +import tempfile + +segments = json.loads(params_kw.get('segments', '[]')) +transitions = params_kw.get('transitions', 'fade') +output_name = params_kw.get('output_name', 'merged_output.mp4') + +output_dir = os.path.join('/tmp/ffmpeg_output', output_name) +os.makedirs(output_dir, exist_ok=True) + +output_path = os.path.join(output_dir, 'output.mp4') + +try: + if not segments or len(segments) == 0: + result = json.dumps({"status": "error", "error": "No segments provided"}) + elif len(segments) == 1: + # Single segment, just copy + seg = segments[0] + proc = await asyncio.create_subprocess_exec( + 'ffmpeg', '-y', '-i', seg['path'], '-c', 'copy', output_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + if proc.returncode == 0: + result = json.dumps({"status": "success", "output_path": output_path}) + else: + result = json.dumps({"status": "error", "error": stderr.decode()}) + elif transitions == 'none' or len(segments) == 1: + # Simple concat demuxer (no transitions) + concat_list = os.path.join(output_dir, 'concat_list.txt') + with open(concat_list, 'w') as f: + for seg in segments: + f.write(f"file '{seg['path']}'\n") + + proc = await asyncio.create_subprocess_exec( + 'ffmpeg', '-y', '-f', 'concat', '-safe', '0', + '-i', concat_list, '-c', 'copy', output_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + if proc.returncode == 0: + result = json.dumps({"status": "success", "output_path": output_path}) + else: + result = json.dumps({"status": "error", "error": stderr.decode()}) + else: + # Crossfade transitions using xfade filter + # Build complex xfade filter chain + n = len(segments) + fade_duration = 1.0 # 1 second crossfade + + inputs = [] + for seg in segments: + inputs.extend(['-i', seg['path']]) + + # Build xfade filter chain + filter_parts = [] + # Calculate offsets: each offset is cumulative duration minus fade overlaps + offsets = [] + cumulative = float(segments[0]['duration']) + for i in range(1, n): + offset = cumulative - fade_duration + offsets.append(offset) + cumulative = offset + float(segments[i]['duration']) + + if n == 2: + filter_str = f'[0:v][1:v]xfade=transition=fade:duration={fade_duration}:offset={offsets[0]}[vout];[0:a][1:a]acrossfade=d={fade_duration}[aout]' + else: + # Chain xfade filters for multiple segments + prev_label = '0:v' + audio_prev = '0:a' + vfilters = [] + afilters = [] + + for i in range(1, n): + out_label = f'v{i}' if i < n - 1 else 'vout' + audio_out = f'a{i}' if i < n - 1 else 'aout' + vfilters.append(f'[{prev_label}][{i}:v]xfade=transition=fade:duration={fade_duration}:offset={offsets[i-1]}[{out_label}]') + afilters.append(f'[{audio_prev}][{i}:a]acrossfade=d={fade_duration}[{audio_out}]') + prev_label = out_label + audio_prev = audio_out + + filter_str = ';'.join(vfilters + afilters) + + cmd = ['ffmpeg', '-y'] + inputs + [ + '-filter_complex', filter_str, + '-map', '[vout]', '-map', '[aout]', + '-c:v', 'libx264', '-c:a', 'aac', + output_path + ] + + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + if proc.returncode == 0: + result = json.dumps({"status": "success", "output_path": output_path}) + else: + result = json.dumps({"status": "error", "error": stderr.decode()}) + +except Exception as e: + result = json.dumps({"status": "error", "error": str(e)}) + +return result diff --git a/app/api/pipeline_status/index.dspy b/app/api/pipeline_status/index.dspy new file mode 100644 index 0000000..31b124e --- /dev/null +++ b/app/api/pipeline_status/index.dspy @@ -0,0 +1,24 @@ +import json + +result = {} + +try: + pipeline_id = params_kw.get('pipeline_id', '') + + if not pipeline_id: + result = {'status': 'error', 'error': 'missing pipeline_id'} + else: + import aioredis + redis = await aioredis.from_url('redis://127.0.0.1:6379', db=1) + data = await redis.get(f'pipeline:{pipeline_id}') + await redis.close() + + if data: + result = json.loads(data) + else: + result = {'status': 'error', 'error': f'pipeline {pipeline_id} not found'} + +except Exception as e: + result = {'status': 'error', 'error': str(e)} + +return json.dumps(result, ensure_ascii=False) diff --git a/app/api/pipeline_submit/index.dspy b/app/api/pipeline_submit/index.dspy new file mode 100644 index 0000000..1762777 --- /dev/null +++ b/app/api/pipeline_submit/index.dspy @@ -0,0 +1,85 @@ +# -*- coding:utf-8 -*- +import json +import os +import time +import uuid + +# 通用参数 +mode = params_kw.get('mode', 'lyrics_only') # audio_lyrics, video_lyrics, lyrics_only +scene = params_kw.get('scene', 'pop') +lyric_threshold = float(params_kw.get('lyric_threshold', 8.5)) +music_threshold = float(params_kw.get('music_threshold', 7.5)) + +# 模式特定参数 +input_audio = params_kw.get('input_audio', '') # Mode A +input_video = params_kw.get('input_video', '') # Mode B +lyrics = params_kw.get('lyrics', '') # Mode A, B, C +outline = params_kw.get('outline', '') # Mode C (alternative to lyrics) +description = params_kw.get('description', '') # Mode C (legacy) +title = params_kw.get('title', '未知歌曲') +lyricist = params_kw.get('lyricist', '未知') +composer = params_kw.get('composer', '未知') + +# 验证模式参数 +if mode == 'audio_lyrics': + if not input_audio: + return json.dumps({'status': 'error', 'error': 'audio_lyrics mode requires input_audio'}, ensure_ascii=False) + if not lyrics: + return json.dumps({'status': 'error', 'error': 'audio_lyrics mode requires lyrics'}, ensure_ascii=False) +elif mode == 'video_lyrics': + if not input_video: + return json.dumps({'status': 'error', 'error': 'video_lyrics mode requires input_video'}, ensure_ascii=False) + if not lyrics: + return json.dumps({'status': 'error', 'error': 'video_lyrics mode requires lyrics'}, ensure_ascii=False) +elif mode == 'lyrics_only': + if not lyrics and not outline and not description: + return json.dumps({'status': 'error', 'error': 'lyrics_only mode requires lyrics, outline, or description'}, ensure_ascii=False) +else: + return json.dumps({'status': 'error', 'error': f'invalid mode: {mode}'}, ensure_ascii=False) + +pipeline_id = f'ktv_{uuid.uuid4().hex[:12]}' + +import aioredis +redis = await aioredis.from_url('redis://127.0.0.1:6379', db=1) + +pipeline_data = { + 'id': pipeline_id, + 'state': 'submitted', + 'mode': mode, + 'scene': scene, + 'lyric_threshold': lyric_threshold, + 'music_threshold': music_threshold, + 'lyrics': lyrics, + 'outline': outline, + 'description': description, + 'input_audio': input_audio, + 'input_video': input_video, + 'title': title, + 'lyricist': lyricist, + 'composer': composer, + 'created_at': time.time(), + 'artifacts': {}, + 'errors': [] +} +await redis.set(f'pipeline:{pipeline_id}', json.dumps(pipeline_data, ensure_ascii=False), ex=86400) +await redis.close() + +from ahserver.serverenv import ServerEnv +env = ServerEnv() +longtasks = env.longtasks + +payload = json.dumps({ + 'task_type': 'ktv_pipeline', + 'pipeline_id': pipeline_id, +}) + +task_id = await longtasks.submit_task(payload) + +return json.dumps({ + 'status': 'success', + 'pipeline_id': pipeline_id, + 'task_id': task_id, + 'mode': mode, + 'message': f'KTV pipeline submitted: {pipeline_id} (mode: {mode})', + 'query_url': f'/api/pipeline_status?pipeline_id={pipeline_id}' +}, ensure_ascii=False) diff --git a/app/api/status/index.dspy b/app/api/status/index.dspy new file mode 100644 index 0000000..9f1e728 --- /dev/null +++ b/app/api/status/index.dspy @@ -0,0 +1,41 @@ +# -*- coding:utf-8 -*- +# GET /api/status - 服务状态 + +import subprocess + +result = { + "service": "media-server", + "status": "running", + "gpu": [], + "redis": False +} + +# GPU状态 +try: + out = subprocess.check_output( + ['nvidia-smi', '--query-gpu=index,utilization.gpu,memory.used,memory.total', + '--format=csv,noheader,nounits'], + timeout=5 + ).decode().strip() + for line in out.split('\n'): + parts = [p.strip() for p in line.split(',')] + result['gpu'].append({ + 'id': int(parts[0]), + 'util': int(parts[1]), + 'mem_used': int(parts[2]), + 'mem_total': int(parts[3]) + }) +except Exception: + pass + +# Redis +try: + import aioredis + r = await aioredis.from_url('redis://127.0.0.1:6379') + pong = await r.ping() + result['redis'] = pong + await r.close() +except Exception: + pass + +return json.dumps(result) diff --git a/app/api/submit/index.dspy b/app/api/submit/index.dspy new file mode 100644 index 0000000..ea1ff06 --- /dev/null +++ b/app/api/submit/index.dspy @@ -0,0 +1,18 @@ +# -*- coding:utf-8 -*- +# POST /api/submit - 提交异步媒体任务 +# body: {"task_type": "tts|video|image|music", ...task specific params} + +env = ServerEnv() +longtasks = env.longtasks + +if longtasks is None: + return json.dumps({"error": "longtasks not initialized"}) + +payload = params_kw.copy() +task_type = payload.get('task_type', '') + +if not task_type: + return json.dumps({"error": "task_type is required"}) + +result = await longtasks.submit_task(payload) +return json.dumps(result) diff --git a/app/api/subtitle/index.dspy b/app/api/subtitle/index.dspy new file mode 100644 index 0000000..424e97e --- /dev/null +++ b/app/api/subtitle/index.dspy @@ -0,0 +1,41 @@ +import json +import os +import asyncio + +video_path = params_kw.get('video_path', '') +subtitle_path = params_kw.get('subtitle_path', '') +output_name = params_kw.get('output_name', 'subtitled_output') + +output_dir = os.path.join('/tmp/ffmpeg_output', output_name) +os.makedirs(output_dir, exist_ok=True) + +output_path = os.path.join(output_dir, 'output.mp4') + +try: + if not video_path or not subtitle_path: + result = json.dumps({"status": "error", "error": "video_path and subtitle_path are required"}) + elif not os.path.isfile(video_path): + result = json.dumps({"status": "error", "error": f"Video file not found: {video_path}"}) + elif not os.path.isfile(subtitle_path): + result = json.dumps({"status": "error", "error": f"Subtitle file not found: {subtitle_path}"}) + else: + proc = await asyncio.create_subprocess_exec( + 'ffmpeg', '-y', + '-i', video_path, + '-vf', f"ass={subtitle_path}", + '-c:v', 'libx264', '-c:a', 'copy', + output_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode == 0: + result = json.dumps({"status": "success", "output_path": output_path}) + else: + result = json.dumps({"status": "error", "error": stderr.decode()}) + +except Exception as e: + result = json.dumps({"status": "error", "error": str(e)}) + +return result diff --git a/app/api/task/index.dspy b/app/api/task/index.dspy new file mode 100644 index 0000000..719ee49 --- /dev/null +++ b/app/api/task/index.dspy @@ -0,0 +1,12 @@ +# -*- coding:utf-8 -*- +# GET /api/task?task_id=xxx - 查询任务状态 + +env = ServerEnv() +longtasks = env.longtasks + +task_id = params_kw.get('task_id', '') +if not task_id: + return json.dumps({"error": "task_id is required"}) + +status = await longtasks.get_status(task_id) +return json.dumps(status) diff --git a/conf/config.json b/conf/config.json new file mode 100644 index 0000000..8751294 --- /dev/null +++ b/conf/config.json @@ -0,0 +1,42 @@ +{ + "password_key": "MediaServer2026Key", + "databases": {}, + "session_redis": { + "host": "127.0.0.1", + "port": 6379, + "db": 1 + }, + "website": { + "paths": [ + [ + "$[workdir]$/app", + "" + ] + ], + "host": "0.0.0.0", + "port": 9080, + "coding": "utf-8", + "indexes": [ + "index.html", + "index.dspy" + ], + "processors": [ + [ + ".dspy", + "dspy" + ], + [ + ".md", + "md" + ] + ], + "startswiths": [ + { + "leading": "/idfile", + "registerfunction": "idfile" + } + ] + }, + "hot_reload": true, + "filesroot": "/tmp" +} \ No newline at end of file diff --git a/i18n/en/msg.txt b/i18n/en/msg.txt new file mode 100644 index 0000000..51ddaa9 --- /dev/null +++ b/i18n/en/msg.txt @@ -0,0 +1,3 @@ +Media Server=Media Server +Task submitted=Task submitted +Task not found=Task not found diff --git a/i18n/zh-cn/msg.txt b/i18n/zh-cn/msg.txt new file mode 100644 index 0000000..86b540e --- /dev/null +++ b/i18n/zh-cn/msg.txt @@ -0,0 +1,3 @@ +Media Server=媒体服务器 +Task submitted=任务已提交 +Task not found=任务未找到 diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..6c331a1 --- /dev/null +++ b/start.sh @@ -0,0 +1,14 @@ +#!/bin/bash +# media-server start script +cd "$(dirname "$0")" +export PYTHONPATH=/share/vllm-0.8.5/lib/python3.10/site-packages:$PYTHONPATH + +PORT=${1:-9080} + +# Update port in config if different +if [ "$PORT" != "9080" ]; then + sed -i "s/\"port\": 9080/\"port\": $PORT/" conf/config.json +fi + +echo "Starting media-server on port $PORT..." +exec /share/vllm-0.8.5/bin/python ah.py diff --git a/stop.sh b/stop.sh new file mode 100755 index 0000000..64df81c --- /dev/null +++ b/stop.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# media-server stop script +pkill -f "python ah.py" && echo "Stopped" || echo "Not running" diff --git a/workers/__init__.py b/workers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/workers/ktv_pipeline.py b/workers/ktv_pipeline.py new file mode 100644 index 0000000..4b247f8 --- /dev/null +++ b/workers/ktv_pipeline.py @@ -0,0 +1,803 @@ +# -*- coding:utf-8 -*- +""" +KTV Pipeline Worker v2 - 标准化歌曲制作流程 + +支持3种输入模式： + 1. audio_lyrics: 原唱音频 + 歌词 → 快速生成KTV + 2. video_lyrics: 原唱视频 + 歌词 → 使用原视频画面 + 3. lyrics_only: 仅歌词/大纲 → 完整AI生成（音乐+MV） + +输出产物： + 1. KTV视频（双轨音频：伴奏默认播放 + 原唱可切换，卡拉OK字幕） + 2. 单轨字幕视频（原声混合 + 烧录字幕） + +状态机： + Mode A (audio_lyrics): + submitted → audio_preparing → demucs_separating → lyric_calibrating → subtitle_rendering → ktv_synthesizing → completed + + Mode B (video_lyrics): + submitted → video_preparing → demucs_separating → lyric_calibrating → subtitle_rendering → ktv_synthesizing → completed + + Mode C (lyrics_only): + submitted → lyric_generating → lyric_evaluating → music_generating → music_polling → + demucs_separating → lyric_calibrating → subtitle_rendering → ktv_synthesizing → completed + + 任何状态均可转换到 'failed'（错误时） + lyric_evaluating 可循环回 lyric_generating（阈值未达标时，最多3次） +""" +import json +import os +import asyncio +import time +import aiohttp +import subprocess +from pathlib import Path + +SERVICES = { + 'demucs': 'http://127.0.0.1:9080/api/demucs', + 'lyric_calibrate': 'http://127.0.0.1:9080/api/lyric_calibrate', + 'songrate': 'http://127.0.0.1:8900/api/evaluate', +} + +LLM_API = os.environ.get('LLM_API_BASE', 'https://token.opencomputing.cn/llmage/v1') +WORK_DIR = '/tmp/ktv_pipelines' + + +async def update_state(redis, pipeline_id, state, **kwargs): + """更新Pipeline状态到Redis""" + data = await redis.get(f'pipeline:{pipeline_id}') + if data: + pipeline = json.loads(data) + pipeline['state'] = state + pipeline['updated_at'] = time.time() + if 'artifacts' in kwargs: + pipeline['artifacts'].update(kwargs['artifacts']) + if 'error' in kwargs: + pipeline['errors'].append({'state': state, 'error': kwargs['error'], 'time': time.time()}) + pipeline.update({k: v for k, v in kwargs.items() if k not in ('artifacts', 'error')}) + await redis.set(f'pipeline:{pipeline_id}', json.dumps(pipeline, ensure_ascii=False), ex=86400) + + +async def call_llm(session, prompt, model='qwen3-235b-a22b', temperature=0.7, max_tokens=4096): + """调用LLM API""" + api_key = os.environ.get('LLM_API_KEY', '') + if not api_key: + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + api_key = getattr(env, 'llm_api_key', '') or '' + except: + pass + + headers = { + 'Authorization': f'Bearer {api_key}', + 'Content-Type': 'application/json' + } + payload = { + 'model': model, + 'catelogid': 't2t', + 'messages': [{'role': 'user', 'content': prompt}], + 'temperature': temperature, + 'max_tokens': max_tokens + } + async with session.post(f'{LLM_API}/chat/completions', json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=120)) as resp: + data = await resp.json() + return data.get('choices', [{}])[0].get('message', {}).get('content', '') + + +async def call_service(session, url, data, timeout=300): + """调用内部服务""" + async with session.post(url, json=data, timeout=aiohttp.ClientTimeout(total=timeout)) as resp: + return await resp.json() + + +async def call_service_multipart(session, url, audio_data, audio_filename, lyrics, timeout=600): + """调用lyric_calibrate服务（multipart/form-data）""" + boundary = f'Boundary{int(time.time()*1000)}' + + body_parts = [] + # audio_file field + body_parts.append(('--' + boundary + '\r\n').encode()) + body_parts.append(f'Content-Disposition: form-data; name="audio_file"; filename="{audio_filename}"\r\n'.encode()) + body_parts.append(b'Content-Type: audio/wav\r\n\r\n') + body_parts.append(audio_data) + body_parts.append(b'\r\n') + + # lyrics field + body_parts.append(('--' + boundary + '\r\n').encode()) + body_parts.append(b'Content-Disposition: form-data; name="lyrics"\r\n\r\n') + body_parts.append(lyrics.encode('utf-8')) + body_parts.append(b'\r\n') + + # closing boundary + body_parts.append(('--' + boundary + '--\r\n').encode()) + + body = b''.join(body_parts) + + headers = { + 'Content-Type': f'multipart/form-data; boundary={boundary}' + } + + async with session.post(url, data=body, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)) as resp: + return await resp.json() + + +# ============================================================================ +# Mode A: audio_lyrics - 原唱音频 + 歌词 +# ============================================================================ + +async def step_audio_preparing(pipeline, session): + """准备原唱音频：提取音频并确保格式正确""" + audio_path = pipeline.get('input_audio', '') + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + os.makedirs(pipeline_dir, exist_ok=True) + + # 复制输入音频到工作目录 + input_audio_local = os.path.join(pipeline_dir, 'input_audio.mp3') + + if audio_path.startswith('http'): + # 下载远程音频 + async with session.get(audio_path, timeout=aiohttp.ClientTimeout(total=120)) as resp: + with open(input_audio_local, 'wb') as f: + f.write(await resp.read()) + else: + # 本地文件，复制 + if not os.path.exists(audio_path): + raise ValueError(f'Input audio not found: {audio_path}') + subprocess.run(['cp', audio_path, input_audio_local], check=True) + + # 验证音频文件 + probe = subprocess.run( + ['ffprobe', '-v', 'error', '-show_entries', 'format=duration', + '-of', 'default=noprint_wrappers=1:nokey=1', input_audio_local], + capture_output=True, text=True + ) + duration = float(probe.stdout.strip()) + + return { + 'audio_path': input_audio_local, + 'audio_duration': duration, + 'mode': 'audio_lyrics' + } + + +# ============================================================================ +# Mode B: video_lyrics - 原唱视频 + 歌词 +# ============================================================================ + +async def step_video_preparing(pipeline, session): + """准备原唱视频：提取音频并确保格式正确""" + video_path = pipeline.get('input_video', '') + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + os.makedirs(pipeline_dir, exist_ok=True) + + # 复制输入视频到工作目录 + input_video_local = os.path.join(pipeline_dir, 'input_video.mp4') + input_audio_local = os.path.join(pipeline_dir, 'input_audio.mp3') + + if video_path.startswith('http'): + # 下载远程视频 + async with session.get(video_path, timeout=aiohttp.ClientTimeout(total=300)) as resp: + with open(input_video_local, 'wb') as f: + f.write(await resp.read()) + else: + # 本地文件，复制 + if not os.path.exists(video_path): + raise ValueError(f'Input video not found: {video_path}') + subprocess.run(['cp', video_path, input_video_local], check=True) + + # 从视频提取音频 + subprocess.run([ + 'ffmpeg', '-y', '-i', input_video_local, + '-vn', '-acodec', 'libmp3lame', '-ab', '192k', + input_audio_local + ], check=True, capture_output=True) + + # 验证视频文件 + probe = subprocess.run( + ['ffprobe', '-v', 'error', '-show_entries', 'format=duration', + '-of', 'default=noprint_wrappers=1:nokey=1', input_video_local], + capture_output=True, text=True + ) + duration = float(probe.stdout.strip()) + + return { + 'video_path': input_video_local, + 'audio_path': input_audio_local, + 'audio_duration': duration, + 'mode': 'video_lyrics' + } + + +# ============================================================================ +# Mode C: lyrics_only - 仅歌词（完整AI生成） +# ============================================================================ + +async def step_lyric_generate(pipeline, session): + """生成歌词（如果用户只提供了大纲）""" + outline = pipeline.get('outline', '') + if outline: + # 用户提供了大纲，生成完整歌词 + prompt = f"""你是一位专业的华语歌词创作者。 + +请根据以下大纲创作一首完整的歌词： + +大纲：{outline} +风格：{pipeline.get('scene', 'pop')} + +要求： +1. 歌词结构完整：包含主歌(verse)、副歌(chorus)、桥段(bridge) +2. 每行歌词节奏感强，适合演唱 +3. 标注段落类型，如 [Verse1], [Chorus], [Bridge] 等 +4. 总共16-24行歌词 +5. 注意押韵和情感表达 + +请直接输出歌词（不要markdown标记）： +""" + lyrics = await call_llm(session, prompt) + return {'lyrics': lyrics.strip()} + else: + # 用户已提供歌词，跳过生成 + return {'lyrics': pipeline.get('lyrics', '')} + + +async def step_lyric_evaluate(pipeline, session): + """评估歌词质量""" + lyrics = pipeline['artifacts'].get('lyrics', '') + prompt = f"""你是一位严格的歌词评审专家。请评估以下歌词的质量，满分10分。 + +评分维度： +1. 结构与韵律 (2分) +2. 情感表达 (2分) +3. 意象与画面感 (2分) +4. 语言质量 (2分) +5. 可唱性 (2分) + +歌词： +{lyrics} + +请严格按JSON格式返回： +{{"total_score": 7.5, "dimensions": {{"structure": 1.5, "emotion": 1.5, "imagery": 1.5, "language": 1.5, "singability": 1.5}}, "comment": "简短评语"}} +""" + result = await call_llm(session, prompt, temperature=0.3) + try: + result = result.strip() + if result.startswith('```'): + result = result.split('```')[1] + if result.startswith('json'): + result = result[4:] + score_data = json.loads(result.strip()) + except: + score_data = {'total_score': 7.0, 'dimensions': {}, 'comment': result[:200]} + return {'lyric_score': score_data} + + +async def step_music_generate(pipeline, session): + """生成音乐（Suno API）""" + lyrics = pipeline['artifacts'].get('lyrics', '') + scene = pipeline.get('scene', 'pop') + + api_key = os.environ.get('LLM_API_KEY', '') + if not api_key: + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + api_key = getattr(env, 'llm_api_key', '') or '' + except: + pass + + headers = { + 'Authorization': f'Bearer {api_key}', + 'Content-Type': 'application/json' + } + payload = { + 'model': 'suno-v4', + 'catelogid': 'music_gen', + 'prompt': f'{scene} style music', + 'tags': scene, + 'title': f"AI Song - {pipeline['id']}", + } + async with session.post(f'{LLM_API}/audio/generations', json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=60)) as resp: + data = await resp.json() + taskid = data.get('taskid', data.get('id', '')) + return {'music_task_id': taskid, 'music_submit_response': data} + + +async def step_music_poll(pipeline, session): + """轮询音乐生成结果""" + task_id = pipeline['artifacts'].get('music_task_id', '') + if not task_id: + raise ValueError('No music_task_id found') + + api_key = os.environ.get('LLM_API_KEY', '') + headers = {'Authorization': f'Bearer {api_key}'} + + # 最多等待10分钟 + for attempt in range(60): + async with session.get(f'{LLM_API}/tasks?taskid={task_id}', headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp: + data = await resp.json() + status = data.get('status', '') + if status == 'SUCCEEDED': + return {'music_url': data.get('result_url', ''), 'music_status': 'SUCCEEDED'} + elif status == 'FAILED': + raise ValueError(f'Music generation failed: {data}') + await asyncio.sleep(10) + + raise ValueError('Music generation timed out') + + +async def step_demucs_separate(pipeline, session): + """Demucs人声分离（通用步骤，适用于所有模式）""" + # 确定输入音频路径 + audio_path = pipeline['artifacts'].get('audio_path', '') + music_url = pipeline['artifacts'].get('music_url', '') + + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + os.makedirs(pipeline_dir, exist_ok=True) + + # Mode C: 需要先下载音乐 + if not audio_path and music_url: + audio_path = os.path.join(pipeline_dir, 'music.mp3') + async with session.get(music_url, timeout=aiohttp.ClientTimeout(total=120)) as resp: + with open(audio_path, 'wb') as f: + f.write(await resp.read()) + + if not audio_path: + raise ValueError('No audio source for demucs separation') + if not os.path.exists(audio_path): + raise ValueError(f'Audio file not found: {audio_path}') + + # 调用demucs分离 + result = await call_service(session, SERVICES['demucs'], {'filepath': audio_path}) + + if result.get('status') != 'success': + raise ValueError(f"Demucs failed: {result.get('error', result)}") + + # demucs API 返回 vocals_url/no_vocals_url (格式: /idfile?path=...) + # 需要转换为本地文件路径 + import re + + vocals_url = result.get('vocals_url', '') + no_vocals_url = result.get('no_vocals_url', '') + + def url_to_path(url): + """Convert /idfile?path=xxx to /tmp/xxx""" + match = re.search(r'path=(.+)', url) + if match: + return '/tmp' + match.group(1) + return '' + + vocals_path = url_to_path(vocals_url) + no_vocals_path = url_to_path(no_vocals_url) + + if not vocals_path or not os.path.exists(vocals_path): + raise ValueError(f'Demucs failed: vocals not found at {vocals_path} (from {vocals_url})') + if not no_vocals_path or not os.path.exists(no_vocals_path): + raise ValueError(f'Demucs failed: accompaniment not found at {no_vocals_path} (from {no_vocals_url})') + + return { + 'vocals_path': vocals_path, + 'no_vocals_path': no_vocals_path, + 'audio_path': audio_path # 保留原唱路径 + } + + +# ============================================================================ +# 通用步骤：歌词校准 + 字幕渲染 + 合成 +# ============================================================================ + +async def step_lyric_calibrating(pipeline, session): + """歌词时间校准（ASR + LLM）""" + vocals_path = pipeline['artifacts'].get('vocals_path', '') + audio_path = pipeline['artifacts'].get('audio_path', '') + lyrics = pipeline['artifacts'].get('lyrics', pipeline.get('lyrics', '')) + + # 优先使用分离后的纯人声，否则使用原音频 + source_audio = vocals_path if vocals_path else audio_path + + if not source_audio: + raise ValueError('No audio source for calibration') + if not lyrics: + raise ValueError('No lyrics for calibration') + + # 读取音频文件 + with open(source_audio, 'rb') as f: + audio_data = f.read() + + audio_filename = os.path.basename(source_audio) + + # 调用lyric_calibrate服务 + result = await call_service_multipart( + session, + SERVICES['lyric_calibrate'], + audio_data, + audio_filename, + lyrics, + timeout=600 + ) + + if result.get('status') != 'ok': + error_msg = result.get('error', 'Unknown error') + raise ValueError(f'Lyric calibration failed: {error_msg}') + + # 提取校准结果 + calibrated_lines = result.get('calibrated_lines', 0) + ass_file_url = result.get('ass_file', '') + json_data_url = result.get('json_data', '') + + # 下载并解析校准后的JSON + calibrated_json = None + if json_data_url: + # json_data_url 格式: '/idfile?path=lyric_calibrate/123456/calibrated.json' + import re + match = re.search(r'path=([^\s&]+)', json_data_url) + if match: + json_path = '/tmp/' + match.group(1) + if os.path.exists(json_path): + with open(json_path, 'r', encoding='utf-8') as f: + calibrated_json = json.load(f) + + return { + 'calibrated_subs': calibrated_json, + 'calibrated_ass_file': ass_file_url, + 'calibrated_lines': calibrated_lines, + } + + +async def step_subtitle_rendering(pipeline, session): + """渲染ASS字幕（使用build_ass.py的5-style方案）""" + calibrated = pipeline['artifacts'].get('calibrated_subs', []) + if isinstance(calibrated, dict): + segments = calibrated.get('segments', []) + elif isinstance(calibrated, list): + segments = calibrated + else: + segments = [] + + if not segments: + raise ValueError('No calibrated segments for subtitle rendering') + + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + ass_path = os.path.join(pipeline_dir, 'karaoke.ass') + + # 获取视频时长 + video_duration = pipeline['artifacts'].get('audio_duration', 0) + if not video_duration: + # 估算：最后一段歌词结束时间 + 10s + last_end = max(seg.get('end', 0) for seg in segments) + video_duration = last_end + 10 + + # 获取标题和词曲信息 + title = pipeline.get('title', '未知歌曲') + lyricist = pipeline.get('lyricist', '未知') + composer = pipeline.get('composer', '未知') + credit_text = f"词：{lyricist} 曲：{composer}" + + def sec2ass(s): + h = int(s // 3600) + m = int((s % 3600) // 60) + sec = s % 60 + return f"{h}:{m:02d}:{sec:05.2f}" + + # ASS Header（5个样式） + 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: Title,WenQuanYi Zen Hei,140,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,2,0,1,4,2,5,30,30,30,1 +Style: Credit,WenQuanYi Zen Hei,56,&H00CCCCCC,&H0000FFFF,&H00000000,&H80000000,0,0,0,0,100,100,1,0,1,3,1,5,30,30,40,1 +Style: TitleSmall,WenQuanYi Zen Hei,42,&H00DDDDDD,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,3,1,7,30,30,60,1 +Style: CreditSmall,WenQuanYi Zen Hei,36,&H00888888,&H0000FFFF,&H00000000,&H80000000,0,0,0,0,100,100,0,0,1,2,1,7,30,30,30,1 +Style: Karaoke,WenQuanYi Zen Hei,80,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,4,2,2,30,30,60,1 + +[Events] +Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text +""" + + events = [] + + # 标题：0:00 - 0:15 居中（15秒） + events.append(f"Dialogue: 0,{sec2ass(0)},{sec2ass(15)},Title,,0,0,0,,{{\\fad(1500,1500)}}{title}") + + # 词曲：0:05 - 0:15 居中（10秒） + events.append(f"Dialogue: 0,{sec2ass(5)},{sec2ass(15)},Credit,,0,0,0,,{{\\fad(1000,1000)}}{credit_text}") + + # 标题小字：0:15 - 结束，左上角常驻 + events.append(f"Dialogue: 0,{sec2ass(15)},{sec2ass(video_duration)},TitleSmall,,0,0,0,,{title}") + + # 词曲小字：0:15 - 结束，左上角常驻（在标题下方） + events.append(f"Dialogue: 0,{sec2ass(15)},{sec2ass(video_duration)},CreditSmall,,0,0,0,,{credit_text}") + + # 卡拉OK歌词：逐字高亮 + for line_data in segments: + line = line_data.get('line', line_data.get('text', '')) + start = line_data.get('start', 0) + end = line_data.get('end', 0) + chars = line_data.get('chars', []) + + if start < 0 or end <= start: + continue + + # 构建 \kf 卡拉OK计时（每字厘秒数） + parts = [] + for ch in chars: + cs = ch.get('start', 0) + ce = ch.get('end', cs + 0.1) + duration_cs = int(max((ce - cs) * 100, 1)) # 厘秒 + parts.append(f"{{\\kf{duration_cs}}}{ch.get('char', '')}") + + if parts: + text = "".join(parts) + events.append(f"Dialogue: 0,{sec2ass(start)},{sec2ass(end)},Karaoke,,0,0,0,,{text}") + + ass_content = header + "\n".join(events) + "\n" + + with open(ass_path, 'w', encoding='utf-8') as f: + f.write(ass_content) + + return {'subtitle_path': ass_path} + + +async def step_ktv_synthesizing(pipeline, session): + """合成KTV视频（双轨音频 + 单轨字幕视频）""" + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + + # 获取视频源（如果有） + video_path = pipeline['artifacts'].get('video_path', '') + vocals_path = pipeline['artifacts'].get('vocals_path', '') + no_vocals_path = pipeline['artifacts'].get('no_vocals_path', '') + audio_path = pipeline['artifacts'].get('audio_path', '') + subtitle_path = pipeline['artifacts'].get('subtitle_path', '') + + if not subtitle_path: + raise ValueError('Missing subtitle_path') + + # 输出文件 + ktv_output = os.path.join(pipeline_dir, 'ktv_dual_track.mp4') + single_output = os.path.join(pipeline_dir, 'ktv_single_track.mp4') + + # 确定音频源 + original_audio = audio_path # 原唱（完整混合） + accompaniment = no_vocals_path # 伴奏 + + if not os.path.exists(original_audio): + raise ValueError(f'Original audio not found: {original_audio}') + if accompaniment and not os.path.exists(accompaniment): + raise ValueError(f'Accompaniment not found: {accompaniment}') + + # ======================================================================== + # 1. 合成KTV双轨视频（伴奏默认 + 原唱可切换） + # ======================================================================== + if video_path and os.path.exists(video_path): + # 有视频：视频 + 伴奏（默认） + 原唱 + cmd_ktv = [ + 'ffmpeg', '-y', + '-i', video_path, + '-i', accompaniment if accompaniment else original_audio, + '-i', original_audio, + '-vf', f'ass={subtitle_path}', + '-map', '0:v', + '-map', '1:a', + '-map', '2:a', + '-c:v', 'libx264', '-preset', 'medium', '-crf', '23', + '-c:a', 'aac', '-b:a', '192k', + '-metadata:s:a:0', 'handler_name=伴奏(Accompaniment)', + '-metadata:s:a:1', 'handler_name=原唱(Original)', + '-disposition:a:0', 'default', + '-disposition:a:1', '0', + ktv_output + ] + else: + # 无视频：黑屏 + 伴奏 + 原唱 + video_duration = pipeline['artifacts'].get('audio_duration', 0) + if not video_duration: + # 估算时长 + calibrated = pipeline['artifacts'].get('calibrated_subs', []) + last_end = max(seg.get('end', 0) for seg in calibrated) if calibrated else 180 + video_duration = last_end + 10 + + cmd_ktv = [ + 'ffmpeg', '-y', + '-f', 'lavfi', '-i', f'color=c=black:s=1920x1080:r=30:d={video_duration}', + '-i', accompaniment if accompaniment else original_audio, + '-i', original_audio, + '-vf', f'ass={subtitle_path}', + '-map', '0:v', + '-map', '1:a', + '-map', '2:a', + '-c:v', 'libx264', '-preset', 'medium', '-crf', '23', + '-c:a', 'aac', '-b:a', '192k', + '-shortest', + '-metadata:s:a:0', 'handler_name=伴奏(Accompaniment)', + '-metadata:s:a:1', 'handler_name=原唱(Original)', + '-disposition:a:0', 'default', + '-disposition:a:1', '0', + ktv_output + ] + + # 执行KTV合成 + proc = await asyncio.create_subprocess_exec( + *cmd_ktv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode != 0: + raise ValueError(f'KTV synthesis failed: {stderr.decode()[:500]}') + + # ======================================================================== + # 2. 合成单轨字幕视频（原声混合 + 烧录字幕） + # ======================================================================== + if video_path and os.path.exists(video_path): + # 有视频：视频 + 原唱音频 + cmd_single = [ + 'ffmpeg', '-y', + '-i', video_path, + '-i', original_audio, + '-vf', f'ass={subtitle_path}', + '-map', '0:v', + '-map', '1:a', + '-c:v', 'libx264', '-preset', 'medium', '-crf', '23', + '-c:a', 'aac', '-b:a', '192k', + single_output + ] + else: + # 无视频：黑屏 + 原唱音频 + video_duration = pipeline['artifacts'].get('audio_duration', 0) + if not video_duration: + calibrated = pipeline['artifacts'].get('calibrated_subs', []) + last_end = max(seg.get('end', 0) for seg in calibrated) if calibrated else 180 + video_duration = last_end + 10 + + cmd_single = [ + 'ffmpeg', '-y', + '-f', 'lavfi', '-i', f'color=c=black:s=1920x1080:r=30:d={video_duration}', + '-i', original_audio, + '-vf', f'ass={subtitle_path}', + '-map', '0:v', + '-map', '1:a', + '-c:v', 'libx264', '-preset', 'medium', '-crf', '23', + '-c:a', 'aac', '-b:a', '192k', + '-shortest', + single_output + ] + + # 执行单轨合成 + proc = await asyncio.create_subprocess_exec( + *cmd_single, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await proc.communicate() + + if proc.returncode != 0: + raise ValueError(f'Single track synthesis failed: {stderr.decode()[:500]}') + + # 验证输出文件 + if not os.path.exists(ktv_output): + raise ValueError(f'KTV output not created: {ktv_output}') + if not os.path.exists(single_output): + raise ValueError(f'Single output not created: {single_output}') + + # 获取文件大小 + ktv_size = os.path.getsize(ktv_output) / (1024 * 1024) # MB + single_size = os.path.getsize(single_output) / (1024 * 1024) # MB + + return { + 'ktv_dual_track_path': ktv_output, + 'ktv_single_track_path': single_output, + 'ktv_size_mb': round(ktv_size, 2), + 'single_size_mb': round(single_size, 2), + } + + +# ============================================================================ +# 状态机定义（按模式分支） +# ============================================================================ + +TRANSITIONS_MODE_A = { + 'submitted': ('audio_preparing', None), + 'audio_preparing': ('demucs_separating', step_audio_preparing), + 'demucs_separating': ('lyric_calibrating', step_demucs_separate), + 'lyric_calibrating': ('subtitle_rendering', step_lyric_calibrating), + 'subtitle_rendering': ('ktv_synthesizing', step_subtitle_rendering), + 'ktv_synthesizing': ('completed', step_ktv_synthesizing), +} + +TRANSITIONS_MODE_B = { + 'submitted': ('video_preparing', None), + 'video_preparing': ('demucs_separating', step_video_preparing), + 'demucs_separating': ('lyric_calibrating', step_demucs_separate), + 'lyric_calibrating': ('subtitle_rendering', step_lyric_calibrating), + 'subtitle_rendering': ('ktv_synthesizing', step_subtitle_rendering), + 'ktv_synthesizing': ('completed', step_ktv_synthesizing), +} + +TRANSITIONS_MODE_C = { + 'submitted': ('lyric_generating', None), + 'lyric_generating': ('lyric_evaluating', step_lyric_generate), + 'lyric_evaluating': ('music_generating', step_lyric_evaluate), # 或循环回 lyric_generating + 'music_generating': ('music_polling', step_music_generate), + 'music_polling': ('demucs_separating', step_music_poll), + 'demucs_separating': ('lyric_calibrating', step_demucs_separate), + 'lyric_calibrating': ('subtitle_rendering', step_lyric_calibrating), + 'subtitle_rendering': ('ktv_synthesizing', step_subtitle_rendering), + 'ktv_synthesizing': ('completed', step_ktv_synthesizing), +} + + +def get_transitions(mode): + """根据模式获取状态转换表""" + if mode == 'audio_lyrics': + return TRANSITIONS_MODE_A + elif mode == 'video_lyrics': + return TRANSITIONS_MODE_B + else: # lyrics_only (default) + return TRANSITIONS_MODE_C + + +async def run_pipeline(pipeline_id): + """执行KTV Pipeline""" + import aioredis + redis = await aioredis.from_url('redis://127.0.0.1:6379', db=1) + + try: + data = await redis.get(f'pipeline:{pipeline_id}') + if not data: + raise ValueError(f'Pipeline {pipeline_id} not found') + + pipeline = json.loads(data) + mode = pipeline.get('mode', 'lyrics_only') + transitions = get_transitions(mode) + + pipeline_dir = os.path.join(WORK_DIR, pipeline_id) + os.makedirs(pipeline_dir, exist_ok=True) + + async with aiohttp.ClientSession() as session: + state = pipeline.get('state', 'submitted') + retry_counts = {} + + while state != 'completed' and state != 'failed': + await update_state(redis, pipeline_id, state) + + transition = transitions.get(state) + if not transition: + await update_state(redis, pipeline_id, 'failed', error=f'No transition for state: {state}') + break + + next_state, handler = transition + + if handler: + try: + result = await handler(pipeline, session) + pipeline['artifacts'].update(result) + + # 歌词评估：阈值检查 + if state == 'lyric_evaluating': + score = result.get('lyric_score', {}).get('total_score', 0) + retry_counts['lyric'] = retry_counts.get('lyric', 0) + 1 + if score < pipeline.get('lyric_threshold', 8.5) and retry_counts['lyric'] < 3: + next_state = 'lyric_generating' # 循环回生成 + else: + next_state = 'music_generating' # 通过，继续 + + except Exception as e: + await update_state(redis, pipeline_id, state, error=str(e)) + # 重试机制 + retry_counts[state] = retry_counts.get(state, 0) + 1 + if retry_counts[state] > 2: + await update_state(redis, pipeline_id, 'failed', error=f'{state} failed after retries: {e}') + break + await asyncio.sleep(5) + continue + + state = next_state + + finally: + await redis.close() diff --git a/workers/ktv_pipeline.py.bak b/workers/ktv_pipeline.py.bak new file mode 100644 index 0000000..14ac0d9 --- /dev/null +++ b/workers/ktv_pipeline.py.bak @@ -0,0 +1,539 @@ +# -*- coding:utf-8 -*- +""" +KTV Pipeline Worker - 状态机驱动的完整歌曲制作流程 + +State machine: + submitted -> lyric_generating -> lyric_evaluating -> lyric_done + -> music_generating -> music_separating -> music_aligning -> music_calibrating -> music_evaluating -> music_done + -> mv_story_generating -> mv_designing -> mv_frames_generating -> mv_video_generating -> mv_video_evaluating -> mv_merging -> mv_done + -> subtitle_rendering -> ktv_synthesizing -> completed + + Any state can transition to 'failed' on error. + Evaluation states can loop back to the generating state if threshold not met. +""" +import json +import os +import asyncio +import time +import aiohttp + +SERVICES = { + 'align': 'http://127.0.0.1:8080/api/align', + 'fastwhisper': 'http://127.0.0.1:9925/api/asr', + 'fastwhisper_status': 'http://127.0.0.1:9925/api/status', + 'songrate': 'http://127.0.0.1:8900/api/evaluate', + 'demucs': 'http://127.0.0.1:9080/api/demucs', + 'lyric_calibrate': 'http://127.0.0.1:9080/api/lyric_calibrate', + 'merge': 'http://127.0.0.1:9080/api/merge', + 'subtitle': 'http://127.0.0.1:9080/api/subtitle', + 'ktv': 'http://127.0.0.1:9080/api/ktv', +} + +LLM_API = os.environ.get('LLM_API_BASE', 'https://token.opencomputing.cn/llmage/v1') +WORK_DIR = '/tmp/ktv_pipelines' + + +async def update_state(redis, pipeline_id, state, **kwargs): + """Update pipeline state in Redis""" + data = await redis.get(f'pipeline:{pipeline_id}') + if data: + pipeline = json.loads(data) + pipeline['state'] = state + pipeline['updated_at'] = time.time() + if 'artifacts' in kwargs: + pipeline['artifacts'].update(kwargs['artifacts']) + if 'error' in kwargs: + pipeline['errors'].append({'state': state, 'error': kwargs['error'], 'time': time.time()}) + pipeline.update({k: v for k, v in kwargs.items() if k not in ('artifacts', 'error')}) + await redis.set(f'pipeline:{pipeline_id}', json.dumps(pipeline, ensure_ascii=False), ex=86400) + + +async def call_llm(session, prompt, model='qwen3-235b-a22b', temperature=0.7, max_tokens=4096): + """Call Sage LLM API""" + api_key = os.environ.get('LLM_API_KEY', '') + if not api_key: + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + api_key = getattr(env, 'llm_api_key', '') or '' + except: + pass + + headers = { + 'Authorization': f'Bearer {api_key}', + 'Content-Type': 'application/json' + } + payload = { + 'model': model, + 'catelogid': 't2t', + 'messages': [{'role': 'user', 'content': prompt}], + 'temperature': temperature, + 'max_tokens': max_tokens + } + async with session.post(f'{LLM_API}/chat/completions', json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=120)) as resp: + data = await resp.json() + return data.get('choices', [{}])[0].get('message', {}).get('content', '') + + +async def call_service(session, url, data, timeout=300): + """Call an internal service""" + async with session.post(url, json=data, timeout=aiohttp.ClientTimeout(total=timeout)) as resp: + return await resp.json() + + +async def step_lyric_generate(pipeline, session): + """Step 1: Generate lyrics using LLM""" + prompt = f"""你是一位专业的华语歌词创作者。 + +请根据以下描述创作一首完整的歌词： + +描述：{pipeline['description']} +风格：{pipeline.get('scene', 'pop')} + +要求： +1. 歌词结构完整：包含主歌(verse)、副歌(chorus)、桥段(bridge) +2. 每行歌词节奏感强，适合演唱 +3. 标注段落类型，如 [Verse1], [Chorus], [Bridge] 等 +4. 总共16-24行歌词 +5. 注意押韵和情感表达 + +请直接输出歌词（不要markdown标记）： +""" + lyrics = await call_llm(session, prompt) + return {'lyrics': lyrics.strip()} + + +async def step_lyric_evaluate(pipeline, session): + """Step 2: Evaluate lyrics (call lyric-evaluator via Hermes or LLM fallback)""" + lyrics = pipeline['artifacts'].get('lyrics', '') + # Use LLM as evaluator since lyric-evaluator is a Hermes skill + prompt = f"""你是一位严格的歌词评审专家。请评估以下歌词的质量，满分10分。 + +评分维度： +1. 结构与韵律 (2分) +2. 情感表达 (2分) +3. 意象与画面感 (2分) +4. 语言质量 (2分) +5. 可唱性 (2分) + +歌词： +{lyrics} + +请严格按JSON格式返回： +{{"total_score": 7.5, "dimensions": {{"structure": 1.5, "emotion": 1.5, "imagery": 1.5, "language": 1.5, "singability": 1.5}}, "comment": "简短评语"}} +""" + result = await call_llm(session, prompt, temperature=0.3) + try: + result = result.strip() + if result.startswith('```'): + result = result.split('```')[1] + if result.startswith('json'): + result = result[4:] + score_data = json.loads(result.strip()) + except: + score_data = {'total_score': 7.0, 'dimensions': {}, 'comment': result[:200]} + return {'lyric_score': score_data} + + +async def step_music_generate(pipeline, session): + """Step 4: Generate music via Suno API (through Sage llmage)""" + lyrics = pipeline['artifacts'].get('lyrics', '') + scene = pipeline.get('scene', 'pop') + + # Call Suno through Sage + api_key = os.environ.get('LLM_API_KEY', '') + if not api_key: + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + api_key = getattr(env, 'llm_api_key', '') or '' + except: + pass + + headers = { + 'Authorization': f'Bearer {api_key}', + 'Content-Type': 'application/json' + } + payload = { + 'model': 'suno-v4', + 'catelogid': 'music_gen', + 'prompt': f'{scene} style music', + 'tags': scene, + 'title': f"AI Song - {pipeline['id']}", + } + async with session.post(f'{LLM_API}/audio/generations', json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=60)) as resp: + data = await resp.json() + taskid = data.get('taskid', data.get('id', '')) + return {'music_task_id': taskid, 'music_submit_response': data} + + +async def step_music_poll(pipeline, session): + """Poll for music generation result""" + task_id = pipeline['artifacts'].get('music_task_id', '') + if not task_id: + raise ValueError('No music_task_id found') + + api_key = os.environ.get('LLM_API_KEY', '') + headers = {'Authorization': f'Bearer {api_key}'} + + # Poll Sage task endpoint + for attempt in range(60): # max 10 minutes + async with session.get(f'{LLM_API}/tasks?taskid={task_id}', headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp: + data = await resp.json() + status = data.get('status', '') + if status == 'SUCCEEDED': + return {'music_url': data.get('result_url', ''), 'music_status': 'SUCCEEDED'} + elif status == 'FAILED': + raise ValueError(f'Music generation failed: {data}') + await asyncio.sleep(10) + + raise ValueError('Music generation timed out') + + +async def step_music_separate(pipeline, session): + """Step 5: Demucs vocal separation""" + music_url = pipeline['artifacts'].get('music_url', '') + if not music_url: + raise ValueError('No music_url found') + + # Download the music file first + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + os.makedirs(pipeline_dir, exist_ok=True) + music_path = os.path.join(pipeline_dir, 'music.mp3') + + async with session.get(music_url, timeout=aiohttp.ClientTimeout(total=120)) as resp: + with open(music_path, 'wb') as f: + f.write(await resp.read()) + + # Call demucs + result = await call_service(session, SERVICES['demucs'], {'filepath': music_path}) + return {'vocals_path': result.get('vocals_path', ''), 'no_vocals_path': result.get('no_vocals_path', '')} + + +async def step_music_align(pipeline, session): + """Step 6: WhisperX transcription + alignment""" + vocals_path = pipeline['artifacts'].get('vocals_path', '') + lyrics = pipeline['artifacts'].get('lyrics', '') + + # Submit to fastwhisper + result = await call_service(session, SERVICES['fastwhisper'], {'audio_path': vocals_path}) + task_id = result.get('task_id', '') + + if not task_id: + raise ValueError(f'WhisperX submit failed: {result}') + + # Poll for result + for attempt in range(30): + await asyncio.sleep(5) + status_result = await call_service(session, SERVICES['fastwhisper_status'], {'task_id': task_id}) + status = status_result.get('status', '') + if status == 'SUCCEEDED': + return {'whisperx_json': status_result.get('data', status_result)} + elif status == 'FAILED': + raise ValueError(f'WhisperX failed: {status_result}') + + raise ValueError('WhisperX alignment timed out') + + +async def step_music_calibrate(pipeline, session): + """Step 7: ASR + LLM lyric calibration (integrated service)""" + vocals_path = pipeline['artifacts'].get('vocals_path', '') + lyrics = pipeline['artifacts'].get('lyrics', '') + + if not vocals_path: + raise ValueError('Missing vocals_path for calibration') + if not lyrics: + raise ValueError('Missing lyrics for calibration') + + # Read vocals file + with open(vocals_path, 'rb') as f: + audio_data = f.read() + + # Build multipart/form-data + boundary = f'Boundary{int(time.time()*1000)}' + + body_parts = [] + # audio_file field + body_parts.append(('--' + boundary + '\r\n').encode()) + body_parts.append(b'Content-Disposition: form-data; name="audio_file"; filename="vocals.wav"\r\n') + body_parts.append(b'Content-Type: audio/wav\r\n\r\n') + body_parts.append(audio_data) + body_parts.append(b'\r\n') + + # lyrics field + body_parts.append(('--' + boundary + '\r\n').encode()) + body_parts.append(b'Content-Disposition: form-data; name="lyrics"\r\n\r\n') + body_parts.append(lyrics.encode('utf-8')) + body_parts.append(b'\r\n') + + # closing boundary + body_parts.append(('--' + boundary + '--\r\n').encode()) + + body = b''.join(body_parts) + + headers = { + 'Content-Type': f'multipart/form-data; boundary={boundary}' + } + + # Call lyric_calibrate service (integrated ASR + LLM) + url = SERVICES['lyric_calibrate'] + timeout = aiohttp.ClientTimeout(total=600) # 10 minutes for ASR + LLM + + async with session.post(url, data=body, headers=headers, timeout=timeout) as resp: + result = await resp.json() + + if result.get('status') != 'ok': + error_msg = result.get('error', 'Unknown error') + raise ValueError(f'Lyric calibration failed: {error_msg}') + + # Extract calibrated data + calibrated_lines = result.get('calibrated_lines', 0) + ass_file_url = result.get('ass_file', '') + json_data_url = result.get('json_data', '') + segments_used = result.get('segments_used', 0) + segments_total = result.get('segments_total', 0) + + # Download and parse calibrated JSON + calibrated_json = None + if json_data_url: + # json_data_url is like '/idfile?path=lyric_calibrate/123456/calibrated.json' + # We need to extract the actual file path + import re + match = re.search(r'path=([^\s&]+)', json_data_url) + if match: + json_path = '/tmp/' + match.group(1) + if os.path.exists(json_path): + with open(json_path, 'r', encoding='utf-8') as f: + calibrated_json = json.load(f) + + return { + 'calibrated_subs': calibrated_json, + 'calibrated_ass_file': ass_file_url, + 'calibrated_lines': calibrated_lines, + 'segments_used': segments_used, + 'segments_total': segments_total + } + + +async def step_music_evaluate(pipeline, session): + """Step 8: Song quality evaluation""" + music_path = os.path.join(WORK_DIR, pipeline['id'], 'music.mp3') + + result = await call_service(session, SERVICES['songrate'], { + 'filepath': music_path, + 'scene': pipeline.get('scene', 'pop') + }) + + total = result.get('total_score', 0) + return {'music_score': result, 'music_total_score': total} + + +async def step_mv_story(pipeline, session): + """Step 10: LLM generates MV storyline + storyboard""" + lyrics = pipeline['artifacts'].get('lyrics', '') + + prompt = f"""你是一位MV导演。请根据以下歌词，创作MV的分镜脚本。 + +歌词： +{lyrics} + +要求： +1. 为每个段落设计1-2个镜头 +2. 每个镜头描述：场景、人物动作、镜头运动、情绪氛围 +3. 输出JSON数组格式 + +请严格按JSON格式输出： +{{"scenes": [ + {{"lyric_line": "歌词行", "description": "场景描述", "camera": "镜头运动", "mood": "情绪", "needs_reference": false}}, + ... +]}} +""" + result = await call_llm(session, prompt, temperature=0.8) + try: + result = result.strip() + if result.startswith('```'): + result = result.split('```')[1] + if result.startswith('json'): + result = result[4:] + storyboard = json.loads(result.strip()) + except: + storyboard = {'scenes': [{'description': result[:500], 'camera': 'static', 'mood': 'neutral', 'needs_reference': False}]} + return {'storyboard': storyboard} + + +async def step_mv_generate_video(pipeline, session): + """Steps 11-14: Generate MV video segments via Sage""" + # This step calls Sage video generation API (t2v/i2v/r2v) + # For now, create placeholder - actual video gen needs Sage integration + storyboard = pipeline['artifacts'].get('storyboard', {}) + return {'mv_segments': [], 'mv_note': 'Video generation pending Sage integration'} + + +async def step_subtitle_render(pipeline, session): + """Step 18: Generate ASS subtitle file with karaoke effects""" + # New format: calibrated_subs is a list of {line, start, end, chars} + # Old format: calibrated_subs was a dict with segments key + calibrated = pipeline['artifacts'].get('calibrated_subs', []) + if isinstance(calibrated, dict): + segments = calibrated.get('segments', []) + elif isinstance(calibrated, list): + segments = calibrated + else: + segments = [] + + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + ass_path = os.path.join(pipeline_dir, 'karaoke.ass') + + # Generate ASS file + ass_content = """[Script Info] +Title: KTV Karaoke Subtitles +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: Karaoke,Noto Sans CJK SC,72,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,3,1,2,20,20,50,1 + +[Events] +Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text +""" + for seg in segments: + start = seg.get('start', 0) + end = seg.get('end', 0) + chars = seg.get('chars', []) + text = seg.get('line', seg.get('text', '')) + + if chars: + # Build karaoke tags + kar_parts = [] + for c in chars: + dur = max(1, int((c.get('end', c.get('start', 0)) - c.get('start', 0)) * 100)) + kar_parts.append(f'{{\\k{dur}}}{c.get("char", "")}') + line_text = ''.join(kar_parts) + else: + line_text = text + + def fmt_time(t): + h = int(t // 3600) + m = int((t % 3600) // 60) + s = t % 60 + return f'{h}:{m:02d}:{s:05.2f}' + + ass_content += f'Dialogue: 0,{fmt_time(start)},{fmt_time(end)},Karaoke,,0,0,0,,{line_text}\n' + + with open(ass_path, 'w', encoding='utf-8') as f: + f.write(ass_content) + + return {'subtitle_path': ass_path} + + +async def step_ktv_synthesize(pipeline, session): + """Step 19: Final KTV synthesis""" + pipeline_dir = os.path.join(WORK_DIR, pipeline['id']) + + video_path = pipeline['artifacts'].get('mv_merged_path', '') + vocals_path = pipeline['artifacts'].get('vocals_path', '') + no_vocals_path = pipeline['artifacts'].get('no_vocals_path', '') + subtitle_path = pipeline['artifacts'].get('subtitle_path', '') + + if not all([video_path, no_vocals_path, subtitle_path]): + # If no MV, just create audio KTV + result = {'status': 'partial', 'note': 'Missing some artifacts, creating audio-only KTV'} + return {'ktv_result': result} + + result = await call_service(session, SERVICES['ktv'], { + 'video_path': video_path, + 'vocals_path': vocals_path, + 'accompaniment_path': no_vocals_path, + 'subtitle_path': subtitle_path, + 'output_name': pipeline['id'] + }, timeout=600) + + return {'ktv_result': result} + + +# State transition table +TRANSITIONS = { + 'submitted': ('lyric_generating', step_lyric_generate), + 'lyric_generating': ('lyric_evaluating', None), + 'lyric_evaluating': ('lyric_done', None), # or back to lyric_generating + 'lyric_done': ('music_generating', step_music_generate), + 'music_generating': ('music_separating', step_music_poll), + 'music_separating': ('music_aligning', step_music_separate), + 'music_aligning': ('music_calibrating', None), # Skipped - lyric_calibrate handles ASR+LLM + 'music_calibrating': ('music_evaluating', step_music_calibrate), + 'music_evaluating': ('music_done', step_music_evaluate), # or back + 'music_done': ('mv_story_generating', step_mv_story), + 'mv_story_generating': ('mv_video_generating', None), + 'mv_video_generating': ('subtitle_rendering', step_mv_generate_video), + 'subtitle_rendering': ('ktv_synthesizing', step_subtitle_render), + 'ktv_synthesizing': ('completed', step_ktv_synthesize), +} + + +async def run_pipeline(pipeline_id): + """Execute the full KTV pipeline""" + import aioredis + redis = await aioredis.from_url('redis://127.0.0.1:6379', db=1) + + try: + data = await redis.get(f'pipeline:{pipeline_id}') + if not data: + raise ValueError(f'Pipeline {pipeline_id} not found') + + pipeline = json.loads(data) + pipeline_dir = os.path.join(WORK_DIR, pipeline_id) + os.makedirs(pipeline_dir, exist_ok=True) + + async with aiohttp.ClientSession() as session: + state = pipeline['state'] + retry_counts = {} + + while state != 'completed' and state != 'failed': + await update_state(redis, pipeline_id, state) + + transition = TRANSITIONS.get(state) + if not transition: + await update_state(redis, pipeline_id, 'failed', error=f'No transition for state: {state}') + break + + next_state, handler = transition + + if handler: + try: + result = await handler(pipeline, session) + pipeline['artifacts'].update(result) + + # Check thresholds for evaluation states + if state == 'lyric_evaluating': + score = result.get('lyric_score', {}).get('total_score', 0) + retry_counts['lyric'] = retry_counts.get('lyric', 0) + 1 + if score < pipeline.get('lyric_threshold', 8.5) and retry_counts['lyric'] < 3: + next_state = 'lyric_generating' + else: + next_state = 'lyric_done' + + elif state == 'music_evaluating': + score = result.get('music_total_score', 0) + retry_counts['music'] = retry_counts.get('music', 0) + 1 + if score < pipeline.get('music_threshold', 7.5) and retry_counts['music'] < 2: + next_state = 'music_generating' + else: + next_state = 'music_done' + + except Exception as e: + await update_state(redis, pipeline_id, state, error=str(e)) + # Retry once on error + retry_counts[state] = retry_counts.get(state, 0) + 1 + if retry_counts[state] > 2: + await update_state(redis, pipeline_id, 'failed', error=f'{state} failed after retries: {e}') + break + await asyncio.sleep(5) + continue + + state = next_state + + finally: + await redis.close()