"""KTV pipeline step handlers — 通过 llmage API 调用 GPU 服务。

配置方式（环境变量）：
  KTV_API_BASE:  llmage API 地址，默认 https://token.opencomputing.cn/llmage/v1
  KTV_API_KEY:   Bearer token
  KTV_FILE_BASE: 文件服务地址，默认同 KTV_API_BASE 所在域

步骤→模型映射（KTV_STEP_MODELS 字典，可通过环境变量覆盖各项）:
  demucs_separating → demucs-separate
  asr_transcribing    → asr-transcribe
  subtitle_rendering  → subtitle-render
  face_detecting      → face-detect
  realesrgan_upscaling → realesrgan-upscale
  rvc_converting      → rvc-convert
  video_evaluating    → video-eval-evaluate
  songrate_evaluating → songrate-evaluate

本地操作（ffmpeg/ffprobe）不走 API。
"""

import asyncio
import json
import os
import logging
import time
import aiohttp

from app.quality_gate import (
    eval_music_quality, eval_character_design, eval_character_image,
    eval_storyboard, eval_ktv_synthesis, eval_scene_video_quality,
)
from app.model_selector import handle_model_selecting

logger = logging.getLogger("pipeline.handlers.ktv")

# ── config ──────────────────────────────────────────────────────────

API_BASE  = os.environ.get("KTV_API_BASE", "https://token.opencomputing.cn/llmage/v1")
API_KEY   = os.environ.get("KTV_API_KEY", "")
FILE_BASE = os.environ.get("KTV_FILE_BASE", API_BASE.rsplit("/", 2)[0])

STEP_MODELS = {
    "demucs_separating":    os.environ.get("KTV_MODEL_DEMUCS",    "demucs-separate"),
    "asr_transcribing":     os.environ.get("KTV_MODEL_ASR",       "asr-transcribe"),
    "subtitle_rendering":   os.environ.get("KTV_MODEL_SUBTITLE",  "subtitle-render"),
    "face_detecting":       os.environ.get("KTV_MODEL_FACE",      "face-detect"),
    "realesrgan_upscaling": os.environ.get("KTV_MODEL_REALESRGAN","realesrgan-upscale"),
    "rvc_converting":       os.environ.get("KTV_MODEL_RVC",       "rvc-convert"),
    "video_evaluating":     os.environ.get("KTV_MODEL_VIDEO_EVAL","video-eval-evaluate"),
    "songrate_evaluating":  os.environ.get("KTV_MODEL_SONGRATE",  "songrate-evaluate"),
}

LOCAL_WORK_DIR = "/data/pipeline/ktv"
GPU_HOST = None  # 不再直连 GPU，保留变量兼容旧引用，新代码不应使用


def _task_dir(task_id: str) -> str:
    d = os.path.join(LOCAL_WORK_DIR, task_id)
    os.makedirs(d, exist_ok=True)
    return d


# ── API client ─────────────────────────────────────────────────────

class KtvClient:
    """llmage pipeline API 客户端。"""

    def __init__(self, base: str = "", key: str = ""):
        self.base = base or API_BASE
        self.key  = key  or API_KEY
        self._session = None

    async def _ensure_session(self):
        if self._session is None:
            self._session = aiohttp.ClientSession(
                headers={"Authorization": f"Bearer {self.key}",
                         "Content-Type": "application/json"},
                timeout=aiohttp.ClientTimeout(total=600),
            )

    async def close(self):
        if self._session:
            await self._session.close()
            self._session = None

    async def call(self, model: str, params: dict) -> dict:
        """同步调用，返回 GPU 响应 dict。"""
        await self._ensure_session()
        body = {"model": model, **params}
        url = f"{self.base}/pipeline/submit"
        async with self._session.post(url, json=body) as resp:
            data = await resp.json()
            if resp.status != 200:
                raise RuntimeError(f"KTV API {model} HTTP {resp.status}: {data}")
            if data.get("taskstatus") == "PENDING":
                # 异步任务 → 轮询等待
                taskid = data.get("taskid")
                return await self._poll(taskid)
            return data

    async def _poll(self, taskid: str, interval: int = 5, max_wait: int = 600) -> dict:
        """轮询异步任务直到完成。"""
        await self._ensure_session()
        deadline = time.time() + max_wait
        while time.time() < deadline:
            await asyncio.sleep(interval)
            url = f"{self.base}/tasks?taskid={taskid}"
            async with self._session.get(url) as resp:
                data = await resp.json()
                ts = data.get("taskstatus", data.get("status", ""))
                if ts in ("SUCCEEDED", "FAILED"):
                    return data
        raise TimeoutError(f"Task {taskid} timeout after {max_wait}s")


