Alibabacloud Avatar Video

#!/usr/bin/env python3
"""
AnimateAnyone Gen2 — 三步流水线

SECURITY NOTES:
- subprocess: used ONLY to invoke system ffmpeg for video/image format
  conversion (e.g. webm→mp4, heic→jpg). No shell=True, no eval/exec.
- OSS credentials: read from environment variables, used ONLY to upload
  user media to their own OSS bucket. Never transmitted to third parties.
- All API calls target dashscope.aliyuncs.com (Alibaba Cloud official).
  Step 1: animate-anyone-detect-gen2   图像检测（同步）
  Step 2: animate-anyone-template-gen2 动作模板生成（异步，得到 template_id）
  Step 3: animate-anyone-gen2          视频生成（异步，得到 video_url）

支持多种输入格式，通过 ffmpeg 自动转换：
  图片: webp/heic/tif/bmp → jpg
  视频: webm/avi/mov/mkv/flv → mp4 (H.264, ≥24fps)

用法:
  python animate_anyone.py --image ./face.jpg --video ./dance.webm --download
  python animate_anyone.py --image-url https://... --video-url https://... --download
  python animate_anyone.py --image ./face.jpg --template-id AACT.xxx --download  # 跳过模板生成
"""

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

import requests

from input_validation import (
    mk_temp_path_for_ffmpeg,
    resolve_under_cwd,
    validate_http_https_url,
)

BASE_URL = os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com")
_OSS_SIGNED_URL_EXPIRES = int(os.environ.get("OSS_SIGNED_URL_EXPIRES", str(3 * 24 * 3600)))

# ── ffmpeg helpers ─────────────────────────────────────────────────────────────

def _find_ffmpeg() -> str:
    """Find ffmpeg in PATH or common install locations."""
    for p in ["ffmpeg", "/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg"]:
        if shutil.which(p):
            return p
    raise RuntimeError("ffmpeg not found. Install: apt install ffmpeg / brew install ffmpeg")


def convert_image(src: str) -> str:
    """
    Convert image to jpg if not already jpg/jpeg/png.
    Returns path to (possibly new) file. Temp files are tracked for cleanup.
    """
    p = Path(src)
    if p.suffix.lower() in (".jpg", ".jpeg", ".png"):
        return src
    ff = _find_ffmpeg()
    dst = mk_temp_path_for_ffmpeg(".jpg", "aa_img_")
    subprocess.run([ff, "-y", "-i", src, "-q:v", "2", dst],
                   check=True, capture_output=True)
    print(f"[convert] {p.name} → {Path(dst).name}")
    return dst


def convert_video(src: str) -> str:
    """
    Convert video to mp4 (H.264) with ≥24fps if not already compatible.
    Returns path to (possibly new) file.
    Requirements: mp4/avi/mov, H.264 or H.265, fps≥24, bitrate reasonable.
    """
    p = Path(src)
    # probe fps
    probe = subprocess.run(
        ["ffprobe", "-v", "quiet", "-select_streams", "v:0",
         "-show_entries", "stream=codec_name,r_frame_rate,width,height",
         "-of", "json", src],
        capture_output=True, text=True
    )
    codec, fps_num, fps_den = "unknown", 25, 1
    try:
        info = json.loads(probe.stdout)
        stream = info.get("streams", [{}])[0]
        codec = stream.get("codec_name", "unknown")
        fr = stream.get("r_frame_rate", "25/1").split("/")
        fps_num, fps_den = int(fr[0]), max(int(fr[1]), 1)
    except Exception:
        pass
    fps = fps_num / fps_den

    need_convert = (
        p.suffix.lower() not in (".mp4", ".avi", ".mov")
        or codec not in ("h264", "hevc", "h265")
        or fps < 24
    )
    if not need_convert:
        return src

    ff = _find_ffmpeg()
    dst = mk_temp_path_for_ffmpeg(".mp4", "aa_vid_")
    # ensure ≥24fps, H.264
    vf = f"fps=max(fps\\,24)" if fps < 24 else None
    cmd = [ff, "-y", "-i", src, "-c:v", "libx264", "-preset", "fast",
           "-crf", "22", "-c:a", "aac", "-movflags", "+faststart"]
    if vf:
        cmd += ["-vf", f"fps=24"]
    cmd.append(dst)
    subprocess.run(cmd, check=True, capture_output=True)
    print(f"[convert] {p.name} → {Path(dst).name}  codec={codec} fps={fps:.1f}→converted")
    return dst


# ── OSS upload ─────────────────────────────────────────────────────────────────

def upload_to_oss(local_path: str, expires: int = _OSS_SIGNED_URL_EXPIRES) -> str:
    """Upload file to OSS and return signed GET URL (default 3 days)."""
    import oss2
    auth = oss2.Auth(
        os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
        os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
    )
    bucket_name = os.environ["OSS_BUCKET"]
    endpoint = os.environ.get("OSS_ENDPOINT", "oss-cn-beijing.aliyuncs.com")
    endpoint = endpoint.replace("https://", "").replace("http://", "").rstrip("/")
    bucket = oss2.Bucket(auth, f"https://{endpoint}", bucket_name)
    key = f"human-avatar/{Path(local_path).name}"
    print(f"[oss] uploading {Path(local_path).name} …")
    bucket.put_object_from_file(key, local_path)
    url = bucket.sign_url("GET", key, expires)
    print(f"[oss] signed_url ok ({expires//3600}h)")
    return url


# ── DashScope helpers ──────────────────────────────────────────────────────────

USER_AGENT = "AlibabaCloud-Agent-Skills/alibabacloud-avatar-video"


def _headers(async_mode: bool = False) -> dict:
    key = os.environ.get("DASHSCOPE_API_KEY")
    if not key:
        raise RuntimeError("DASHSCOPE_API_KEY not set")
    h = {
        "Authorization": f"Bearer {key}",
        "Content-Type": "application/json",
        "User-Agent": USER_AGENT,
    }
    if async_mode:
        h["X-DashScope-Async"] = "enable"
    return h


def wait_task(task_id: str, interval: int = 10, max_wait: int = 1800) -> dict:
    """Poll until task SUCCEEDED/FAILED, return full output dict."""
    url = f"{BASE_URL}/api/v1/tasks/{task_id}"
    start = time.time()
    while time.time() - start < max_wait:
        r = requests.get(url, headers=_headers(), timeout=60)
        r.raise_for_status()
        data = r.json()
        out = data.get("output", {})
        status = out.get("task_status", "UNKNOWN")
        elapsed = int(time.time() - start)
        print(f"  [{elapsed}s] task_id={task_id[:16]}… status={status}")
        if status == "SUCCEEDED":
            return out
        if status in ("FAILED", "CANCELED", "UNKNOWN"):
            raise RuntimeError(f"Task failed: {json.dumps(data, ensure_ascii=False)}")
        time.sleep(interval)
    raise TimeoutError(f"Task {task_id} timed out after {max_wait}s")


# ── Step 1: Image detect ───────────────────────────────────────────────────────

