@@ -20,7 +20,7 @@ def __init__(self, api_token, api_url):
20
20
'Content-Type' : 'application/json'
21
21
}
22
22
23
- def kling_generate_image (self , model_name , prompt , negative_prompt , output_format , n , aspect_ratio , callback_url ):
23
+ def _kling_generate_image (self , model_name , prompt , negative_prompt , output_format , n , aspect_ratio , callback_url ):
24
24
"""使用 kling 生成图像
25
25
26
26
参数:
@@ -55,7 +55,7 @@ def kling_generate_image(self, model_name, prompt, negative_prompt, output_forma
55
55
# 从 json 文件中返回任务 ID
56
56
return json_data ['data' ]['task_id' ]
57
57
58
- def query_kling_image_url (self , task_id ):
58
+ def _query_kling_image_url (self , task_id ):
59
59
"""使用查询接口获取生成图像 url
60
60
61
61
输入参数:
@@ -77,11 +77,12 @@ def query_kling_image_url(self, task_id):
77
77
json_data = json .loads (res .read ().decode ("utf-8" ))
78
78
# 如果任务状态为成功,则返回图像 url
79
79
if json_data ['data' ]['task_status' ] == "succeed" :
80
- return json_data ['data' ]['task_result' ]['images' ][0 ]['url' ]
80
+ image_urls = [image ['url' ] for image in json_data ['data' ]['task_result' ]['images' ]]
81
+ return image_urls
81
82
else :
82
83
return None
83
84
84
- def generate_image (self , model_name , prompt , negative_prompt = "" , output_format = "png" , n = 1 , aspect_ratio = "16:9" , callback_url = "" ):
85
+ def generate_image (self , model_name , prompt , negative_prompt = "" , output_format = "png" , n = 1 , aspect_ratio = "16:9" , callback_url = "" , timeout = 60 ):
85
86
"""实现功能,直接根据预设的参数返回生成图像的 url
86
87
87
88
参数:
@@ -96,14 +97,12 @@ def generate_image(self, model_name, prompt, negative_prompt="", output_format="
96
97
image_url: 图像 url
97
98
"""
98
99
# 调用生成图像 api 提交图像生成任务,返回获取 task_id。
99
- task_id = self .kling_generate_image (model_name , prompt , negative_prompt , output_format , n , aspect_ratio , callback_url )
100
+ task_id = self ._kling_generate_image (model_name , prompt , negative_prompt , output_format , n , aspect_ratio , callback_url )
100
101
start_time = time .time ()
101
- # 队列等待超时时间
102
- timeout = 60
103
102
# 轮询等待生成完成
104
103
while True :
105
104
# 根据 task_id 调用查询图像api 查看图像生成任务是否完成。
106
- image_url = self .query_kling_image_url (task_id )
105
+ image_url = self ._query_kling_image_url (task_id )
107
106
# 如果图像生成任务完成,则返回图像 url
108
107
if image_url is not None :
109
108
return image_url
@@ -119,7 +118,7 @@ def generate_image(self, model_name, prompt, negative_prompt="", output_format="
119
118
# 使用示例
120
119
if __name__ == "__main__" :
121
120
API_URL = "www.dmxapi.cn" # API 节点地址
122
- DMX_API_TOKEN = "sk-XXXXXXXXXXX " # API 密钥
121
+ DMX_API_TOKEN = "sk-XXXXXXXXXXXXXX " # API 密钥
123
122
124
123
# 创建图像生成器实例
125
124
kling_text_to_image = KlingTextToImage (api_token = DMX_API_TOKEN , api_url = API_URL )
@@ -132,7 +131,8 @@ def generate_image(self, model_name, prompt, negative_prompt="", output_format="
132
131
# output_format="png", # 输出格式:png 或 jpg
133
132
# n=1, # int, 生成数量 [1, 9]
134
133
# aspect_ratio="16:9", # 输出比例:16:9, 9:16, 1:1, 4:3, 3:4, 3:2, 2:3
135
- # callback_url="" # 回调地址,可以用于 webhook 等通知场景
134
+ # callback_url="", # 回调地址,可以用于 webhook 等通知场景
135
+ # timeout=120 # 队列等待超时时间
136
136
)
137
137
138
138
print (image_url )
0 commit comments