api/ai.py

from __future__ import annotations

import mimetypes
from decimal import Decimal, ROUND_HALF_UP

from typing import Any

from fastapi import APIRouter, Header, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel, ConfigDict, Field
from curl_cffi import requests as curl_requests

from api.image_inputs import parse_image_edit_request, read_image_sources
from api.support import require_identity, resolve_image_base_url
from services.auth_service import DAILY_GIFT_ACCOUNT_POOL_MODEL_ID, auth_service
from services.chat_model_service import resolve_chat_models
from services.image_model_service import resolve_image_models
from services.system_model_service import resolve_system_models
from services.content_filter import check_request, request_text
from services.config import config
from services.editable_file_task_service import editable_file_task_service
from services.image_annotation_service import identify_image_region
from services.log_service import LoggedCall
from services.pool_pricing_service import pool_pricing_service
from services.skill_service import skill_service
from services.protocol import (
    anthropic_v1_messages,
    openai_v1_chat_complete,
    openai_v1_image_edit,
    openai_v1_image_generations,
    openai_v1_models,
    openai_v1_response,
    openai_search,
)
from services.third_party_image_service import third_party_image_service
from utils.helper import has_response_image_generation_tool, is_image_chat_request, parse_image_count


class ImageGenerationRequest(BaseModel):
    prompt: str = Field(..., min_length=1)
    model: str = "gpt-image-2"
    n: int = Field(default=1, ge=1, le=4)
    size: str | None = None
    quality: str | None = None
    resolution: str | None = None
    image_resolution: str | None = None
    third_party_image_group_id: str | None = None
    response_format: str = "url"
    account_pool_strategy: str = "own_first"
    history_disabled: bool = True
    stream: bool | None = None


class ChatCompletionRequest(BaseModel):
    model_config = ConfigDict(extra="allow")
    model: str | None = None
    prompt: str | None = None
    n: int | None = None
    stream: bool | None = None
    modalities: list[str] | None = None
    messages: list[dict[str, object]] | None = None


class ResponseCreateRequest(BaseModel):
    model_config = ConfigDict(extra="allow")
    model: str | None = None
    input: object | None = None
    tools: list[dict[str, object]] | None = None
    tool_choice: object | None = None
    stream: bool | None = None


class AnthropicMessageRequest(BaseModel):
    model_config = ConfigDict(extra="allow")
    model: str | None = None
    messages: list[dict[str, object]] | None = None
    system: object | None = None
    stream: bool | None = None


class PromptOptimizationRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=30000)


class ReversePromptRequest(BaseModel):
    image: str = Field(..., min_length=1, max_length=40_000_000)


class ImageRegionBounds(BaseModel):
    x: float = Field(..., ge=0, le=100)
    y: float = Field(..., ge=0, le=100)
    width: float = Field(..., gt=0, le=100)
    height: float = Field(..., gt=0, le=100)


class ImageRegionRecognitionRequest(BaseModel):
    image: str = Field(..., min_length=1, max_length=40_000_000)
    bounds: ImageRegionBounds


class SearchRequest(BaseModel):
    prompt: str = Field(..., min_length=1)


class EditableFileTaskRequest(BaseModel):
    prompt: str = ""
    base64_images: list[str] = Field(default_factory=list)
    client_task_id: str | None = None


async def filter_or_log(call: LoggedCall, text: str) -> None:
    try:
        await run_in_threadpool(check_request, text)
    except HTTPException as exc:
        call.log("调用失败", status="failed", error=str(exc.detail))
        raise


def _to_positive_image_count(value: object, default: int = 1) -> int:
    try:
        return max(1, int(value or default))
    except (TypeError, ValueError):
        return default


def _credit_amount(value: object) -> int | float:
    try:
        amount = max(Decimal("0"), Decimal(str(value or 0))).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    except Exception:
        amount = Decimal("0")
    return int(amount) if amount == amount.to_integral_value() else float(amount)


def _third_party_activity_quote(identity: dict[str, object], quantity: int, *, group_id: str, base_unit_price: object) -> dict[str, object]:
    pricing_config = config.normal_user_third_party_image_pricing
    if str(identity.get("role") or "").strip().lower() != "normal" or not bool(pricing_config.get("enabled")):
        base = _credit_amount(base_unit_price)
        return {
            "applies": False,
            "target_type": "third_party_group",
            "third_party_group_id": group_id,
            "base_unit_price": base,
            "pay_percentage": 100,
            "unit_price": base,
            "quantity": quantity,
            "estimated_total": _credit_amount(float(base) * quantity),
            "rule_id": "",
            "rule_name": "",
        }
    return pool_pricing_service.quote(
        identity,
        quantity=quantity,
        pricing_config=pricing_config,
        target_type="third_party_group",
        third_party_group_id=group_id,
        base_unit_price=base_unit_price,
    )


