图像编辑 - Wan2.7/2.6/2.5

import base64
import mimetypes
from http import HTTPStatus
from urllib.parse import urlparse, unquote
from pathlib import PurePosixPath

import dashscope
import requests
from dashscope import ImageSynthesis
import os

dashscope.base_http_api_url = 'https://dashscope.aliyuncs.com/api/v1'

# 如果未设置环境变量，请将下行替换为：api_key="sk-xxx"
api_key = os.getenv("DASHSCOPE_API_KEY")

# --- 输入图片：Base64编码 ---
# Base64格式：data:{MIME_type};base64,{base64_data}
def encode_file(file_path):
  mime_type, _ = mimetypes.guess_type(file_path)
  if not mime_type or not mime_type.startswith("image/"):
    raise ValueError("不支持或无法识别的图片格式")
  with open(file_path, "rb") as image_file:
    encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
  return f"data:{mime_type};base64,{encoded_string}"

"""
图像输入方式：
选择以下其中一种：

1. 公网URL — 适用于公开可访问的图片
2. 本地文件 — 适用于本地开发和测试
3. Base64编码 — 适用于私有图片或需要加密传输的场景
"""

# [方式一] 公网图片URL
image_url_1 = "https://img.alicdn.com/imgextra/i3/O1CN0157XGE51l6iL9441yX_!!6000000004770-49-tps-1104-1472.webp"
image_url_2 = "https://img.alicdn.com/imgextra/i3/O1CN01SfG4J41UYn9WNt4X1_!!6000000002530-49-tps-1696-960.webp"

# [方式二] 本地文件（支持绝对路径和相对路径）
# 格式：file:// + 文件路径
# 示例（绝对路径）：
# image_url_1 = "file://" + "/path/to/your/image_1.png"     # Linux/macOS
# image_url_2 = "file://" + "C:/path/to/your/image_2.png"  # Windows
# 示例（相对路径）：
# image_url_1 = "file://" + "./image_1.png"
# image_url_2 = "file://" + "./image_2.png"

# [方式三] Base64编码图片
# image_url_1 = encode_file("./image_1.png")
# image_url_2 = encode_file("./image_2.png")

print('----同步调用，请稍候----')
rsp = ImageSynthesis.call(api_key=api_key,
                          model="wan2.5-i2i-preview",
                          prompt="将图1中的闹钟放置到图2的餐桌的花瓶旁边位置",
                          images=[image_url_1, image_url_2],
                          negative_prompt="",
                          n=1,
                          # size="1280*1280",
                          prompt_extend=True,
                          watermark=False,
                          seed=12345)
print('response: %s' % rsp)
if rsp.status_code == HTTPStatus.OK:
  # 保存图片到当前目录
  for result in rsp.output.results:
    file_name = PurePosixPath(unquote(urlparse(result.url).path)).parts[-1]
    with open('./%s' % file_name, 'wb+') as f:
      f.write(requests.get(result.url).content)
else:
  print('同步调用失败, status_code: %s, code: %s, message: %s' %
        (rsp.status_code, rsp.code, rsp.message))

import os
from http import HTTPStatus
from urllib.parse import urlparse, unquote
from pathlib import PurePosixPath
import dashscope
import requests
from dashscope import ImageSynthesis

dashscope.base_http_api_url = 'https://dashscope.aliyuncs.com/api/v1'

# 如果未设置环境变量，请将下行替换为：api_key="sk-xxx"
api_key = os.getenv("DASHSCOPE_API_KEY")

# 公网图片URL
image_url_1 = "https://img.alicdn.com/imgextra/i3/O1CN0157XGE51l6iL9441yX_!!6000000004770-49-tps-1104-1472.webp"
image_url_2 = "https://img.alicdn.com/imgextra/i3/O1CN01SfG4J41UYn9WNt4X1_!!6000000002530-49-tps-1696-960.webp"

def async_call():
  print('----创建任务----')
  task_info = create_async_task()
  print('----等待任务----')
  wait_async_task(task_info)

# 创建异步任务
def create_async_task():
  rsp = ImageSynthesis.async_call(api_key=api_key,
                                  model="wan2.5-i2i-preview",
                                  prompt="将图1中的闹钟放置到图2的餐桌的花瓶旁边位置",
                                  images=[image_url_1, image_url_2],
                                  negative_prompt="",
                                  n=1,
                                  # size="1280*1280",
                                  prompt_extend=True,
                                  watermark=False,
                                  seed=12345)
  print(rsp)
  if rsp.status_code == HTTPStatus.OK:
    print(rsp.output)
  else:
    print('失败, status_code: %s, code: %s, message: %s' %
          (rsp.status_code, rsp.code, rsp.message))
  return rsp

# 等待异步任务完成
def wait_async_task(task):
  rsp = ImageSynthesis.wait(task=task, api_key=api_key)
  print(rsp)
  if rsp.status_code == HTTPStatus.OK:
    print(rsp.output)
    # 保存文件到当前目录
    for result in rsp.output.results:
      file_name = PurePosixPath(unquote(urlparse(result.url).path)).parts[-1]
      with open('./%s' % file_name, 'wb+') as f:
        f.write(requests.get(result.url).content)
  else:
    print('失败, status_code: %s, code: %s, message: %s' %
          (rsp.status_code, rsp.code, rsp.message))

# 查询异步任务状态
def fetch_task_status(task):
  status = ImageSynthesis.fetch(task=task, api_key=api_key)
  print(status)
  if status.status_code == HTTPStatus.OK:
    print(status.output.task_status)
  else:
    print('失败, status_code: %s, code: %s, message: %s' %
          (status.status_code, status.code, status.message))

# 取消异步任务，仅PENDING状态的任务可以取消
def cancel_task(task):
  rsp = ImageSynthesis.cancel(task=task, api_key=api_key)
  print(rsp)
  if rsp.status_code == HTTPStatus.OK:
    print(rsp.output.task_status)
  else:
    print('失败, status_code: %s, code: %s, message: %s' %
          (rsp.status_code, rsp.code, rsp.message))

if __name__ == '__main__':
  async_call()