pipeline-app/app/cost_estimator.py

"""
Video generation cost estimator - multi-model pricing and cost prediction.

Estimates cost for generating scene videos using different models:
- Wan2.2 (local GPU, fast, lower cost)
- Wan2.7 (local GPU, higher quality, medium cost)
- Vidu 2.0 (cloud API, highest quality, premium cost)

Cost factors:
- Scene count from storyboard
- Duration per scene (frames)
- Model pricing (per frame or per second)
- GPU compute cost (for local models)
"""

import logging
from typing import List, Dict, Any

logger = logging.getLogger("pipeline.cost_estimator")


# Model pricing configuration — llmage v1 video models
MODEL_PRICING = {
    "wan2.7-t2v": {
        "name": "通义万象 WAN 2.7 (T2V) ⭐推荐",
        "price_per_second": 0.05,
        "quality_score": 8.5,
        "generation_speed": "medium",
        "description": "最新版文生视频，质量更高",
        "features": ["T2V", "I2V", "Ref2V"],
        "recommended_for": ["quality", "professional"],
    },
    "happyhorse-1.0-t2v": {
        "name": "快乐马 HappyHorse 1.0 (T2V)",
        "price_per_second": 0.03,
        "quality_score": 7.5,
        "generation_speed": "fast",
        "description": "快速文生视频，经济实惠",
        "features": ["T2V", "I2V"],
        "recommended_for": ["standard", "batch"],
    },
    "doubao-seedance-2-0-260128": {
        "name": "豆包 Seedance 2.0 (T2V)",
        "price_per_second": 0.08,
        "quality_score": 9.0,
        "generation_speed": "slow",
        "description": "字节跳动旗舰视频生成，支持音效同步",
        "features": ["T2V", "I2V", "Ref2V", "音效"],
        "recommended_for": ["premium", "cinematic"],
    },
}


def estimate_scene_cost(storyboard: List[Dict[str, Any]], fps: int = 24) -> int:
    """
    Estimate total frames needed for all scenes.
    
    Args:
        storyboard: List of scene dicts with start_time/end_time
        fps: Frames per second (default 24)
    
    Returns:
        Total frame count
    """
    total_frames = 0
    for scene in storyboard:
        duration = scene.get("end_time", 10) - scene.get("start_time", 0)
        frames = int(duration * fps)
        total_frames += frames
    return total_frames


def estimate_model_cost(model_id: str, total_frames: int, fps: int = 24) -> Dict[str, Any]:
    """
    Estimate cost for a specific model.
    
    Args:
        model_id: Model identifier (e.g., wan2.7-t2v, doubao-seedance-2-0-260128)
        total_frames: Total frames to generate
    
    Returns:
        Cost estimate dict with breakdown
    """
    if model_id not in MODEL_PRICING:
        raise ValueError(f"Unknown model: {model_id}")
    
    model = MODEL_PRICING[model_id]
    
    return {
        "model_id": model_id,
        "model_name": model["name"],
        "total_frames": total_frames,
        "fps": fps,
        "price_per_second": model["price_per_second"],
        "total_cost": round(total_frames * model["price_per_second"] / fps, 4),
        "quality_score": model["quality_score"],
        "generation_speed": model["generation_speed"],
        "description": model["description"],
        "features": model["features"],
        "recommended_for": model["recommended_for"],
    }


def estimate_all_models(storyboard: List[Dict[str, Any]], fps: int = 24) -> List[Dict[str, Any]]:
    """
    Estimate cost for all available models.
    
    Args:
        storyboard: List of scene dicts
        fps: Frames per second
    
    Returns:
        List of cost estimates sorted by total_cost (ascending)
    """
    total_frames = estimate_scene_cost(storyboard, fps)
    scene_count = len(storyboard)
    
    estimates = []
    for model_id in MODEL_PRICING.keys():
        try:
            estimate = estimate_model_cost(model_id, total_frames, fps)
            estimate["scene_count"] = scene_count
            estimate["fps"] = fps
            estimates.append(estimate)
        except Exception as e:
            logger.error(f"Failed to estimate cost for {model_id}: {e}")
    
    # Sort by total_cost (cheapest first)
    estimates.sort(key=lambda x: x["total_cost"])
    
    # Add recommendation flag
    if estimates:
        # Recommend middle option (best value)
        mid_idx = len(estimates) // 2
        estimates[mid_idx]["recommended"] = True
        for i, est in enumerate(estimates):
            if i != mid_idx:
                est["recommended"] = False
    
    return estimates


def format_cost_summary(estimates: List[Dict[str, Any]]) -> str:
    """
    Format cost estimates as human-readable summary.
    
    Args:
        estimates: List of cost estimate dicts
    
    Returns:
        Formatted string summary
    """
    lines = []
    lines.append("=" * 70)
    lines.append("视频生成模型费用预估")
    lines.append("=" * 70)
    
    if not estimates:
        lines.append("无可用模型")
        return "\n".join(lines)
    
    scene_count = estimates[0].get("scene_count", 0)
    total_frames = estimates[0].get("total_frames", 0)
    fps = estimates[0].get("fps", 24)
    
    lines.append(f"分镜数量: {scene_count}")
    lines.append(f"总帧数: {total_frames} (at {fps}fps)")
    lines.append("")
    
    for est in estimates:
        rec = " ⭐ 推荐" if est.get("recommended") else ""
        lines.append(f"【{est['model_name']}】{rec}")
        lines.append(f"  质量评分: {est['quality_score']}/10")
        lines.append(f"  生成速度: {est['generation_speed']}")
        lines.append(f"  功能: {', '.join(est['features'])}")
        lines.append(f"  描述: {est['description']}")
        lines.append(f"  费用: 每秒 ${est['price_per_second']:.4f} × {est['total_frames']}帧@{est['fps']}fps = ${est['total_cost']:.4f}")
        lines.append("")
    
    lines.append("=" * 70)
    return "\n".join(lines)