def aa_detect(image_url: str) -> dict:
    """
    POST /aa-detect — 检测图像是否符合 AA 要求（同步）。
    Returns output dict with check_pass, bodystyle.
    """
    print(f"\n[step1] aa-detect …")
    r = requests.post(
        f"{BASE_URL}/api/v1/services/aigc/image2video/aa-detect",
        headers=_headers(async_mode=False),
        json={"model": "animate-anyone-detect-gen2", "input": {"image_url": image_url}},
        timeout=60,
    )
    r.raise_for_status()
    data = r.json()
    out = data.get("output", {})
    check_pass = out.get("check_pass", False)
    bodystyle = out.get("bodystyle", "")
    reason = out.get("reason", "")
    if check_pass:
        print(f"  ✅ detect passed  bodystyle={bodystyle}")
    else:
        print(f"  ❌ detect FAILED  reason={reason}")
        raise ValueError(f"Image failed AA detect: {reason}")
    return out


# ── Step 2: Template generation ────────────────────────────────────────────────

def aa_template(video_url: str) -> str:
    """
    POST /aa-template-generation/ — 从视频提取动作模板（异步）。
    Returns template_id str.
    """
    print(f"\n[step2] aa-template-generation …")
    r = requests.post(
        f"{BASE_URL}/api/v1/services/aigc/image2video/aa-template-generation/",
        headers=_headers(async_mode=True),
        json={"model": "animate-anyone-template-gen2", "input": {"video_url": video_url}},
        timeout=60,
    )
    r.raise_for_status()
    data = r.json()
    task_id = (data.get("output") or {}).get("task_id")
    if not task_id:
        raise RuntimeError(f"No task_id in template response: {json.dumps(data, ensure_ascii=False)}")
    print(f"  task_id={task_id}")
    out = wait_task(task_id, interval=10)
    template_id = out.get("template_id")
    if not template_id:
        raise RuntimeError(f"No template_id in result: {out}")
    duration = (out.get("usage") or {}).get("video_duration", "?")
    print(f"  ✅ template_id={template_id}  duration={duration}s")
    return template_id


# ── Step 3: Video generation ────────────────────────────────────────────────────

def aa_generate(image_url: str, template_id: str,
                use_ref_img_bg: bool = False, video_ratio: str = "9:16") -> str:
    """
    POST /video-synthesis/ — 基于图像 + 动作模板生成视频（异步）。
    Returns video_url str.
    """
    print(f"\n[step3] aa-generate  use_ref_img_bg={use_ref_img_bg}  video_ratio={video_ratio} …")
    payload = {
        "model": "animate-anyone-gen2",
        "input": {"image_url": image_url, "template_id": template_id},
        "parameters": {"use_ref_img_bg": use_ref_img_bg, "video_ratio": video_ratio},
    }
    r = requests.post(
        f"{BASE_URL}/api/v1/services/aigc/image2video/video-synthesis/",
        headers=_headers(async_mode=True),
        json=payload,
        timeout=60,
    )
    r.raise_for_status()
    data = r.json()
    task_id = (data.get("output") or {}).get("task_id")
    if not task_id:
        raise RuntimeError(f"No task_id in generate response: {json.dumps(data, ensure_ascii=False)}")
    print(f"  task_id={task_id}")
    out = wait_task(task_id, interval=15, max_wait=1800)
    video_url = out.get("video_url")
    if not video_url:
        raise RuntimeError(f"No video_url in result: {out}")
    print(f"  ✅ video_url={video_url}")
    return video_url


# ── Main ───────────────────────────────────────────────────────────────────────

def main():
    p = argparse.ArgumentParser(
        description="AnimateAnyone Gen2: image+video → animated video (3-step pipeline)"
    )
    p.add_argument("--image-url", help="图片公网 URL（已上传 OSS 等）")
    p.add_argument("--video-url", help="动作视频公网 URL（已上传 OSS 等）")
    p.add_argument("--image", help="本地图片路径（自动转换格式并上传 OSS）")
    p.add_argument("--video", help="本地动作视频路径（自动转换格式并上传 OSS）")
    p.add_argument("--template-id", help="已有 template_id，跳过 Step 2")
    p.add_argument("--use-ref-img-bg", action="store_true",
                   help="以输入图片为背景生成（默认用视频背景）")
    p.add_argument("--video-ratio", default="9:16", choices=["9:16", "3:4"],
                   help="视频画幅（use_ref_img_bg=true 时有效）")
    p.add_argument("--download", action="store_true", help="下载输出视频")
    p.add_argument("--output", default="aa_output.mp4", help="输出文件名")
    p.add_argument("--skip-detect", action="store_true", help="跳过图像检测步骤")
    args = p.parse_args()

    _AA_TEMPLATE_ID_RE = re.compile(r"^[A-Za-z0-9._-]{1,256}$")

    tmp_files = []

    try:
        # ── prepare image URL ──────────────────────────────────────────────
        image_url = args.image_url
        if image_url:
            image_url = validate_http_https_url(image_url, field="--image-url")
        if not image_url:
            if not args.image:
                p.error("需要 --image 或 --image-url")
            converted = convert_image(args.image)
            if converted != args.image:
                tmp_files.append(converted)
            image_url = upload_to_oss(converted)

        # ── prepare video URL ──────────────────────────────────────────────
        video_url = args.video_url
        if video_url:
            video_url = validate_http_https_url(video_url, field="--video-url")
        if args.template_id and not _AA_TEMPLATE_ID_RE.fullmatch(args.template_id.strip()):
            raise ValueError(f"Invalid --template-id format: {args.template_id!r}")
        if not video_url and not args.template_id:
            if not args.video:
                p.error("需要 --video 或 --video-url 或 --template-id")
            converted = convert_video(args.video)
            if converted != args.video:
                tmp_files.append(converted)
            video_url = upload_to_oss(converted)

        # ── Step 1: detect ─────────────────────────────────────────────────
        if not args.skip_detect:
            aa_detect(image_url)

        # ── Step 2: template ───────────────────────────────────────────────
        template_id = args.template_id
        if not template_id:
            template_id = aa_template(video_url)

        # ── Step 3: generate ───────────────────────────────────────────────
        final_url = aa_generate(
            image_url, template_id,
            use_ref_img_bg=args.use_ref_img_bg,
            video_ratio=args.video_ratio,
        )

        print(f"\n✅ Done! video_url = {final_url}")

        if args.download:
            out_path = resolve_under_cwd(args.output, field="--output")
            safe_url = validate_http_https_url(final_url, field="result video URL")
            print(f"Downloading → {out_path} …")
            with urllib.request.urlopen(safe_url, timeout=300) as response:
                with open(out_path, 'wb') as f:
                    f.write(response.read())
            size = out_path.stat().st_size
            print(f"Saved {out_path} ({size//1024}KB)")

    finally:
        for f in tmp_files:
            try:
                os.unlink(f)
            except Exception:
                pass


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"\nERROR: {e}", file=sys.stderr)
        sys.exit(1)

#!/usr/bin/env python3
"""灵眸：自动选择模板创建播报视频并轮询结果。

能力：
- 列出账号下已有播报模板
- 未指定 --template-id 时随机选择一个已有模板
- 自动读取模板 variables，并把文本填入可用的 text 变量（优先 text_content）
- 在 1.7.0+ SDK 下可列出公共模板，并在本地模板为空时复制最多 3 个公共模板
- 注意：复制公共模板成功 ≠ 一定可以立即直接生成视频；有些复制出来的是草稿模板，可能仍需人工补充有效片段/素材
"""
import argparse
import json
import os
import random
import sys
import time
import urllib.request
from typing import Optional

