重构 Pixelle-Video：公有 AI 视频生成方案

import requests

def generate_script(topic: str) -> dict:
    response = requests.post(
        "https://api.minimaxi.com/v1/text/chatcompletion_v2",
        headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
        json={
            "model": "MiniMax-Text-01",
            "messages": [{"role": "user", "content": prompt_template.format(user_topic=topic)}]
        }
    )
    return response.json()["choices"][0]["message"]["content"]

import requests

def generate_image(prompt: str, output_path: str) -> str:
    response = requests.post(
        "https://ark.cn-beijing.volces.com/api/v3/images/generations",
        headers={"Authorization": f"Bearer {DOUBAN_API_KEY}"},
        json={
            "model": "doubao-vision-01",
            "prompt": prompt,
            "size": "1080x1920",
            "response_format": "url"
        }
    )
    image_url = response.json()["data"][0]["url"]
    img_data = requests.get(image_url).content
    with open(output_path, "wb") as f:
        f.write(img_data)
    return output_path

import requests, time

def generate_video(image_path: str) -> str:
    # 1. 上传图片获取 asset_id
    with open(image_path, "rb") as f:
        upload_resp = requests.post(
            "https://api.minimaxi.com/v1/files/upload",
            headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
            files={"file": f}
        )
    asset_id = upload_resp.json()["file"]["file_id"]

    # 2. 发起视频生成任务
    task_resp = requests.post(
        "https://api.minimaxi.com/v1/video_generation",
        headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
        json={"model": "minimax-video-01", "asset_id": asset_id, "duration": 5}
    )
    task_id = task_resp.json()["task_id"]

    # 3. 轮询直到完成
    while True:
        status = requests.get(
            f"https://api.minimaxi.com/v1/video_generation/{task_id}",
            headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"}
        ).json()
        if status["status"] == "completed":
            return status["video"]["url"]
        elif status["status"] == "failed":
            raise Exception("视频生成失败")
        time.sleep(3)

import requests, time

def generate_music(mood: str, duration: int = 30) -> str:
    response = requests.post(
        "https://api.minimaxi.com/v1/music_generation",
        headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
        json={
            "model": "minimax-music-01",
            "prompt": f"生成一段{mood}风格的背景音乐",
            "duration": duration,
            "format": "mp3"
        }
    )
    task_id = response.json()["task_id"]
    # 轮询直到完成
    while True:
        status = requests.get(
            f"https://api.minimaxi.com/v1/music_generation/{task_id}",
            headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"}
        ).json()
        if status["status"] == "completed":
            music_url = status["music"]["audio_url"]
            mp3_data = requests.get(music_url).content
            output_path = f"/tmp/bgm_{mood}_{int(time.time())}.mp3"
            with open(output_path, "wb") as f:
                f.write(mp3_data)
            return output_path
        elif status["status"] == "failed":
            raise Exception("音乐生成失败")
        time.sleep(3)

import asyncio, edge_tts

async def generate_voice(narration: str, output_path: str, voice: str = "zh-CN-XiaoxiaoNeural") -> str:
    communicate = edge_tts.Communicate(narration, voice)
    await communicate.save(output_path)
    return output_path

# 常用中文音色
VOICES = {
    "女-晓晓": "zh-CN-XiaoxiaoNeural",
    "男-云扬": "zh-CN-YunyanNeural",
    "女-云希": "zh-CN-YunxiNeural",
}

import oss2, hashlib
from pathlib import Path

def upload_to_oss(local_path: str, tenant_id: str, category: str) -> str:
    """上传本地素材到OSS，返回CDN URL"""
    oss_config = get_oss_config(tenant_id)
    bucket = oss2.Bucket(oss2.Auth(oss_config['key'], oss_config['secret']),
                         oss_config['endpoint'], oss_config['bucket'])

    # 计算文件MD5去重
    with open(local_path, 'rb') as f:
        file_hash = hashlib.md5(f.read()).hexdigest()

    oss_key = f"brand_assets/{tenant_id}/{category}/{file_hash}_{Path(local_path).name}"

    # 已存在则跳过
    if bucket.object_exists(oss_key):
        return f"https://{oss_config['cdn_domain']}/{oss_key}"

    bucket.put_object_from_file(oss_key, local_path)
    return f"https://{oss_config['cdn_domain']}/{oss_key}"

