1+ import http .client
2+ import json
3+ import base64
4+ import time
5+
6+ class KlingImageToImage :
7+ def __init__ (self , api_token , api_url ):
8+ """初始化 Kling 图生图转换器
9+
10+ 参数:
11+ api_token: API 密钥
12+ api_url: API 节点地址
13+ """
14+ self .api_url = api_url
15+ self .api_token = api_token
16+ # 初始化 HTTP 连接
17+ self .conn = http .client .HTTPSConnection (self .api_url )
18+ # 设置请求头
19+ self .headers = {
20+ 'Authorization' : f'Bearer { self .api_token } ,
21+ 'Content-Type' : 'application/json'
22+ }
23+
24+ @staticmethod
25+ def get_image_base64 (image_path ):
26+ """将图片转换为 base64 编码形式
27+
28+ 参数:
29+ image_path: 图片路径
30+ 返回:
31+ base64 编码后的图片字符串
32+ """
33+ with open (image_path , "rb" ) as image_file :
34+ return base64 .b64encode (image_file .read ()).decode ("utf-8" )
35+
36+ def _kling_generate_image (self , model_name , prompt , image , image_reference ,
37+ image_fidelity = 0.5 , human_fidelity = 0.5 ,
38+ output_format = "png" , n = 1 , aspect_ratio = "16:9" , callback_url = "" ):
39+ """使用 kling 进行图生图
40+
41+ 参数:
42+ model_name: str, 模型名称，可选择 kling-v1-5 或 kling-v1
43+ prompt: str, 文本提示词
44+ image: str, 参考图片，base64编码或URL
45+ image_reference: str, 参考图片类型，可选值：subject（角色特征参考）, face（人物长相参考）
46+ image_fidelity: float, 参考图片强度，取值范围：[0,1]，数值越大参考强度越大
47+ human_fidelity: float, 面部参考强度，取值范围：[0,1]，数值越大参考强度越大
48+ output_format: str, 输出格式：png 或 jpg
49+ n: int, 生成数量 [1, 9]
50+ aspect_ratio: str, 输出比例：16:9, 9:16, 1:1, 4:3, 3:4, 3:2, 2:3
51+ callback_url: str, 回调地址，可以用于 webhook 等通知场景
52+ 返回:
53+ task_id: 生成任务的 id
54+ """
55+ # 构建请求体，请求的核心参数
56+ payload = {
57+ "model_name" : model_name ,
58+ "prompt" : prompt ,
59+ "image" : image ,
60+ "image_reference" : image_reference ,
61+ "image_fidelity" : image_fidelity ,
62+ "human_fidelity" : human_fidelity ,
63+ "output_format" : output_format ,
64+ "n" : n ,
65+ "aspect_ratio" : aspect_ratio ,
66+ "callback_url" : callback_url
67+ }
68+
69+ # 发送 POST 请求，提交图像生成任务
70+ self .conn .request ("POST" , "/kling/v1/images/generations" , json .dumps (payload ), self .headers )
71+ # 获取响应
72+ res = self .conn .getresponse ()
73+ # 读取响应内容并解析为 JSON
74+ json_data = json .loads (res .read ().decode ("utf-8" ))
75+ # print(json_data)
76+ if 'code' in json_data and json_data ['code' ] == 0 :
77+ # 成功则返回提交的任务 id
78+ return json_data ['data' ]['task_id' ]
79+ else :
80+ # 失败则返回错误信息
81+ raise Exception (f"API调用失败：{ json_data ['message' ]} )
82+
83+ def _query_kling_image_url (self , task_id ):
84+ """使用查询接口获取生成图像 url
85+
86+ 参数:
87+ task_id: 生成任务的 id
88+ 返回:
89+ image_url: 图像 url，任务未完成时返回 None
90+ """
91+ # 构建查询路径
92+ query_path = f"/kling/v1/images/generations/{ task_id }
93+
94+ # 发送 GET 请求，查询图像生成任务状态
95+ self .conn .request ("GET" , query_path , None , self .headers )
96+ # 获取响应
97+ res = self .conn .getresponse ()
98+ # 读取响应内容并解析为 JSON
99+ json_data = json .loads (res .read ().decode ("utf-8" ))
100+ # 如果任务状态为成功，则返回图像 url
101+ if json_data ['data' ]['task_status' ] == "succeed" :
102+ image_urls = [image ['url' ] for image in json_data ['data' ]['task_result' ]['images' ]]
103+ return image_urls
104+ else :
105+ return None
106+
107+ def generate_image (self , model_name , prompt , image ,
108+ image_reference = "subject" , image_fidelity = 0.5 , human_fidelity = 0.5 ,
109+ output_format = "png" , n = 1 , aspect_ratio = "16:9" , callback_url = "" , timeout = 120 ):
110+ """实现功能，直接根据预设的参数返回生成图像的 url
111+
112+ 参数:
113+ model_name: str, 模型名称，可选择 kling-v1-5 或 kling-v1
114+ prompt: str, 文本提示词
115+ image: str, 参考图片的URL或本地文件路径
116+ image_reference: str, 参考图片类型，可选值：subject（角色特征参考）, face（人物长相参考）
117+ image_fidelity: float, 参考图片强度，取值范围：[0,1]，数值越大参考强度越大
118+ human_fidelity: float, 面部参考强度，取值范围：[0,1]，数值越大参考强度越大
119+ output_format: str, 输出格式：png 或 jpg
120+ n: int, 生成数量 [1, 9]
121+ aspect_ratio: str, 输出比例：16:9, 9:16, 1:1, 4:3, 3:4, 3:2, 2:3
122+ callback_url: str, 回调地址，可以用于 webhook 等通知场景
123+ timeout: int, 等待生成完成的超时时间（秒）
124+ 返回:
125+ image_url: 图像 url
126+ """
127+ # 处理图像输入 - 自动判断是URL还是本地文件路径
128+ if image .startswith (('http://' , 'https://' , 'ftp://' )):
129+ # 如果是URL，直接使用
130+ image_data = image
131+ else :
132+ # 否则当作本地文件路径处理，转换为base64
133+ try :
134+ image_data = KlingImageToImage .get_image_base64 (image )
135+ except Exception as e :
136+ raise ValueError (f"无法读取图像文件: { str (e )} )
137+
138+ # 调用生成图像 api 提交图像生成任务，返回获取 task_id
139+ task_id = self ._kling_generate_image (
140+ model_name , prompt , image_data , image_reference ,
141+ image_fidelity , human_fidelity , output_format , n , aspect_ratio , callback_url
142+ )
143+
144+ start_time = time .time ()
145+
146+ # 轮询等待生成完成
147+ while True :
148+ # 根据 task_id 调用查询图像api 查看图像生成任务是否完成
149+ image_url = self ._query_kling_image_url (task_id )
150+ # 如果图像生成任务完成，则返回图像 url
151+ if image_url is not None :
152+ return image_url
153+ # 如果轮询超时，则返回 None
154+ if time .time () - start_time > timeout :
155+ print (f"请求达到 { timeout } )
156+ return None
157+ # 轮询间隔 1 秒
158+ time .sleep (1 )
159+ print (f"等待图像生成，{ int (time .time () - start_time )} , flush = True )
160+
161+
162+ # 使用示例
163+ if __name__ == "__main__" :
164+ API_URL = "www.dmxapi.cn" # API 节点地址
165+ DMX_API_TOKEN = "sk-XXXXXXXXXXXXXX" # API 密钥
166+
167+ # 创建图生图转换器实例
168+ kling_image_to_image = KlingImageToImage (api_token = DMX_API_TOKEN , api_url = API_URL )
169+
170+ # 生成图像
171+ image_urls = kling_image_to_image .generate_image (
172+ model_name = "kling-v1-5" , # [必选]模型名称 参数 kling-v1 或者 kling-v1-5 （注意 v2 没有图生图能力）
173+ prompt = "请生成这张照片的梵高风格迁移后的图像" , # [必选]文本提示词
174+ image = "/Users/dmxapi/Desktop/dmxapi.png" , # [必选]参考图片路径 参数可以是 图片的URL https://image.jpg 或者 图片的本地路径 即可
175+ image_reference = "subject" , # [必选]参考图片类型，可选值：subject（角色特征参考）, face（人物长相参考）
176+ # image_fidelity=0.5, # 参考图片强度 参数范围 0-1 默认 0.5
177+ # human_fidelity=0.5, # 面部参考强度 参数范围 0-1 默认 0.5
178+ # output_format="png", # 输出格式 参数范围 png 或者 jpg
179+ # n=1, # 生成数量 参数范围 1-9 默认 1
180+ # aspect_ratio="16:9", # 输出比例 参数范围 16:9, 9:16, 1:1, 4:3, 3:4, 3:2, 2:3 默认 16:9
181+ # callback_url="", # 回调地址
182+ # timeout=120 # 等待超时时间
183+ )
184+
185+ print (image_urls )
0 commit comments