.gitignore

@@ -0,0 +1,10 @@
+__pycache__/
+*.pyc
+*.pyo
+logs/
+*.log
+nohup.out
+nohup_gpu*.out
+py3/
+*.egg-info/
+*.pid

README.md

@@ -1,2 +0,0 @@
-# media-server
-

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)

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)

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)

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)

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

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)

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

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)

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)

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)

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)

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

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)

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"
+}

i18n/en/msg.txt

@@ -0,0 +1,3 @@
+Media Server=Media Server
+Task submitted=Task submitted
+Task not found=Task not found

i18n/zh-cn/msg.txt

@@ -0,0 +1,3 @@
+Media Server=媒体服务器
+Task submitted=任务已提交
+Task not found=任务未找到

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

stop.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+# media-server stop script
+pkill -f "python ah.py" && echo "Stopped" || echo "Not running"

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()

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()