def resolve_product_image(product_id: str, scene_prompt: str) -> str:
    """将 prompt 中的 {product_id} 替换为实际素材路径"""
    if product_id in scene_prompt:
        local_path = get_product_image(product_id)  # 从素材库获取
        return scene_prompt.replace(f"{{{product_id}}}", local_path)
    return scene_prompt  # 无 product_id，使用 AI 生成

import requests, time, os

MINIMAX_API_KEY = os.environ["MINIMAX_API_KEY"]

def clone_voice(audio_path: str, voice_name: str, tenant_id: str) -> str:
    """
    上传样本音频，发起克隆任务，返回 voice_id
    """
    # Step 1: 上传音频文件
    with open(audio_path, "rb") as f:
        upload_resp = requests.post(
            "https://api.minimaxi.com/v1/files/upload",
            headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
            files={"file": f}
        )
    file_id = upload_resp.json()["file"]["file_id"]

    # Step 2: 发起克隆任务
    clone_resp = requests.post(
        "https://api.minimaxi.com/v1/voice_clone",
        headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
        json={
            "file_id": file_id,
            "name": voice_name,
            "model": "minimax-voice-clone-01"
        }
    )
    task_id = clone_resp.json()["task_id"]

    # Step 3: 轮询直到完成（每5秒一次，超时30分钟）
    for _ in range(360):
        status = requests.get(
            f"https://api.minimaxi.com/v1/voice_clone/{task_id}",
            headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"}
        ).json()
        if status["status"] == "completed":
            voice_id = status["voice_id"]
            save_voice_to_library(tenant_id, voice_id, voice_name)
            return voice_id
        elif status["status"] == "failed":
            raise Exception(f"音色克隆失败: {status.get('error', 'unknown')}")
        time.sleep(5)

    raise Exception("音色克隆超时")


def generate_voice_with_clone(narration: str, voice_id: str, output_path: str) -> str:
    """使用克隆音色生成配音"""
    response = requests.post(
        "https://api.minimaxi.com/v1/tts_with_voice_id",
        headers={"Authorization": f"Bearer {MINIMAX_API_KEY}"},
        json={
            "text": narration,
            "voice_id": voice_id,
            "model": "minimax-tts-01",
            "response_format": "mp3"
        }
    )
    audio_url = response.json()["audio_url"]
    audio_data = requests.get(audio_url).content
    with open(output_path, "wb") as f:
        f.write(audio_data)
    return output_path

# 音色库表结构
"""
CREATE TABLE voice_library (
    id SERIAL PRIMARY KEY,
    tenant_id TEXT NOT NULL,
    voice_id TEXT NOT NULL,
    voice_name TEXT NOT NULL,
    sample_duration INT,
    created_at TIMESTAMP DEFAULT NOW(),
    status TEXT DEFAULT 'active',
    usage_count INT DEFAULT 0
);
"""

def list_voices(tenant_id: str) -> list:
    return db.query(
        "SELECT * FROM voice_library WHERE tenant_id = ? AND status = 'active'",
        tenant_id
    )

def delete_voice(tenant_id: str, voice_id: str) -> bool:
    affected = db.execute(
        "DELETE FROM voice_library WHERE tenant_id = ? AND voice_id = ?",
        tenant_id, voice_id
    )
    return affected > 0

def generate_srt(scenes, output_path):
    def ms_to_srt(ms):
        h, rem = divmod(int(ms), 3600000)
        m, rem = divmod(rem, 60000)
        s, ms = divmod(rem, 1000)
        return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
    with open(output_path, "w", encoding="utf-8") as f:
        for i, scene in enumerate(scenes, 1):
            start = sum(s["duration"] for s in scenes[:i-1])
            end = start + scene["duration"]
            f.write(f"{i}\n{ms_to_srt(start*1000)} --> {ms_to_srt(end*1000)}\n{scene['narration']}\n\n")

import subprocess
from pathlib import Path