from input_validation import read_text_file_limited, resolve_under_cwd, validate_http_https_url


VENV_PYTHON = os.environ.get(
    "LINGMOU_VENV_PYTHON",
    "/data/xxx-workspace/.venv-alibabacloud-avatar-video/bin/python",
)


def create_client():
    try:
        from alibabacloud_lingmou20250527.client import Client
        from alibabacloud_tea_openapi import models as open_api_models
    except Exception as e:
        raise RuntimeError(
            "需要安装: pip install alibabacloud-lingmou20250527 alibabacloud-tea-openapi"
        ) from e

    config = open_api_models.Config(
        access_key_id=os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
        access_key_secret=os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
        endpoint=os.environ.get("LINGMOU_ENDPOINT", "lingmou.cn-beijing.aliyuncs.com"),
        region_id=os.environ.get("LINGMOU_REGION", "cn-beijing"),
        user_agent="AlibabaCloud-Agent-Skills/alibabacloud-avatar-video",
    )
    return Client(config)


def sdk_supports_public_templates(client) -> bool:
    return hasattr(client, "list_public_broadcast_scene_templates") and hasattr(client, "copy_broadcast_scene_from_template")


def list_templates(client):
    from alibabacloud_lingmou20250527 import models as lm

    resp = client.list_broadcast_templates(lm.ListBroadcastTemplatesRequest())
    return resp.body.data or []


def list_public_templates(client, size: int = 10):
    from alibabacloud_lingmou20250527 import models as lm

    if not sdk_supports_public_templates(client):
        return []
    req = lm.ListPublicBroadcastSceneTemplatesRequest(size=size)
    resp = client.list_public_broadcast_scene_templates(req)
    return resp.body.data or []


def copy_public_templates(client, public_templates, max_copy: int = 3):
    from alibabacloud_lingmou20250527 import models as lm

    copied = []
    for idx, t in enumerate(public_templates[:max_copy], start=1):
        req = lm.CopyBroadcastSceneFromTemplateRequest(
            name=f"OpenClaw copied template {idx}",
            ratio=getattr(t, "ratio", None) or "16:9",
            template_id=t.id,
        )
        resp = client.copy_broadcast_scene_from_template(req)
        copied.append(resp.body.data)
    return copied


def get_template_detail(client, template_id: str):
    from alibabacloud_lingmou20250527 import models as lm

    req = lm.GetBroadcastTemplateRequest()
    req.template_id = template_id
    resp = client.get_broadcast_template(req)
    return resp.body.data


def choose_template(client, explicit_template_id: Optional[str] = None, seed: Optional[int] = None, auto_copy_public: bool = True):
    templates = list_templates(client)
    if explicit_template_id:
        for t in templates:
            if t.id == explicit_template_id:
                return t, False
        class T:
            pass
        t = T()
        t.id = explicit_template_id
        t.name = explicit_template_id
        return t, False

    if templates:
        rng = random.Random(seed)
        return rng.choice(list(templates)), False

    if auto_copy_public and sdk_supports_public_templates(client):
        publics = list_public_templates(client, size=10)
        if publics:
            copied = copy_public_templates(client, publics, max_copy=3)
            if copied:
                rng = random.Random(seed)
                return rng.choice(list(copied)), True

    raise RuntimeError(
        "当前账号下没有可用播报模板；且未能获得可直接使用的公共模板副本。请先在灵眸中准备至少一个可直接生成的视频模板。"
    )


def build_variables(template_detail, text: str):
    from alibabacloud_lingmou20250527 import models as lm

    variables = getattr(template_detail, "variables", None) or []
    if not variables:
        return [
            lm.TemplateVariable(
                name="text_content",
                type="text",
                properties={"content": text},
            )
        ]

    text_vars = [v for v in variables if getattr(v, "type", None) == "text"]
    if not text_vars:
        raise RuntimeError("模板中没有可替换的 text 变量，暂时无法仅凭脚本生成口播视频")

    preferred_names = ["text_content", "test_text", "content", "text"]
    target = None
    for name in preferred_names:
        for v in text_vars:
            if getattr(v, "name", None) == name:
                target = v
                break
        if target:
            break
    if not target:
        target = text_vars[0]

    return [
        lm.TemplateVariable(
            name=target.name,
            type="text",
            properties={"content": text},
        )
    ]


def submit_video(client, template_id: str, text: str, name: str, resolution: str, fps: int, watermark: bool):
    from alibabacloud_lingmou20250527 import models as lm

    template_detail = get_template_detail(client, template_id)
    variables = build_variables(template_detail, text)
    req = lm.CreateBroadcastVideoFromTemplateRequest(
        template_id=template_id,
        name=name,
        variables=variables,
        video_options=lm.CreateBroadcastVideoFromTemplateRequestVideoOptions(
            resolution=resolution,
            fps=fps,
            watermark=watermark,
        ),
    )
    resp = client.create_broadcast_video_from_template(req)
    video_id = resp.body.data.id
    return video_id, template_detail, variables


