import time
import requests
bearer_token = "jenius_bearer_token" ## 可在 Jenius 账户管理>接口密钥 中查看
headers = {"Authorization": f"Bearer {bearer_token}"}
# --- Step 1: Submit a poster generation task ---
# This function calls the API to start an asynchronous template-based poster generation task.
# Upon success, the API immediately returns a task_id for querying task status.
def make_poster_task() -> str:
"""Submit a template-based poster generation task and return the task ID"""
url = "https://www.jenius.cn/api/plugins/poster/tasks"
payload = {
"prompt": "youqian", # pls check "图像特效生成参考列表" for more selection
"image_url": "https://example.com/user-photo.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 poster generation is asynchronous, you need to check status periodically using task_id.
# Once status becomes "completed", the function returns the final URL; if failed, an exception is raised.
def query_task_status(task_id: str):
url = "https://www.jenius.cn/api/plugins/poster/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["live_poster_url"]
elif status == "failed":
raise Exception(f"Poster generation failed: {response_json}")
# --- Step 3: Save the poster file ---
# This helper function downloads the generated poster from the provided URL and saves it locally.
def save_from_url(poster_url: str):
print(f"Downloading poster file from {poster_url}...")
response = requests.get(poster_url)
response.raise_for_status()
with open("output.mp4", "wb") as f:
f.write(response.content)
print("Poster 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_poster_task()
print(f"Poster generation task submitted successfully, task_id: {task_id}")
final_poster_url = query_task_status(task_id)
print(f"Task completed successfully, poster URL: {final_poster_url}")
save_from_url(final_poster_url)