def synthesize(video_or_images, voices, music_path, scenes, output_path):
    # Step 1: 图片/视频转片段
    for i, src in enumerate(video_or_images):
        ext = Path(src).suffix
        if ext in [".jpg", ".jpeg", ".png"]:
            subprocess.run(["ffmpeg", "-y", "-loop", "1", "-i", src,
                "-t", str(scenes[i]["duration"]),
                "-vf", "scale=1080:1920,crop=1080:1920",
                "-r", "30", "-c:v", "libx264", "-preset", "fast",
                f"/tmp/scene_{i}.mp4"])
        else:
            subprocess.run(["cp", src, f"/tmp/scene_{i}.mp4"])

    # Step 2: 拼接
    with open("/tmp/filelist.txt", "w") as f:
        for i in range(len(video_or_images)):
            f.write(f"file '/tmp/scene_{i}.mp4'\n")
    subprocess.run(["ffmpeg", "-y", "-f", "concat", "-safe", "0",
        "-i", "/tmp/filelist.txt", "-c", "copy", "/tmp/video_only.mp4"])

    # Step 3: 字幕
    generate_srt(scenes, "/tmp/subtitle.srt")
    subprocess.run(["ffmpeg", "-y", "-i", "/tmp/video_only.mp4",
        "-vf", "subtitles=subtitle.srt", "/tmp/video_with_subs.mp4"])

    # Step 4: 混音绑定
    total_dur = sum(s["duration"] for s in scenes)
    subprocess.run(["ffmpeg", "-y", "-i", "/tmp/video_with_subs.mp4",
        "-i", music_path, "-c:v", "copy", "-c:a", "aac",
        "-t", str(total_dur), output_path])
    return output_path

def blend_product_image(product_img: str, duration: int, output: str):
    """将产品图作为固定背景，持续 duration 秒"""
    subprocess.run([
        "ffmpeg", "-y",
        "-loop", "1", "-i", product_img,
        "-i", f"/tmp/voice_{idx}.wav",
        "-i", f"/tmp/bgm.mp3",
        "-t", str(duration),
        "-vf", (
            "scale=1080:1920:force_original_aspect_ratio=increase,"
            "crop=1080:1920,"
            "boxblur=3"
        ),
        "-i", "logo.png",
        "-filter_complex",
        "[0:v][3:v]overlay=W-w-20:H-h-20[bg];"
        "[bg]ass=subtitle.ass[out]",
        "-map", "[out]", "-map", "1:a", "-map", "2:a",
        "-c:v", "libx264", "-preset", "fast",
        "-shortest", output
    ], check=True)

import asyncio

async def generate_assets_concurrently(scenes, music_mood):
    image_tasks = [generate_image(s["visual_prompt"]) for s in scenes]
    voice_tasks = [generate_voice(s["narration"]) for s in scenes]
    music_task = generate_music(music_mood)
    images, voices, music = await asyncio.gather(
        asyncio.gather(*image_tasks),
        asyncio.gather(*voice_tasks),
        music_task
    )
    return list(images), list(voices), music

MAX_RETRIES = 3

def generate_with_retry(func, *args, **kwargs):
    for attempt in range(MAX_RETRIES):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if attempt == MAX_RETRIES - 1:
                raise
            time.sleep(5 * (2 ** attempt))

import streamlit as st

st.title("🎬 AI 全自动视频生成")

topic = st.text_input("输入视频主题", placeholder="例如：五金工具促销，冲击钻特价活动")
video_type = st.selectbox("视频类型", ["产品展示", "促销活动", "使用教程"])

if st.button("🚀 一键生成"):
    with st.spinner("文案生成中..."):
        script = generate_script(topic)

    col1, col2, col3 = st.columns(3)
    with col1:
        st.info("✅ 文案完成")
    with col2:
        with st.spinner("配图+配音+音乐生成中..."):
            assets = generate_assets(script)
    with col3:
        with st.spinner("视频合成中..."):
            output = synthesize_video(script, assets)

    st.success("生成完成！")
    st.video(output)
    st.download_button("📥 下载视频", open(output, "rb"), f"{topic}.mp4")

@hermes.on_message(channel="wecom")
async def handle_video_request(message: str, user_id: str):
    topic = message.strip()
    await hermes.send_text(
        f"收到！正在为您生成视频：{topic}\n预计 3-5 分钟完成，请稍候...",
        to=user_id
    )
    loop = asyncio.get_event_loop()
    loop.run_in_executor(None, generate_video_task, topic, user_id)

