pip install requests
export BLACKPIG_API_KEY="sk-your-api-key"
完整示例
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")
生產環境應將任務 ID 和業務訂單關聯儲存,由後臺任務持續輪詢;不要讓一個 Web 請求長時間阻塞等待影片完成。
