"""根据配置创建 Browser Use 可用的 LLM 实例；主 Key 余额不足时自动切兜底 Key（见 llm_keys）。"""

from __future__ import annotations

import os
import asyncio
from dataclasses import dataclass, field
from typing import Any

from backend.browser import llm_keys
from backend.browser import diagnostics
from backend.browser import structured_output
from backend.config import MODELS_BY_PROVIDER, settings
from backend.db import database as db
from backend.prospecting import usage as model_usage


def _setting(key: str, env_key: str, default: str = "") -> str:
    return db.get_setting(key) or getattr(settings, key, "") or os.getenv(env_key, default)


def get_llm_config() -> dict:
    provider = _setting("llm_provider", "LLM_PROVIDER", settings.llm_provider)
    model = _setting("llm_model", "LLM_MODEL", settings.llm_model)
    base_url = _setting("llm_base_url", "LLM_BASE_URL", settings.llm_base_url)

    if provider == "qwen":
        primary = _setting("dashscope_api_key", "DASHSCOPE_API_KEY", settings.dashscope_api_key)
        if not base_url or "deepseek" in base_url:
            base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"
    else:
        provider = "deepseek"
        primary = _setting("llm_api_key", "DEEPSEEK_API_KEY", settings.deepseek_api_key)
        if not primary:
            primary = _setting("deepseek_api_key", "DEEPSEEK_API_KEY", settings.deepseek_api_key)
        if not base_url:
            base_url = "https://api.deepseek.com/v1"

    fallback = llm_keys.fallback_key(provider)
    if fallback == primary:
        fallback = ""
    api_key, using_fallback = llm_keys.choose(provider, primary)

    return {
        "provider": provider,
        "model": model,
        # 本次应使用的 Key（主 Key 已标记耗尽且有兜底时就是兜底）
        "api_key": api_key,
        "primary_api_key": primary,
        "fallback_api_key": fallback,
        "using_fallback": using_fallback,
        "base_url": base_url,
        "available_models": MODELS_BY_PROVIDER.get(provider, MODELS_BY_PROVIDER["deepseek"]),
    }


def _build_model(cfg: dict, api_key: str):
    if cfg["provider"] == "qwen":
        from browser_use import ChatOpenAI

        return ObservedChatModel(ChatOpenAI(model=cfg["model"], api_key=api_key, base_url=cfg["base_url"], temperature=0.1), cfg["model"])

    if cfg["model"] in {"deepseek-flash", "deepseek-v4-flash", "deepseek-v4-flash-vision-exp", "deepseek-v4-pro"}:
        return DeepSeekBrowserChat(model=cfg["model"], api_key=api_key, base_url=cfg["base_url"])

    from browser_use import ChatDeepSeek

    return ObservedChatModel(ChatDeepSeek(model=cfg["model"], api_key=api_key, base_url=cfg["base_url"], temperature=0.1), cfg["model"])


@dataclass
class ObservedChatModel:
    """Observe legacy adapters without changing their model or request options.

    Their internal failed validation may not expose provider usage. Those calls
    remain explicitly unknown instead of being reported as free requests.
    """
    wrapped: Any = field(repr=False)
    model: str
    _verified_api_keys: bool = False

    @property
    def provider(self) -> str:
        return str(self.wrapped.provider)

    @property
    def name(self) -> str:
        return str(self.wrapped.name)

    @property
    def model_name(self) -> str:
        return self.model

    async def ainvoke(self, messages: list, output_format: Any = None, **kwargs: Any):
        call = model_usage.start_call(self.model)
        call_status, status_code = "connection_error", None
        try:
            result = await self.wrapped.ainvoke(messages, output_format, **kwargs)
            model_usage.record_response(call, getattr(result, "usage", None), model=self.model,
                                        finish_reason=getattr(result, "stop_reason", None))
            call_status = "succeeded"
            return result
        except asyncio.CancelledError:
            call_status = "cancelled"
            raise
        except Exception as error:
            status_code = getattr(error, "status_code", None)
            if type(status_code) is int:
                call_status = "http_error"
            elif type(error).__name__ == "ModelOutputTruncatedError":
                call_status = "truncated"
            raise
        finally:
            model_usage.finish_call(call, call_status, http_status=status_code)


