1
2
3
4
5
6
7
8
9

def my_function():
    results = process_dressing(
        model_image_path="1.jpg",
        garment_image_path="2.jpg",
        output_dir="./output",
        access_key="YOUR_KEY",
        secret_key="YOUR_SECRET"
    )
    return results

1
2
3
4
5
6

def my_function():
    api = JiMengDressingAPI("YOUR_KEY", "YOUR_SECRET")
    task_id = api.submit_task("1.jpg", "2.jpg")
    result = api.query_result(task_id)
    paths = api.save_result_images(result, "./output")
    return paths

1
2
3

def process_multiple():
    for model, clothes in image_pairs:
        process_dressing(model, clothes, "./output", ak, sk)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from change_clothes import process_dressing  # 导入函数

def my_app():
    results = process_dressing(
        "model.jpg", 
        "clothes.jpg", 
        "./results",
        "your_ak",
        "your_sk"
    )
    print(f"生成了 {len(results)} 张图片")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324

# change_clothes.py
import base64
import json
import time
import requests
from typing import Optional, Dict, List
from pathlib import Path

try:
    from volcengine.visual.VisualService import VisualService

    USE_SDK = True
except ImportError:
    USE_SDK = False
    print("警告: 未安装volcengine SDK，将使用基础HTTP请求")
    print("建议安装: pip install volcengine")


class JiMengDressingAPI:
    """即梦图片换装API封装类"""

    def __init__(self, access_key: str, secret_key: str):
        """
        初始化API客户端

        Args:
            access_key: 火山引擎访问密钥ID
            secret_key: 火山引擎访问密钥
        """
        self.access_key = access_key
        self.secret_key = secret_key

        if USE_SDK:
            # 使用官方SDK
            self.visual_service = VisualService()
            self.visual_service.set_ak(access_key)
            self.visual_service.set_sk(secret_key)
        else:
            self.base_url = "https://visual.volcengineapi.com"
            self.region = "cn-north-1"
            self.service = "cv"

    def _encode_image_to_base64(self, image_path: str) -> str:
        """
        将图片编码为base64字符串

        Args:
            image_path: 图片文件路径

        Returns:
            base64编码的字符串
        """
        with open(image_path, 'rb') as f:
            image_data = f.read()
            base64_str = base64.b64encode(image_data).decode('utf-8')
        return base64_str

    def submit_task(
            self,
            model_image_path: str,
            garment_image_path: str,
            garment_type: str = "full",
            inference_config: Optional[Dict] = None
    ) -> str:
        """
        提交换装任务

        Args:
            model_image_path: 模特图片路径
            garment_image_path: 服装图片路径
            garment_type: 服装类型 ("upper"上衣, "bottom"下衣, "full"上衣+下衣)
            inference_config: 推理配置参数

        Returns:
            任务ID (task_id)
        """
        # 将图片转换为base64
        model_base64 = self._encode_image_to_base64(model_image_path)
        garment_base64 = self._encode_image_to_base64(garment_image_path)

        # 构建请求体
        request_body = {
            "req_key": "dressing_diffusionV2",
            "binary_data_base64": [model_base64, garment_base64],
            "req_image_store_type": 0,
            "garment": {
                "data": [
                    {
                        "type": garment_type
                    }
                ]
            }
        }

        # 添加推理配置
        if inference_config:
            request_body["inference_config"] = inference_config

        if USE_SDK:
            # 使用官方SDK
            try:
                response = self.visual_service.cv_submit_task(request_body)
                result = response
            except Exception as e:
                raise Exception(f"SDK调用失败: {str(e)}")
        else:
            # 使用基础HTTP请求（需要手动签名，此方法可能不work）
            url = f"{self.base_url}?Action=CVSubmitTask&Version=2022-08-31"
            headers = {'Content-Type': 'application/json'}

            try:
                response = requests.post(url, headers=headers, json=request_body, timeout=30)
                result = response.json()
            except Exception as e:
                raise Exception(f"HTTP请求失败: {str(e)}")

        print(f"API响应: {json.dumps(result, ensure_ascii=False, indent=2)}")

        if result.get('code') != 10000:
            error_msg = result.get('message', '未知错误')
            raise Exception(f"提交任务失败 (code: {result.get('code')}): {error_msg}")

        task_id = result['data']['task_id']
        print(f"任务提交成功，任务ID: {task_id}")
        return task_id

    def query_result(
            self,
            task_id: str,
            return_url: bool = True,
            max_retries: int = 30,
            retry_interval: int = 2
    ) -> Dict:
        """
        查询任务结果

        Args:
            task_id: 任务ID
            return_url: 是否返回图片URL
            max_retries: 最大重试次数
            retry_interval: 重试间隔（秒）

        Returns:
            包含图片数据的字典
        """
        req_json = {
            "return_url": return_url,
            "logo_info": {
                "add_logo": False
            }
        }

        request_body = {
            "req_key": "dressing_diffusionV2",
            "task_id": task_id,
            "req_json": json.dumps(req_json)
        }

        # 轮询查询结果
        for i in range(max_retries):
            if USE_SDK:
                try:
                    response = self.visual_service.cv_get_result(request_body)
                    result = response
                except Exception as e:
                    raise Exception(f"SDK查询失败: {str(e)}")
            else:
                url = f"{self.base_url}?Action=CVGetResult&Version=2022-08-31"
                headers = {'Content-Type': 'application/json'}

                try:
                    response = requests.post(url, headers=headers, json=request_body, timeout=30)
                    result = response.json()
                except Exception as e:
                    raise Exception(f"HTTP查询失败: {str(e)}")

            if result.get('code') != 10000:
                error_msg = result.get('message', '未知错误')
                raise Exception(f"查询任务失败 (code: {result.get('code')}): {error_msg}")

            status = result['data']['status']

            if status == 'done':
                print("任务处理完成")
                return result['data']
            elif status == 'not_found':
                raise Exception("任务未找到")
            elif status == 'expired':
                raise Exception("任务已过期")
            elif status in ['in_queue', 'generating']:
                print(f"任务处理中... ({status}) [{i + 1}/{max_retries}]")
                time.sleep(retry_interval)
            else:
                raise Exception(f"未知状态: {status}")

        raise Exception("查询超时")

    def save_result_images(
            self,
            result_data: Dict,
            output_dir: str = "./output"
    ) -> List[str]:
        """
        保存结果图片

        Args:
            result_data: 查询结果返回的数据
            output_dir: 输出目录

        Returns:
            保存的图片路径列表
        """
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)

        saved_paths = []

        # 保存base64格式的图片
        if result_data.get('binary_data_base64'):
            for idx, base64_str in enumerate(result_data['binary_data_base64']):
                image_data = base64.b64decode(base64_str)
                file_path = output_path / f"result_{idx + 1}.png"

                with open(file_path, 'wb') as f:
                    f.write(image_data)

                saved_paths.append(str(file_path))
                print(f"图片已保存: {file_path}")

        # 如果有URL，也可以下载
        if result_data.get('image_urls'):
            for idx, url in enumerate(result_data['image_urls']):
                response = requests.get(url)
                file_path = output_path / f"result_from_url_{idx + 1}.png"

                with open(file_path, 'wb') as f:
                    f.write(response.content)

                saved_paths.append(str(file_path))
                print(f"图片已保存: {file_path}")

        return saved_paths