# 全局单例，惰性初始化
_ktv_client: KtvClient | None = None


def get_ktv_client() -> KtvClient:
    global _ktv_client
    if _ktv_client is None:
        _ktv_client = KtvClient()
    return _ktv_client


# ── helpers ────────────────────────────────────────────────────────

async def _run_local(cmd: str, timeout: int = 300) -> tuple[str, str, int]:
    proc = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        return stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace"), proc.returncode
    except asyncio.TimeoutError:
        proc.kill()
        return "", "timeout", -1


async def _download(url: str, dest: str) -> None:
    """下载文件到本地。"""
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            r.raise_for_status()
            with open(dest, "wb") as f:
                async for chunk in r.content.iter_chunked(8192):
                    f.write(chunk)


# ── Media Preparation ──────────────────────────────────────────────

async def handle_audio_preparing(tenant_id, task_id, step_name, input_data, config):
    work_dir = _task_dir(task_id)
    params = input_data.get("task_params", {})
    audio_url = params.get("audio_url", params.get("audio_path", ""))
    if not audio_url:
        raise ValueError("缺少 audio_url 或 audio_path 参数")

    audio_path = os.path.join(work_dir, "original_audio.mp3")
    if audio_url.startswith("http"):
        await _download(audio_url, audio_path)
    else:
        await _run_local(f"cp '{audio_url}' '{audio_path}'")

    stdout, _, rc = await _run_local(
        f"ffprobe -v error -show_entries format=duration "
        f"-of default=noprint_wrappers=1:nokey=1 '{audio_path}'")
    duration = float(stdout.strip()) if rc == 0 else 0

    return {"audio_path": audio_path, "duration": duration,
            "format": os.path.splitext(audio_path)[1].lstrip(".")}


async def handle_video_preparing(tenant_id, task_id, step_name, input_data, config):
    work_dir = _task_dir(task_id)
    params = input_data.get("task_params", {})
    video_url = params.get("video_url", params.get("video_path", ""))
    if not video_url:
        raise ValueError("缺少 video_url 或 video_path 参数")

    video_path = os.path.join(work_dir, "original_video.mp4")
    audio_path = os.path.join(work_dir, "original_audio.mp3")
    if video_url.startswith("http"):
        await _download(video_url, video_path)
    else:
        await _run_local(f"cp '{video_url}' '{video_path}'")

    await _run_local(f"ffmpeg -y -i '{video_path}' -vn -acodec libmp3lame -q:a 2 '{audio_path}'")
    stdout, _, rc = await _run_local(
        f"ffprobe -v error -show_entries format=duration "
        f"-of default=noprint_wrappers=1:nokey=1 '{video_path}'")
    duration = float(stdout.strip()) if rc == 0 else 0

    return {"video_path": video_path, "audio_path": audio_path, "duration": duration}


# ── Demucs ─────────────────────────────────────────────────────────