@dataclass
class DeepSeekBrowserChat:
    """Browser-use 0.13.10 adapter for current DeepSeek models.

    Preserve images, disabled thinking and usage accounting while returning the
    complete step through the provider's forced named-function-call protocol.
    Free text JSON mode does not enforce the required browser action envelope.
    """

    model: str
    api_key: str = field(repr=False)
    base_url: str = "https://api.deepseek.com/v1"
    timeout: float = 75.0
    max_tokens: int = 8192
    _verified_api_keys: bool = False

    @property
    def provider(self) -> str:
        return "deepseek"

    @property
    def name(self) -> str:
        return self.model

    @property
    def model_name(self) -> str:
        return self.model

    async def ainvoke(self, messages: list, output_format: Any = None, **kwargs: Any):
        from openai import APIConnectionError, APIStatusError, AsyncOpenAI, RateLimitError
        from pydantic import ValidationError

        from browser_use.llm.exceptions import ModelOutputTruncatedError, ModelProviderError, ModelRateLimitError
        from browser_use.llm.openai.serializer import OpenAIMessageSerializer
        from browser_use.llm.views import ChatInvokeCompletion, ChatInvokeUsage

        prepared = OpenAIMessageSerializer.serialize_messages(messages)
        params: dict[str, Any] = {
            "model": self.model,
            "temperature": 0.1,
            "max_tokens": self.max_tokens,
            "extra_body": {"thinking": {"type": "disabled"}},
        }
        if output_format is not None:
            function, instruction = structured_output.request_contract(output_format)
            # Resolve the framework's raw-JSON examples without duplicating the
            # schema or changing the original conversation's terminal role.
            prepared = structured_output.attach_instruction(prepared, instruction)
            params["tools"] = [function]
            params["tool_choice"] = {"type": "function", "function": {"name": structured_output.FUNCTION_NAME}}
        call = model_usage.start_call(self.model)
        call_status = "connection_error"
        status_code = None
        try:
            if output_format is not None:
                diagnostics.record_request_contract(call, function)
            async with AsyncOpenAI(
                api_key=self.api_key, base_url=self.base_url, timeout=self.timeout, max_retries=0,
            ) as client:
                response = await client.chat.completions.create(messages=prepared, **params)
            # Record what the provider actually billed before JSON/schema
            # validation can reject the action and hide its consumed tokens.
            choice = response.choices[0] if response.choices else None
            model_usage.record_response(call, response.usage, model=getattr(response, "model", None) or self.model,
                                        finish_reason=getattr(choice, "finish_reason", None))
            call_status = "invalid_response"
            if not response.choices:
                raise ModelProviderError("DeepSeek 未返回有效选择。", status_code=502, model=self.model)
            choice = response.choices[0]
            if choice.finish_reason == "length":
                call_status = "truncated"
                raise ModelOutputTruncatedError("DeepSeek 输出超过单步限制，未执行截断动作。", model=self.model)
            if choice.finish_reason not in {"stop", "tool_calls"}:
                raise ModelProviderError("DeepSeek 本次输出未正常完成，未执行动作。", status_code=422, model=self.model)
            if output_format is not None:
                completion = structured_output.validated_tool_result(choice.message, output_format)
            else:
                completion = choice.message.content
                if not isinstance(completion, str) or not completion.strip():
                    raise ModelProviderError("DeepSeek 未返回动作内容。", status_code=502, model=self.model)
            usage = None
            if response.usage is not None:
                counters = model_usage.normalize_usage(response.usage)
                usage = ChatInvokeUsage(
                    prompt_tokens=response.usage.prompt_tokens,
                    prompt_cached_tokens=counters["cache_hit_tokens"],
                    prompt_cache_creation_tokens=None,
                    prompt_image_tokens=None,
                    completion_tokens=response.usage.completion_tokens,
                    total_tokens=response.usage.total_tokens,
                )
            call_status = "succeeded"
            diagnostics.validated_response(getattr(call, "phase", "browser"))
            return ChatInvokeCompletion(completion=completion, usage=usage, stop_reason=choice.finish_reason)
        except RateLimitError as error:
            call_status, status_code = "http_error", 429
            raise ModelRateLimitError("DeepSeek 请求限流。", model=self.model) from error
        except APIStatusError as error:
            call_status, status_code = "http_error", error.status_code
            raise ModelProviderError(
                f"Error code: {error.status_code} - DeepSeek 请求失败，请检查 Key、余额和服务状态。",
                status_code=error.status_code, model=self.model,
            ) from error
        except APIConnectionError as error:
            call_status = "connection_error"
            raise ModelProviderError("无法连接 DeepSeek，本次请求未自动重试。", model=self.model) from error
        except ValidationError as error:
            call_status = "schema_error"
            diagnostic = diagnostics.schema_error(call, error, output_format)
            raise ModelProviderError("DeepSeek 返回的动作不符合任务结构，未执行。" + diagnostics.correction_hint(diagnostic), status_code=422, model=self.model) from None
        except structured_output.ResponseContractError as error:
            call_status = "schema_error"
            diagnostic = diagnostics.response_contract_error(call, error.reason)
            raise ModelProviderError("DeepSeek 返回的动作不符合任务结构，未执行。" + diagnostics.correction_hint(diagnostic), status_code=422, model=self.model) from None
        except asyncio.CancelledError:
            call_status = "cancelled"
            raise
        finally:
            if call_status != "schema_error":
                diagnostics.validated_response(getattr(call, "phase", "browser"))
            model_usage.finish_call(call, call_status, http_status=status_code)


