图像画面扩展

import os
import requests
import time
from http import HTTPStatus

# 从环境变量获取阿里云百炼API Key，或直接在代码中赋值
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")

# 请将 workspace_id 替换为您的百炼业务空间ID
workspace_id = "your-workspace-id"
# 以下为华北2（北京）地域的URL。请将 {WorkspaceId} 替换为您的百炼业务空间ID，各地域的URL不同。
base_url = f"https://{workspace_id}.cn-beijing.maas.aliyuncs.com"

def submit_task():
    """提交一个扩图任务"""
    url = f"{base_url}/api/v1/services/aigc/image2image/out-painting"

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "X-DashScope-Async": "enable"  # 异步调用
    }

    body = {
        "model": "image-out-painting",
        "input": {
            "image_url": "https://huarong123.oss-cn-hangzhou.aliyuncs.com/image/%E5%9B%BE%E5%83%8F%E7%94%BB%E9%9D%A2%E6%89%A9%E5%B1%95.png"
        },
        "parameters": {
            "angle": 90,
            "x_scale": 1.5,
            "y_scale": 1.5
        }
    }

    response = requests.post(url, headers=headers, json=body)

    if response.status_code == HTTPStatus.OK:
        task_id = response.json().get('output', {}).get('task_id')
        print(f"任务提交成功，任务ID为: {task_id}")
        return task_id
    else:
        print(f"任务提交失败，状态码: {response.status_code}, 响应: {response.text}")
        return None

def query_task_result(task_id):
    """根据任务ID轮询查询结果"""
    if not task_id:
        return

    url = f"{base_url}/api/v1/tasks/{task_id}"
    headers = {"Authorization": f"Bearer {api_key}"}

    print("开始查询任务状态...")
    while True:
        response = requests.get(url, headers=headers)
        if response.status_code != HTTPStatus.OK:
            print(f"查询失败，状态码: {response.status_code}, 响应: {response.text}")
            break

        response_data = response.json()
        task_status = response_data.get('output', {}).get('task_status')

        if task_status == 'SUCCEEDED':
            print("任务成功完成！")
            print(f"任务成功响应数据: {response_data}")
            url = response_data.get('output', {}).get('output_image_url', "")
            print(f"生成图片 URL: {url}")
            break
        elif task_status == 'FAILED':
            print(f"任务失败。错误信息: {response_data}")
            break
        else:
            print(f"任务正在处理中，当前状态: {task_status}...")
            # 这是一个简单轮询方式，建议在生产中采用更优雅的轮询机制
            time.sleep(5)  # 等待5秒后再次查询

if __name__ == '__main__':
    task_id = submit_task()
    if task_id:
        query_task_result(task_id)