import requests import time from WxMini.Milvus.Config.MulvusConfig import * # DashScope API 配置 DASHSCOPE_API_KEY = MODEL_API_KEY # 替换为你的 API Key CREATE_TASK_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/image-generation/generation" GET_TASK_RESULT_URL = "https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}" # 创建人像风格重绘任务 def create_style_repaint_task(image_url, style_index): headers = { "Authorization": f"Bearer {DASHSCOPE_API_KEY}", "X-DashScope-Async": "enable", "Content-Type": "application/json" } payload = { "model": "wanx-style-repaint-v1", "input": { "image_url": image_url, "style_index": style_index } } response = requests.post(CREATE_TASK_URL, headers=headers, json=payload) if response.status_code == 200: task_id = response.json().get("output", {}).get("task_id") if task_id: print(f"任务创建成功,任务ID: {task_id}") return task_id else: print("未获取到任务ID") return None else: print(f"任务创建失败,状态码: {response.status_code}, 响应: {response.text}") return None # 根据任务ID查询结果 def get_task_result(task_id): headers = { "Authorization": f"Bearer {DASHSCOPE_API_KEY}" } url = GET_TASK_RESULT_URL.format(task_id=task_id) response = requests.get(url, headers=headers) if response.status_code == 200: task_status = response.json().get("output", {}).get("task_status") if task_status == "SUCCEEDED": result_image_url = response.json().get("output", {}).get("results", [{}])[0].get("url") print(f"任务成功,生成图片URL: {result_image_url}") return result_image_url else: print(f"任务状态: {task_status}") return None else: print(f"查询任务失败,状态码: {response.status_code}, 响应: {response.text}") return None # 示例调用 if __name__ == "__main__": # 创建任务 image_url = "https://public-vigen-video.oss-cn-shanghai.aliyuncs.com/public/dashscope/test.png" #style_index = 5 # 国画古风 style_index = 6 # 将军百战 # 更多类型参数参考这里 #https://help.aliyun.com/zh/model-studio/user-guide/style-repaint?spm=a2c4g.11186623.help-menu-2400256.d_1_0_6_1.5c227ceefBYDrn#236650d306qt8 task_id = create_style_repaint_task(image_url, style_index) if task_id: # 轮询查询任务结果 while True: result_image_url = get_task_result(task_id) if result_image_url: break time.sleep(5) # 每隔 5 秒查询一次