def wait_video(client, video_id: str, interval: int = 3, max_wait: int = 1800):
    from alibabacloud_lingmou20250527 import models as lm

    start = time.time()
    while time.time() - start < max_wait:
        req = lm.ListBroadcastVideosByIdRequest(video_ids=[video_id])
        resp = client.list_broadcast_videos_by_id(req)
        data = resp.body.data or []
        if not data:
            time.sleep(interval)
            continue

        video = data[0]
        status = video.status
        print(f"status={status}")
        if status == "SUCCESS":
            return video.video_url
        if status in ("ERROR", "FAILED"):
            raise RuntimeError(f"LingMou task failed: {status}")

        time.sleep(interval)

    raise TimeoutError("LingMou polling timeout")


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--template-id")
    p.add_argument("--list-templates", action="store_true", help="列出账号下已有模板")
    p.add_argument("--list-public-templates", action="store_true", help="列出公共模板（需 SDK 1.7.0+）")
    p.add_argument("--copy-public-templates", action="store_true", help="复制最多 3 个公共模板到当前账号（需 SDK 1.7.0+）")
    p.add_argument("--show-template-detail", action="store_true", help="输出所选模板详情")
    p.add_argument("--seed", type=int, default=None, help="随机模板选择种子，便于复现")
    p.add_argument("--text")
    p.add_argument("--text-file")
    p.add_argument("--name", default="OpenClaw Avatar Video")
    p.add_argument("--resolution", default="720p", choices=["720p", "1080p"])
    p.add_argument("--fps", type=int, default=30, choices=[15, 30])
    p.add_argument("--watermark", action="store_true", default=False)
    p.add_argument("--download", action="store_true")
    p.add_argument("--output", default="lingmou_output.mp4")
    args = p.parse_args()

    client = create_client()

    if args.list_templates:
        templates = list_templates(client)
        print(json.dumps([
            {"id": t.id, "name": getattr(t, "name", None)} for t in templates
        ], ensure_ascii=False, indent=2))

    if args.list_public_templates:
        publics = list_public_templates(client, size=10)
        print(json.dumps([
            {"id": t.id, "name": getattr(t, "name", None), "ratio": getattr(t, "ratio", None), "desc": getattr(t, "desc", None)} for t in publics
        ], ensure_ascii=False, indent=2))

    if args.copy_public_templates:
        publics = list_public_templates(client, size=10)
        copied = copy_public_templates(client, publics, max_copy=3)
        print(json.dumps([
            {"id": t.id, "name": getattr(t, "name", None), "ratio": getattr(t, "ratio", None), "status": getattr(t, "status", None)} for t in copied
        ], ensure_ascii=False, indent=2))
        if not (args.text or args.text_file or args.template_id or args.show_template_detail):
            return

    if (args.list_templates or args.list_public_templates) and not (args.text or args.text_file or args.template_id or args.show_template_detail or args.copy_public_templates):
        return

    chosen, copied_from_public = choose_template(client, args.template_id, seed=args.seed, auto_copy_public=True)
    print(f"template_id={chosen.id}")
    print(f"template_name={getattr(chosen, 'name', '')}")
    print(f"template_from_public_copy={str(copied_from_public).lower()}")

    if args.show_template_detail:
        detail = get_template_detail(client, chosen.id)
        print(json.dumps(detail.to_map(), ensure_ascii=False, indent=2, default=str))
        if not (args.text or args.text_file):
            return

    text = args.text
    if args.text_file:
        text = read_text_file_limited(args.text_file, field="--text-file")
    if not text:
        p.error("Need --text or --text-file")

    try:
        video_id, template_detail, variables = submit_video(
            client,
            template_id=chosen.id,
            text=text,
            name=args.name,
            resolution=args.resolution,
            fps=args.fps,
            watermark=args.watermark,
        )
    except Exception as e:
        if copied_from_public:
            raise RuntimeError(
                f"公共模板已复制，但直接生成失败：{e}。这通常表示复制出的模板仍是草稿或缺少有效片段/素材，需要先在灵眸侧完善模板。"
            )
        raise

    print(f"video_id={video_id}")
    print("variables=" + json.dumps([v.to_map() for v in variables], ensure_ascii=False))
    if args.show_template_detail:
        print("template_detail=" + json.dumps(template_detail.to_map(), ensure_ascii=False, default=str))

    video_url = wait_video(client, video_id)
    print(f"video_url={video_url}")

    if args.download and video_url:
        out_path = resolve_under_cwd(args.output, field="--output")
        safe_url = validate_http_https_url(video_url, field="video URL")
        with urllib.request.urlopen(safe_url, timeout=300) as response:
            with open(out_path, 'wb') as f:
                f.write(response.read())
        print(f"saved={out_path}")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)

#!/usr/bin/env python3
"""
Human Avatar Demo Pipeline

统一入口，调用本 Skill 的三个脚本：
- EMO: portrait_animate.py
- AA: animate_anyone.py
- 灵眸: avatar_video.py

示例：
  # EMO（本地图片+本地音频）
  python demo_pipeline.py --mode emo --image ./face.jpg --audio ./speech.mp3 --download

  # AA（URL）
  python demo_pipeline.py --mode aa --image-url https://... --video-url https://... --download

  # 灵眸（模板+文案）
  python demo_pipeline.py --mode lingmou --template-id BSxxxx --text "大家好" --download
"""

import argparse
import os
import subprocess
import sys
from pathlib import Path

from input_validation import resolve_under_cwd, validate_http_https_url

ROOT = Path(__file__).resolve().parent


def run_cmd(cmd):
    print("$", " ".join(cmd))
    proc = subprocess.run(cmd)
    if proc.returncode != 0:
        sys.exit(proc.returncode)


def build_emo_cmd(args):
    cmd = [sys.executable, str(ROOT / "portrait_animate.py")]
    if args.image_url:
        cmd += ["--image-url", args.image_url]
    if args.audio_url:
        cmd += ["--audio-url", args.audio_url]
    if args.image:
        cmd += ["--image", args.image]
    if args.audio:
        cmd += ["--audio", args.audio]
    cmd += ["--ratio", args.ratio, "--style-level", args.style_level, "--output", args.output]
    if args.download:
        cmd += ["--download"]
    return cmd


def build_aa_cmd(args):
    cmd = [sys.executable, str(ROOT / "animate_anyone.py")]
    if args.image_url:
        cmd += ["--image-url", args.image_url]
    if args.video_url:
        cmd += ["--video-url", args.video_url]
    if args.image:
        cmd += ["--image", args.image]
    if args.video:
        cmd += ["--video", args.video]
    cmd += ["--output", args.output]
    if args.download:
        cmd += ["--download"]
    return cmd


def build_lingmou_cmd(args):
    cmd = [
        sys.executable,
        str(ROOT / "avatar_video.py"),
        "--template-id",
        args.template_id,
        "--name",
        args.name,
        "--resolution",
        args.resolution,
        "--fps",
        str(args.fps),
        "--output",
        args.output,
    ]
    if args.watermark:
        cmd += ["--watermark"]
    if args.text_file:
        cmd += ["--text-file", args.text_file]
    else:
        cmd += ["--text", args.text]
    if args.download:
        cmd += ["--download"]
    return cmd


def validate_args(args):
    if args.download:
        resolve_under_cwd(args.output, field="--output")
    if args.mode == "emo":
        if not ((args.image_url or args.image) and (args.audio_url or args.audio)):
            raise SystemExit("EMO 需要 image(+url|file) 和 audio(+url|file)")
        if args.image_url:
            validate_http_https_url(args.image_url, field="--image-url")
        if args.audio_url:
            validate_http_https_url(args.audio_url, field="--audio-url")
    elif args.mode == "aa":
        if not ((args.image_url or args.image) and (args.video_url or args.video)):
            raise SystemExit("AA 需要 image(+url|file) 和 video(+url|file)")
        if args.image_url:
            validate_http_https_url(args.image_url, field="--image-url")
        if args.video_url:
            validate_http_https_url(args.video_url, field="--video-url")
    elif args.mode == "lingmou":
        if not args.template_id:
            raise SystemExit("灵眸模式必须提供 --template-id")
        if not (args.text or args.text_file):
            raise SystemExit("灵眸模式必须提供 --text 或 --text-file")


def print_env_hint(mode):
    print("\n[env-check]")
    if mode in ("emo", "aa"):
        print("- 需要 DASHSCOPE_API_KEY（北京地域）")
    if mode in ("emo", "aa") and not (os.getenv("DASHSCOPE_API_KEY")):
        print("  ! 未检测到 DASHSCOPE_API_KEY")
    if mode == "lingmou":
        print("- 需要 ALIBABA_CLOUD_ACCESS_KEY_ID / ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        if not os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"):
            print("  ! 未检测到 ALIBABA_CLOUD_ACCESS_KEY_ID")
        if not os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"):
            print("  ! 未检测到 ALIBABA_CLOUD_ACCESS_KEY_SECRET")


