基于腾讯云高性能应用服务（HAI）与混元大模型的AI漫剧自动化生产流水线

# 通过 HAI 控制台创建实例后，获取公网 IP 和 API Key
# 默认 WebUI 监听 :7860，启用 --api 参数
# 测试 API 是否正常
curl http://<HAI_IP>:7860/sdapi/v1/sd-models

pip install cos-python-sdk-v5 tencentcloud-sdk-python

import json
import tencentcloud.common as common
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models

class ScriptGenerator:
    def __init__(self, secret_id, secret_key, region="ap-guangzhou"):
        cred = common.Credential(secret_id, secret_key)
        self.client = hunyuan_client.HunyuanClient(cred, region)
    
    def generate_script(self, topic: str, num_shots: int = 6) -> list:
        """
        返回分镜列表，每个元素包含：shot_id, scene_description, character_action, dialogue
        """
        prompt = f"""
        你是一位漫剧编剧。根据主题「{topic}」，生成一个包含{num_shots}个分镜的短剧脚本。
        每个分镜需包含：场景描述、角色动作、对白（若有）。
        输出格式必须是合法的 JSON 数组，每个元素有字段：shot_id(1-based), scene, action, dialogue(若无则空字符串)。
        示例：[{{"shot_id":1,"scene":"雨夜城市街道","action":"主角撑伞奔跑","dialogue":"我必须找到她"}}, ...]
        只输出 JSON，不要任何额外文字。
        """
        req = models.ChatCompletionsRequest()
        req.Model = "hunyuan-standard"
        req.Messages = [{"Role": "user", "Content": prompt}]
        req.Temperature = 0.8
        resp = self.client.ChatCompletions(req)
        content = resp.Choices[0].Message.Content
        # 提取 JSON（去除可能的 markdown 代码块）
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0]
        elif "```" in content:
            content = content.split("```")[1].split("```")[0]
        shots = json.loads(content)
        return shots

def build_sd_payload(shot: dict, character_prompt: str = "1girl, long black hair, red jacket, determined expression") -> dict:
    """
    根据分镜信息生成 Stable Diffusion WebUI API 所需的 payload
    """
    # 正向提示词：角色锚定 + 场景 + 动作 + 品质词
    pos = f"{character_prompt}, {shot['scene']}, {shot['action']}, masterpiece, best quality, anime style, intricate details, 8k"
    # 反向提示词：常见负面
    neg = "worst quality, low quality, blurry, bad anatomy, extra limbs, deformed, watermark, signature"
    
    payload = {
        "prompt": pos,
        "negative_prompt": neg,
        "steps": 25,
        "sampler_index": "Euler a",
        "cfg_scale": 7,
        "width": 768,
        "height": 432,   # 16:9 宽屏
        "batch_size": 1,
        "n_iter": 1,
        "seed": -1,      # 随机种子，但可固定保持一致性
        "override_settings": {
            "sd_model_checkpoint": "majicmixRealistic_v6.safetensors"  # 推荐动漫写实混合模型
        }
    }
    return payload

def call_sd_api(endpoint: str, payload: dict) -> bytes:
    import requests, base64
    url = f"{endpoint}/sdapi/v1/txt2img"
    response = requests.post(url, json=payload, timeout=120)
    if response.status_code == 200:
        data = response.json()
        img_b64 = data['images'][0]
        return base64.b64decode(img_b64)
    else:
        raise Exception(f"SD API error: {response.text}")

import asyncio
import aiohttp
from qcloud_cos import CosConfig, CosS3Client
import hashlib

