SD / Seedance 视频接入

curl --fail-with-body --max-time 120 \
  https://api.manxiaobai.online/v1/videos \
  -H "Authorization: Bearer $MXB_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "dreamina-seedance-2-0-mini-260615",
    "prompt": "一只蓝色纸船漂浮在平静的水面，柔和晨光，镜头缓慢推进",
    "seconds": "4",
    "resolution": "480p",
    "ratio": "16:9",
    "generate_audio": false,
    "watermark": false
  }'

{
  "id": "task_EXAMPLE",
  "object": "video",
  "model": "dreamina-seedance-2-0-mini-260615",
  "status": "queued",
  "progress": 0
}

import json
import os
from pathlib import Path

content = [{"type": "text", "text": "参考@图片1的主体与@视频1的运镜，结合@音频1的节奏生成短片"}]
for kind, variable in [("image", "MXB_IMAGE_URL"), ("video", "MXB_VIDEO_URL"), ("audio", "MXB_AUDIO_URL")]:
    content.append({
        "type": kind + "_url",
        kind + "_url": {"url": os.environ[variable]},
        "role": "reference_" + kind,
    })
payload = {
    "model": "dreamina-seedance-2-0-mini-260615",
    "seconds": "4",
    "resolution": "480p",
    "ratio": "16:9",
    "generate_audio": True,
    "watermark": False,
    "metadata": {"content": content},
}
Path("seedance-request.json").write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")

import json
import os
from pathlib import Path

payload = {
    "model": "dreamina-seedance-2-0-fast-260128",
    "prompt": "镜头平滑地从首帧过渡到尾帧，保持主体一致",
    "seconds": "5",
    "resolution": "720p",
    "ratio": "16:9",
    "generate_audio": False,
    "watermark": False,
    "metadata": {
        "first_frame_image": os.environ["MXB_FIRST_FRAME_URL"],
        "last_frame_image": os.environ["MXB_LAST_FRAME_URL"],
    },
}
Path("seedance-frames.json").write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")

curl --fail-with-body --max-time 30 \
  "https://api.manxiaobai.online/v1/videos/$MXB_VIDEO_TASK_ID" \
  -H "Authorization: Bearer $MXB_API_KEY"

{
  "id": "task_EXAMPLE",
  "object": "video",
  "status": "completed",
  "progress": 100,
  "usage": {"completion_tokens": 40594, "total_tokens": 40594}
}

curl --fail --max-time 180 \
  "https://api.manxiaobai.online/v1/videos/$MXB_VIDEO_TASK_ID/content" \
  -H "Authorization: Bearer $MXB_API_KEY" \
  --output seedance-result.mp4

# 已设置服务端环境变量 MXB_API_KEY
python mxb-seedance.py --prompt "蓝色纸船在水面上缓缓漂浮"

# 再次执行自动从 seedance-task.json 续查，不重新生成
python mxb-seedance.py

# 新素材任务应指定一个新的状态文件
python mxb-seedance.py --request seedance-request.json --state reference-task.json

# 使用已知任务 ID，只查询与下载
python mxb-seedance.py --task "$MXB_VIDEO_TASK_ID" --output result.mp4

"""Python 3.10+; submit once, resume a saved task, download from this site's API."""
import argparse
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None  # Never forward the API key to another URL.


class ApiError(Exception):
    def __init__(self, status):
        self.status = status
        super().__init__(f"HTTP {status}; check the task and usage logs")