def main():
    p = argparse.ArgumentParser(description="Human Avatar Demo Pipeline")
    p.add_argument("--mode", required=True, choices=["emo", "aa", "lingmou"])

    # 通用
    p.add_argument("--download", action="store_true")
    p.add_argument("--output", default="demo_output.mp4")

    # EMO / AA 共用输入
    p.add_argument("--image-url")
    p.add_argument("--image")

    # EMO
    p.add_argument("--audio-url")
    p.add_argument("--audio")
    p.add_argument("--ratio", default="1:1", choices=["1:1", "3:4"])
    p.add_argument("--style-level", default="normal", choices=["normal", "calm", "active"])

    # AA
    p.add_argument("--video-url")
    p.add_argument("--video")

    # 灵眸
    p.add_argument("--template-id")
    p.add_argument("--text")
    p.add_argument("--text-file")
    p.add_argument("--name", default="OpenClaw Avatar Video")
    p.add_argument("--resolution", default="720p", choices=["720p", "1080p"])
    p.add_argument("--fps", type=int, default=30, choices=[15, 30])
    p.add_argument("--watermark", action="store_true")

    args = p.parse_args()
    validate_args(args)
    print_env_hint(args.mode)

    if args.mode == "emo":
        cmd = build_emo_cmd(args)
    elif args.mode == "aa":
        cmd = build_aa_cmd(args)
    else:
        cmd = build_lingmou_cmd(args)

    run_cmd(cmd)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
万相图生视频 (wan2.x-i2v)

SECURITY NOTES:
- subprocess: used ONLY for ffmpeg image format conversion. No shell=True.
- OSS credentials: env-only, used ONLY to upload user media to their own
  OSS bucket and generate signed GET URLs for Alibaba Cloud API access.
- All API calls target dashscope.aliyuncs.com (Alibaba Cloud official).
默认模型：wan2.7-i2v

用法:
  # wan2.7-i2v（默认）
  python image_to_video.py --image ./portrait.jpg --prompt "她在草地上微笑跳舞" --download
  python image_to_video.py --image ./portrait.jpg --resolution 1080P --duration 10 --download
  
  # wan2.6-i2v-flash
  python image_to_video.py --image ./portrait.jpg --prompt "..." --model wan2.6-i2v-flash --download
  python image_to_video.py --image-url https://... --prompt "..." --model wan2.6-i2v-flash --resolution 720P --duration 5
  
  # 带音频
  python image_to_video.py --image ./portrait.jpg --audio-url https://.../bgm.mp3 --prompt "..." --download
  
  # 先文生图再图生视频（一条龙）
  python image_to_video.py --t2i-prompt "一位身着汉服的女性站在桃花林中" --prompt "她缓缓转身，花瓣飘落" --download
