Blame for app.py

import httpx
import re
import base64
import os
import asyncio
import time
import gc
from datetime import datetime
from collections import deque
from PIL import Image
from io import BytesIO
from openai import AsyncOpenAI
from typing import Optional, List
from fastapi import FastAPI, HTTPException, Request
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel
import uvicorn

# ==========================================
# 1. 配置区域
# ==========================================
CONCURRENCY_LIMITER = asyncio.Semaphore(5)

DYNAMIC_BLOCK_CACHE = deque(maxlen=200)

SENSITIVE_WORDS = [
    "nsfw", "nude", "naked", "sex", "porn", "hentai", 
    "裸", "色情", "全裸", "blood", "kill", "murder", 
    "血腥", "尸体", "毒品", "杀"
]

# 精确的枚举表
RESOLUTION_MAP = {
    "1:1":  {"1k": (1024, 1024), "2k": (2048, 2048), "4k": (4096, 4096)},
    "16:9": {"1k": (1280, 720),  "2k": (2560, 1440), "4k": (3840, 2160)},
    "9:16": {"1k": (720, 1280),  "2k": (1440, 2560), "4k": (2160, 3840)},
    "4:3":  {"1k": (1152, 864),  "2k": (2048, 1536), "4k": (2880, 2160)},
    "3:4":  {"1k": (864, 1152),  "2k": (1536, 2048), "4k": (2160, 2880)},
    "21:9": {"1k": (1680, 720),  "2k": (3360, 1440), "4k": (5040, 2160)},
    "2:3":  {"1k": (852, 1280),  "2k": (1704, 2560), "4k": (2440, 3660)}
}

IP_REQUEST_HISTORY = {}
app_base_url = os.getenv("APP_BASE_URL", "").strip()

# ==========================================
# 2. 辅助功能
# ==========================================

def get_real_ip(request: Request):
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.client.host

def check_ip_rate_limit(ip: str, limit: int = 5, window: int = 60):
    now = time.time()
    history = IP_REQUEST_HISTORY.get(ip, [])
    valid_history = [t for t in history if now - t < window]
    if len(valid_history) >= limit:
        IP_REQUEST_HISTORY[ip] = valid_history
        return True
    valid_history.append(now)
    IP_REQUEST_HISTORY[ip] = valid_history
    if len(IP_REQUEST_HISTORY) > 5000: IP_REQUEST_HISTORY.clear()
    return False

def contains_local_sensitive_words(text: str):
    if text in DYNAMIC_BLOCK_CACHE: return True, text
    text_lower = text.lower()
    for word in SENSITIVE_WORDS:
        if word and word.lower() in text_lower: return True, word
    return False, None

# ==========================================
# 3. 百度审核 & 图片处理逻辑
# ==========================================

_BAIDU_TOKEN_CACHE = {"token": None, "expires_at": 0}

async def get_baidu_access_token():
    ak = os.getenv("BAIDU_API_KEY", "").strip()
    sk = os.getenv("BAIDU_SECRET_KEY", "").strip()
    if not ak or not sk: return None
    now = time.time()
    if _BAIDU_TOKEN_CACHE["token"] and now < _BAIDU_TOKEN_CACHE["expires_at"]:
        return _BAIDU_TOKEN_CACHE["token"]
    
    url = "https://aip.baidubce.com/oauth/2.0/token"
    params = {"grant_type": "client_credentials", "client_id": ak, "client_secret": sk}
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.post(url, params=params)
            token = resp.json().get("access_token")
            if token:
                _BAIDU_TOKEN_CACHE["token"] = token
                _BAIDU_TOKEN_CACHE["expires_at"] = now + 1728000
                return token
    except: return None

async def check_baidu_text_censor(text):
    token = await get_baidu_access_token()
    if not token: return False, None

    url = f"https://aip.baidubce.com/rest/2.0/solution/v1/text_censor/v2/user_defined?access_token={token}"
    headers = {'Content-Type': 'application/x-www-form-urlencoded'}
    data = {'text': text}
    
    try:
        async with httpx.AsyncClient(timeout=20) as client:
            resp = await client.post(url, data=data, headers=headers)
            result = resp.json()
            if result.get('conclusionType') == 2:
                msg = [i.get('msg') for i in result.get('data', [])]
                reason = ",".join(msg) if msg else "违规"
                if text not in DYNAMIC_BLOCK_CACHE:
                    DYNAMIC_BLOCK_CACHE.append(text)
                    print(f"🔒 [Cache] 新增违规: {text[:10]}...")
                return True, reason
            elif result.get('conclusionType') == 3: return True, "疑似违规"
            return False, None
    except: return False, None

