server.py

"""
手机端视频生成服务

在电脑上运行这个服务，手机在同一 WiFi 下用浏览器打开即可使用。

用法:
  cd D:\zero_track\Reset\video_maker_app\server
  pip install fastapi uvicorn python-multipart
  python server.py

然后在手机浏览器打开 http://你的电脑IP:8000
"""

import os
import sys
import uuid
import json
import shutil
from pathlib import Path
from typing import Optional
import uvicorn
from fastapi import FastAPI, UploadFile, File, Form, BackgroundTasks
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles

# ── 路径 ──
BASE_DIR = Path(__file__).parent
UPLOAD_DIR = BASE_DIR / "uploads"
OUTPUT_DIR = BASE_DIR / "output"
TEMPLATES_DIR = BASE_DIR / "templates"
VIDEO_MAKER = BASE_DIR
TTS_CACHE = VIDEO_MAKER / ".tts_cache"
MUSIC_DIR = VIDEO_MAKER / "input" / "music"

os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)
os.makedirs(TTS_CACHE, exist_ok=True)

app = FastAPI(title="视频生成器")


# ── API ──

@app.get("/", response_class=HTMLResponse)
async def index():
    """返回网页界面"""
    html = (TEMPLATES_DIR / "index.html").read_text(encoding="utf-8")
    return html


@app.post("/api/generate")
async def generate(
    background_tasks: BackgroundTasks,
    images: list[UploadFile] = File(...),
    subtitles: str = Form(...),  # JSON 数组: ["字幕1","字幕2",...]
    bgm: Optional[UploadFile] = File(None),
    voice_clone: bool = Form(True),  # 是否用声音克隆
    duration: float = Form(15.0),
):
    """
    生成视频接口

    - images: 图片文件（按顺序上传）
    - subtitles: JSON 字符串，如 ["字幕1","字幕2"]
    - bgm: 背景音乐文件（可选）
    - voice_clone: 是否使用声音克隆（默认 True）
    """
    task_id = uuid.uuid4().hex[:12]
    work_dir = UPLOAD_DIR / task_id
    img_dir = work_dir / "images"
    os.makedirs(img_dir)

    try:
        # 1. 保存图片
        sub_list = json.loads(subtitles)
        for i, img in enumerate(images):
            ext = Path(img.filename).suffix or ".jpg"
            path = img_dir / f"{i}{ext}"
            with open(path, "wb") as f:
                f.write(await img.read())

        # 2. 保存背景音乐
        bgm_path = None
        if bgm and bgm.filename:
            bgm_path = work_dir / "bgm" / bgm.filename
            os.makedirs(bgm_path.parent)
            with open(bgm_path, "wb") as f:
                f.write(await bgm.read())

        # 3. 生成字幕文件
        sub_file = work_dir / "subtitles.txt"
        sub_file.write_text("\n".join(sub_list), encoding="utf-8")

        # 4. 输出路径
        output_path = OUTPUT_DIR / f"{task_id}.mp4"

        # 5. 异步渲染
        background_tasks.add_task(
            render_video,
            img_dir=str(img_dir),
            sub_file=str(sub_file),
            bgm_path=str(bgm_path) if bgm_path else None,
            output_path=str(output_path),
            voice_clone=voice_clone,
            duration=duration,
        )

        return {
            "task_id": task_id,
            "status": "processing",
            "download_url": f"/api/download/{task_id}",
            "check_url": f"/api/status/{task_id}",
        }

    except Exception as e:
        return {"error": str(e)}


@app.get("/api/status/{task_id}")
async def status(task_id: str):
    """检查任务状态"""
    output_path = OUTPUT_DIR / f"{task_id}.mp4"
    if output_path.exists():
        return {"status": "done"}
    return {"status": "processing"}


@app.get("/api/download/{task_id}")
async def download(task_id: str):
    """下载视频"""
    output_path = OUTPUT_DIR / f"{task_id}.mp4"
    if not output_path.exists():
        return {"error": "文件未就绪"}
    return FileResponse(str(output_path), media_type="video/mp4",
                        filename="output.mp4")


# ── 渲染函数 ──

def render_video(img_dir: str, sub_file: str, bgm_path: str | None,
                 output_path: str, voice_clone: bool, duration: float):
    """调用本地渲染引擎生成视频"""
    import subprocess

    # 复制图片到 video_maker/input/images
    img_target = VIDEO_MAKER / "input" / "images"
    img_target.mkdir(parents=True, exist_ok=True)
    for f in Path(img_dir).iterdir():
        if f.is_file():
            shutil.copy2(str(f), str(img_target / f.name))

    # 复制字幕
    shutil.copy2(sub_file, str(VIDEO_MAKER / "input" / "subtitles.txt"))

    # 复制背景音乐
    if bgm_path:
        bgm_target = MUSIC_DIR / "background_custom.wav"
        shutil.copy2(bgm_path, str(bgm_target))

    # 调用 make_fast.py
    cmd = [
        sys.executable,
        str(VIDEO_MAKER / "make_fast.py"),
        "--local-tts" if voice_clone else "--no-tts",
        "--output", output_path,
        "--duration", str(duration),
    ]
    # 如果有自定义 BGM，传参
    if bgm_path:
        cmd += ["--music", str(VIDEO_MAKER / "input" / "music" / "background_custom.wav")]
    else:
        cmd += ["--no-music"]

    # 设置代理（需要时）
    env = os.environ.copy()
    env["HTTP_PROXY"] = "http://localhost:9910"
    env["HTTPS_PROXY"] = "http://localhost:9910"
    env["PYTHONIOENCODING"] = "utf-8"

    log_path = Path(output_path).with_suffix(".log")
    with open(log_path, "w", encoding="utf-8") as log:
        result = subprocess.run(cmd, stdout=log, stderr=log, env=env)
    if result.returncode != 0:
        print(f"❌ 渲染失败 (code {result.returncode})，日志: {log_path}")
    else:
        print(f"✅ 渲染成功: {output_path}")


if __name__ == "__main__":
    print("=" * 60)
    print("📱 视频生成服务启动")
    print("   手机浏览器打开 http://你的电脑IP:8000")
    print("   本地打开 http://localhost:8000")
    print("=" * 60)
    uvicorn.run(app, host="0.0.0.0", port=8000)