AI 造岗：快手可灵背后，15 个新职业是怎么冒出来的

"""
模拟可灵 AI 视频生成工作流
实际使用时需替换为可灵 API 的真实 endpoint 和认证信息
运行前：pip install requests pyyaml
"""

import requests
import yaml
import json
import time
import os
from pathlib import Path

# ---------- 配置 ----------
# 实际使用时替换为可灵开放平台的真实地址
API_BASE = "https://api.klingai.com/v1"  # 示例地址，需按官方文档替换
API_KEY = os.getenv("KLING_API_KEY", "your-api-key-here")

# prompt 模板：短视频带货场景
PROMPT_TEMPLATES = {
    "food_review": (
        "一位美食博主坐在暖色调餐厅中，面前摆着一碗热气腾腾的牛肉面，"
        "她拿起筷子夹起一片牛肉，表情惊喜，镜头从侧面缓慢推近，"
        "背景音乐轻快，画面风格为竖屏短视频，时长5秒"
    ),
    "product_demo": (
        "一只手在白色桌面上拆开一个科技产品包装盒，"
        "产品外壳金属质感，镜头从俯视角度缓慢旋转，"
        "光影干净，产品细节清晰，竖屏构图，时长5秒"
    ),
    "travel_vlog": (
        "一个背包客站在山顶俯瞰云海，风吹动头发，"
        "镜头从背后缓慢拉远展现全景，色调偏冷蓝，"
        "竖屏短视频风格，时长5秒"
    ),
}

# ---------- 核心函数 ----------
def generate_video(prompt: str, style: str = "vertical_short", duration: int = 5) -> dict:
    """调用视频生成 API（模拟）"""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "prompt": prompt,
        "style": style,
        "duration": duration,
        "resolution": "1080x1920",  # 竖屏
    }
    # 实际调用：resp = requests.post(f"{API_BASE}/video/generate", headers=headers, json=payload)
    # 此处模拟返回
    task_id = f"task_{hash(prompt) % 100000}"
    print(f"[提交] prompt={prompt[:30]}... → task_id={task_id}")
    return {"task_id": task_id, "status": "pending"}


def poll_task(task_id: str, max_wait: int = 120) -> dict:
    """轮询任务状态直到完成（模拟）"""
    print(f"[轮询] task_id={task_id}, 最多等待 {max_wait}s")
    # 实际调用：resp = requests.get(f"{API_BASE}/video/status/{task_id}", headers=...)
    # 模拟：直接返回成功
    time.sleep(2)  # 模拟等待
    return {
        "task_id": task_id,
        "status": "completed",
        "video_url": f"https://cdn.klingai.com/output/{task_id}.mp4",
        "metadata": {"duration": 5, "resolution": "1080x1920"},
    }


def batch_generate(template_key: str, count: int = 3) -> list:
    """批量生成同模板的变体视频"""
    base_prompt = PROMPT_TEMPLATES[template_key]
    results = []
    for i in range(count):
        # 微调 prompt 产生变体
        variant = f"{base_prompt}，变体{i+1}：光线略有变化"
        task = generate_video(variant)
        result = poll_task(task["task_id"])
        results.append(result)
    return results


def save_manifest(results: list, output_dir: str = "./kling_output") -> str:
    """将生成结果保存为 YAML 清单，方便后续审核和分发"""
    Path(output_dir).mkdir(exist_ok=True)
    manifest = {
        "generated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
        "total_videos": len(results),
        "videos": [],
    }
    for r in results:
        manifest["videos"].append({
            "task_id": r["task_id"],
            "url": r["video_url"],
            "status": r["status"],
            "duration": r["metadata"]["duration"],
        })
    manifest_path = os.path.join(output_dir, "manifest.yaml")
    with open(manifest_path, "w") as f:
        yaml.dump(manifest, f, allow_unicode=True, default_flow_style=False)
    print(f"[归档] 清单已保存至 {manifest_path}")
    return manifest_path


# ---------- 执行 ----------
if __name__ == "__main__":
    # 选一个模板，批量生成 3 个变体
    results = batch_generate("food_review", count=3)
    # 保存归档清单
    manifest_path = save_manifest(results)
    # 打印摘要
    print(f"\n生成完成！共 {len(results)} 个视频，清单文件：{manifest_path}")
    print("下一步：人工审核清单中的视频，筛掉合规风险内容，再分发到快手账号。")