def _third_party_wallet_remaining(identity: dict[str, object], group_id: str, model_id: str) -> int | float:
    try:
        summary = auth_service.credit_summary(identity)
    except Exception:
        return 0
    wallets = summary.get("third_party_image_wallets") if isinstance(summary, dict) else []
    wallet_key = f"{group_id}:{model_id}"
    wallet_items = wallets if isinstance(wallets, list) else []
    balances: list[int | float] = []
    for wallet in wallet_items:
        if not isinstance(wallet, dict):
            continue
        if str(wallet.get("id") or "") == wallet_key or (
            str(wallet.get("group_id") or "") == group_id
            and str(wallet.get("model_id") or "").strip() == model_id
        ):
            try:
                balances.append(_credit_amount(wallet.get("remaining")))
            except (TypeError, ValueError):
                balances.append(0)
    return max(balances) if balances else 0


def _count_data_images(value: object) -> int:
    if not isinstance(value, list):
        return 0
    return sum(
        1
        for item in value
        if isinstance(item, dict) and (str(item.get("b64_json") or "").strip() or str(item.get("url") or "").strip())
    )


def _count_image_generation_items(value: object) -> int:
    if not isinstance(value, list):
        return 0
    count = 0
    for item in value:
        if not isinstance(item, dict):
            continue
        if str(item.get("type") or "") == "image_generation_call" and str(item.get("result") or "").strip():
            count += 1
    return count


def _count_data_url_images(text: object) -> int:
    return str(text or "").count("data:image/")


def _count_successful_images(value: object) -> int:
    if not isinstance(value, dict):
        return 0

    count = _count_data_images(value.get("data"))
    if count:
        return count

    output = value.get("output")
    count = _count_image_generation_items(output)
    if count:
        return count

    item = value.get("item")
    if isinstance(item, dict):
        count = _count_image_generation_items([item])
        if count:
            return count

    choices = value.get("choices")
    if isinstance(choices, list):
        return sum(
            _count_data_url_images(
                ((choice.get("message") if isinstance(choice, dict) else {}) or {}).get("content")
                if isinstance(choice, dict)
                else ""
            )
            + _count_data_url_images(
                ((choice.get("delta") if isinstance(choice, dict) else {}) or {}).get("content")
                if isinstance(choice, dict)
                else ""
            )
            for choice in choices
        )

    response = value.get("response")
    if isinstance(response, dict):
        return _count_successful_images(response)

    return 0


def _attach_account_pool(payload: dict[str, object], identity: dict[str, object]) -> None:
    has_personal_pool = identity.get("role") == "admin" or bool(identity.get("account_pool_enabled"))
    payload["account_owner_id"] = str(identity.get("id") or "").strip()
    if not has_personal_pool:
        payload["account_pool_strategy"] = "admin_only"
        payload["account_allow_shared"] = False
        return
    payload["account_pool_strategy"] = str(payload.get("account_pool_strategy") or "own_first")
    payload["account_allow_shared"] = bool(payload.get("account_allow_shared", True))


def _chat_completion_text(value: object) -> str:
    if not isinstance(value, dict):
        return ""
    choices = value.get("choices")
    if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict):
        return ""
    message = choices[0].get("message")
    if not isinstance(message, dict):
        return ""
    content = message.get("content")
    if isinstance(content, str):
        return content.strip()
    if not isinstance(content, list):
        return ""
    return "".join(
        str(item.get("text") or "")
        for item in content
        if isinstance(item, dict) and str(item.get("type") or "") in {"text", "output_text"}
    ).strip()


def _response_text(value: object) -> str:
    if not isinstance(value, dict):
        return ""
    direct = value.get("output_text")
    if isinstance(direct, str) and direct.strip():
        return direct.strip()
    output = value.get("output")
    if not isinstance(output, list):
        return ""
    parts: list[str] = []
    for item in output:
        if not isinstance(item, dict):
            continue
        content = item.get("content")
        if not isinstance(content, list):
            continue
        parts.extend(
            str(part.get("text") or "")
            for part in content
            if isinstance(part, dict) and str(part.get("type") or "") in {"text", "output_text"}
        )
    return "".join(parts).strip()


def _split_grants(grants: list[dict[str, object]], amount: int | float) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    remaining = Decimal(str(_credit_amount(amount)))
    used: list[dict[str, object]] = []
    refunded: list[dict[str, object]] = []
    for grant in grants:
        if not isinstance(grant, dict):
            continue
        grant_amount = Decimal(str(_credit_amount(grant.get("amount"))))
        grant_id = str(grant.get("id") or "").strip()
        if not grant_id or grant_amount <= 0:
            continue
        take = min(grant_amount, remaining)
        if take > 0:
            used.append({"id": grant_id, "amount": _credit_amount(take)})
        if grant_amount - take > 0:
            refunded.append({"id": grant_id, "amount": _credit_amount(grant_amount - take)})
        remaining -= take
    return used, refunded


