| | import os |
| | import time |
| | import requests |
| | import json |
| |
|
| | api_key = "Fill in the API Key" |
| |
|
| | prompt = "Description of your video" |
| | model = "video-01" |
| | output_file_name = "output.mp4" |
| |
|
| | def invoke_video_generation()->str: |
| | print("-----------------Submit video generation task-----------------") |
| | url = "https://api.minimaxi.chat/v1/video_generation" |
| | payload = json.dumps({ |
| | "prompt": prompt, |
| | "model": model |
| | }) |
| | headers = { |
| | 'authorization': 'Bearer ' + api_key, |
| | 'content-type': 'application/json', |
| | } |
| |
|
| | response = requests.request("POST", url, headers=headers, data=payload) |
| | print(response.text) |
| | task_id = response.json()['task_id'] |
| | print("Video generation task submitted successfully, task ID.:"+task_id) |
| | return task_id |
| |
|
| | def query_video_generation(task_id: str): |
| | url = "https://api.minimaxi.chat/v1/query/video_generation?task_id="+task_id |
| | headers = { |
| | 'authorization': 'Bearer ' + api_key |
| | } |
| | response = requests.request("GET", url, headers=headers) |
| | status = response.json()['status'] |
| | if status == 'Preparing': |
| | print("...Preparing...") |
| | return "", 'Preparing' |
| | elif status == 'Queueing': |
| | print("...In the queue...") |
| | return "", 'Queueing' |
| | elif status == 'Processing': |
| | print("...Generating...") |
| | return "", 'Processing' |
| | elif status == 'Success': |
| | return response.json()['file_id'], "Finished" |
| | elif status == 'Fail': |
| | return "", "Fail" |
| | else: |
| | return "", "Unknown" |
| |
|
| |
|
| | def fetch_video_result(file_id: str): |
| | print("---------------Video generated successfully, downloading now---------------") |
| | url = "https://api.minimaxi.chat/v1/files/retrieve?file_id="+file_id |
| | headers = { |
| | 'authorization': 'Bearer '+api_key, |
| | } |
| |
|
| | response = requests.request("GET", url, headers=headers) |
| | print(response.text) |
| |
|
| | download_url = response.json()['file']['download_url'] |
| | print("Video download link:" + download_url) |
| | with open(output_file_name, 'wb') as f: |
| | f.write(requests.get(download_url).content) |
| | print("THe video has been downloaded in:"+os.getcwd()+'/'+output_file_name) |
| |
|
| |
|
| | if __name__ == '__main__': |
| | task_id = invoke_video_generation() |
| | print("-----------------Video generation task submitted -----------------") |
| | while True: |
| | time.sleep(10) |
| |
|
| | file_id, status = query_video_generation(task_id) |
| | if file_id != "": |
| | fetch_video_result(file_id) |
| | print("---------------Successful---------------") |
| | break |
| | elif status == "Fail" or status == "Unknown": |
| | print("---------------Failed---------------") |
| | break |