class VideoClient:
    def __init__(self, key, base="https://api.manxiaobai.online"):
        if base.rstrip("/") not in ("https://api.manxiaobai.online", "https://openpix.cc"):
            raise ValueError("Use one of the documented site origins")
        if not key or "\n" in key or "\r" in key:
            raise ValueError("Set MXB_API_KEY in your server environment")
        self.base, self.key = base.rstrip("/"), key
        self.opener = urllib.request.build_opener(NoRedirect())

    def open(self, path, payload=None, timeout=30):
        data = None if payload is None else json.dumps(payload).encode("utf-8")
        headers = {"Authorization": "Bearer " + self.key}
        if data is not None:
            headers["Content-Type"] = "application/json"
        request = urllib.request.Request(self.base + path, data=data, headers=headers)
        try:
            return self.opener.open(request, timeout=timeout)
        except urllib.error.HTTPError as error:
            error.close()
            raise ApiError(error.code) from None

    def json(self, path, payload=None):
        with self.open(path, payload, timeout=120 if payload is not None else 30) as response:
            result = json.load(response)
        if not isinstance(result, dict):
            raise ValueError("Unexpected API response; check logs before submitting again")
        return result

    @staticmethod
    def task_path(task):
        if not isinstance(task, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,200}", task):
            raise ValueError("Invalid task ID")
        return "/v1/videos/" + urllib.parse.quote(task, safe="")

    def wait(self, task, seconds):
        deadline = time.monotonic() + seconds
        while time.monotonic() < deadline:
            try:
                result = self.json(self.task_path(task))
            except (urllib.error.URLError, TimeoutError):
                time.sleep(5)
                continue
            except ApiError as error:
                if error.status not in (429, 500, 502, 503, 504):
                    raise
                time.sleep(5)
                continue
            if result.get("id") != task:
                raise ValueError("Response task ID mismatch; saved task remains available")
            status = str(result.get("status", "")).lower()
            if status in ("completed", "succeeded", "success"):
                return result
            if status in ("failed", "failure", "cancelled", "canceled", "expired"):
                raise ValueError("Task failed; inspect its error and refund in the site logs")
            if status not in ("queued", "pending", "submitted", "in_progress", "running", "processing"):
                raise ValueError("Unknown task state; keep the ID and check site logs")
            time.sleep(5)
        raise TimeoutError("Waiting stopped; run again with the saved state or --task to resume")

    def download(self, task, output):
        temporary = output.with_name(output.name + ".part")
        try:
            with self.open(self.task_path(task) + "/content", timeout=180) as response:
                if response.status != 200:
                    raise ValueError("Unexpected download response")
                first = response.read(32)
                if b"ftyp" not in first:
                    raise ValueError("The content response is not an MP4 file")
                with temporary.open("wb") as stream:
                    stream.write(first)
                    size = len(first)
                    while chunk := response.read(1024 * 1024):
                        size += len(chunk)
                        if size > 512 * 1024 * 1024:
                            raise ValueError("Download exceeds the example's 512 MiB limit")
                        stream.write(chunk)
            temporary.replace(output)
        except BaseException:
            temporary.unlink(missing_ok=True)
            raise


def get_task(client, state, payload):
    if state.exists():
        saved = json.loads(state.read_text(encoding="utf-8"))
        if saved.get("base") != client.base:
            raise ValueError("Saved origin differs; use the same --base as the original task")
        if not saved.get("task_id"):
            raise ValueError("Submission result unknown: check site logs before creating another task")
        client.task_path(saved["task_id"])
        return saved["task_id"]
    # Reserve the state file BEFORE sending POST. A crash must not trigger a duplicate.
    with state.open("x", encoding="utf-8") as stream:
        json.dump({"base": client.base, "status": "submission_unknown"}, stream)
        stream.flush()
        os.fsync(stream.fileno())
    result = client.json("/v1/videos", payload)
    task = result.get("id")
    client.task_path(task)
    print("Accepted task:", task, flush=True)
    temporary = state.with_name(state.name + ".tmp")
    temporary.write_text(json.dumps({"base": client.base, "task_id": task}), encoding="utf-8")
    temporary.replace(state)
    return task


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base", default="https://api.manxiaobai.online")
    parser.add_argument("--prompt", default="A blue paper boat floating gently on calm water")
    parser.add_argument("--model", choices=["dreamina-seedance-2-0-mini-260615", "dreamina-seedance-2-0-fast-260128"], default="dreamina-seedance-2-0-mini-260615")
    parser.add_argument("--request", type=Path, help="UTF-8 JSON request file")
    parser.add_argument("--state", type=Path, default=Path("seedance-task.json"))
    parser.add_argument("--task", help="Existing task ID; never submits a generation")
    parser.add_argument("--output", type=Path, default=Path("seedance-result.mp4"))
    parser.add_argument("--wait", type=int, default=1800, help="Polling deadline in seconds")
    args = parser.parse_args()
    client = VideoClient(os.environ.get("MXB_API_KEY", ""), args.base)
    payload = None
    if not args.task and not args.state.exists():
        payload = json.loads(args.request.read_text(encoding="utf-8-sig")) if args.request else {
            "model": args.model, "prompt": args.prompt, "seconds": "4",
            "resolution": "480p", "ratio": "16:9", "generate_audio": False, "watermark": False,
        }
        if not isinstance(payload, dict):
            raise ValueError("Request file must contain a JSON object")
    task = args.task or get_task(client, args.state, payload)
    client.task_path(task)
    print("Querying task:", task, flush=True)
    client.wait(task, args.wait)
    client.download(task, args.output)
    print("Saved video:", args.output)


if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, ApiError) as error:
        # Do not dump request headers, raw API responses or signed media addresses.
        print(f"Stopped ({type(error).__name__}). Keep the task state and check site logs; do not auto-resubmit.", file=sys.stderr)
        sys.exit(1)