class CreditReservation:
    def __init__(self, user_id: str = "", amount: int | float = 0):
        self.user_id = user_id
        self.amount = _credit_amount(amount)
        self.daily_amount: int | float = 0
        self.paid_amount: int | float = 0
        self.package_reservations: list[dict[str, object]] = []
        self.grant_reservations: list[dict[str, object]] = []
        self.settled = False

    @classmethod
    def reserve(
        cls,
        identity: dict[str, object],
        amount: int | float,
        *,
        model_id: str = DAILY_GIFT_ACCOUNT_POOL_MODEL_ID,
    ) -> "CreditReservation":
        if str(identity.get("role") or "").strip().lower() != "normal":
            return cls()
        reservation = cls(str(identity.get("id") or "").strip(), amount)
        if not reservation.user_id or reservation.amount <= 0:
            return reservation
        details = auth_service.reserve_credits(
            reservation.user_id,
            reservation.amount,
            model_id=model_id,
        ).get("_credit_reservation", {})
        if not isinstance(details, dict):
            details = {}
        reservation.daily_amount = _credit_amount(details.get("daily"))
        reservation.paid_amount = _credit_amount(details.get("permanent_paid") if "permanent_paid" in details else details.get("paid"))
        reservation.package_reservations = details.get("packages") if isinstance(details.get("packages"), list) else []
        reservation.grant_reservations = details.get("grants") if isinstance(details.get("grants"), list) else []
        return reservation

    def settle(self, consume_amount: int | float) -> None:
        if self.settled:
            return
        if not self.user_id or self.amount <= 0:
            self.settled = True
            return

        consume_count = _credit_amount(min(float(self.amount), float(_credit_amount(consume_amount))))
        daily_to_consume = _credit_amount(min(float(self.daily_amount), float(consume_count)))
        remaining = _credit_amount(float(consume_count) - float(daily_to_consume))
        packages_to_consume, packages_to_refund = _split_grants(self.package_reservations, remaining)
        remaining = _credit_amount(float(remaining) - sum(float(_credit_amount(item.get("amount"))) for item in packages_to_consume))
        grants_to_consume, grants_to_refund = _split_grants(self.grant_reservations, remaining)
        remaining = _credit_amount(float(remaining) - sum(float(_credit_amount(item.get("amount"))) for item in grants_to_consume))
        paid_to_consume = _credit_amount(min(float(self.paid_amount), float(remaining)))

        daily_to_refund = _credit_amount(float(self.daily_amount) - float(daily_to_consume))
        paid_to_refund = _credit_amount(float(self.paid_amount) - float(paid_to_consume))

        if consume_count > 0:
            auth_service.consume_reserved_credits(
                self.user_id,
                consume_count,
                daily_amount=daily_to_consume,
                paid_amount=paid_to_consume,
                package_reservations=packages_to_consume,
                grant_reservations=grants_to_consume,
            )
        refund_count = _credit_amount(float(self.amount) - float(consume_count))
        if refund_count > 0:
            auth_service.refund_reserved_credits(
                self.user_id,
                refund_count,
                daily_amount=daily_to_refund,
                paid_amount=paid_to_refund,
                package_reservations=packages_to_refund,
                grant_reservations=grants_to_refund,
            )
        self.settled = True


class InternalImageModelDailyLimitReservation:
    def __init__(self, *, user_id: str = "", model_id: str = "", amount: int = 0, scope: str = ""):
        self.user_id = user_id
        self.model_id = model_id
        self.amount = max(0, int(amount or 0))
        self.scope = scope
        self.settled = False

    @classmethod
    def reserve(
        cls,
        identity: dict[str, object],
        model_id: object,
        amount: int,
    ) -> "InternalImageModelDailyLimitReservation":
        if str(identity.get("role") or "").strip().lower() != "user":
            return cls()
        user_id = str(identity.get("id") or "").strip()
        normalized_model_id = str(model_id or "gpt-image-2").strip() or "gpt-image-2"
        if not user_id:
            return cls()
        result = auth_service.reserve_internal_image_model_daily_limit(
            user_id,
            model_id=normalized_model_id,
            amount=amount,
        )
        reservation = result.get("_internal_image_model_daily_limit_reservation") if isinstance(result, dict) else {}
        reserved_amount = max(0, int(reservation.get("amount") or 0)) if isinstance(reservation, dict) else 0
        reserved_scope = str(reservation.get("scope") or "").strip() if isinstance(reservation, dict) else ""
        return cls(user_id=user_id, model_id=normalized_model_id, amount=reserved_amount, scope=reserved_scope)

    def settle(self, success_count: int) -> None:
        if self.settled:
            return
        success_amount = min(self.amount, max(0, int(success_count or 0)))
        refund_amount = self.amount - success_amount
        if self.user_id and self.model_id and success_amount > 0:
            consume_kwargs: dict[str, object] = {
                "model_id": self.model_id,
                "amount": success_amount,
            }
            if self.scope:
                consume_kwargs["scope"] = self.scope
            auth_service.consume_internal_image_model_daily_limit(
                self.user_id,
                **consume_kwargs,
            )
        if self.user_id and self.model_id and refund_amount > 0:
            refund_kwargs: dict[str, object] = {
                "model_id": self.model_id,
                "amount": refund_amount,
            }
            if self.scope:
                refund_kwargs["scope"] = self.scope
            auth_service.refund_internal_image_model_daily_limit(
                self.user_id,
                **refund_kwargs,
            )
        self.settled = True


