SIUBIU commited on
Commit
fefd44d
1 Parent(s): e00eb5c

Update kolorsvton1024.py

Browse files
Files changed (1) hide show
  1. kolorsvton1024.py +171 -170
kolorsvton1024.py CHANGED
@@ -1,171 +1,172 @@
1
- import time
2
- import jwt
3
- import requests
4
- import json
5
- import base64
6
- import os
7
- def image_to_base64(image_path):
8
- """
9
- 将图片文件转换为Base64编码格式
10
- :param image_path: 图片文件路径
11
- :return: Base64编码字符串
12
- """
13
- with open(image_path, "rb") as image_file:
14
- return base64.b64encode(image_file.read()).decode('utf-8')
15
-
16
- ak = "c1748494bc5d42ed8db2b2d24ceb1a2b" # 填写access key
17
- sk = "1cf0bbef768746f79d30ef13630b393b" # 填写secret key
18
-
19
- def encode_jwt_token(ak, sk):
20
- headers = {
21
- "alg": "HS256",
22
- "typ": "JWT"
23
- }
24
- payload = {
25
- "iss": ak,
26
- "exp": int(time.time()) + 1800, # 有效时间,此处示例代表当前时间+1800s(30min)
27
- "nbf": int(time.time()) - 5 # 开始生效的时间,此处示例代表当前时间-5秒
28
- }
29
- token = jwt.encode(payload, sk, headers=headers)
30
- return token
31
-
32
-
33
-
34
-
35
- # 图片及回调配置
36
- # HUMAN_IMAGE = image_to_base64(r"F:\codetest24\0903_lf\vton\people.jpg") # 替换为人物图片的Base64或URL
37
- # CLOTH_IMAGE = image_to_base64(r"F:\codetest24\0903_lf\vton\hanfu.jpg") # 替换为服饰图片的Base64或URL
38
- # CALLBACK_URL = "" # 可选,任务结果回调通知URL
39
- SAVE_DIRECTORY = "downloads/kolor"
40
- # ========== 步骤 1:创建虚拟试穿任务 ==========
41
-
42
- def create_virtual_tryon_task(humen, cloth):
43
- """
44
- 创建虚拟试穿任务
45
- """
46
-
47
- # ========== 配置参数 ==========
48
-
49
- # API 请求的基础配置
50
- api_token = encode_jwt_token(ak, sk)
51
- print(api_token) # 打印生成的API_TOKEN
52
- BASE_URL = "https://api.klingai.com/v1/images/kolors-virtual-try-on"
53
-
54
- # 请求头
55
- HEADERS = {
56
- "Content-Type": "application/json",
57
- "Authorization": f"Bearer {api_token}"
58
- }
59
- # 请求体数据
60
- data = {
61
- "model_name": "kolors-virtual-try-on-v1",
62
- "human_image": image_to_base64(humen),
63
- "cloth_image": image_to_base64(cloth),
64
- }
65
-
66
- # 发起POST请求创建任务
67
- response = requests.post(BASE_URL, headers=HEADERS, data=json.dumps(data))
68
-
69
- # 处理响应
70
- if response.status_code == 200:
71
- result = response.json()
72
- task_id = result['data']['task_id']
73
- print(f"创建任务成功,任务ID: {task_id}")
74
- return task_id
75
- else:
76
- print(f"创建任务失败: {response.status_code}, {response.text}")
77
- return None
78
-
79
- # ========== 步骤 2:查询单个虚拟试穿任务状态并保存图片 ==========
80
-
81
- def query_virtual_tryon_task(task_id, i):
82
- """
83
- 根据任务ID查询任务状态,成功后保存生成的图片
84
- """
85
- # 请求URL
86
- url = f"{BASE_URL}/{task_id}"
87
-
88
- # 发起GET请求查询任务状态
89
- response = requests.get(url, headers=HEADERS)
90
-
91
- # 处理响应
92
- if response.status_code == 200:
93
- result = response.json()
94
- task_status = result['data']['task_status']
95
- print(f"任务状态: {task_status}")
96
-
97
- # 如果任务成功,下载并保存生成的图片
98
- if task_status == 'succeed':
99
- images = result['data']['task_result']['images']
100
- for image in images:
101
- image_url = image['url']
102
- image_index = i
103
- save_image(image_url, image_index)
104
-
105
- # 返回任务状态
106
- return task_status
107
- else:
108
- print(f"查询任务失败: {response.status_code}, {response.text}")
109
- return None
110
-
111
- # ========== 步骤 3:下载并保存图片 ==========
112
-
113
- def save_image(image_url, image_index):
114
- """
115
- 根据图片URL下载并保存到本地
116
- """
117
- response = requests.get(image_url)
118
-
119
- if response.status_code == 200:
120
- # 保存图片到本地
121
- image_path = os.path.join(SAVE_DIRECTORY, f"kolor_{image_index}.png")
122
- with open(image_path, "wb") as file:
123
- file.write(response.content)
124
- print(f"图片已保存到: {image_path}")
125
- else:
126
- print(f"下载图片失败: {response.status_code}, {response.text}")
127
-
128
- # ========== 步骤 4:查询任务列表 ==========
129
-
130
- def query_task_list(page_num=1, page_size=30):
131
- """
132
- 查询任务列表
133
- """
134
- # 查询参数
135
- params = {
136
- "pageNum": page_num,
137
- "pageSize": page_size
138
- }
139
-
140
- # 发起GET请求查询任务列表
141
- response = requests.get(BASE_URL, headers=HEADERS, params=params)
142
-
143
- # 处理响应
144
- if response.status_code == 200:
145
- result = response.json()
146
- print("任务列表:")
147
- for task in result['data']:
148
- print(f"任务ID: {task['task_id']}, 状态: {task['task_status']}, 创建时间: {task['created_at']}")
149
- else:
150
- print(f"查询任务列表失败: {response.status_code}, {response.text}")
151
-
152
- # ========== 主流程 ==========
153
-
154
- def kolor_vton(humen, cloth, i):
155
- # 1. 创建虚拟试穿任务
156
- task_id = create_virtual_tryon_task(humen, cloth)
157
-
158
- # 如果任务创建成功,继续执行查询步骤
159
- if task_id:
160
- # 2. 定期查询该任务状态,直到任务完成
161
- while True:
162
- status = query_virtual_tryon_task(task_id, i)
163
- if status in ['succeed', 'failed']:
164
- break
165
- print("任务正在处理中,等待5秒后重试...")
166
- time.sleep(5) # 等待5秒后重试
167
-
168
- # 3. 查询任务列表
169
- query_task_list(page_num=1, page_size=10)
170
-
 