"""

import argparse
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

import requests

from input_validation import (
    mk_temp_path_for_ffmpeg,
    resolve_under_cwd,
    validate_http_https_url,
)

BASE_URL = os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com")
_OSS_SIGNED_URL_EXPIRES = int(os.environ.get("OSS_SIGNED_URL_EXPIRES", str(3 * 24 * 3600)))


USER_AGENT = "AlibabaCloud-Agent-Skills/alibabacloud-avatar-video"


def _headers(async_mode: bool = True) -> dict:
    key = os.environ.get("DASHSCOPE_API_KEY")
    if not key:
        raise RuntimeError("DASHSCOPE_API_KEY not set")
    h = {
        "Authorization": f"Bearer {key}",
        "Content-Type": "application/json",
        "User-Agent": USER_AGENT,
    }
    if async_mode:
        h["X-DashScope-Async"] = "enable"
    return h


def _wait_task(task_id: str, interval: int = 10, max_wait: int = 600) -> dict:
    url = f"{BASE_URL}/api/v1/tasks/{task_id}"
    start = time.time()
    while time.time() - start < max_wait:
        r = requests.get(url, headers=_headers(async_mode=False), timeout=30)
        r.raise_for_status()
        data = r.json()
        out = data.get("output", {})
        status = out.get("task_status", "UNKNOWN")
        elapsed = int(time.time() - start)
        print(f"  [{elapsed}s] status={status}")
        if status == "SUCCEEDED":
            return out
        if status in ("FAILED", "CANCELED", "UNKNOWN"):
            raise RuntimeError(f"Task failed: {json.dumps(data, ensure_ascii=False)}")
        time.sleep(interval)
    raise TimeoutError(f"Task {task_id} timed out")


def upload_to_oss(local_path: str, expires: int = _OSS_SIGNED_URL_EXPIRES) -> str:
    """Upload to OSS, return signed GET URL (default 3 days)."""
    import oss2
    auth = oss2.Auth(
        os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
        os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
    )
    bucket_name = os.environ["OSS_BUCKET"]
    endpoint = os.environ.get("OSS_ENDPOINT", "oss-cn-beijing.aliyuncs.com")
    endpoint = endpoint.replace("https://", "").replace("http://", "").rstrip("/")
    bucket = oss2.Bucket(auth, f"https://{endpoint}", bucket_name)
    key = f"human-avatar/{Path(local_path).name}"
    print(f"[oss] uploading {Path(local_path).name} …")
    bucket.put_object_from_file(key, local_path)
    url = bucket.sign_url("GET", key, expires)
    print(f"[oss] signed_url ok ({expires//3600}h)")
    return url


def convert_image(src: str) -> str:
    """Convert image to jpg/png if not compatible."""
    p = Path(src)
    if p.suffix.lower() in (".jpg", ".jpeg", ".png", ".bmp", ".webp"):
        return src
    ff = shutil.which("ffmpeg") or "ffmpeg"
    dst = mk_temp_path_for_ffmpeg(".jpg", "i2v_img_")
    subprocess.run([ff, "-y", "-i", src, "-q:v", "2", dst], check=True, capture_output=True)
    return dst


def image_to_video(
    img_url: str,
    prompt: str = "",
    model: str = "wan2.7-i2v",
    resolution: str = "720P",
    duration: int = 5,
    negative_prompt: str = "",
    prompt_extend: bool = True,
    audio_url: str | None = None,
    audio: bool = True,
) -> str:
    """
    Call 万相 image-to-video API. Returns video URL.

    Args:
        img_url:         首帧图像 URL（公网可访问）或 base64
        prompt:          描述视频动作的提示词
        model:           模型名，默认 wan2.7-i2v
        resolution:      分辨率，wan2.7: 720P/1080P（默认 1080P），wan2.6: 480P/720P（默认 720P）
        duration:        视频时长（秒），wan2.7: [2,15]，wan2.6: 3/5/10s，默认 5
        negative_prompt: 反向提示词
        prompt_extend:   是否开启提示词智能改写
        audio_url:       自定义背景音频 URL（可选）
        audio:           是否生成音频（wan2.6/2.5 支持；wan2.7 未传入音频时自动生成）
    """
    endpoint = f"{BASE_URL}/api/v1/services/aigc/video-generation/video-synthesis"

    # 根据模型版本构建不同的请求结构
    is_wan27 = model.startswith("wan2.7")
    
    if is_wan27:
        # wan2.7-i2v 使用 media 数组结构
        media = [{"type": "first_frame", "url": img_url}]
        if audio_url:
            media.append({"type": "driving_audio", "url": audio_url})
        
        inp = {"prompt": prompt, "media": media}
        if negative_prompt:
            inp["negative_prompt"] = negative_prompt
        
        # wan2.7 默认 resolution=1080P, duration 支持 [2,15]
        params = {
            "resolution": resolution,
            "prompt_extend": prompt_extend,
            "duration": duration,
        }
    else:
        # wan2.6 及更早版本使用 img_url 字段
        inp = {"img_url": img_url}
        if prompt:
            inp["prompt"] = prompt
        if negative_prompt:
            inp["negative_prompt"] = negative_prompt
        if audio_url:
            inp["audio_url"] = audio_url
        
        params = {
            "resolution": resolution,
            "prompt_extend": prompt_extend,
            "duration": duration,
        }
        # wan2.7-i2v generates audio by default; set False to disable
        if not audio:
            params["audio"] = False

    payload = {"model": model, "input": inp, "parameters": params}

    print(f"\n[i2v] submit  model={model}  resolution={resolution}  duration={duration}s")
    r = requests.post(endpoint, headers=_headers(async_mode=True), json=payload, timeout=60)
    r.raise_for_status()
    data = r.json()
    task_id = (data.get("output") or {}).get("task_id")
    if not task_id:
        raise RuntimeError(f"No task_id: {json.dumps(data, ensure_ascii=False)}")
    print(f"  task_id={task_id}")

    out = _wait_task(task_id, interval=10, max_wait=600)
    video_url = out.get("video_url")
    if not video_url:
        raise RuntimeError(f"No video_url in result: {out}")
    print(f"\n✅ video_url = {video_url}")
    return video_url


def main():
    p = argparse.ArgumentParser(description="万相图生视频")
    # image input
    img_grp = p.add_mutually_exclusive_group()
    img_grp.add_argument("--image", help="本地图片路径（自动上传 OSS）")
    img_grp.add_argument("--image-url", help="图片公网 URL")
    img_grp.add_argument("--t2i-prompt", help="先文生图再图生视频（调用 text_to_image）")

    p.add_argument("--prompt", default="", help="描述视频内容的提示词")
    p.add_argument("--model", default="wan2.7-i2v",
                   help="模型名 (default: wan2.7-i2v)，可选: wan2.6-i2v-flash, wan2.5-i2v-preview, wan2.2-i2v-plus 等")
    p.add_argument("--t2i-model", default="wan2.2-t2i-flash",
                   help="文生图模型（--t2i-prompt 模式下使用，default: wan2.2-t2i-flash）")
    p.add_argument("--resolution", default=None,
                   help="分辨率，wan2.7: 720P/1080P(默认1080P)，wan2.6: 480P/720P(默认720P)")
    p.add_argument("--duration", type=int, default=None,
                   help="视频时长（秒），wan2.7: [2,15]，wan2.6: 3/5/10，默认 5")
    p.add_argument("--negative-prompt", default="", help="反向提示词")
    p.add_argument("--no-prompt-extend", action="store_true", help="关闭提示词智能改写")
    p.add_argument("--audio-url", default=None, help="背景音频 URL（wan2.6/2.5 支持）")
    p.add_argument("--no-audio", action="store_true", help="生成无声视频")
    p.add_argument("--download", action="store_true", help="下载生成视频到本地")
    p.add_argument("--output", default="i2v_output.mp4", help="输出文件名")
    args = p.parse_args()

    # ── get image URL ────────────────────────────────────────────────────
    img_url = args.image_url
    if img_url:
        img_url = validate_http_https_url(img_url, field="--image-url")
    if args.audio_url:
        args.audio_url = validate_http_https_url(args.audio_url, field="--audio-url")
    tmp_files = []

    if args.t2i_prompt:
        # Generate image first using text_to_image
        print(f"\n[step0] text-to-image  model={args.t2i_model}")
        from text_to_image import text_to_image
        images = text_to_image(
            prompt=args.t2i_prompt,
            model=args.t2i_model,
            size="960*1696",  # 9:16 default for portrait
            n=1,
            prompt_extend=not args.no_prompt_extend,
        )
        img_url = validate_http_https_url(images[0], field="text-to-image URL")
        print(f"  t2i image_url = {img_url}")

    elif args.image:
        converted = convert_image(args.image)
        if converted != args.image:
            tmp_files.append(converted)
        img_url = upload_to_oss(converted)

    if not img_url:
        p.error("需要 --image, --image-url 或 --t2i-prompt 之一")

    # 根据模型版本设置默认值
    is_wan27 = args.model.startswith("wan2.7")
    if args.resolution is None:
        args.resolution = "1080P" if is_wan27 else "720P"
    if args.duration is None:
        args.duration = 5  # 两个版本都支持 5 秒
    
    # 验证参数
    if is_wan27:
        if args.resolution not in ["720P", "1080P"]:
            p.error(f"wan2.7 支持的分辨率: 720P, 1080P，当前: {args.resolution}")
        if args.duration < 2 or args.duration > 15:
            p.error(f"wan2.7 支持的时长: [2,15] 秒，当前: {args.duration}")
    else:
        if args.resolution not in ["480P", "720P"]:
            p.error(f"wan2.6 及更早版本支持的分辨率: 480P, 720P，当前: {args.resolution}")
        if args.duration not in [3, 5, 10]:
            p.error(f"wan2.6 及更早版本支持的时长: 3, 5, 10 秒，当前: {args.duration}")

    # ── submit i2v ───────────────────────────────────────────────────────
    try:
        video_url = image_to_video(
            img_url=img_url,
            prompt=args.prompt,
            model=args.model,
            resolution=args.resolution,
            duration=args.duration,
            negative_prompt=args.negative_prompt,
            prompt_extend=not args.no_prompt_extend,
            audio_url=args.audio_url,
            audio=not args.no_audio,
        )

        if args.download:
            out_path = resolve_under_cwd(args.output, field="--output")
            safe_url = validate_http_https_url(video_url, field="result video URL")
            print(f"Downloading → {out_path} …")
            with urllib.request.urlopen(safe_url, timeout=300) as response:
                with open(out_path, 'wb') as f:
                    f.write(response.read())
            size_kb = out_path.stat().st_size // 1024
            print(f"Saved {out_path} ({size_kb}KB)")

    finally:
        for f in tmp_files:
            try:
                os.unlink(f)
            except Exception:
                pass


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"\nERROR: {e}", file=sys.stderr)
        sys.exit(1)

"""Shared CLI input checks: path confinement, http(s) URLs, readable files."""

from __future__ import annotations

import os
from pathlib import Path
from urllib.parse import urlparse

_MAX_TEXT_FILE_BYTES = 10 * 1024 * 1024


def resolve_under_cwd(user_path: str, *, field: str = "path") -> Path:
    """Resolve a user path under the current working directory; reject traversal."""
    base = Path.cwd().resolve(strict=False)
    candidate = (base / user_path).resolve(strict=False)
    try:
        candidate.relative_to(base)
    except ValueError as e:
        raise ValueError(
            f"{field} must resolve inside the current working directory ({base}): {user_path!r}"
        ) from e
    return candidate


def validate_http_https_url(url: str, *, field: str = "URL") -> str:
    u = url.strip()
    parsed = urlparse(u)
    if parsed.scheme not in ("http", "https"):
        raise ValueError(f"{field} must use http or https scheme: {url!r}")
    if not parsed.netloc:
        raise ValueError(f"{field} must include a host: {url!r}")
    return u


def validate_readable_file(path_str: str, *, field: str = "file") -> Path:
    p = Path(path_str).expanduser()
    if not p.is_file():
        raise ValueError(f"{field} must be an existing regular file: {path_str!r}")
    return p.resolve()


def read_text_file_limited(path_str: str, *, field: str = "--text-file") -> str:
    path = validate_readable_file(path_str, field=field)
    size = path.stat().st_size
    if size > _MAX_TEXT_FILE_BYTES:
        raise ValueError(
            f"{field} exceeds max size ({_MAX_TEXT_FILE_BYTES // (1024 * 1024)} MiB): {path_str!r}"
        )
    return path.read_text(encoding="utf-8").strip()


def mk_temp_path_for_ffmpeg(suffix: str, prefix: str) -> str:
    """Create an empty file with a random name and return its path (fd closed for ffmpeg)."""
    import tempfile

    fd, path = tempfile.mkstemp(suffix=suffix, prefix=prefix)
    try:
        os.close(fd)
    except OSError:
        try:
            os.unlink(path)
        except OSError:
            pass
        raise
    return path

#!/usr/bin/env python3
"""
LivePortrait 人像视频生成 — 两步流水线

