Image editing - 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-intl.aliyuncs.com/api/v1'

# If you have not configured an environment variable, replace the following line with your API key: api_key="sk-xxx"
api_key = os.getenv("DASHSCOPE_API_KEY")

# --- Image input: Use Base64 encoding ---
# The Base64 encoding format is 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("Unsupported or unrecognized image format")
  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}"

"""
Image input methods:
Choose one of the following three methods.

1. Use a public URL - Suitable for publicly accessible images.
2. Use a local file - Suitable for local development and testing.
3. Use Base64 encoding - Suitable for private images or scenarios that require encrypted transmission.
"""

# [Method 1] Use a public image 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"

# [Method 2] Use a local file (supports absolute and relative paths)
# Format requirement: file:// + file path
# Example (absolute path):
# image_url_1 = "file://" + "/path/to/your/image_1.png"     # Linux/macOS
# image_url_2 = "file://" + "C:/path/to/your/image_2.png"  # Windows
# Example (relative path):
# image_url_1 = "file://" + "./image_1.png"                 # Use your actual path
# image_url_2 = "file://" + "./image_2.png"                # Use your actual path

# [Method 3] Use a Base64-encoded image
# image_url_1 = encode_file("./image_1.png")               # Use your actual path
# image_url_2 = encode_file("./image_2.png")              # Use your actual path

print('----sync call, please wait a moment----')
rsp = ImageSynthesis.call(api_key=api_key,
                          model="wan2.5-i2i-preview",
                          prompt="Place the alarm clock from Image 1 next to the vase on the dining table in Image 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:
  # Save the image in the current directory
  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('sync_call Failed, 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-intl.aliyuncs.com/api/v1'

# If you have not configured an environment variable, replace the following line with your API key: api_key="sk-xxx"
api_key = os.getenv("DASHSCOPE_API_KEY")

# Use a public image 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('----create task----')
  task_info = create_async_task()
  print('----wait task----')
  wait_async_task(task_info)


# Create an asynchronous task
def create_async_task():
  rsp = ImageSynthesis.async_call(api_key=api_key,
                  model="wan2.5-i2i-preview",
                  prompt="Place the alarm clock from Image 1 next to the vase on the dining table in Image 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('Failed, status_code: %s, code: %s, message: %s' %
              (rsp.status_code, rsp.code, rsp.message))
  return rsp


# Wait for the asynchronous task to complete
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)
    # save file to current directory
    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('Failed, status_code: %s, code: %s, message: %s' %
              (rsp.status_code, rsp.code, rsp.message))


# Get asynchronous task information
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('Failed, status_code: %s, code: %s, message: %s' %
              (status.status_code, status.code, status.message))


# Cancel the asynchronous task. Only tasks in the PENDING state can be canceled.
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('Failed, status_code: %s, code: %s, message: %s' %
              (rsp.status_code, rsp.code, rsp.message))


if __name__ == '__main__':
  async_call()