class MangaProducer:
    def __init__(self, cos_secret_id, cos_secret_key, cos_region, bucket_name, sd_endpoint):
        self.sd_endpoint = sd_endpoint
        self.bucket = bucket_name
        config = CosConfig(Region=cos_region, SecretId=cos_secret_id, SecretKey=cos_secret_key)
        self.cos_client = CosS3Client(config)
    
    async def generate_one_shot(self, session, shot: dict, shot_index: int, task_id: str):
        payload = build_sd_payload(shot)
        # 固定种子保证角色一致（用 shot_id 作为种子基准）
        payload['seed'] = 1000 + shot_index  
        url = f"{self.sd_endpoint}/sdapi/v1/txt2img"
        async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=120)) as resp:
            if resp.status != 200:
                raise Exception(await resp.text())
            data = await resp.json()
            img_bytes = base64.b64decode(data['images'][0])
            # 上传到 COS，路径：{task_id}/shot_{shot_index}.png
            key = f"{task_id}/shot_{shot_index:03d}.png"
            self.cos_client.put_object(
                Bucket=self.bucket,
                Body=img_bytes,
                Key=key,
                ContentType="image/png"
            )
            cos_url = f"https://{self.bucket}.cos.{self.cos_client._conf.region}.myqcloud.com/{key}"
            return cos_url
    
    async def produce_all(self, shots: list, task_id: str) -> list:
        async with aiohttp.ClientSession() as session:
            tasks = [self.generate_one_shot(session, shot, i, task_id) for i, shot in enumerate(shots)]
            cos_urls = await asyncio.gather(*tasks, return_exceptions=True)
            # 简单处理异常：重试一次（可扩展）
            final_urls = []
            for idx, result in enumerate(cos_urls):
                if isinstance(result, Exception):
                    print(f"Shot {idx} failed: {result}, retrying...")
                    # 重试逻辑（略）
                else:
                    final_urls.append(result)
            return final_urls

import subprocess
import os
from tencentcloud.tts.v20190823 import tts_client, models

class VideoComposer:
    def __init__(self, tts_secret_id, tts_secret_key):
        self.tts_client = tts_client.TtsClient(common.Credential(tts_secret_id, tts_secret_key), "ap-guangzhou")
    
    def text_to_speech(self, text: str, output_path: str):
        req = models.TextToVoiceRequest()
        req.Text = text
        req.SessionId = "manga_" + hashlib.md5(text.encode()).hexdigest()[:8]
        req.VoiceType = 1001  # 智侠（男声），可换
        resp = self.tts_client.TextToVoice(req)
        with open(output_path, "wb") as f:
            f.write(base64.b64decode(resp.Audio))
        return output_path
    
    def compose_video(self, image_urls: list, dialogues: list, output_mp4: str):
        """
        从 COS 下载图片，合成带音频的视频（每图停留3秒，音频对齐）
        实际实现需先下载图片，此处提供 FFmpeg 命令参考：
        ffmpeg -framerate 1/3 -i shot_%03d.png -c:v libx264 -r 30 -pix_fmt yuv420p out.mp4
        合并音频：ffmpeg -i out.mp4 -i audio.wav -c:v copy -c:a aac final.mp4
        """
        # 完整代码见附录
        pass

# state_machine.yaml (简化)
States:
  - GenerateScript:
      Type: Task
      Resource: "arn:tencent:scf:ap-guangzhou:function:gen-script"
      Next: BatchGenerateImages
  - BatchGenerateImages:
      Type: Task
      Resource: "arn:tencent:scf:ap-guangzhou:function:batch-sd"
      Next: ComposeVideo
  - ComposeVideo:
      Type: Task
      Resource: "arn:tencent:scf:ap-guangzhou:function:composer"
      End: true

import os
from dotenv import load_dotenv
load_dotenv()

# 配置
TENCENT_SECRET_ID = os.getenv("TENCENT_SECRET_ID")
TENCENT_SECRET_KEY = os.getenv("TENCENT_SECRET_KEY")
COS_REGION = "ap-guangzhou"
COS_BUCKET = "ai-manga-1234567890"
SD_ENDPOINT = "http://<your-hai-ip>:7860"

# 1. 生成剧本
script_gen = ScriptGenerator(TENCENT_SECRET_ID, TENCENT_SECRET_KEY)
shots = script_gen.generate_script("赛博朋克外卖员与猫咪", num_shots=5)
print("分镜脚本：", json.dumps(shots, indent=2, ensure_ascii=False))

# 2. 批量生产
producer = MangaProducer(TENCENT_SECRET_ID, TENCENT_SECRET_KEY, COS_REGION, COS_BUCKET, SD_ENDPOINT)
task_id = f"task_{int(time.time())}"
urls = asyncio.run(producer.produce_all(shots, task_id))
print("图片URLs：", urls)

# 3. 合成视频（示例）
composer = VideoComposer(TENCENT_SECRET_ID, TENCENT_SECRET_KEY)
# 假设下载图片到本地，此处省略
# composer.compose_video(urls, [s['dialogue'] for s in shots], f"{task_id}.mp4")

python launch.py --api --xformers --opt-split-attention --listen