class ImageBillingReservation:
    def __init__(
        self,
        *,
        mode: str = "",
        user_id: str = "",
        amount: int | float = 0,
        per_image_cost: int | float = 1,
        group_id: str = "",
        model_id: str = "",
        credit_reservation: CreditReservation | None = None,
        internal_limit_reservation: InternalImageModelDailyLimitReservation | None = None,
        pricing: dict[str, object] | None = None,
    ):
        self.mode = mode
        self.user_id = user_id
        self.amount = _credit_amount(amount)
        self.per_image_cost = _credit_amount(per_image_cost)
        self.group_id = group_id
        self.model_id = model_id
        self.credit_reservation = credit_reservation
        self.internal_limit_reservation = internal_limit_reservation
        self.pricing = dict(pricing or {})
        self.actual = 0
        self.refund = 0
        self.log_detail: dict[str, object] = {}
        if self.pricing:
            self.log_detail.update({
                "billing_base_unit_price": self.pricing.get("base_unit_price", 1),
                "billing_effective_unit_price": self.pricing.get("unit_price", 1),
                "billing_pay_percentage": self.pricing.get("pay_percentage", 100),
                "billing_rule_id": self.pricing.get("rule_id", ""),
                "billing_rule_name": self.pricing.get("rule_name", ""),
                "billing_estimated": self.pricing.get("estimated_total", self.amount),
                "billing_actual": 0,
                "billing_refund": 0,
                "billing_priced_at": self.pricing.get("priced_at") or self.pricing.get("evaluated_at", ""),
            })
        self.settled = False

    @classmethod
    def reserve(cls, identity: dict[str, object], payload: dict[str, object], *, requested_amount: int | None = None) -> "ImageBillingReservation":
        amount = _to_positive_image_count(requested_amount if requested_amount is not None else payload.get("n"), 1)
        resolution = str(payload.get("resolution") or payload.get("image_resolution") or "").strip().lower()
        group_id = str(payload.get("third_party_image_group_id") or "").strip()
        use_third_party = resolution in {"1k", "2k", "4k"} or bool(group_id)
        route = None
        if not use_third_party:
            find_named_route = getattr(third_party_image_service, "find_named_route", None)
            route = find_named_route(payload.get("model"), advance=False) if callable(find_named_route) else None
            use_third_party = route is not None
        if not use_third_party:
            model_id = str(payload.get("model") or "gpt-image-2").strip() or "gpt-image-2"
            is_normal = str(identity.get("role") or "").strip().lower() == "normal"
            pricing = pool_pricing_service.quote(identity, quantity=amount) if is_normal else {}
            estimated_total = _credit_amount(pricing.get("estimated_total", amount))
            return cls(
                mode="general",
                credit_reservation=CreditReservation.reserve(
                    identity,
                    estimated_total,
                    model_id=DAILY_GIFT_ACCOUNT_POOL_MODEL_ID,
                ),
                internal_limit_reservation=InternalImageModelDailyLimitReservation.reserve(identity, model_id, amount),
                amount=estimated_total,
                per_image_cost=_credit_amount(pricing.get("unit_price", 1)),
                pricing=pricing,
            )

        require_4k = resolution == "4k"
        if route is None and str(identity.get("role") or "").strip().lower() == "normal":
            find_wallet_route = getattr(third_party_image_service, "find_wallet_route_for_identity", None)
            if callable(find_wallet_route):
                route = find_wallet_route(
                    identity,
                    payload.get("model"),
                    group_id=group_id,
                    require_4k=require_4k,
                    resolution=resolution,
                    advance=False,
                )
        if route is None:
            route = third_party_image_service.find_route(
                payload.get("model"),
                group_id=group_id,
                require_4k=require_4k,
                resolution=resolution or "1k",
                advance=False,
            )
        if route is None:
            raise ValueError("No enabled image API model route")
        try:
            third_party_image_service.assert_access(identity, route)
        except PermissionError as exc:
            raise PermissionError(str(exc)) from exc
        payload["_third_party_image"] = True
        payload["_third_party_image_group_id"] = str(route.group.get("id") or "")
        payload["third_party_image_group_id"] = str(route.group.get("id") or "")
        payload["resolution"] = str(route.model.get("resolution") or resolution or "1k")
        per_image_cost = max(0, third_party_image_service.cost_for_route(route, payload.get("quality") or payload.get("image_quality"), 1))
        total_cost = per_image_cost * amount
        group_id = str(route.group.get("id") or "")
        model_id = str(route.model.get("id") or "")
        billing_mode = str(route.group.get("billing_mode") or "")
        if billing_mode == "general_multiplier" and _third_party_wallet_remaining(identity, group_id, model_id) > 0:
            # A stale group setting must not strand an existing model-specific
            # dedicated balance. Treat it as "both" until that balance is used.
            billing_mode = "both"
        internal_limit_reservation = InternalImageModelDailyLimitReservation.reserve(identity, model_id, amount)
        if str(identity.get("role") or "").strip().lower() != "normal":
            return cls(
                mode=billing_mode,
                amount=0,
                per_image_cost=per_image_cost,
                group_id=group_id,
                model_id=model_id,
                internal_limit_reservation=internal_limit_reservation,
            )
        user_id = str(identity.get("id") or "").strip()
        wallet_pricing = _third_party_activity_quote(identity, amount, group_id=group_id, base_unit_price=1)
        wallet_total_cost = _credit_amount(wallet_pricing.get("estimated_total", amount))
        wallet_unit_cost = _credit_amount(wallet_pricing.get("unit_price", 1))
        if billing_mode == "dedicated_wallet" or (
            billing_mode == "both" and per_image_cost > 0 and _third_party_wallet_remaining(identity, group_id, model_id) >= wallet_total_cost
        ):
            if wallet_total_cost > 0:
                try:
                    auth_service.reserve_third_party_image_wallet(
                        user_id,
                        group_id=group_id,
                        model_id=model_id,
                        amount=wallet_total_cost,
                    )
                except ValueError:
                    if billing_mode != "both":
                        raise
                else:
                    return cls(
                        mode="dedicated_wallet",
                        user_id=user_id,
                        amount=wallet_total_cost,
                        per_image_cost=wallet_unit_cost,
                        group_id=group_id,
                        model_id=model_id,
                        internal_limit_reservation=internal_limit_reservation,
                        pricing=wallet_pricing,
                    )
            else:
                return cls(
                    mode="dedicated_wallet",
                    user_id=user_id,
                    amount=0,
                    per_image_cost=0,
                    group_id=group_id,
                    model_id=model_id,
                    internal_limit_reservation=internal_limit_reservation,
                    pricing=wallet_pricing,
                )
        credit_pricing = _third_party_activity_quote(identity, amount, group_id=group_id, base_unit_price=per_image_cost)
        total_cost = _credit_amount(credit_pricing.get("estimated_total", total_cost))
        effective_unit_cost = _credit_amount(credit_pricing.get("unit_price", per_image_cost))
        return cls(
            mode=billing_mode if billing_mode == "both" else "general_multiplier",
            amount=total_cost,
            per_image_cost=effective_unit_cost,
            credit_reservation=CreditReservation.reserve(
                identity,
                total_cost,
                model_id=f"{group_id}:{model_id}",
            ),
            internal_limit_reservation=internal_limit_reservation,
            group_id=group_id,
            model_id=model_id,
            pricing=credit_pricing,
        )

    def settle(self, success_count: int) -> None:
        if self.settled:
            return
        if self.internal_limit_reservation is not None:
            self.internal_limit_reservation.settle(success_count)
        success_cost = _credit_amount(min(float(self.amount), max(0, int(success_count or 0)) * float(self.per_image_cost)))
        if self.mode == "dedicated_wallet" and self.user_id and self.amount > 0:
            if success_cost > 0:
                auth_service.consume_third_party_image_wallet(
                    self.user_id,
                    group_id=self.group_id,
                    model_id=self.model_id,
                    amount=success_cost,
                )
            refund = self.amount - success_cost
            if refund > 0:
                auth_service.refund_third_party_image_wallet(
                    self.user_id,
                    group_id=self.group_id,
                    model_id=self.model_id,
                    amount=refund,
                )
        elif self.credit_reservation is not None:
            self.credit_reservation.settle(success_cost)
        self.actual = success_cost
        self.refund = _credit_amount(float(self.amount) - float(success_cost))
        if self.pricing:
            self.log_detail["billing_actual"] = self.actual
            self.log_detail["billing_refund"] = self.refund
        self.settled = True