171
  # kolor_vton("uploads/user_image.jpg", "uploads/user_image.jpg", 1)
 
1
+ import time
2
+ import jwt
3
+ import requests
4
+ import json
5
+ import base64
6
+ import os
7
+ def image_to_base64(image_path):
8
+ """
9
+ 将图片文件转换为Base64编码格式
10
+ :param image_path: 图片文件路径
11
+ :return: Base64编码字符串
12
+ """
13
+ with open(image_path, "rb") as image_file:
14
+ return base64.b64encode(image_file.read()).decode('utf-8')
15
+
16
+ ak = "c1748494bc5d42ed8db2b2d24ceb1a2b" # 填写access key
17
+ sk = "1cf0bbef768746f79d30ef13630b393b" # 填写secret key
18
+
19
+ def encode_jwt_token(ak, sk):
20
+ headers = {
21
+ "alg": "HS256",
22
+ "typ": "JWT"
23
+ }
24
+ payload = {
25
+ "iss": ak,
26
+ "exp": int(time.time()) + 1800, # 有效时间,此处示例代表当前时间+1800s(30min)
27
+ "nbf": int(time.time()) - 5 # 开始生效的时间,此处示例代表当前时间-5秒
28
+ }
29
+ token = jwt.encode(payload, sk, headers=headers)
30
+ return token
31
+
32
+
33
+
34
+
35
+ # 图片及回调配置
36
+ # HUMAN_IMAGE = image_to_base64(r"F:\codetest24\0903_lf\vton\people.jpg") # 替换为人物图片的Base64或URL
37
+ # CLOTH_IMAGE = image_to_base64(r"F:\codetest24\0903_lf\vton\hanfu.jpg") # 替换为服饰图片的Base64或URL
38
+ # CALLBACK_URL = "" # 可选,任务结果回调通知URL
39
+ SAVE_DIRECTORY = "downloads/kolor"
40
+ BASE_URL = "https://api.klingai.com/v1/images/kolors-virtual-try-on"
41
+ # ========== 步骤 1:创建虚拟试穿任务 ==========
42
+
43
+ def create_virtual_tryon_task(humen, cloth):
44
+ """
45
+ 创建虚拟试穿任务
46
+ """
47
+
48
+ # ========== 配置参数 ==========
49
+
50
+ # API 请求的基础配置
51
+ api_token = encode_jwt_token(ak, sk)
52
+ print(api_token) # 打印生成的API_TOKEN
53
+
54
+
55
+ # 请求头
56
+ HEADERS = {
57
+ "Content-Type": "application/json",
58
+ "Authorization": f"Bearer {api_token}"
59
+ }
60
+ # 请求体数据
61
+ data = {
62
+ "model_name": "kolors-virtual-try-on-v1",
63
+ "human_image": image_to_base64(humen),
64
+ "cloth_image": image_to_base64(cloth),
65
+ }
66
+
67
+ # 发起POST请求创建任务
68
+ response = requests.post(BASE_URL, headers=HEADERS, data=json.dumps(data))
69
+
70
+ # 处理响应
71
+ if response.status_code == 200:
72
+ result = response.json()
73
+ task_id = result['data']['task_id']
74
+ print(f"创建任务成功,任务ID: {task_id}")
75
+ return task_id
76
+ else:
77
+ print(f"创建任务失败: {response.status_code}, {response.text}")
78
+ return None
79
+
80
+ # ========== 步骤 2:查询单个虚拟试穿任务状态并保存图片 ==========
81
+
82
+ def query_virtual_tryon_task(task_id, i):
83
+ """
84
+ 根据任务ID查询任务状态,成功后保存生成的图片
85
+ """
86
+ # 请求URL
87
+ url = f"{BASE_URL}/{task_id}"
88
+
89
+ # 发起GET请求查询任务状态
90
+ response = requests.get(url, headers=HEADERS)
91
+
92
+ # 处理响应
93
+ if response.status_code == 200:
94
+ result = response.json()
95
+ task_status = result['data']['task_status']
96
+ print(f"任务状态: {task_status}")
97
+
98
+ # 如果任务成功,下载并保存生成的图片
99
+ if task_status == 'succeed':
100
+ images = result['data']['task_result']['images']
101
+ for image in images:
102
+ image_url = image['url']
103
+ image_index = i
104
+ save_image(image_url, image_index)
105
+
106
+ # 返回任务状态
107
+ return task_status
108
+ else:
109
+ print(f"查询任务失败: {response.status_code}, {response.text}")
110
+ return None
111
+
112
+ # ========== 步骤 3:下载并保存图片 ==========
113
+
114
+ def save_image(image_url, image_index):
115
+ """
116
+ 根据图片URL下载并保存到本地
117
+ """
118
+ response = requests.get(image_url)
119
+
120
+ if response.status_code == 200:
121
+ # 保存图片到本地
122
+ image_path = os.path.join(SAVE_DIRECTORY, f"kolor_{image_index}.png")
123
+ with open(image_path, "wb") as file:
124
+ file.write(response.content)
125
+ print(f"图片已保存到: {image_path}")
126
+ else:
127
+ print(f"下载图片失败: {response.status_code}, {response.text}")
128
+
129
+ # ========== 步骤 4:查询任务列表 ==========
130
+
131
+ def query_task_list(page_num=1, page_size=30):
132
+ """
133
+ 查询任务列表
134
+ """
135
+ # 查询参数
136
+ params = {
137
+ "pageNum": page_num,
138
+ "pageSize": page_size
139
+ }
140
+
141
+ # 发起GET请求查询任务列表
142
+ response = requests.get(BASE_URL, headers=HEADERS, params=params)
143
+
144
+ # 处理响应
145
+ if response.status_code == 200:
146
+ result = response.json()
147
+ print("任务列表:")
148
+ for task in result['data']:
149
+ print(f"任务ID: {task['task_id']}, 状态: {task['task_status']}, 创建时间: {task['created_at']}")
150
+ else:
151
+ print(f"查询任务列表失败: {response.status_code}, {response.text}")
152
+
153
+ # ========== 主流程 ==========
154
+
155
+ def kolor_vton(humen, cloth, i):
156
+ # 1. 创建虚拟试穿任务
157
+ task_id = create_virtual_tryon_task(humen, cloth)
158
+
159
+ # 如果任务创建成功,继续执行查询步骤
160
+ if task_id:
161
+ # 2. 定期查询该任务状态,直到任务完成
162
+ while True:
163
+ status = query_virtual_tryon_task(task_id, i)
164
+ if status in ['succeed', 'failed']:
165
+ break
166
+ print("任务正在处理中,等待5秒后重试...")
167
+ time.sleep(5) # 等待5秒后重试
168
+
169
+ # 3. 查询任务列表
170
+ query_task_list(page_num=1, page_size=10)
171
+
172
  # kolor_vton("uploads/user_image.jpg", "uploads/user_image.jpg", 1)