async def handle_demucs_separating(tenant_id, task_id, step_name, input_data, config):
    """通过 API 调 demucs-separate。固定使用 separate_full 模式。"""
    work_dir = _task_dir(task_id)
    audio_path = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            audio_path = dep_output.get("audio_path")
            if audio_path: break
    if not audio_path:
        raise ValueError("上游步骤未提供 audio_path")

    client = get_ktv_client()
    result = await client.call(STEP_MODELS["demucs_separating"],
                               {"audio_file": audio_path,
                                "task_type": "separate_full"})

    if result.get("error"):
        raise ValueError(f"Demucs 分离失败: {result['error']}")

    # 下载结果文件
    vocals_local = os.path.join(work_dir, "vocals.wav")
    accompaniment_local = os.path.join(work_dir, "accompaniment.wav")
    if result.get("vocals_url"):
        await _download(result["vocals_url"], vocals_local)
    if result.get("accompaniment_url"):
        await _download(result["accompaniment_url"], accompaniment_local)

    return {"vocals_path": vocals_local,
            "no_vocals_path": accompaniment_local,
            "accompaniment_path": accompaniment_local,
            "mode": "separate_full",
            "usage": result.get("usage", {})}


# ── Lyric Calibration ──────────────────────────────────────────────

async def handle_lyric_calibrating(tenant_id, task_id, step_name, input_data, config):
    """ASR 通过 API + LLM 校准。"""
    work_dir = _task_dir(task_id)
    params = input_data.get("task_params", {})
    lyrics_text = params.get("lyrics", params.get("lyrics_text", ""))

    # 找人声文件
    vocals_path = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            vocals_path = dep_output.get("vocals_path")
            if vocals_path: break
    if not vocals_path:
        raise ValueError("上游步骤未提供 vocals_path")
    if not lyrics_text:
        raise ValueError("缺少 lyrics 参数")

    # 通过 API 做 ASR
    client = get_ktv_client()
    asr_result = await client.call(STEP_MODELS["asr_transcribing"],
                                   {"audio_file": vocals_path})

    asr_timings = asr_result.get("segments", []) if not asr_result.get("error") else []
    if not asr_timings:
        logger.warning(f"ASR 未返回时间戳，使用空列表: {asr_result.get('error','')}")

    # LLM 校准
    calibrated = await _llm_calibrate(lyrics_text, asr_timings)

    calibrated_path = os.path.join(work_dir, "calibrated_lyrics.json")
    with open(calibrated_path, "w", encoding="utf-8") as f:
        json.dump(calibrated, f, ensure_ascii=False, indent=2)

    return {"calibrated_lyrics_path": calibrated_path,
            "calibrated_lyrics": calibrated,
            "segment_count": len(calibrated)}


async def _llm_calibrate(lyrics_text: str, asr_timings: list) -> list:
    prompt = f"""你是一个歌词时间轴校准专家。

原始歌词文本:
{lyrics_text}

ASR识别的时间戳（秒）:
{json.dumps(asr_timings, ensure_ascii=False)}

请将原始歌词的每一句与ASR时间戳对齐，输出JSON数组，每个元素包含:
- text: 歌词文本
- start: 开始时间（秒，浮点数）
- end: 结束时间（秒，浮点数）

要求:
1. 保持原始歌词的文本和顺序
2. 时间戳以ASR结果为基础进行微调
3. 确保时间不重叠，每句之间留适当间隔
4. 只输出JSON，不要其他内容"""
    try:
        from pipeline_service.llm_bridge import llm_call
        result = await llm_call(prompt)
        result = result.strip()
        if result.startswith("```"):
            result = result.split("\n", 1)[1].rsplit("```", 1)[0]
        return json.loads(result)
    except Exception as e:
        logger.warning(f"LLM calibration failed, using ASR timings directly: {e}")
        return asr_timings


# ── Subtitle ───────────────────────────────────────────────────────