class GenerateRequest(BaseModel):
    api_key: str
    model: str
    prompt: str
    ratio: str
    scale: str
    format: str 
    init_images: List[str] = []  # 修改为支持列表

def process_image_in_memory(image_bytes: bytes, target_format: str, target_size: tuple = None) -> str:
    try:
        with Image.open(BytesIO(image_bytes)) as img:
            out = BytesIO()
            final_img = img
            status_msg = f"[原图直出] 尺寸: {img.size}"

            if target_size:
                w_diff = abs(img.size[0] - target_size[0])
                h_diff = abs(img.size[1] - target_size[1])
                if w_diff > 50 or h_diff > 50:
                    print(f"⚠️ [Resize] {img.size} -> {target_size}")
                    final_img = img.resize(target_size, Image.Resampling.LANCZOS)
                    status_msg = f"[触发放大] -> {target_size}"
                else:
                    status_msg = f"[原图直出] 符合预期"
            print(f"✅ {status_msg}")

            fmt = "JPEG" if target_format == "jpg" else "PNG"
            final_img = final_img.convert("RGB") if target_format == "jpg" else final_img
            final_img.save(out, format=fmt, quality=95)
            return base64.b64encode(out.getvalue()).decode('utf-8')
    except Exception as e: 
        raise Exception(f"Image Error: {e}")
    finally: 
        gc.collect()

def sanitize_input_image(base64_str: str) -> str:
    """

    清洗上传的图片

    """
    if not base64_str: return None
    
    if "base64," in base64_str:
        base64_data = base64_str.split("base64,")[1]
    else:
        base64_data = base64_str

    try:
        img_data = base64.b64decode(base64_data)
        with Image.open(BytesIO(img_data)) as img:
            # 强制转为 RGB
            if img.mode != "RGB":
                img = img.convert("RGB")
            
            # 缩放过大的图片
            max_size = 1024
            if max(img.size) > max_size:
                img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
                print(f"📉 [Input] 参考图已压缩至: {img.size}")

            out = BytesIO()
            img.save(out, format="JPEG", quality=85)
            return base64.b64encode(out.getvalue()).decode('utf-8')
    except Exception as e:
        print(f"⚠️ 图片预处理失败，将尝试使用原始数据: {e}")
        return base64_data

async def download_image_as_bytes(url):
    print(f"📥 下载: {url}")
    async with httpx.AsyncClient(timeout=360) as client:
        r = await client.get(url)
        if r.status_code == 200: return r.content
        raise Exception(f"HTTP {r.status_code}")

