app.py

# app.py
# HuggingFace Spaces (Gradio + ZeroGPU) 单文件示例：
# - 自动下载 LongCat-Video GitHub 代码（zip）
# - 自动下载 LongCat-Video / LongCat-Video-Avatar 权重（HF Hub）
# - 通过 spaces.GPU 在 ZeroGPU 环境下按需申请 GPU 执行推理
# - 支持单人：AT2V / AI2V
#
# 说明：
# 1) 官方示例使用 torchrun nproc=2（多进程/可能更快）：
#    这里默认改为 nproc=1 + context_parallel_size=1，更适合 Spaces。
# 2) FlashAttention 默认在 config 开启，但在 Spaces 上未必能顺利安装；
#    本示例会尝试把 config 里所有包含 "flash" 的 attention backend 字段递归替换为 "sdpa"。
#
# 参考：
# - ZeroGPU 官方用法：@spaces.GPU(duration=...)  :contentReference[oaicite:5]{index=5}（用户侧不需要引用，代码内不写引用）
# - LongCat-Video-Avatar 模型卡：推理命令/参数/权重目录结构 :contentReference[oaicite:6]{index=6}

import os
import re
import sys
import json
import time
import shutil
import zipfile
import hashlib
import subprocess
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, Tuple, Optional

# ----------------------------
# 运行时“尽量单文件”的依赖安装
# ----------------------------
def _pip_install(pkgs):
    """在 Spaces 里尽量避免反复安装：用一个标记文件 + 简单 import 探测。"""
    cmd = [sys.executable, "-m", "pip", "install", "-U"] + pkgs
    print("[pip]", " ".join(cmd))
    subprocess.check_call(cmd)

def _ensure_imports():
    """
    只安装本 app 直接需要的包。
    LongCat-Video 自身依赖很多（官方 requirements），这里不强制全量预装，
    而是交给官方脚本在运行时 import；若缺包会在日志里体现，再按需加到下面列表。
    """
    try:
        import gradio as gr  # noqa
    except Exception:
        _pip_install(["gradio>=4.0.0"])

    try:
        import requests  # noqa
    except Exception:
        _pip_install(["requests>=2.31.0"])

    try:
        from huggingface_hub import snapshot_download  # noqa
    except Exception:
        _pip_install(["huggingface_hub[cli]>=0.24.0"])

    # ZeroGPU 推荐的 spaces 包：多数 ZeroGPU 环境自带；没有就装
    try:
        import spaces  # noqa
    except Exception:
        _pip_install(["spaces>=0.27.0"])

_ensure_imports()

import gradio as gr
import requests
from huggingface_hub import snapshot_download

# spaces 在非 ZeroGPU 环境也应可安全使用；若导入失败已在上面安装
import spaces


# ----------------------------
# 配置区（可按需改）
# ----------------------------
GITHUB_ZIP_URL = "https://github.com/meituan-longcat/LongCat-Video/archive/refs/heads/main.zip"

# HF 权重（模型卡说明的目录） :contentReference[oaicite:7]{index=7}
HF_MODEL_LONGCAT_VIDEO = "meituan-longcat/LongCat-Video"
HF_MODEL_LONGCAT_AVATAR = "meituan-longcat/LongCat-Video-Avatar"

# 本地缓存目录：Spaces 上建议放到 /home/user 或当前目录
BASE_DIR = Path(__file__).parent.resolve()
CACHE_DIR = BASE_DIR / "_cache"
REPO_DIR = CACHE_DIR / "LongCat-Video-main"  # zip 解压后的目录名
WEIGHTS_DIR = CACHE_DIR / "weights"
WEIGHTS_LONGCAT_VIDEO = WEIGHTS_DIR / "LongCat-Video"
WEIGHTS_LONGCAT_AVATAR = WEIGHTS_DIR / "LongCat-Video-Avatar"
OUTPUT_DIR = CACHE_DIR / "outputs"
TMP_DIR = CACHE_DIR / "tmp"

# 为了减少 torch CUDA 内存碎片（有时有用）
os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")