async def run_metered_image_call(
    call: LoggedCall,
    handler,
    payload: dict[str, object],
    identity: dict[str, object],
    *,
    requested_amount: int | None = None,
):
    try:
        reservation = ImageBillingReservation.reserve(identity, payload, requested_amount=requested_amount)
    except PermissionError as exc:
        raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
    except ValueError as exc:
        raise HTTPException(status_code=402, detail={"error": str(exc)}) from exc

    call.billing_detail = reservation.log_detail

    stream_success_count = 0

    def on_stream_item(item: object) -> None:
        nonlocal stream_success_count
        stream_success_count += _count_successful_images(item)

    def on_stream_complete() -> None:
        reservation.settle(stream_success_count)

    def on_stream_error(_exc: BaseException) -> None:
        reservation.settle(stream_success_count)

    try:
        result = await call.run(
            handler,
            payload,
            on_stream_item=on_stream_item,
            on_stream_complete=on_stream_complete,
            on_stream_error=on_stream_error,
            on_result=lambda value: reservation.settle(_count_successful_images(value)),
        )
    except Exception:
        reservation.settle(0)
        raise
    if isinstance(result, StreamingResponse):
        return result
    reservation.settle(_count_successful_images(result))
    return result


def _external_prompt_optimization(prompt: str) -> tuple[str, str]:
    base_url = config.prompt_optimizer_base_url.rstrip("/")
    model = config.prompt_optimizer_model
    if not base_url:
        raise ValueError("第三方提示词优化 Base URL 不能为空")
    endpoint = base_url if base_url.endswith("/chat/completions") else f"{base_url}/chat/completions"
    headers = {"Content-Type": "application/json"}
    if config.prompt_optimizer_api_key:
        headers["Authorization"] = f"Bearer {config.prompt_optimizer_api_key}"
    response = curl_requests.post(
        endpoint,
        headers=headers,
        json={
            "model": model,
            "stream": False,
            "messages": [
                {"role": "system", "content": config.prompt_optimizer_system_prompt},
                {
                    "role": "user",
                    "content": "只优化下面的原始提示词，保持原语言并仅返回改写后的提示词：\n" + prompt,
                },
            ],
        },
        timeout=60,
    )
    if response.status_code >= 400:
        raise RuntimeError(f"第三方提示词优化接口返回 HTTP {response.status_code}: {response.text[:500]}")
    data = response.json()
    text = _chat_completion_text(data) if isinstance(data, dict) else ""
    if not text:
        raise RuntimeError("第三方提示词优化接口未返回有效内容")
    return text, model


