完整 Python 示例

pip install requests
export LINGGAN_API_KEY="<PUBLIC_API_KEY>"
python linggan_example.py ./product.png

import os
import random
import sys
import time
from pathlib import Path

import requests

BASE_URL = "https://api.ch88.cn"
API_KEY = os.environ["LINGGAN_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


def require_success(response: requests.Response) -> dict:
    if response.ok:
        return response.json()
    raise RuntimeError(
        f"HTTP {response.status_code}: {response.text}; "
        f"request-id={response.headers.get('X-Linggan-Request-Id')}"
    )


def upload_image(path: Path) -> str:
    with path.open("rb") as image_file:
        response = requests.post(
            f"{BASE_URL}/v1/open/uploads/images",
            headers=HEADERS,
            files={"file": (path.name, image_file)},
            timeout=120,
        )
    result = require_success(response)
    print("图片临时地址有效至：", result["expiresAt"])
    return result["data"]["url"]


def submit_generation(image_url: str) -> str:
    response = requests.post(
        f"{BASE_URL}/v1/open/product-images/generations",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={
            "prompt": "生成简洁高级的电商主图，产品居中，背景干净，保留真实包装细节。",
            "image_urls": [image_url],
            "product_info": {
                "name": "示例商品",
                "sellingPoints": "突出真实外观和清晰主体",
                "platform": "Amazon",
            },
            "size": "1:1",
            "resolution": "1k",
            "output_language": "zh-cn",
        },
        timeout=120,
    )
    result = require_success(response)
    return result["data"]["task_id"]


def wait_for_task(task_id: str) -> dict:
    delays = [5, 8, 13, 20, 30]
    started_at = time.monotonic()
    attempt = 0

    while time.monotonic() - started_at < 15 * 60:
        time.sleep(delays[min(attempt, len(delays) - 1)])
        response = requests.get(
            f"{BASE_URL}/v1/open/tasks/{task_id}",
            headers=HEADERS,
            timeout=30,
        )

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", "30"))
            time.sleep(retry_after + random.random())
            continue

        task = require_success(response)["data"]
        print("任务状态：", task["status"])

        if task["status"] == "succeeded":
            return task
        if task["status"] in {"failed", "cancelled"}:
            raise RuntimeError(f"任务终止：{task.get('error')}")

        attempt += 1

    raise TimeoutError("等待任务超过 15 分钟，任务可能仍在后台处理")


def download_result(task: dict, output_path: Path) -> None:
    image = task["result"]["images"][0]
    response = requests.get(image["url"], timeout=120)
    response.raise_for_status()
    output_path.write_bytes(response.content)
    print("结果已保存：", output_path)
    print("平台临时地址预计有效至：", image.get("expires_at"))


if __name__ == "__main__":
    source_path = Path(sys.argv[1])
    uploaded_url = upload_image(source_path)
    task_id = submit_generation(uploaded_url)
    print("任务编号：", task_id)
    completed_task = wait_for_task(task_id)
    download_result(completed_task, Path("result.png"))