@dataclass
class FallbackChatModel:
    """对 browser-use Agent 表现为普通 BaseChatModel：主 Key 402 时当场切到兜底 Key 重发这一次请求。

    Agent 构造时拿到的是这个包装对象，所以任务跑到一半余额耗尽也不会失败，只会多一次调用。
    """

    primary: Any
    fallback: Any
    model: str
    _verified_api_keys: bool = False
    _active: Any = field(init=False, default=None, repr=False)

    def __post_init__(self) -> None:
        self._active = self.fallback if llm_keys.primary_exhausted() else self.primary

    @property
    def provider(self) -> str:
        return str(self._active.provider)

    @property
    def name(self) -> str:
        return str(self._active.name)

    @property
    def model_name(self) -> str:
        return self.model

    @property
    def using_fallback(self) -> bool:
        return self._active is self.fallback

    async def ainvoke(self, messages: list, output_format: Any = None, **kwargs: Any):
        try:
            result = await self._active.ainvoke(messages, output_format, **kwargs)
        except Exception as error:  # noqa: BLE001  browser-use 把所有供应商错误都包成 ModelProviderError
            if self._active is self.primary and llm_keys.is_balance_error(error):
                llm_keys.mark_primary_exhausted(str(error))
                self._active = self.fallback
                return await self._active.ainvoke(messages, output_format, **kwargs)
            raise
        if self._active is self.primary:
            llm_keys.clear_primary_exhausted()
        return result


def create_llm():
    cfg = get_llm_config()
    if not cfg["api_key"]:
        raise RuntimeError(
            "未配置 API Key。DeepSeek 请在 .env 设置 DEEPSEEK_API_KEY，或在设置页保存。"
        )
    if cfg["primary_api_key"] and cfg["fallback_api_key"]:
        return FallbackChatModel(
            primary=_build_model(cfg, cfg["primary_api_key"]),
            fallback=_build_model(cfg, cfg["fallback_api_key"]),
            model=cfg["model"],
        )
    return _build_model(cfg, cfg["api_key"])


def use_vision_for_model(model: str, provider: str) -> bool:
    if provider == "qwen":
        return "vl" in model.lower()
    if provider == "deepseek":
        return model.lower() in {"deepseek-flash", "deepseek-v4-flash", "deepseek-v4-flash-vision-exp"}
    return False