import time
import requests
bearer_token = "jenius_bearer_token" ## 可在 Jenius 账户管理>接口密钥 中查看
headers = {"Authorization": f"Bearer {bearer_token}"}
# --- Step 1: Submit a video generation task ---
# This function calls the API to start an asynchronous template-based video generation task.
# Upon success, the API immediately returns a task_id for querying task status.
def make_video_task() -> str:
"""Submit a template-based video generation task and return the task ID"""
url = "https://www.jenius.cn/api/plugins/video/tasks"
payload = {
# 'scene' specifies the video scene.
"scene": "celebrity",
"image_url": "https://statics.jenius.cn/vector/202601/db3d1a21-1022-400d-82e9-c6ae7d132105/eddc58eb0ac636d0997e1880ce286a10.png",
## some scene may require mulitple images, use list of "image_urls" instead of "image_url"
"image_urls": [
"https://example.com/photo1.jpg",
"https://example.com/photo2.jpg",
"https://example.com/photo3.jpg"
]
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
task_id = response.json()["data"]["uuid"]
return task_id
# --- Step 2: Poll task status ---
# Since video generation is asynchronous, you need to check status periodically using task_id.
# Once status becomes "completed", the function returns the video URL; if failed, an exception is raised.
def query_task_status(task_id: str):
url = "https://www.jenius.cn/api/plugins/video/tasks/{task_id}"
while True:
# A reasonable polling interval is recommended to avoid excessive requests.
time.sleep(10)
response = requests.get(url, headers=headers)
response.raise_for_status()
response_json = response.json()["data"]
status = response_json["status"]
print(f"Current task status: {status}")
if status == "completed":
return response_json["video_url"]
elif status == "failed":
raise Exception(f"Video generation failed: {response_json}")
# --- Step 3: Save the video file ---
# This helper function downloads the generated video from the provided URL and saves it locally.
def save_video_from_url(video_url: str):
print(f"Downloading video from {video_url}...")
response = requests.get(video_url)
response.raise_for_status()
with open("output.mp4", "wb") as f:
f.write(response.content)
print("Video successfully saved as output.mp4")
# --- Main process: Full workflow ---
# Execute the entire flow in the order: submit -> poll -> save.
if __name__ == "__main__":
task_id = make_video_task()
print(f"Video generation task submitted successfully, task_id: {task_id}")
final_video_url = query_task_status(task_id)
print(f"Task completed successfully, video URL: {final_video_url}")
save_video_from_url(final_video_url)