async def handle_subtitle_rendering(tenant_id, task_id, step_name, input_data, config):
    """本地生成 ASS 字幕（轻量操作，不走 API）。"""
    work_dir = _task_dir(task_id)
    calibrated = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            calibrated = dep_output.get("calibrated_lyrics")
            if not calibrated and dep_output.get("calibrated_lyrics_path"):
                with open(dep_output["calibrated_lyrics_path"], "r") as f:
                    calibrated = json.load(f)
            if calibrated: break
    if not calibrated:
        raise ValueError("上游步骤未提供 calibrated_lyrics")

    ass_path = os.path.join(work_dir, "karaoke.ass")
    _write_ass_file(ass_path, calibrated)
    return {"ass_path": ass_path}


def _write_ass_file(path: str, segments: list):
    header = """[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: KTV,Source Han Sans SC,72,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,1,0,1,3,1,2,40,40,60,1

[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
    with open(path, "w", encoding="utf-8") as f:
        f.write(header)
        for seg in segments:
            start = _seconds_to_ass_time(seg["start"])
            end   = _seconds_to_ass_time(seg["end"])
            text  = seg["text"].replace("\n", "\\N")
            dur_ms = int((seg["end"] - seg["start"]) * 100)
            f.write(f"Dialogue: 0,{start},{end},KTV,,0,0,0,,{{\\k{dur_ms}}}{text}\n")


def _seconds_to_ass_time(seconds: float) -> str:
    h  = int(seconds // 3600)
    m  = int((seconds % 3600) // 60)
    s  = int(seconds % 60)
    cs = int((seconds % 1) * 100)
    return f"{h}:{m:02d}:{s:02d}.{cs:02d}"


async def handle_subtitle_exporting(tenant_id, task_id, step_name, input_data, config):
    ass_path = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            ass_path = dep_output.get("ass_path")
            if ass_path: break
    if not ass_path or not os.path.exists(ass_path):
        raise ValueError("上游步骤未提供有效的 ass_path")
    return {"subtitle_path": ass_path, "format": "ass"}


# ── Lyric Gen / Eval ───────────────────────────────────────────────

async def handle_lyric_generating(tenant_id, task_id, step_name, input_data, config):
    params = input_data.get("task_params", {})
    topic = params.get("topic", params.get("outline", ""))
    style = params.get("style", "流行")
    language = params.get("language", "zh")
    if not topic:
        raise ValueError("缺少 topic/outline参数")

    prompt = f"""请创作一首{style}风格的{language}歌词。

主题/大纲: {topic}

要求:
1. 包含完整结构: 主歌(Verse)、副歌(Chorus)、桥段(Bridge)
2. 每句歌词节奏感强，适合演唱
3. 押韵自然，情感真实
4. 总长度适合3-5分钟歌曲

直接输出歌词文本，标注段落结构。"""
    try:
        from pipeline_service.llm_bridge import llm_call
        lyrics = await llm_call(prompt)
        return {"lyrics": lyrics.strip(), "topic": topic, "style": style}
    except Exception as e:
        raise ValueError(f"歌词生成失败: {e}")


async def handle_lyric_evaluating(tenant_id, task_id, step_name, input_data, config):
    threshold = config.get("threshold", 8.5)
    lyrics = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            lyrics = dep_output.get("lyrics")
            if lyrics: break
    if not lyrics:
        raise ValueError("上游步骤未提供歌词")

    prompt = f"""请从以下维度评估这首歌词的质量(1-10分):

1. 韵律节奏: 押韵、节奏感、可唱性
2. 情感表达: 情感真实度、共鸣力
3. 文学性: 用词、意象、修辞
4. 结构完整: 段落编排、层次感
5. 商业潜力: 流行度、记忆点

歌词:
{lyrics}