def generate_video_task(topic: str, user_id: str):
    try:
        output_path = main_pipeline(topic)
        hermes.send_file(output_path, to=user_id)
    except Exception as e:
        hermes.send_text(f"生成失败：{str(e)}", to=user_id)

def on_video_completed(task_id: str, video_path: str):
    requests.post(
        "https://your-crm.com/webhook/video",
        json={
            "task_id": task_id,
            "video_path": video_path,
            "completed_at": datetime.now().isoformat()
        }
    )

# POST /api/v1/materials/upload
from fastapi import UploadFile, File
import hashlib, uuid, asyncio

async def upload_material(tenant_id: str, category: str, file: UploadFile = File(...)):
    ext = file.filename.split('.')[-1].lower()
    content = await file.read()
    file_hash = hashlib.md5(content).hexdigest()

    # MD5去重
    existing = db.query(
        "SELECT id FROM materials WHERE tenant_id=? AND hash=?", tenant_id, file_hash
    )
    if existing:
        return {"id": existing[0]['id'], "duplicate": True}

    material_id = str(uuid.uuid4())
    path = f"/data/brand_assets/{tenant_id}/{category}/{material_id}.{ext}"
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, 'wb') as f:
        f.write(content)

    db.execute(
        "INSERT INTO materials(id,tenant_id,category,hash,path,filename,size) VALUES(?,?,?,?,?,?,?)",
        material_id, tenant_id, category, file_hash, path, file.filename, len(content)
    )

    # OSS备份（异步）
    asyncio.create_task(upload_to_oss(path, tenant_id, category))
    return {"id": material_id, "url": f"/materials/{material_id}"}

# GET /api/v1/materials?category=product&page=1&page_size=20
def list_materials(tenant_id: str, category: str = None,
                   tag: str = None, page: int = 1, page_size: int = 20):
    query = "SELECT * FROM materials WHERE tenant_id=? AND deleted_at IS NULL"
    params = [tenant_id]
    if category:
        query += " AND category=?"
        params.append(category)
    if tag:
        query += " AND tags LIKE ?"
        params.append(f"%{tag}%")
    query += " ORDER BY created_at DESC LIMIT ? OFFSET ?"
    offset = (page - 1) * page_size
    params.extend([page_size, offset])
    items = db.query(query, *params)
    total = db.query(
        "SELECT COUNT(*) FROM materials WHERE tenant_id=? AND deleted_at IS NULL"
        + (" AND category=?" if category else ""),
        tenant_id, *([category] if category else [])
    )[0]['count']
    return {"items": items, "total": total, "page": page, "page_size": page_size}

# DELETE /api/v1/materials/{id}
def delete_material(tenant_id: str, material_id: str):
    # 软删除（保留30天）
    db.execute(
        "UPDATE materials SET deleted_at=NOW() WHERE id=? AND tenant_id=?",
        material_id, tenant_id
    )
    ref_count = db.query(
        "SELECT COUNT(*) FROM video_tasks WHERE tenant_id=? AND materials LIKE ?",
        tenant_id, f"%{material_id}%"
    )[0]['count']
    if ref_count > 0:
        return {"warning": f"该素材被{ref_count}个任务引用，30天内可恢复"}
    return {"status": "deleted"}

# POST /api/v1/voices/clone
def create_voice_clone(tenant_id: str, sample_id: str, voice_name: str):
    sample = db.query("SELECT * FROM voice_samples WHERE id=?", sample_id)[0]
    voice_id = clone_voice(sample['file_path'], voice_name, tenant_id)

    voice_record = {
        'id': str(uuid.uuid4()),
        'tenant_id': tenant_id,
        'voice_id': voice_id,
        'voice_name': voice_name,
        'sample_duration': sample['duration'],
        'status': 'active',
        'created_at': datetime.now()
    }
    db.execute("INSERT INTO voice_library(...) VALUES(...)", *voice_record.values())
    return voice_record

# GET /api/v1/voices/{voice_id}/preview
def preview_voice(voice_id: str, text: str = "欢迎使用品牌配音"):
    return generate_voice_with_clone(
        narration=text[:50],
        voice_id=voice_id,
        output_path=f"/tmp/preview_{voice_id}.mp3"
    )