def process_dressing(
        model_image_path: str,
        garment_image_path: str,
        output_dir: str = "./output",
        access_key: str = "YOUR_ACCESS_KEY",
        secret_key: str = "YOUR_SECRET_KEY",
        garment_type: str = "full",
        inference_config: Optional[Dict] = None
) -> List[str]:
    """
    图片换装处理函数

    Args:
        model_image_path: 模特图片路径（如: 1.jpg）
        garment_image_path: 服装图片路径（如: 2.jpg）
        output_dir: 输出目录
        access_key: 火山引擎访问密钥ID
        secret_key: 火山引擎访问密钥
        garment_type: 服装类型 ("upper", "bottom", "full")
        inference_config: 推理配置

    Returns:
        保存的图片路径列表
    """
    # 创建API客户端
    api = JiMengDressingAPI(access_key, secret_key)

    # 提交任务
    print(f"正在提交换装任务...")
    print(f"模特图片: {model_image_path}")
    print(f"服装图片: {garment_image_path}")

    task_id = api.submit_task(
        model_image_path=model_image_path,
        garment_image_path=garment_image_path,
        garment_type=garment_type,
        inference_config=inference_config
    )

    # 查询结果
    print(f"正在查询任务结果...")
    result_data = api.query_result(task_id)

    # 保存图片
    print(f"正在保存结果图片到: {output_dir}")
    saved_paths = api.save_result_images(result_data, output_dir)

    print(f"\n处理完成！共保存 {len(saved_paths)} 张图片")
    return saved_paths


# 使用示例
if __name__ == "__main__":
    # 配置你的密钥
    ACCESS_KEY = <YOUR_ACCESS_KEY>  # 替换为你的access_key
    SECRET_KEY =  <YOUR_ACCESS_KEY>  # 替换为你的secret_key

    # 调用换装函数
    try:
        result_paths = process_dressing(
            model_image_path="1.jpg",  # 模特图片
            garment_image_path="2.jpg",  # 服装图片
            output_dir="./output",  # 输出目录
            access_key=ACCESS_KEY,
            secret_key=SECRET_KEY,
            garment_type="full",  # 服装类型
            inference_config={  # 可选的推理配置
                "do_sr": False,  # 是否超分
                "keep_head": True,  # 保持头部
                "keep_hand": False,  # 保持手部
                "num_steps": 16  # 推理步数
            }
        )

        print("\n保存的图片路径:")
        for path in result_paths:
            print(f"  - {path}")

    except Exception as e:
        print(f"错误: {e}")