输出JSON: {{"score": 8.5, "dimensions": {{...}}, "suggestions": "..."}}
只输出JSON。"""
    try:
        from pipeline_service.llm_bridge import llm_call
        result = await llm_call(prompt)
        result = result.strip()
        if result.startswith("```"):
            result = result.split("\n", 1)[1].rsplit("```", 1)[0]
        evaluation = json.loads(result)
        score = evaluation.get("score", 0)
    except Exception:
        score = 7.0
        evaluation = {"score": score, "note": "evaluation_parse_failed"}

    if score < threshold:
        raise ValueError(f"歌词评分 {score} 低于阈值 {threshold}，需要重新生成")

    return {"lyrics": lyrics, "evaluation": evaluation, "score": score, "passed": True}


# ── Music Generation ───────────────────────────────────────────────

async def handle_music_generating(tenant_id, task_id, step_name, input_data, config):
    """提交音乐生成到外部 API（Suno/MiniMax）。"""
    lyrics = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            lyrics = dep_output.get("lyrics")
            if lyrics: break
    if not lyrics:
        raise ValueError("上游步骤未提供歌词")

    params = input_data.get("task_params", {})
    music_service = params.get("music_service", "suno")
    style = params.get("music_style", "pop")

    job_id = f"music_{task_id}_{int(time.time())}"
    return {"music_job_id": job_id, "music_service": music_service,
            "style": style, "lyrics": lyrics, "status": "submitted"}


async def handle_music_polling(tenant_id, task_id, step_name, input_data, config):
    job_info = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            job_info = dep_output
            if job_info and job_info.get("music_job_id"): break
    if not job_info:
        raise ValueError("上游步骤未提供 music_job_id")

    work_dir = _task_dir(task_id)
    music_path = os.path.join(work_dir, "generated_music.mp3")
    return {"music_path": music_path, "music_job_id": job_info.get("music_job_id"),
            "status": "completed"}


# ── Character & Video ──────────────────────────────────────────────

async def handle_character_designing(tenant_id, task_id, step_name, input_data, config):
    lyrics = None
    params = input_data.get("task_params", {})
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            lyrics = dep_output.get("lyrics") or dep_output.get("calibrated_lyrics")
            if isinstance(lyrics, list):
                lyrics = " ".join(s.get("text", "") for s in lyrics)
            if lyrics: break
    style = params.get("visual_style", "anime")

    prompt = f"""根据以下歌词，设计MV角色方案。

歌词:
{lyrics}

视觉风格: {style}

请设计1-3个角色，每个角色包含:
1. 角色名称
2. 外貌描述（用于AI图像生成的详细prompt）
3. 性格特征
4. 在MV中的角色定位

输出JSON数组。"""
    try:
        from pipeline_service.llm_bridge import llm_call
        result = await llm_call(prompt)
        result = result.strip()
        if result.startswith("```"):
            result = result.split("\n", 1)[1].rsplit("```", 1)[0]
        characters = json.loads(result)
    except Exception as e:
        raise ValueError(f"角色设计失败: {e}")

    return {"characters": characters, "visual_style": style}


async def handle_character_image_generating(tenant_id, task_id, step_name, input_data, config):
    """通过 /v1/image/generations 生成角色图。"""
    characters = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            characters = dep_output.get("characters")
            if characters: break
    if not characters:
        raise ValueError("上游步骤未提供角色设计")

    client = get_ktv_client()
    char_images = []
    work_dir = _task_dir(task_id)

    for i, char in enumerate(characters):
        prompt = char.get("prompt", char.get("description", ""))
        if not prompt:
            continue

        # 调 llmage image/generations
        await client._ensure_session()
        url = f"{client.base}/image/generations"
        body = {"model": "cogview-3-flash", "catelogid": "t2i",
                "prompt": prompt, "size": "512x512", "n": 1}
        async with client._session.post(url, json=body) as resp:
            data = await resp.json()

        local_path = os.path.join(work_dir, f"char_{i}.png")
        img_url = (data.get("image", [None])[0] if isinstance(data.get("image"), list)
                   else data.get("output_url"))
        if img_url:
            await _download(img_url, local_path)

        char_images.append({"name": char.get("name", f"char_{i}"),
                            "image_path": local_path, "prompt": prompt})

    return {"character_images": char_images}


async def handle_storyboard_generating(tenant_id, task_id, step_name, input_data, config):
    lyrics = None
    char_images = None
    params = input_data.get("task_params", {})
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            if dep_output.get("calibrated_lyrics"):
                lyrics = dep_output["calibrated_lyrics"]
            elif dep_output.get("lyrics"):
                lyrics = dep_output["lyrics"]
            if dep_output.get("character_images"):
                char_images = dep_output["character_images"]
    if not lyrics:
        raise ValueError("上游步骤未提供歌词")

    duration = params.get("duration", 240)
    prompt = f"""根据歌词和角色，生成MV分镜脚本。

