---
title: Python 示例
description: 使用 requests 建立、輪詢並下載影片任務。
icon: file-code-2
---

安裝依賴：

```bash
pip install requests
export BLACKPIG_API_KEY="sk-your-api-key"
```

## 完整示例

```python
import os
import time
from pathlib import Path

import requests

BASE_URL = "https://api.heizhu.lol"
API_KEY = os.environ["BLACKPIG_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


def raise_for_api_error(response: requests.Response) -> None:
    if response.ok:
        return
    raise RuntimeError(f"HTTP {response.status_code}: {response.text}")


def create_video() -> dict:
    response = requests.post(
        f"{BASE_URL}/v1/videos",
        headers=HEADERS,
        json={
            "model": "L-stable-seedance-2-0-933-720p",
            "prompt": "电影感近景，人物在雨夜街道缓慢回头，镜头平稳推进",
            "duration": 1,
            "video_duration": 15,
            "aspect_ratio": "9:16",
            "resolution": "720p",
            "image_urls": ["https://example.com/reference.jpg"],
        },
        timeout=60,
    )
    raise_for_api_error(response)
    return response.json()


def wait_for_video(task_id: str) -> dict:
    while True:
        response = requests.get(
            f"{BASE_URL}/v1/videos/{task_id}",
            headers=HEADERS,
            timeout=30,
        )
        raise_for_api_error(response)
        task = response.json()

        print(task["status"], f'{task.get("progress", 0)}%')

        if task["status"] == "completed":
            return task
        if task["status"] == "failed":
            message = task.get("error", {}).get("message", "Video generation failed")
            raise RuntimeError(message)

        time.sleep(5)


def download_video(task_id: str, output_path: Path) -> None:
    with requests.get(
        f"{BASE_URL}/v1/videos/{task_id}/content",
        headers=HEADERS,
        timeout=300,
        stream=True,
    ) as response:
        raise_for_api_error(response)
        with output_path.open("wb") as output:
            for chunk in response.iter_content(chunk_size=1024 * 1024):
                output.write(chunk)


task = create_video()
print("Created:", task["id"])

wait_for_video(task["id"])
download_video(task["id"], Path("result.mp4"))
print("Saved: result.mp4")
```

<Note>
生產環境應將任務 ID 和業務訂單關聯儲存，由後臺任務持續輪詢；不要讓一個 Web 請求長時間阻塞等待影片完成。
</Note>