出圖進階：工作流編排與去 AI 味

import os
import requests

BASE = "https://api.apiyi.com/v1"
API_KEY = os.environ["APIYI_API_KEY"]          # 絕不要把 Key 寫死在程式碼裡

REWRITE_SYSTEM = """你是圖片提示詞工程師。把使用者的口語需求改寫成一段結構化的中文出圖提示詞。

必須補齊六要素，缺什麼補什麼，不要詢問使用者：
1 主體：具體到材質、顏色、數量、狀態
2 環境：背景是什麼、虛實關係
3 光線：光源方向、軟硬、有沒有補光，必須是可指認的單一主光
4 鏡頭與視角：焦段、光圈、機位高度、俯仰角
5 色調與介質：白平衡傾向、飽和度、膠片或數碼質感
6 構圖：主體在畫面什麼位置，留白在哪

規則：
- 只輸出提示詞正文，不要解釋、不要分點、不要加標題
- 不要出現品牌名、商標、可識別的文字內容，除非使用者明確要求
- 不要使用 8K、超高畫質、傑作、完美 這類空泛的品質詞
- 使用者明確指定過的要素原樣保留，不要改寫"""


def rewrite(user_prompt: str) -> str:
    r = requests.post(
        f"{BASE}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "gemini-3.5-flash",
            "messages": [
                {"role": "system", "content": REWRITE_SYSTEM},
                {"role": "user", "content": user_prompt},
            ],
        },
        timeout=60,
    )
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"].strip()

import base64
import json
import os
from concurrent.futures import ThreadPoolExecutor

import requests

BASE = "https://api.apiyi.com"
API_KEY = os.environ["APIYI_API_KEY"]
HEAD = {"Authorization": f"Bearer {API_KEY}"}

STYLE_CONST = "冷調中性白平衡，整體偏低飽和，畫面乾淨留白充足。"   # 你的 preset


def draw(prompt: str, size: str = "2K", aspect: str = "1:1") -> bytes:
    """出一張圖（Nano Banana Pro 原生 Gemini 端點）"""
    url = f"{BASE}/v1beta/models/gemini-3-pro-image:generateContent"
    body = {
        "contents": [{"parts": [{"text": prompt}]}],
        "generationConfig": {
            "responseModalities": ["IMAGE"],
            "imageConfig": {"aspectRatio": aspect, "imageSize": size},
        },
    }
    r = requests.post(url, headers=HEAD, json=body, timeout=600)   # 4K 按 600s 留餘量
    r.raise_for_status()
    parts = r.json()["candidates"][0]["content"]["parts"]
    part = next((p for p in parts if p.get("inlineData")), None)
    if part is None:                                   # HTTP 200 但沒圖 = 多半被稽核攔了
        raise RuntimeError("未返回圖片：" + json.dumps(parts, ensure_ascii=False)[:300])
    return base64.b64decode(part["inlineData"]["data"])


def score(image: bytes, prompt: str) -> dict:
    """用視覺模型給候選圖打分，返回 {總分, 各維度分, 一句話講問題}"""
    data_url = "data:image/png;base64," + base64.b64encode(image).decode()
    rubric = (
        "給這張圖打分，嚴格返回 JSON："
        '{"instruction":0-10,"anatomy":0-10,"text":0-10,"texture":0-10,'
        '"composition":0-10,"total":0-50,"issue":"一句話"}。'
        "instruction=是否滿足下面這條需求；anatomy=手指、肢體、物體結構有無錯誤；"
        "text=畫面內文字是否正確（沒有文字給 10）；texture=質感是否像真實拍攝而非渲染；"
        "composition=構圖是否可用。需求如下：\n" + prompt
    )
    r = requests.post(
        f"{BASE}/v1/chat/completions",
        headers=HEAD,
        json={
            "model": "gemini-3.5-flash",
            "messages": [{"role": "user", "content": [
                {"type": "text", "text": rubric},
                {"type": "image_url", "image_url": {"url": data_url}},
            ]}],
            "response_format": {"type": "json_object"},
        },
        timeout=120,
    )
    r.raise_for_status()
    return json.loads(r.json()["choices"][0]["message"]["content"])


def best_of(user_input: str, n: int = 4) -> bytes:
    prompt = rewrite(user_input) + "\n" + STYLE_CONST        # Step 1 + Step 2
    with ThreadPoolExecutor(max_workers=n) as pool:          # Step 3：客戶端併發，不用 n 引數
        results = list(pool.map(lambda _: _safe(draw, prompt), range(n)))

    cands = [img for ok, img in results if ok]
    if not cands:
        raise RuntimeError("全部失敗，檢查稽核攔截或降級到別的模型")

    with ThreadPoolExecutor(max_workers=len(cands)) as pool:  # Step 4：併發打分
        scores = list(pool.map(lambda im: score(im, prompt), cands))

    ranked = sorted(zip(cands, scores), key=lambda x: x[1]["total"], reverse=True)
    return ranked[0][0]                                      # Step 5 的精修與轉存按業務接上


def _safe(fn, *args):
    try:
        return True, fn(*args)
    except Exception as e:                # 單張失敗不拖垮整批
        return False, str(e)