def _external_reverse_prompt(image: str) -> tuple[str, str]:
    base_url = config.reverse_prompt_base_url.rstrip("/")
    model = config.reverse_prompt_model
    if not base_url:
        raise ValueError("第三方反推提示词 Base URL 不能为空")
    endpoint = base_url if base_url.endswith("/chat/completions") else f"{base_url}/chat/completions"
    headers = {"Content-Type": "application/json"}
    if config.reverse_prompt_api_key:
        headers["Authorization"] = f"Bearer {config.reverse_prompt_api_key}"
    instruction = config.reverse_prompt_system_prompt
    response = curl_requests.post(
        endpoint,
        headers=headers,
        json={
            "model": model,
            "stream": False,
            "messages": [
                {"role": "system", "content": instruction},
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": instruction},
                        {"type": "image_url", "image_url": {"url": image}},
                    ],
                },
            ],
        },
        timeout=120,
    )
    if response.status_code >= 400:
        raise RuntimeError(f"第三方反推提示词接口返回 HTTP {response.status_code}: {response.text[:500]}")
    data = response.json()
    text = _chat_completion_text(data) if isinstance(data, dict) else ""
    if not text:
        raise RuntimeError("第三方反推提示词接口未返回有效内容")
    return text, model


def collect_models_result(identity: dict[str, object]) -> dict[str, Any]:
    """上游账号池模型 + 第三方接入模型，合并且按 id 去重后的完整目录。

    `/v1/models`（对外 OpenAI 兼容）与 `/api/chat/models`（对话页、
    管理页的「对话可用模型」）必须用同一份目录，否则两边口径会不一致——
    之前 `/api/chat/models` 只取 `openai_v1_models.list_models()`，
    漏掉了在这里额外拼接的第三方模型，导致第三方模型在对话候选里根本不出现。
    """

    result = openai_v1_models.list_models()
    data = result.get("data") if isinstance(result, dict) else None
    if not isinstance(data, list):
        return result
    seen = {str(item.get("id") or "").strip() for item in data if isinstance(item, dict)}
    for item in third_party_image_service.public_models_for_identity(identity):
        model_id = str(item.get("id") or "").strip()
        if not model_id or model_id in seen:
            continue
        data.append(item)
        seen.add(model_id)
    return result