# ----------------------------
# 工具函数
# ----------------------------
def _sha1(s: str) -> str:
    return hashlib.sha1(s.encode("utf-8")).hexdigest()[:10]

def _run(cmd, cwd: Optional[Path] = None, env: Optional[Dict[str, str]] = None) -> Tuple[int, str]:
    """运行命令并返回 (code, stdout+stderr)。"""
    print("[run]", " ".join(cmd))
    p = subprocess.Popen(
        cmd,
        cwd=str(cwd) if cwd else None,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
        universal_newlines=True,
    )
    out_lines = []
    while True:
        line = p.stdout.readline()
        if not line and p.poll() is not None:
            break
        if line:
            out_lines.append(line)
    code = p.wait()
    return code, "".join(out_lines)

def _download_and_extract_repo():
    """下载并解压 GitHub zip 到 CACHE_DIR。"""
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    zip_path = CACHE_DIR / "LongCat-Video-main.zip"

    if REPO_DIR.exists() and (REPO_DIR / "run_demo_avatar_single_audio_to_video.py").exists():
        return

    # 清理旧目录
    if REPO_DIR.exists():
        shutil.rmtree(REPO_DIR, ignore_errors=True)

    # 下载 zip
    if not zip_path.exists():
        r = requests.get(GITHUB_ZIP_URL, stream=True, timeout=120)
        r.raise_for_status()
        with open(zip_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1024 * 1024):
                if chunk:
                    f.write(chunk)

    # 解压
    with zipfile.ZipFile(zip_path, "r") as zf:
        zf.extractall(CACHE_DIR)

    # 基本校验
    if not (REPO_DIR / "run_demo_avatar_single_audio_to_video.py").exists():
        raise RuntimeError("仓库解压后未找到 run_demo_avatar_single_audio_to_video.py，可能 GitHub 结构变化。")

def _download_weights():
    """下载 HF 权重到 WEIGHTS_DIR。"""
    WEIGHTS_DIR.mkdir(parents=True, exist_ok=True)

    # 使用 token（若你在 Space Secrets 里配置了 HF_TOKEN）
    token = os.environ.get("HF_TOKEN", None)

    if not WEIGHTS_LONGCAT_VIDEO.exists():
        snapshot_download(
            repo_id=HF_MODEL_LONGCAT_VIDEO,
            local_dir=str(WEIGHTS_LONGCAT_VIDEO),
            token=token,
            local_dir_use_symlinks=False,
        )

    if not WEIGHTS_LONGCAT_AVATAR.exists():
        snapshot_download(
            repo_id=HF_MODEL_LONGCAT_AVATAR,
            local_dir=str(WEIGHTS_LONGCAT_AVATAR),
            token=token,
            local_dir_use_symlinks=False,
        )

def _recursive_patch_attention_backend(obj: Any) -> Any:
    """
    递归把 config 里疑似 flash-attn backend 的字段替换为 sdpa。
    不依赖具体 key 名，尽量“宽松匹配”：
      - key 或 value 里出现 flash / flashattn / flash_attn => 改成 "sdpa"
    """
    if isinstance(obj, dict):
        new = {}
        for k, v in obj.items():
            lk = str(k).lower()
            if any(x in lk for x in ["attn", "attention", "backend"]):
                # 先递归处理 value
                vv = _recursive_patch_attention_backend(v)
                # 再判断是否需要替换
                if isinstance(vv, str) and ("flash" in vv.lower() or "flash_attn" in vv.lower() or "flashattn" in vv.lower()):
                    new[k] = "sdpa"
                else:
                    new[k] = vv
            else:
                new[k] = _recursive_patch_attention_backend(v)
        return new
    elif isinstance(obj, list):
        return [_recursive_patch_attention_backend(x) for x in obj]
    else:
        # 普通标量
        if isinstance(obj, str):
            lo = obj.lower()
            if "flash_attn" in lo or "flashattn" in lo or lo.strip() == "flash" or "flash" == lo.strip():
                return "sdpa"
        return obj