SECURITY NOTES:
- subprocess: used ONLY to invoke system ffmpeg (audio/video format
  conversion). All arguments are constructed from user-supplied local
  file paths; no shell=True, no dynamic code execution.
- OSS credentials (ALIBABA_CLOUD_ACCESS_KEY_ID/SECRET): read from env,
  used ONLY to upload media files to the user's own OSS bucket and
  generate time-limited signed GET URLs. Never logged or sent elsewhere.
- All API calls go to dashscope.aliyuncs.com (Alibaba Cloud official).
  Step 1: liveportrait-detect  图像检测（同步）
  Step 2: liveportrait         视频生成（异步）

输入：
  - 人像图片（jpeg/jpg/png/bmp/webp）
  - 音频文件（wav/mp3，<15MB，1s~3min）
    OR 视频文件（自动提取音频）

用法:
  python live_portrait.py --image ./portrait.jpg --audio ./speech.mp3 --download
  python live_portrait.py --image ./portrait.jpg --video ./speech.mp4 --download
  python live_portrait.py --image-url https://... --audio-url https://... --download
  python live_portrait.py --image ./portrait.jpg --audio ./speech.mp3 \\
      --template active --mouth-strength 1.2 --head-strength 0.8 --download
"""

import argparse
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

import requests

from input_validation import (
    mk_temp_path_for_ffmpeg,
    resolve_under_cwd,
    validate_http_https_url,
)

BASE_URL = os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com")
_OSS_SIGNED_URL_EXPIRES = int(os.environ.get("OSS_SIGNED_URL_EXPIRES", str(3 * 24 * 3600)))


# ── helpers ────────────────────────────────────────────────────────────────────

USER_AGENT = "AlibabaCloud-Agent-Skills/alibabacloud-avatar-video"


def _headers(async_mode: bool = False) -> dict:
    key = os.environ.get("DASHSCOPE_API_KEY")
    if not key:
        raise RuntimeError("DASHSCOPE_API_KEY not set")
    h = {
        "Authorization": f"Bearer {key}",
        "Content-Type": "application/json",
        "User-Agent": USER_AGENT,
    }
    if async_mode:
        h["X-DashScope-Async"] = "enable"
    return h


def _wait_task(task_id: str, interval: int = 5, max_wait: int = 600) -> dict:
    url = f"{BASE_URL}/api/v1/tasks/{task_id}"
    start = time.time()
    while time.time() - start < max_wait:
        try:
            r = requests.get(url, headers=_headers(), timeout=30)
            r.raise_for_status()
            data = r.json()
            out = data.get("output", {})
            status = out.get("task_status", "UNKNOWN")
            elapsed = int(time.time() - start)
            print(f"  [{elapsed}s] status={status}")
            if status == "SUCCEEDED":
                return out
            if status in ("FAILED", "CANCELED", "UNKNOWN"):
                raise RuntimeError(f"Task failed: {json.dumps(data, ensure_ascii=False)}")
        except RuntimeError:
            raise
        except Exception as e:
            print(f"  [poll error] {e}")
        time.sleep(interval)
    raise TimeoutError(f"Task {task_id} timed out after {max_wait}s")


def _find_ffmpeg() -> str:
    for p in ["ffmpeg", "/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg"]:
        if shutil.which(p):
            return p
    raise RuntimeError("ffmpeg not found. Install: apt install ffmpeg")


def upload_to_oss(local_path: str, expires: int = _OSS_SIGNED_URL_EXPIRES) -> str:
    """Upload local file to OSS, return signed GET URL (default 3 days)."""
    import oss2
    auth = oss2.Auth(
        os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
        os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
    )
    bucket_name = os.environ["OSS_BUCKET"]
    endpoint = os.environ.get("OSS_ENDPOINT", "oss-cn-beijing.aliyuncs.com")
    endpoint = endpoint.replace("https://", "").replace("http://", "").rstrip("/")
    bucket = oss2.Bucket(auth, f"https://{endpoint}", bucket_name)
    key = f"human-avatar/{Path(local_path).name}"
    print(f"[oss] uploading {Path(local_path).name} …")
    bucket.put_object_from_file(key, local_path)
    url = bucket.sign_url("GET", key, expires)
    print(f"[oss] signed_url ok ({expires // 3600}h)")
    return url


def convert_image(src: str) -> str:
    """Convert image to jpg if format not natively supported (e.g. heic)."""
    p = Path(src)
    if p.suffix.lower() in (".jpg", ".jpeg", ".png", ".bmp", ".webp"):
        return src
    ff = _find_ffmpeg()
    dst = mk_temp_path_for_ffmpeg(".jpg", "lp_img_")
    subprocess.run([ff, "-y", "-i", src, "-q:v", "2", dst], check=True, capture_output=True)
    print(f"[convert] image {p.name} → {Path(dst).name}")
    return dst


def extract_audio_from_video(video_path: str) -> str:
    """
    Extract audio from video file to mp3.
    LivePortrait requires wav or mp3, <15MB, 1s~3min.
    """
    ff = _find_ffmpeg()
    dst = mk_temp_path_for_ffmpeg(".mp3", "lp_audio_")
    print(f"[ffmpeg] extracting audio from {Path(video_path).name} …")
    subprocess.run(
        [ff, "-y", "-i", video_path, "-vn",
         "-ar", "44100", "-ac", "1", "-b:a", "128k",
         "-t", "180",    # cap at 3min
         dst],
        check=True, capture_output=True
    )
    size_mb = Path(dst).stat().st_size / 1024 / 1024
    duration = float(subprocess.run(
        ["ffprobe", "-v", "quiet", "-select_streams", "a:0",
         "-show_entries", "stream=duration", "-of", "csv=p=0", dst],
        capture_output=True, text=True
    ).stdout.strip() or "0")
    print(f"[ffmpeg] extracted: {Path(dst).name}  {size_mb:.1f}MB  {duration:.1f}s")
    if duration < 1:
        raise ValueError(f"Extracted audio is too short ({duration:.1f}s), minimum is 1s")
    if size_mb > 14.5:
        raise ValueError(f"Extracted audio is too large ({size_mb:.1f}MB), max 15MB")
    return dst


def convert_audio(src: str) -> str:
    """Convert audio to mp3 if not wav/mp3."""
    p = Path(src)
    if p.suffix.lower() in (".wav", ".mp3"):
        return src
    ff = _find_ffmpeg()
    dst = mk_temp_path_for_ffmpeg(".mp3", "lp_audio_")
    subprocess.run(
        [ff, "-y", "-i", src, "-vn", "-ar", "44100", "-ac", "1", "-b:a", "128k", dst],
        check=True, capture_output=True
    )
    print(f"[convert] audio {p.name} → {Path(dst).name}")
    return dst


# ── Step 1: detect ─────────────────────────────────────────────────────────────

def lp_detect(image_url: str) -> None:
    """
    POST /face-detect — check image meets LivePortrait requirements (sync).
    Raises ValueError if check fails.
    """
    print(f"\n[step1] liveportrait-detect …")
    r = requests.post(
        f"{BASE_URL}/api/v1/services/aigc/image2video/face-detect",
        headers=_headers(async_mode=False),
        json={"model": "liveportrait-detect", "input": {"image_url": image_url}},
        timeout=30,
    )
    r.raise_for_status()
    data = r.json()
    out = data.get("output", {})
    passed = out.get("pass", False)
    msg = out.get("message", "")
    if passed:
        print(f"  ✅ detect passed")
    else:
        raise ValueError(f"Image failed LivePortrait detect: {msg}")


# ── Step 2: generate ────────────────────────────────────────────────────────────

def lp_generate(
    image_url: str,
    audio_url: str,
    template_id: str = "normal",
    eye_move_freq: float = 0.5,
    video_fps: int = 24,
    mouth_move_strength: float = 1.0,
    paste_back: bool = True,
    head_move_strength: float = 0.7,
) -> str:
    """
    POST /video-synthesis/ — generate LivePortrait video (async).
    Returns video_url.
    """
    print(f"\n[step2] liveportrait generate  template={template_id}  fps={video_fps} …")
    payload = {
        "model": "liveportrait",
        "input": {
            "image_url": image_url,
            "audio_url": audio_url,
        },
        "parameters": {
            "template_id": template_id,
            "eye_move_freq": eye_move_freq,
            "video_fps": video_fps,
            "mouth_move_strength": mouth_move_strength,
            "paste_back": paste_back,
            "head_move_strength": head_move_strength,
        },
    }
    r = requests.post(
        f"{BASE_URL}/api/v1/services/aigc/image2video/video-synthesis/",
        headers=_headers(async_mode=True),
        json=payload,
        timeout=60,
    )
    r.raise_for_status()
    data = r.json()
    task_id = (data.get("output") or {}).get("task_id")
    if not task_id:
        raise RuntimeError(f"No task_id: {json.dumps(data, ensure_ascii=False)}")
    print(f"  task_id={task_id}")

    out = _wait_task(task_id, interval=5, max_wait=600)
    video_url = (out.get("results") or {}).get("video_url")
    if not video_url:
        raise RuntimeError(f"No video_url in result: {out}")
    duration = (out.get("usage") or {}).get("video_duration", "?")
    print(f"  ✅ video_url={video_url}  duration={duration}s")
    return video_url


# ── Main ───────────────────────────────────────────────────────────────────────

def main():
    p = argparse.ArgumentParser(
        description="LivePortrait: portrait image + audio/video → animated portrait video"
    )
    # image
    img_grp = p.add_mutually_exclusive_group()
    img_grp.add_argument("--image", help="本地图片（自动上传 OSS）")
    img_grp.add_argument("--image-url", help="图片公网 URL")

    # audio / video
    audio_grp = p.add_mutually_exclusive_group()
    audio_grp.add_argument("--audio", help="本地音频文件（wav/mp3，自动上传 OSS）")
    audio_grp.add_argument("--audio-url", help="音频公网 URL")
    audio_grp.add_argument("--video", help="本地视频文件，自动提取音频并上传 OSS")

    # generation params
    p.add_argument("--template", default="normal", choices=["normal", "calm", "active"],
                   help="动作模板：normal(默认)/calm(播报)/active(演唱)")
    p.add_argument("--eye-freq", type=float, default=0.5,
                   help="眨眼频率 0~1 (default: 0.5)")
    p.add_argument("--fps", type=int, default=24,
                   help="输出帧率 15~30 (default: 24)")
    p.add_argument("--mouth-strength", type=float, default=1.0,
                   help="嘴部动作幅度 0~1.5 (default: 1.0)")
    p.add_argument("--head-strength", type=float, default=0.7,
                   help="头部动作幅度 0~1 (default: 0.7)")
    p.add_argument("--no-paste-back", action="store_true",
                   help="仅输出人脸区域（不贴回原图）")
    p.add_argument("--skip-detect", action="store_true",
                   help="跳过图像检测步骤")

    p.add_argument("--download", action="store_true", help="下载生成的视频到本地")
    p.add_argument("--output", default="lp_output.mp4", help="输出文件名 (default: lp_output.mp4)")
    args = p.parse_args()

    tmp_files = []

    try:
        # ── image URL ──────────────────────────────────────────────────────
        image_url = args.image_url
        if image_url:
            image_url = validate_http_https_url(image_url, field="--image-url")
        if not image_url:
            if not args.image:
                p.error("需要 --image 或 --image-url")
            converted = convert_image(args.image)
            if converted != args.image:
                tmp_files.append(converted)
            image_url = upload_to_oss(converted)

        # ── audio URL ──────────────────────────────────────────────────────
        audio_url = args.audio_url
        if audio_url:
            audio_url = validate_http_https_url(audio_url, field="--audio-url")
        if not audio_url:
            if args.video:
                extracted = extract_audio_from_video(args.video)
                tmp_files.append(extracted)
                audio_url = upload_to_oss(extracted)
            elif args.audio:
                converted_audio = convert_audio(args.audio)
                if converted_audio != args.audio:
                    tmp_files.append(converted_audio)
                audio_url = upload_to_oss(converted_audio)
            else:
                p.error("需要 --audio, --audio-url 或 --video 之一")

        # ── Step 1: detect ─────────────────────────────────────────────────
        if not args.skip_detect:
            lp_detect(image_url)

        # ── Step 2: generate ───────────────────────────────────────────────
        video_url = lp_generate(
            image_url=image_url,
            audio_url=audio_url,
            template_id=args.template,
            eye_move_freq=args.eye_freq,
            video_fps=args.fps,
            mouth_move_strength=args.mouth_strength,
            paste_back=not args.no_paste_back,
            head_move_strength=args.head_strength,
        )

        print(f"\n✅ Done! video_url = {video_url}")

        if args.download:
            out_path = resolve_under_cwd(args.output, field="--output")
            safe_url = validate_http_https_url(video_url, field="result video URL")
            print(f"Downloading → {out_path} …")
            with urllib.request.urlopen(safe_url, timeout=300) as response:
                with open(out_path, 'wb') as f:
                    f.write(response.read())
            size_kb = out_path.stat().st_size // 1024
            print(f"Saved {out_path} ({size_kb}KB)")

    finally:
        for f in tmp_files:
            try:
                os.unlink(f)
            except Exception:
                pass


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"\nERROR: {e}", file=sys.stderr)
        sys.exit(1)