def create_router() -> APIRouter:
    router = APIRouter()

    @router.get("/v1/models")
    async def list_models(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        try:
            return await run_in_threadpool(collect_models_result, identity)
        except Exception as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc

    @router.get("/api/chat/models")
    async def list_chat_models(authorization: str | None = Header(default=None)):
        """对话页模型下拉 / 管理页「对话可用模型」的数据源。

        `/v1/models` 是给外部 API 客户端用的 OpenAI 兼容接口，必须保持全量
        （含第三方生图模型），所以这里另开一个接口：全量候选 + 每个模型的
        `chat_enabled` 标记（自动规则叠加人工覆盖）。

        候选包含第三方接入模型——它们默认隐藏（生图模型），但可以被手动开启；
        将来第三方接入文本 / 多模态模型时，只要不带生图能力就会默认出现在对话下拉里。
        """

        identity = require_identity(authorization)
        try:
            result = await run_in_threadpool(collect_models_result, identity)
        except Exception as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
        items = result.get("data") if isinstance(result, dict) else None
        overrides = config.chat_models
        data = resolve_chat_models(items if isinstance(items, list) else [], overrides)
        return {
            "object": "list",
            "data": data,
            "include": overrides["include"],
            "exclude": overrides["exclude"],
        }

    @router.get("/api/image/models")
    async def list_image_models(authorization: str | None = Header(default=None)):
        """生图页模型下拉 / 管理页「生图可用模型」的数据源。

        只列出能生图的模型（账号池上游内置生图模型 + 第三方接入生图模型），
        每个带 `image_enabled` 标记。默认全部启用 —— 这里的开关只是展示层过滤，
        不改动第三方生图配置（计费 / 路由 / 额度 / 对外 API 均不受影响）。
        """

        identity = require_identity(authorization)
        try:
            result = await run_in_threadpool(collect_models_result, identity)
        except Exception as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
        items = result.get("data") if isinstance(result, dict) else None
        overrides = config.image_models
        data = resolve_image_models(items if isinstance(items, list) else [], overrides)
        return {
            "object": "list",
            "data": data,
            "include": overrides["include"],
            "exclude": overrides["exclude"],
        }

    @router.get("/api/system/models")
    async def list_system_models(authorization: str | None = Header(default=None)):
        """管理页「系统可用模型」的数据源。

        与 `/v1/models` 同一份全量目录（上游账号池 + 第三方接入），
        额外带上 `system_enabled` 标记：管理员可以把不关心的模型从这张列表里收起来。

        注意这**只是展示层过滤**：`/v1/models` 本身仍保持全量，
        上游账号池、第三方配置、计费与路由都不受影响。
        """

        identity = require_identity(authorization)
        try:
            result = await run_in_threadpool(collect_models_result, identity)
        except Exception as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
        items = result.get("data") if isinstance(result, dict) else None
        overrides = config.system_models
        data = resolve_system_models(items if isinstance(items, list) else [], overrides)
        return {
            "object": "list",
            "data": data,
            "include": overrides["include"],
            "exclude": overrides["exclude"],
        }

    @router.post("/v1/images/generations")
    async def generate_images(
            body: ImageGenerationRequest,
            request: Request,
            authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        payload = body.model_dump(mode="python")
        payload["base_url"] = resolve_image_base_url(request)
        _attach_account_pool(payload, identity)
        call = LoggedCall(identity, "/v1/images/generations", body.model, "文生图", request_text=body.prompt)
        await filter_or_log(call, body.prompt)
        return await run_metered_image_call(call, openai_v1_image_generations.handle, payload, identity)

    @router.post("/v1/images/edits")
    async def edit_images(
            request: Request,
            authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        payload, image_sources, mask_sources = await parse_image_edit_request(request)
        prompt = str(payload["prompt"])
        model = str(payload["model"])
        call = LoggedCall(identity, "/v1/images/edits", model, "图生图", request_text=prompt)
        await filter_or_log(call, prompt)
        payload["images"] = await read_image_sources(image_sources)
        if mask_sources:
            payload["mask"] = await read_image_sources(mask_sources)
        payload["base_url"] = resolve_image_base_url(request)
        _attach_account_pool(payload, identity)
        return await run_metered_image_call(call, openai_v1_image_edit.handle, payload, identity)

    @router.post("/v1/chat/completions")
    async def create_chat_completion(
        body: ChatCompletionRequest,
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        payload = body.model_dump(mode="python")
        model = str(payload.get("model") or "auto")
        request_preview = request_text(payload.get("prompt"), payload.get("messages"))
        is_image_request = is_image_chat_request(payload)
        payload["base_url"] = resolve_image_base_url(request)
        if is_image_request:
            _attach_account_pool(payload, identity)
        call = LoggedCall(identity, "/v1/chat/completions", model, "图片生成" if is_image_request else "文本生成", request_text=request_preview)
        if is_image_request:
            # AI review protects image accounts. Plain chat is also used by
            # internal Skill controllers and image planners; reviewing their
            # system protocols creates false positives and never protects an
            # image-account call because no image account is used here.
            await filter_or_log(call, request_preview)
            return await run_metered_image_call(
                call,
                openai_v1_chat_complete.handle,
                payload,
                identity,
                requested_amount=parse_image_count(payload.get("n")),
            )
        return await call.run(openai_v1_chat_complete.handle, payload, identity)

    @router.post("/api/prompt/optimize")
    async def optimize_prompt(
        body: PromptOptimizationRequest,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        prompt = body.prompt.strip()
        if not prompt:
            raise HTTPException(status_code=400, detail={"error": "提示词不能为空"})
        if config.prompt_optimizer_provider == "external":
            try:
                optimized_prompt, model = await run_in_threadpool(_external_prompt_optimization, prompt)
            except (ValueError, RuntimeError) as exc:
                raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
            return {"prompt": optimized_prompt, "model": model}
        payload = {
            "model": config.prompt_optimizer_model,
            "stream": False,
            "messages": [
                {"role": "system", "content": config.prompt_optimizer_system_prompt},
                {
                    "role": "user",
                    "content": (
                        f"{config.prompt_optimizer_system_prompt}\n\n"
                        "Improve only the original prompt below. Preserve its language exactly and return only the rewritten prompt.\n"
                        f"<original_prompt>\n{prompt}\n</original_prompt>"
                    ),
                },
            ],
        }
        call = LoggedCall(identity, "/api/prompt/optimize", config.prompt_optimizer_model, "提示词优化", request_text=prompt)
        result = await call.run(openai_v1_chat_complete.handle, payload, identity)
        if not isinstance(result, dict):
            return result
        optimized_prompt = _chat_completion_text(result)
        if not optimized_prompt:
            raise HTTPException(status_code=502, detail={"error": "模型未返回优化后的提示词"})
        return {"prompt": optimized_prompt, "model": config.prompt_optimizer_model}

    @router.post("/api/prompt/reverse")
    async def reverse_prompt(
        body: ReversePromptRequest,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        image = body.image.strip()
        header, separator, encoded = image.partition(",")
        if not separator or not header.startswith("data:image/") or ";base64" not in header or not encoded.strip():
            raise HTTPException(status_code=400, detail={"error": "图片数据无效"})
        if config.reverse_prompt_provider == "external":
            try:
                prompt, model = await run_in_threadpool(_external_reverse_prompt, image)
            except (ValueError, RuntimeError) as exc:
                raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
            return {"prompt": prompt, "model": model}
        instruction = config.reverse_prompt_system_prompt
        model = config.reverse_prompt_model
        payload = {
            "model": model,
            "stream": False,
            "instructions": instruction,
            "input": [
                {"type": "input_text", "text": instruction},
                {"type": "input_image", "image_url": image},
            ],
        }
        call = LoggedCall(identity, "/api/prompt/reverse", model, "反推提示词", request_text="参考图片")
        result = await call.run(openai_v1_response.handle, payload)
        if not isinstance(result, dict):
            return result
        prompt = _response_text(result)
        if not prompt:
            raise HTTPException(status_code=502, detail={"error": "模型未返回反推提示词"})
        return {"prompt": prompt, "model": model}

    @router.post("/api/image-annotations/identify")
    async def identify_annotation_region(
        body: ImageRegionRecognitionRequest,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        try:
            return await run_in_threadpool(
                identify_image_region,
                identity,
                body.image,
                body.bounds.model_dump(mode="python"),
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        except RuntimeError as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc

    @router.post("/v1/responses")
    async def create_response(
        body: ResponseCreateRequest,
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        payload = body.model_dump(mode="python")
        model = str(payload.get("model") or "auto")
        request_preview = request_text(payload.get("input"), payload.get("instructions"))
        is_image_request = has_response_image_generation_tool(payload)
        if is_image_request:
            payload["base_url"] = resolve_image_base_url(request)
            _attach_account_pool(payload, identity)
        call = LoggedCall(identity, "/v1/responses", model, "Responses 图片生成" if is_image_request else "Responses", request_text=request_preview)
        if is_image_request:
            await filter_or_log(call, request_preview)
            return await run_metered_image_call(
                call,
                openai_v1_response.handle,
                payload,
                identity,
                requested_amount=1,
            )
        return await call.run(openai_v1_response.handle, payload)

    @router.post("/v1/messages")
    async def create_message(
            body: AnthropicMessageRequest,
            authorization: str | None = Header(default=None),
            x_api_key: str | None = Header(default=None, alias="x-api-key"),
            anthropic_version: str | None = Header(default=None, alias="anthropic-version"),
    ):
        identity = require_identity(authorization or (f"Bearer {x_api_key}" if x_api_key else None))
        payload = body.model_dump(mode="python")
        model = str(payload.get("model") or "auto")
        request_preview = request_text(payload.get("system"), payload.get("messages"), payload.get("tools"))
        call = LoggedCall(identity, "/v1/messages", model, "Messages", request_text=request_preview)
        return await call.run(anthropic_v1_messages.handle, payload, sse="anthropic")

    @router.post("/v1/search")
    async def search(body: SearchRequest, authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        call = LoggedCall(identity, "/v1/search", openai_search.MODEL, "搜索", request_text=body.prompt)
        return await call.run(openai_search.handle, body.model_dump(mode="python"))

    @router.get("/v1/editable-file-tasks")
    async def list_editable_file_tasks(ids: str = "", authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        task_ids = [item.strip() for item in ids.split(",") if item.strip()]
        return await run_in_threadpool(editable_file_task_service.list_tasks, identity, task_ids)

    @router.get("/files/{file_path:path}")
    async def download_editable_file(file_path: str, download: bool = False):
        if file_path.startswith("skills/"):
            await run_in_threadpool(skill_service.cleanup_expired_runs)
        try:
            path = await run_in_threadpool(editable_file_task_service.public_file_path, file_path)
        except Exception as exc:
            raise HTTPException(status_code=404, detail={"error": "file not found"}) from exc
        media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
        if path.suffix.lower() == ".svg":
            media_type = "image/svg+xml"
        if download:
            return FileResponse(path, media_type=media_type, filename=path.name, content_disposition_type="attachment")
        return FileResponse(path, media_type=media_type)

    @router.post("/v1/ppt/generations")
    async def create_ppt_task(body: EditableFileTaskRequest, request: Request, authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        await filter_or_log(LoggedCall(identity, "/v1/ppt/generations", "gpt-5-5-thinking", "PPT生成任务", request_text=body.prompt), body.prompt)
        return await run_in_threadpool(
            editable_file_task_service.submit_ppt,
            identity,
            client_task_id=body.client_task_id or "",
            prompt=body.prompt,
            base64_images=body.base64_images,
            base_url=resolve_image_base_url(request),
        )

    @router.post("/v1/psd/generations")
    async def create_psd_task(body: EditableFileTaskRequest, request: Request, authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        await filter_or_log(LoggedCall(identity, "/v1/psd/generations", "gpt-5-5-thinking", "PSD生成任务", request_text=body.prompt), body.prompt)
        return await run_in_threadpool(
            editable_file_task_service.submit_psd,
            identity,
            client_task_id=body.client_task_id or "",
            prompt=body.prompt,
            base64_images=body.base64_images,
            base_url=resolve_image_base_url(request),
        )

    return router