async def core_generate(req: GenerateRequest):
    # 1. 纯查表逻辑：计算目标尺寸
    ratio_key = req.ratio.split(' ')[0]
    target_w, target_h = 1024, 1024
    if ratio_key in RESOLUTION_MAP:
        if req.scale in RESOLUTION_MAP[ratio_key]:
            target_w, target_h = RESOLUTION_MAP[ratio_key][req.scale]
        else:
            target_w, target_h = RESOLUTION_MAP[ratio_key]["1k"]
    api_size_str = f"{target_w}x{target_h}"
    
    # 2. Prompt 处理
    is_chat_model = "gemini" in req.model.lower() or "grok" in req.model.lower()
    final_prompt = req.prompt
    
    if is_chat_model:
        quality_prompt = "standard quality"
        if req.scale == "4k": quality_prompt = "Extreme High Quality, 4K Resolution"
        elif req.scale == "2k": quality_prompt = "High Quality, 2K Resolution"
        suffix = f". (Settings: Aspect Ratio {ratio_key}, Quality {quality_prompt}, Target Size {api_size_str})"
        final_prompt = f"{req.prompt} {suffix}"
    
    # 3. 预处理所有参考图
    clean_images = []
    if req.init_images:
        print(f"🧹 正在预处理 {len(req.init_images)} 张参考图...")
        for raw_img in req.init_images:
            sanitized = sanitize_input_image(raw_img)
            if sanitized:
                clean_images.append(sanitized)

    # 4. 初始化客户端
    client = AsyncOpenAI(api_key=req.api_key, base_url=f"{app_base_url}/v1", max_retries=0, timeout=360.0)
    img_b = None

    # --- 分支 A: Gemini & Grok 模型 (使用 Chat Completion 多模态，支持多图) ---
    if is_chat_model:
        print(f"🟣 对话模型生图: {req.model} | 参考图数量: {len(clean_images)}")
        content_parts = [{"type": "text", "text": final_prompt}]
        
        # 添加所有图片到上下文
        for img_b64 in clean_images:
            content_parts.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}
            })
        
        extra_payload = {"modalities": ["image", "text"]}
        if "grok" in req.model.lower():
            extra_payload["render"] = True 

        # 发起请求
        res = await client.chat.completions.create(
            model=req.model, 
            messages=[{"role":"user","content": content_parts}], 
            extra_body=extra_payload,
            timeout=360.0
        )
        
        if isinstance(res, str):
            c = res
        elif hasattr(res, 'choices') and len(res.choices) > 0:
            c = res.choices[0].message.content
        else:
            try:
                c = res['choices'][0]['message']['content']
            except:
                raise Exception(f"模型返回格式异常: {str(res)}")

        if not c:
            raise Exception("模型返回内容为空")

        u = re.search(r"!\[.*?\]\((https?://[^\)]+)\)", c) or re.search(r"(https?://\S+\.(?:png|jpg|jpeg|webp))", c)
        
        if u: 
            img_b = await download_image_as_bytes(u.group(1))
        elif "base64," in c: 
            base64_str = c.split("base64,")[1].split(")")[0].strip()
            img_b = base64.b64decode(base64_str)
        else: 
            raise Exception(f"{req.model} 响应中未提取到有效图片 URL")

    # --- 分支 B: 非对话类模型 (Nano Banana 等，仅支持单图作为 init_image) ---
    else:
        # 如果有多张图，非对话模型只能取第一张
        primary_image = clean_images[0] if clean_images else None
        
        if primary_image:
            print(f"🔵 [Edit 接口] 图生图模式: {req.model} (使用第一张参考图)")
            image_data = base64.b64decode(primary_image)
            image_file = BytesIO(image_data)
            image_file.name = "init_image.jpg"

            try:
                res = await client.images.edit(
                    model=req.model,
                    image=image_file,
                    prompt=final_prompt,
                    n=1,
                    size=api_size_str,
                    response_format="b64_json",
                    extra_body={"strength": 0.75}
                )
            except Exception as e:
                print(f"⚠️ Edit 失败，回退至 Generations 模式: {e}")
                res = await client.images.generate(
                    model=req.model,
                    prompt=final_prompt,
                    size=api_size_str,
                    response_format="b64_json",
                    extra_body={"image": primary_image, "strength": 0.75}
                )
        else:
            print(f"🔵 [Generate 接口] 文生图模式: {req.model}")
            res = await client.images.generate(
                model=req.model,
                prompt=final_prompt,
                n=1,
                size=api_size_str,
                response_format="b64_json"
            )

        d = res.data[0]
        if getattr(d, 'b64_json', None):
            img_b = base64.b64decode(d.b64_json)
        elif hasattr(d, 'url') and d.url:
            img_b = await download_image_as_bytes(d.url)
        else:
            raise Exception("API 未返回有效的图片数据")

    # 5. 后处理
    return process_image_in_memory(img_b, req.format, target_size=(target_w, target_h))

# ==========================================
# 4. FastAPI App
# ==========================================
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
async def read_index(): return FileResponse('static/index.html')

@app.get("/health")
def health(): return {"status": "ok"}

@app.post("/api/generate")
async def generate_api(req: GenerateRequest, request: Request):
    ip = get_real_ip(request)
    
    key_suffix = req.api_key[-8:] if len(req.api_key) >= 8 else req.api_key
    masked_key = f"******{key_suffix}"
    
    timestamp = datetime.now().strftime('%H:%M:%S')
    mode = f"Img2Img (x{len(req.init_images)})" if req.init_images else "Txt2Img"
    print("\n" + "="*60)
    print(f"🚀 [Req] {timestamp} | IP: {ip}")
    print(f"🔑 Key: {masked_key}")
    print(f"📌 Model: {req.model} | Mode: {mode} | Scale: {req.scale} | Ratio: {req.ratio}")
    print("-" * 60)
    print(f"💡 Prompt (Full):")
    print(req.prompt)
    print("="*60 + "\n")

    if check_ip_rate_limit(ip, 3, 60):
        return JSONResponse({"status": "error", "message": "请求过快"}, 429)

    is_loc, word = contains_local_sensitive_words(req.prompt)
    if is_loc:
        return JSONResponse({"status": "error", "message": f"违规内容: {word}"}, 200)

    is_bd, reason = await check_baidu_text_censor(req.prompt)
    if is_bd:
        return JSONResponse({"status": "error", "message": f"审核未通过: {reason}"}, 200)

    async with CONCURRENCY_LIMITER:
        try:
            start_time = time.time()
            b64 = await core_generate(req)
            print(f"✅ [Success] 耗时: {time.time()-start_time:.2f}s")
            return {"status": "success", "image_base64": b64}
        except Exception as e:
            import traceback
            traceback.print_exc()
            print(f"❌ [Error] {e}")
            return JSONResponse({"status": "error", "message": str(e)}, 200)
        finally:
            gc.collect()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)