def _try_patch_avatar_configs():
    """
    官方说明：avatar_single/config.json 和 avatar_multi/config.json 默认启用 FlashAttention-2 :contentReference[oaicite:8]{index=8}
    这里尽量替换为 sdpa，避免必须安装 flash-attn。
    """
    cfgs = [
        WEIGHTS_LONGCAT_AVATAR / "avatar_single" / "config.json",
        WEIGHTS_LONGCAT_AVATAR / "avatar_multi" / "config.json",
    ]
    for cfg in cfgs:
        if not cfg.exists():
            continue
        try:
            raw = json.loads(cfg.read_text(encoding="utf-8"))
            patched = _recursive_patch_attention_backend(raw)
            if patched != raw:
                cfg.write_text(json.dumps(patched, ensure_ascii=False, indent=2), encoding="utf-8")
        except Exception as e:
            print(f"[warn] patch config failed: {cfg} -> {e}")

def _load_template_json(template_path: Path) -> Dict[str, Any]:
    data = json.loads(template_path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError("模板 JSON 不是 dict 结构，无法安全修改。")
    return data

def _recursive_replace_first_match(data: Any, key_pred, value_pred, new_value) -> Tuple[Any, bool]:
    """
    在任意 JSON 结构中，找到第一个满足条件的 (key, value) 并替换 value。
    返回 (new_data, replaced?)
    """
    if isinstance(data, dict):
        out = {}
        replaced = False
        for k, v in data.items():
            if (not replaced) and key_pred(k) and value_pred(v):
                out[k] = new_value
                replaced = True
            else:
                nv, r = _recursive_replace_first_match(v, key_pred, value_pred, new_value)
                out[k] = nv
                replaced = replaced or r
        return out, replaced
    elif isinstance(data, list):
        out_list = []
        replaced = False
        for item in data:
            if replaced:
                out_list.append(item)
                continue
            nv, r = _recursive_replace_first_match(item, key_pred, value_pred, new_value)
            out_list.append(nv)
            replaced = replaced or r
        return out_list, replaced
    else:
        return data, False

def _build_input_json_single(
    mode: str,
    audio_path: Path,
    prompt: str,
    ref_image_path: Optional[Path],
    seed: int,
    resolution: str
) -> Path:
    """
    基于 assets/avatar/single_example_1.json 模板生成 input_json。
    官方脚本以 --input_json 读取参数 :contentReference[oaicite:9]{index=9}
    """
    template = REPO_DIR / "assets" / "avatar" / "single_example_1.json"
    if not template.exists():
        raise RuntimeError("未找到模板 assets/avatar/single_example_1.json（仓库结构可能变化）。")

    data = _load_template_json(template)

    # 替换 prompt：优先找 key 包含 prompt/text 之类
    data, _ = _recursive_replace_first_match(
        data,
        key_pred=lambda k: "prompt" in str(k).lower() or "text" in str(k).lower(),
        value_pred=lambda v: isinstance(v, str),
        new_value=prompt.strip() if prompt else "A person is talking."
    )

    # 替换 audio path：找 key 包含 audio 且 value 是字符串
    data, _ = _recursive_replace_first_match(
        data,
        key_pred=lambda k: "audio" in str(k).lower(),
        value_pred=lambda v: isinstance(v, str),
        new_value=str(audio_path)
    )

    # 替换 image path（仅 AI2V）
    if mode == "ai2v" and ref_image_path is not None:
        data, _ = _recursive_replace_first_match(
            data,
            key_pred=lambda k: ("image" in str(k).lower()) or ("ref" in str(k).lower()),
            value_pred=lambda v: isinstance(v, str),
            new_value=str(ref_image_path)
        )

    # seed（若模板里有）
    data, _ = _recursive_replace_first_match(
        data,
        key_pred=lambda k: "seed" in str(k).lower(),
        value_pred=lambda v: isinstance(v, (int, float, str)),
        new_value=int(seed)
    )

    # resolution（若模板里有）
    data, _ = _recursive_replace_first_match(
        data,
        key_pred=lambda k: "resolution" in str(k).lower(),
        value_pred=lambda v: isinstance(v, str),
        new_value=str(resolution)
    )

    TMP_DIR.mkdir(parents=True, exist_ok=True)
    out_path = TMP_DIR / f"single_{mode}_{_sha1(str(audio_path) + prompt + str(seed) + str(time.time()))}.json"
    out_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    return out_path

def _find_latest_mp4(since_ts: float) -> Optional[Path]:
    if not OUTPUT_DIR.exists():
        return None
    candidates = []
    for p in OUTPUT_DIR.rglob("*.mp4"):
        try:
            if p.stat().st_mtime >= since_ts - 2:
                candidates.append(p)
        except Exception:
            pass
    if not candidates:
        return None
    candidates.sort(key=lambda x: x.stat().st_mtime, reverse=True)
    return candidates[0]

def _ensure_ready() -> str:
    """
    准备：
    - 下载 repo
    - 下载权重
    - 尝试 patch attention backend
    """
    _download_and_extract_repo()
    _download_weights()
    _try_patch_avatar_configs()

    # 输出目录
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    return "准备完成：代码与权重已就绪。"


# ----------------------------
# GPU 推理函数（ZeroGPU 核心）
# ----------------------------
@spaces.GPU(duration=900)  # 生成视频通常 >60s，给足时间；你可视情况调小/调大 :contentReference[oaicite:10]{index=10}
def generate_single(
    mode: str,
    audio_file: str,
    prompt: str,
    ref_image_file: Optional[str],
    seed: int,
    resolution: str,
    num_segments: int,
    ref_img_index: int,
    mask_frame_range: int,
) -> Tuple[Optional[str], str]:
    """
    返回：(mp4路径 or None, 日志文本)
    """
    t0 = time.time()

    # 文件落盘路径（Gradio 传入的是本地临时文件路径字符串）
    audio_path = Path(audio_file).resolve()
    ref_image_path = Path(ref_image_file).resolve() if ref_image_file else None

    # 构造 input_json
    input_json = _build_input_json_single(
        mode=mode,
        audio_path=audio_path,
        prompt=prompt,
        ref_image_path=ref_image_path,
        seed=seed,
        resolution=resolution,
    )

    # 运行官方脚本（单进程 torchrun）
    # 官方示例：torchrun --nproc_per_node=2 ... --context_parallel_size=2 ... :contentReference[oaicite:11]{index=11}
    # 这里适配 Space：nproc=1, context_parallel_size=1
    cmd = [
        sys.executable, "-m", "torch.distributed.run",
        "--nproc_per_node=1",
        "run_demo_avatar_single_audio_to_video.py",
        "--context_parallel_size=1",
        f"--checkpoint_dir={WEIGHTS_LONGCAT_AVATAR}",
        f"--stage_1={mode}",
        f"--input_json={input_json}",
        f"--resolution={resolution}",
    ]

    # 续写参数（用户设置 >1 才启用）
    if num_segments and int(num_segments) > 1:
        cmd += [
            f"--num_segments={int(num_segments)}",
            f"--ref_img_index={int(ref_img_index)}",
            f"--mask_frame_range={int(mask_frame_range)}",
        ]

    # 环境变量：让脚本能找到模块
    env = dict(os.environ)
    env["PYTHONPATH"] = str(REPO_DIR) + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "")
    env["HF_HOME"] = str(CACHE_DIR / "hf_home")
    env["TORCH_HOME"] = str(CACHE_DIR / "torch_home")

    # 约定输出目录（若脚本支持/或脚本默认输出在当前目录下的 outputs）
    # 我们用 cwd + 输出扫描兜底
    env["OUTPUT_DIR"] = str(OUTPUT_DIR)

    code, log = _run(cmd, cwd=REPO_DIR, env=env)

    # 尝试找到最新 mp4
    mp4 = _find_latest_mp4(t0)
    if mp4 is None:
        # 兜底：在 repo 内也扫一下
        repo_candidates = list(REPO_DIR.rglob("*.mp4"))
        repo_candidates.sort(key=lambda x: x.stat().st_mtime, reverse=True)
        if repo_candidates and repo_candidates[0].stat().st_mtime >= t0 - 2:
            mp4 = repo_candidates[0]

    if code != 0:
        return None, f"执行失败（exit={code}）。日志如下：\n\n{log}"

    if mp4 is None or not mp4.exists():
        return None, f"执行完成，但未找到 mp4 输出文件。日志如下：\n\n{log}"

    return str(mp4), f"执行成功：{mp4}\n\n日志如下：\n\n{log}"


