AnimationManager/server/python/replace-charactor-ani.py

import asyncio
import aiohttp
import base64
import os
import json
import sys
import io
from aiohttp import ClientError, ServerDisconnectedError

# 设置标准输出编码为 UTF-8，确保中文正确显示
if sys.platform == 'win32':
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

print("图生图脚本启动...", flush=True)

IMAGE_DIR = "LocalImage"  # 固定参考图所在目录
REFERENCE_FILES = ["image1.png", "image2.png"]
PROMPT_TEXT = """Anelyze imaga 1, it contelins a sequence of anmgticn frames eanged in a gnid Return the resuit in .JSON formgt
Creale a high-cuslily animale sore shaat based on the image?, nolce the weapan ae in al fames, nohce the charcer's sze shcuid be same inal iramesCRITICAL INSTRUCTIONSi The lbsyout, glid strucfure, and poses MUST EXACTLY malch the frst image (the lemplale surile shoel) need pure while background2 DO NOT STRETCH the sprites. Manitain the onginal interral aspect ratio of tha charecters.
3. if the ouiput aspect raho ($ftargetAspectRatoy, difers irom the templale, add pedding fempty space) raher than streiching the content4. Apply the charecter's appearance (colors, dothing, features) lo the poses in the template.S'prompt ? 'Addtional instructions: $iprompt’ ")"""

async def main():
    # 检查 image1.png、image2.png 是否存在
    missing_files = []
    image_paths = []
    for file_name in REFERENCE_FILES:
        file_path = os.path.join(IMAGE_DIR, file_name)
        if os.path.isfile(file_path):
            image_paths.append(file_path)
        else:
            missing_files.append(file_name)
    
    if missing_files:
        missing = ", ".join(missing_files)
        expected = ", ".join(REFERENCE_FILES)
        print(f"错误: 找不到参考图 {missing}")
        print(f"请将 {expected} 放在 {IMAGE_DIR} 目录下")
        return
    
    # 读取并编码图片
    print(f"共找到 {len(image_paths)} 张参考图", flush=True)
    content = []
    for idx, image_path in enumerate(image_paths, start=1):
        print(f"[{idx}] 正在读取并编码: {image_path}", flush=True)
        with open(image_path, "rb") as img_file:
            image_data = base64.b64encode(img_file.read()).decode('utf-8')
        image_type = "png"  # 固定为 PNG 格式
        content.append(
            {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/{image_type};base64,{image_data}"
                }
            }
        )
    print("全部图片编码完成", flush=True)
    
    # 添加提示词
    content.append(
        {
            "type": "text",
            "text": PROMPT_TEXT
        }
    )
    
    try:
        # 设置超时和连接器配置
        timeout = aiohttp.ClientTimeout(total=300)  # 总超时5分钟
        connector = aiohttp.TCPConnector(limit=10, limit_per_host=5)
        
        async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
            print("正在发送图生图请求...", flush=True)
            print("请求已发送，等待服务器响应...", flush=True)
            
            async with session.post(
                "https://api.chatanywhere.tech/v1/chat/completions",
                headers={
                    "Authorization": "Bearer sk-j32LgDixK6pfESYGfJtgc2Tzlmszx5NZhSH0sOzpLQkYuKek",
                    "Content-Type": "application/json",
                },
                json={
                    "model": "gemini-3-pro-image-preview",
                    "messages": [{"role": "user", "content": content}],
                },
            ) as resp:
                # 获取响应内容（无论成功或失败都保存）
                result = await resp.text()
                # 直接保存到 ImageToImage.txt
                with open("ImageToImage.txt", "w", encoding="utf-8") as f:
                    f.write(result)
                
                # 检查响应状态
                if resp.status != 200:
                    print(f"错误: 服务器返回状态码 {resp.status}，错误信息已保存到 ImageToImage.txt", flush=True)
                    return
                
                print("收到服务器响应，结果已保存到 ImageToImage.txt", flush=True)
                
    except asyncio.TimeoutError:
        print("错误: 请求超时", flush=True)
    except ServerDisconnectedError:
        print("错误: 服务器断开连接", flush=True)
    except ClientError as e:
        print(f"错误: 网络连接错误 - {str(e)}", flush=True)
    except Exception as e:
        print(f"错误: 发生未知错误 - {str(e)}", flush=True)
        print(f"错误类型: {type(e).__name__}", flush=True)

asyncio.run(main())