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