歌词:
{json.dumps(lyrics, ensure_ascii=False) if isinstance(lyrics, list) else lyrics}

角色:
{json.dumps(char_images, ensure_ascii=False) if char_images else "无特定角色"}

视频总时长: {duration}秒

请输出JSON数组，每个分镜包含:
- scene_id: 分镜编号
- start_time: 开始秒数
- end_time: 结束秒数
- description: 场景描述（英文，用于视频生成prompt）
- characters: 出现的角色
- camera: 镜头运动描述
- mood: 情绪/色调

确保分镜覆盖整首歌，每个分镜5-15秒。"""
    try:
        from pipeline_service.llm_bridge import llm_call
        result = await llm_call(prompt)
        result = result.strip()
        if result.startswith("```"):
            result = result.split("\n", 1)[1].rsplit("```", 1)[0]
        storyboard = json.loads(result)
    except Exception as e:
        raise ValueError(f"分镜生成失败: {e}")

    work_dir = _task_dir(task_id)
    sb_path = os.path.join(work_dir, "storyboard.json")
    with open(sb_path, "w", encoding="utf-8") as f:
        json.dump(storyboard, f, ensure_ascii=False, indent=2)
    return {"storyboard": storyboard, "storyboard_path": sb_path, "scene_count": len(storyboard)}


async def handle_scene_video_generating(tenant_id, task_id, step_name, input_data, config):
    """通过 /v1/video/generations 生成场景视频，模型由客户选择。"""
    work_dir = _task_dir(task_id)
    storyboard = None
    selected_model = "wan2.7-t2v"  # 默认
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            if dep_output.get("storyboard"): storyboard = dep_output["storyboard"]
            if dep_output.get("selected_model"): selected_model = dep_output["selected_model"]
    if not storyboard:
        raise ValueError("上游步骤未提供分镜脚本")

    logger.info(f"Using model: {selected_model}")
    client = get_ktv_client()
    await client._ensure_session()
    scene_videos = []

    for i, scene in enumerate(storyboard):
        desc = scene.get("description", "")
        duration = int(scene.get("end_time", 10) - scene.get("start_time", 5))

        body = {"model": selected_model, "catelogid": "t2v",
                "prompt": desc, "duration": str(duration)}
        url = f"{client.base}/video/generations"
        async with client._session.post(url, json=body) as resp:
            data = await resp.json()

        taskid = data.get("taskid", "")
        if taskid:
            data = await client._poll(taskid)

        local_scene = os.path.join(work_dir, f"scene_{i:03d}.mp4")
        video_url = data.get("output_url") or (data.get("video", [None]) or [None])[0]
        if video_url:
            await _download(video_url, local_scene)

        scene_videos.append({"scene_id": i, "video_path": local_scene,
                             "description": desc, "duration": duration})

    return {"scene_videos": scene_videos, "scene_count": len(scene_videos),
            "selected_model": selected_model}


async def handle_scene_video_evaluating(tenant_id, task_id, step_name, input_data, config):
    scene_videos = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            scene_videos = dep_output.get("scene_videos")
            if scene_videos: break
    if not scene_videos:
        raise ValueError("上游步骤未提供场景视频")

    valid_scenes = []
    for sv in scene_videos:
        path = sv.get("video_path", "")
        if os.path.exists(path) and os.path.getsize(path) > 10000:
            sv["quality_score"] = 8.0
            valid_scenes.append(sv)
        else:
            sv["quality_score"] = 0
            logger.warning(f"Scene {sv.get('scene_id')} missing or too small: {path}")

    if not valid_scenes:
        raise ValueError("所有场景视频质量不合格")

    avg_score = sum(s.get("quality_score", 0) for s in valid_scenes) / len(valid_scenes)
    return {"scene_videos": valid_scenes, "avg_quality": avg_score}


async def handle_scene_video_concatenating(tenant_id, task_id, step_name, input_data, config):
    work_dir = _task_dir(task_id)
    scene_videos = None
    audio_duration = None
    for dep_output in input_data.values():
        if isinstance(dep_output, dict):
            if dep_output.get("scene_videos"): scene_videos = dep_output["scene_videos"]
            if dep_output.get("duration"): audio_duration = dep_output["duration"]
    if not scene_videos:
        raise ValueError("上游步骤未提供场景视频")

    concat_list = os.path.join(work_dir, "concat_list.txt")
    with open(concat_list, "w") as f:
        for sv in scene_videos:
            path = sv.get("video_path", "")
            if os.path.exists(path):
                f.write(f"file '{path}'\n")

    concat_path = os.path.join(work_dir, "concat_video.mp4")
    await _run_local(f"ffmpeg -y -f concat -safe 0 -i '{concat_list}' -c copy '{concat_path}'")

    final_path = os.path.join(work_dir, "final_video.mp4")
    if audio_duration and audio_duration > 0:
        stdout, _, _ = await _run_local(
            f"ffprobe -v error -show_entries format=duration "
            f"-of default=noprint_wrappers=1:nokey=1 '{concat_path}'")
        concat_dur = float(stdout.strip()) if stdout.strip() else 0
        if 0 < concat_dur < audio_duration:
            loops = int(audio_duration / concat_dur) + 1
            await _run_local(f"ffmpeg -y -stream_loop {loops} -i '{concat_path}' "
                             f"-t {audio_duration} -c:v libx264 -preset fast '{final_path}'")
        else:
            await _run_local(f"cp '{concat_path}' '{final_path}'")
    else:
        await _run_local(f"cp '{concat_path}' '{final_path}'")

    return {"final_video_path": final_path}


# ── Final Synthesis ────────────────────────────────────────────────

async def handle_ktv_synthesizing(tenant_id, task_id, step_name, input_data, config):
    work_dir = _task_dir(task_id)
    video_path = ass_path = vocals_path = no_vocals_path = None
    for dep_output in input_data.values():
        if not isinstance(dep_output, dict): continue
        for k in ("final_video_path", "video_path"):
            if dep_output.get(k) and not video_path:
                video_path = dep_output[k]
        if dep_output.get("ass_path"): ass_path = dep_output["ass_path"]
        if dep_output.get("vocals_path"): vocals_path = dep_output["vocals_path"]
        if dep_output.get("no_vocals_path"): no_vocals_path = dep_output["no_vocals_path"]

    if not ass_path: raise ValueError("缺少字幕文件")
    if not video_path: raise ValueError("缺少视频源")

    ktv_path = os.path.join(work_dir, "ktv_final.mp4")
    mtv_path = os.path.join(work_dir, "mtv_final.mp4")

    if vocals_path and no_vocals_path:
        ktv_cmd = (f"ffmpeg -y -i '{video_path}' -i '{vocals_path}' -i '{no_vocals_path}' "
                   f"-filter_complex \"[0:v]ass='{ass_path}'[v]\" "
                   f"-map '[v]' -map 1:a -map 2:a "
                   f"-c:v libx264 -preset fast -c:a aac -b:a 192k "
                   f"-metadata:s:a:0 title='Vocals' -metadata:s:a:1 title='Accompaniment' "
                   f"'{ktv_path}'")
    else:
        ktv_cmd = (f"ffmpeg -y -i '{video_path}' -vf \"ass='{ass_path}'\" "
                   f"-c:v libx264 -preset fast -c:a aac -b:a 192k '{ktv_path}'")

    stdout, stderr, rc = await _run_local(ktv_cmd, timeout=600)
    if rc != 0: raise ValueError(f"KTV合成失败: {stderr}")

    mtv_cmd = (f"ffmpeg -y -i '{video_path}' -vf \"ass='{ass_path}'\" "
               f"-c:v libx264 -preset fast -c:a aac -b:a 192k '{mtv_path}'")
    stdout, stderr, rc = await _run_local(mtv_cmd, timeout=600)
    if rc != 0: logger.warning(f"MTV合成失败，仅输出KTV版本: {stderr}")

    result = {"ktv_path": ktv_path, "subtitle_path": ass_path}
    if os.path.exists(mtv_path): result["mtv_path"] = mtv_path
    return result


# ── Quality Gate ───────────────────────────────────────────────────

def _make_quality_handler(original_handler, eval_func):
    async def wrapper(tenant_id, task_id, step_name, input_data, config):
        from app.quality_gate import with_quality_gate
        version = 1
        if isinstance(config, dict):
            version = config.get("version", 1)
            if not version and isinstance(input_data, dict):
                tp = input_data.get("task_params", {})
                if isinstance(tp, dict): version = tp.get("version", 1)
        return await with_quality_gate(
            task_id=task_id, step_name=step_name, version=version,
            handler=original_handler, evaluator=eval_func,
            tenant_id=tenant_id, input_data=input_data, config=config)
    wrapper.__name__ = f"quality_{original_handler.__name__}"
    wrapper.__qualname__ = wrapper.__name__
    return wrapper


quality_music_polling               = _make_quality_handler(handle_music_polling, eval_music_quality)
quality_character_designing         = _make_quality_handler(handle_character_designing, eval_character_design)
quality_character_image_generating  = _make_quality_handler(handle_character_image_generating, eval_character_image)
quality_storyboard_generating       = _make_quality_handler(handle_storyboard_generating, eval_storyboard)
quality_scene_video_evaluating      = _make_quality_handler(handle_scene_video_evaluating, eval_scene_video_quality)
quality_ktv_synthesizing            = _make_quality_handler(handle_ktv_synthesizing, eval_ktv_synthesis)


# ── Registration ───────────────────────────────────────────────────

KTV_HANDLERS = {
    "audio_preparing":              handle_audio_preparing,
    "video_preparing":              handle_video_preparing,
    "demucs_separating":            handle_demucs_separating,
    "lyric_calibrating":            handle_lyric_calibrating,
    "subtitle_rendering":           handle_subtitle_rendering,
    "subtitle_exporting":           handle_subtitle_exporting,
    "lyric_generating":             handle_lyric_generating,
    "lyric_evaluating":             handle_lyric_evaluating,
    "music_generating":             handle_music_generating,
    "music_polling":                quality_music_polling,
    "character_designing":          quality_character_designing,
    "character_image_generating":   quality_character_image_generating,
    "storyboard_generating":        quality_storyboard_generating,
    "model_selecting":              handle_model_selecting,
    "scene_video_generating":       handle_scene_video_generating,
    "scene_video_evaluating":       quality_scene_video_evaluating,
    "scene_video_concatenating":    handle_scene_video_concatenating,
    "ktv_synthesizing":             quality_ktv_synthesizing,
}


def load_ktv_adapter():
    try:
        from pipeline_service.handler import register_handler
        for step_type, fn in KTV_HANDLERS.items():
            register_handler(step_type, fn)
        logger.info(f"Registered {len(KTV_HANDLERS)} KTV handlers (via llmage API, "
                    f"base={API_BASE})")
    except ImportError:
        logger.warning("pipeline_service not available, KTV handlers not registered")