# ----------------------------
# Gradio UI
# ----------------------------
def ui_prepare() -> str:
    try:
        return _ensure_ready()
    except Exception as e:
        return f"准备失败：{e}"

with gr.Blocks(title="LongCat-Video-Avatar (ZeroGPU) - Single File Space") as demo:
    gr.Markdown(
        """
# LongCat-Video-Avatar（ZeroGPU / 单文件 Space）

- 单人模式：**AT2V（音频+文本）** / **AI2V（音频+图片）**
- 续写（Video Continuation）：把 **num_segments** 设为 > 1 即可（官方参数：ref_img_index / mask_frame_range）  
- 提示：为了更自然的口型，prompt 里建议包含 talking/speaking 等动作词（模型卡建议）  
"""
    )

    with gr.Row():
        btn_prepare = gr.Button("一键准备（下载代码+权重）", variant="primary")
        prep_status = gr.Textbox(label="准备状态", value="尚未准备。首次准备会下载较大权重。", lines=2)

    btn_prepare.click(fn=ui_prepare, outputs=prep_status)

    with gr.Row():
        mode = gr.Radio(
            choices=[("Audio-Text-to-Video (AT2V)", "at2v"), ("Audio-Image-to-Video (AI2V)", "ai2v")],
            value="at2v",
            label="模式"
        )

    with gr.Row():
        audio_in = gr.Audio(label="输入音频（wav/mp3等）", type="filepath")
        ref_img = gr.Image(label="参考图（仅 AI2V 需要）", type="filepath")

    prompt = gr.Textbox(
        label="Prompt（建议包含 talking/speaking 等动作词）",
        value="A young person is talking naturally, realistic style.",
        lines=2
    )

    with gr.Row():
        seed = gr.Number(label="Seed", value=0, precision=0)
        resolution = gr.Dropdown(label="分辨率", choices=["480P", "720P"], value="480P")

    with gr.Accordion("高级参数（续写/一致性/防重复）", open=False):
        num_segments = gr.Slider(label="num_segments（>1 启用续写）", minimum=1, maximum=8, step=1, value=1)
        ref_img_index = gr.Slider(label="ref_img_index（默认 10）", minimum=-30, maximum=60, step=1, value=10)
        mask_frame_range = gr.Slider(label="mask_frame_range（默认 3）", minimum=1, maximum=12, step=1, value=3)

    btn = gr.Button("生成视频", variant="primary")

    out_video = gr.Video(label="输出视频（mp4）")
    out_log = gr.Textbox(label="运行日志", lines=18)

    def _validate(mode_v, audio_fp, img_fp):
        if not audio_fp:
            raise gr.Error("请先上传音频。")
        if mode_v == "ai2v" and not img_fp:
            raise gr.Error("AI2V 模式必须上传参考图。")

    def run(mode_v, audio_fp, prompt_v, img_fp, seed_v, res_v, seg_v, idx_v, mask_v):
        _validate(mode_v, audio_fp, img_fp)
        # seed=0 时也允许；如果想随机可自己改成 random
        return generate_single(mode_v, audio_fp, prompt_v, img_fp, int(seed_v), res_v, int(seg_v), int(idx_v), int(mask_v))

    btn.click(
        fn=run,
        inputs=[mode, audio_in, prompt, ref_img, seed, resolution, num_segments, ref_img_index, mask_frame_range],
        outputs=[out_video, out_log],
    )

demo.queue(max_size=12).launch()