# FastAPI 中间件：自动注入 tenant_id
@app.middleware("http")
async def inject_tenant(request: Request, call_next):
    tenant_id = request.headers.get("X-Tenant-ID")
    if not tenant_id:
        raise HTTPException(401, "Missing X-Tenant-ID")
    with get_db_connection() as conn:
        conn.execute(f"SET LOCAL app.tenant_id = '{tenant_id}'")
        response = await call_next(request)
    return response

def audit_log(tenant_id: str, user_id: str, action: str, resource: str, result: str):
    db.execute(
        "INSERT INTO audit_logs(tenant_id,user_id,action,resource,result,ip,created_at) VALUES(?,?,?,?,?,?,NOW())",
        tenant_id, user_id, action, resource, result,
        request.client.host if request else 'system'
    )

def check_cross_tenant(tenant_id: str, resource_tenant_id: str):
    if tenant_id != resource_tenant_id:
        audit_log(tenant_id, "system", "CROSS_TENANT_ACCESS_DENIED",
                  f"attempted access to {resource_tenant_id}", "DENIED")
        raise HTTPException(403, "Access denied")

@app.get("/health")
def health_check():
    return {
        "status": "healthy",
        "redis": check_redis(),
        "ffmpeg": check_ffmpeg(),
    }

#!/usr/bin/env python3
"""
Pixelle-Refactor：公有 API 全自动视频生成流水线
输入：标题 / 关键词
输出：完整 MP4 视频
"""
import os, asyncio
from pathlib import Path
from datetime import datetime

from modules.script import generate_script
from modules.image import generate_image_batch
from modules.video import generate_video_batch
from modules.music import generate_music
from modules.voice import generate_voice_batch
from modules.synthesize import synthesize

class VideoPipeline:
    def __init__(self, config: dict):
        self.config = config
        self.temp_dir = Path(config["TEMP_DIR"])
        self.output_dir = Path(config["OUTPUT_DIR"])
        self.temp_dir.mkdir(parents=True, exist_ok=True)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    async def run(self, topic: str) -> str:
        task_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        print(f"[{task_id}] 开始生成视频，主题：{topic}")

        # Step 1: 文案生成
        print(f"[{task_id}] Step 1/4 文案生成中...")
        script_data = await generate_script(topic)
        scenes = script_data["scenes"]
        total_duration = sum(s["duration"] for s in scenes)

        # Step 2: 素材并发生成
        print(f"[{task_id}] Step 2/4 素材生成中（图片+配音+音乐）...")
        image_tasks = [generate_image(s["visual_prompt"]) for s in scenes]
        voice_tasks = [generate_voice(s["narration"]) for s in scenes]
        music_task = generate_music(script_data.get("bgm_mood", "欢快"), duration=total_duration)

        images, voices, music_path = await asyncio.gather(
            asyncio.gather(*image_tasks),
            asyncio.gather(*voice_tasks),
            music_task
        )

        # Step 3: 视频生成（降级：使用静态图）
        print(f"[{task_id}] Step 3/4 视频生成中...")
        video_paths = []
        for i, img_path in enumerate(images):
            try:
                video_path = await generate_video(img_path)
                video_paths.append(video_path)
            except Exception as e:
                print(f"[{task_id}] 视频生成失败，使用静态图片：{e}")
                video_paths.append(img_path)

        # Step 4: ffmpeg 合成
        print(f"[{task_id}] Step 4/4 视频合成中...")
        output_path = self.output_dir / f"{task_id}_{topic[:20]}.mp4"
        await synthesize(video_paths, voices, music_path, scenes, output_path)

        print(f"[{task_id}] ✅ 生成完成：{output_path}")
        return str(output_path)

def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--topic", "-t", required=True)
    parser.add_argument("--output", "-o", default="/data/videos")
    args = parser.parse_args()

    config = {
        "TEMP_DIR": os.environ.get("TEMP_DIR", "/tmp/video_generation"),
        "OUTPUT_DIR": args.output,
        "MINIMAX_API_KEY": os.environ.get("MINIMAX_API_KEY", ""),
    }
    pipeline = VideoPipeline(config)
    result = asyncio.run(pipeline.run(args.topic))
    print(f"视频路径：{result}")

if __name__ == "__main__":
    main()