prompt: str,

images: list[bytes] = None,

model: str = None) -> bytes | None:

"""

# 更新token

images = images or []

# 逻辑: 如果未指定 model，默认使用 Landscape

use_model = model if model else MODEL_LANDSCAPE

# 1. 构造 Payload

if images:

content_payload = [{"type": "text", "text": prompt}]

print(f"[Backend] 正在处理 {len(images)} 张图片输入...")

for img_bytes in images:

b64_str = base64.b64encode(img_bytes).decode('utf-8')

content_payload.append({

"type": "image_url",

"image_url": {"url": f"data:image/jpeg;base64,{b64_str}"}

})

else:

content_payload = prompt

payload = {

"model": use_model, # 使用选定的模型

"messages": [{"role": "user", "content": content_payload}],

"stream": True

}

headers = {

"Authorization": API_KEY,

"Content-Type": "application/json"

}

image_url = None

print(f"[Backend] Model: {use_model} | 发起请求: {prompt[:20]}...")

try:

async with aiohttp.ClientSession() as session:

async with session.post(BACKEND_URL, json=payload, headers=headers, timeout=120) as response:

if response.status != 200:

err_text = await response.text()

content = response.content

print(f"[Backend Error] Status {response.status}: {err_text} {content}")

raise Exception(f"API Error: {response.status}: {err_text}")

async for line in response.content:

line_str = line.decode('utf-8').strip()

if line_str.startswith('{"error'):

chunk = json.loads(data_str)

delta = chunk.get("choices", [{}])[0].get("delta", {})

msg = delta['reasoning_content']

if '401' in msg:

msg += '\nAccess Token 已失效，需重新配置。'

elif '400' in msg:

msg += '\n返回内容被拦截。'

raise Exception(msg)

if not line_str or not line_str.startswith('data: '):

continue

data_str = line_str[6:]

if data_str == '[DONE]':

break

try:

chunk = json.loads(data_str)

delta = chunk.get("choices", [{}])[0].get("delta", {})

# 打印思考过程

if "reasoning_content" in delta:

print(delta['reasoning_content'], end="", flush=True)

# 提取内容中的图片链接

if "content" in delta:

content_text = delta["content"]

img_match = re.search(r'!\[.*?\]\((.*?)\)', content_text)

if img_match:

image_url = img_match.group(1)

print(f"\n[Backend] 捕获图片链接: {image_url}")

except json.JSONDecodeError:

continue

# 3. 下载生成的图片

if image_url:

async with session.get(image_url) as img_resp:

if img_resp.status == 200:

image_bytes = await img_resp.read()

return image_bytes

else:

print(f"[Backend Error] 图片下载失败: {img_resp.status}")

except Exception as e:

print(f"[Backend Exception] {e}")

raise e

print("=== AI 绘图接口测试 ===")

user_prompt = input("请输入提示词 (例如 '一只猫'): ").strip()

if not user_prompt:

user_prompt = "A cute cat in the garden"

print(f"正在请求: {user_prompt}")

# 这里的 images 传空列表用于测试文生图

# 如果想测试图生图，你需要手动读取本地文件：

# with open("output_test.jpg", "rb") as f: img_data = f.read()

# result = await request_backend_generation(user_prompt, [img_data])

result = await request_backend_generation(user_prompt)

if result:

filename = "output_test.jpg"

with open(filename, "wb") as f:

f.write(result)

print(f"\n[Success] 图片已保存为 {filename}，大小: {len(result)} bytes")

else: