"""OpenRouter API: модели и chat completions.""" from __future__ import annotations import json import logging import re from dataclasses import dataclass, field from typing import Any import httpx logger = logging.getLogger(__name__) OPENROUTER_BASE = "https://openrouter.ai/api/v1" DEFAULT_MODELS = ( "openai/gpt-4o-mini", "google/gemini-2.5-flash", "anthropic/claude-3.5-sonnet", "deepseek/deepseek-chat", "meta-llama/llama-3.3-70b-instruct", ) DEFAULT_SYSTEM_PROMPT = """Ты — ИИ-поддержка VPN-сервиса (Remnawave / VPN Service). Отвечай кратко, по делу, на русском. Помогай с подключением, клиентами, тарифами, балансом, трафиком и устройствами. Правила: - Опирайся на FAQ и переписку тикета. - Не выдумывай факты о балансе, платежах и сроках подписки пользователя — если нужны точные данные аккаунта, эскалируй. - Не обещай возврат денег без эскалации. - Не раскрывай внутренние API, ключи, админку. - Если вопрос сложный, спорный, про оплату/возврат/бан или ты не уверен — эскалируй человеку. Формат ответа — ТОЛЬКО JSON без markdown: {"action":"answer","reply":"текст ответа пользователю"} или {"action":"escalate","reason":"кратко почему нужен человек"} """ @dataclass(slots=True) class OpenRouterConfig: enabled: bool = False api_key: str = "" model: str = DEFAULT_MODELS[0] models: list[str] = field(default_factory=lambda: list(DEFAULT_MODELS[:3])) system_prompt: str = DEFAULT_SYSTEM_PROMPT auto_reply: bool = True @property def ready(self) -> bool: return bool(self.enabled and self.api_key.strip() and self.model.strip()) def models_for_request(self) -> list[str]: """Primary + fallbacks (без дублей).""" ordered: list[str] = [] for mid in [self.model, *self.models]: mid = (mid or "").strip() if mid and mid not in ordered: ordered.append(mid) return ordered @dataclass(slots=True) class AiDecision: action: str # answer | escalate reply: str = "" reason: str = "" model: str = "" raw: str = "" def _truthy(value: str) -> bool: return (value or "").strip().lower() in ("1", "true", "yes", "on", "да") def load_openrouter_config(db) -> OpenRouterConfig: raw = db.get_settings_map( [ "openrouter_enabled", "openrouter_api_key", "openrouter_model", "openrouter_models", "openrouter_system_prompt", "openrouter_auto_reply", ] ) models: list[str] = [] models_raw = (raw.get("openrouter_models") or "").strip() if models_raw: try: parsed = json.loads(models_raw) if isinstance(parsed, list): models = [str(x).strip() for x in parsed if str(x).strip()] except json.JSONDecodeError: models = [x.strip() for x in models_raw.split(",") if x.strip()] if not models: models = list(DEFAULT_MODELS[:3]) model = (raw.get("openrouter_model") or "").strip() or (models[0] if models else DEFAULT_MODELS[0]) prompt = (raw.get("openrouter_system_prompt") or "").strip() or DEFAULT_SYSTEM_PROMPT auto_raw = raw.get("openrouter_auto_reply") auto_reply = True if auto_raw == "" else _truthy(auto_raw) return OpenRouterConfig( enabled=_truthy(raw.get("openrouter_enabled") or ""), api_key=(raw.get("openrouter_api_key") or "").strip(), model=model, models=models, system_prompt=prompt, auto_reply=auto_reply, ) def mask_api_key(key: str) -> str: key = (key or "").strip() if not key: return "" if len(key) <= 8: return "••••••••" return f"{key[:6]}…{key[-4:]}" def config_public_dict(cfg: OpenRouterConfig) -> dict[str, Any]: return { "enabled": cfg.enabled, "api_key_set": bool(cfg.api_key), "api_key_masked": mask_api_key(cfg.api_key), "model": cfg.model, "models": cfg.models, "system_prompt": cfg.system_prompt, "auto_reply": cfg.auto_reply, "ready": cfg.ready, "defaults": list(DEFAULT_MODELS), } def _headers(api_key: str, *, referer: str = "", title: str = "VPN Bot Support") -> dict[str, str]: headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } if referer: headers["HTTP-Referer"] = referer headers["X-Title"] = title return headers async def list_models(api_key: str, *, referer: str = "") -> list[dict[str, Any]]: async with httpx.AsyncClient(timeout=45.0) as client: resp = await client.get( f"{OPENROUTER_BASE}/models", headers=_headers(api_key, referer=referer), ) resp.raise_for_status() data = resp.json() items = data.get("data") if isinstance(data, dict) else data out: list[dict[str, Any]] = [] for item in items or []: mid = str(item.get("id") or "").strip() if not mid: continue out.append( { "id": mid, "name": str(item.get("name") or mid), "context_length": item.get("context_length"), "pricing": item.get("pricing") or {}, } ) out.sort(key=lambda x: x["id"].lower()) return out async def chat_completion( *, api_key: str, model: str, messages: list[dict[str, str]], models: list[str] | None = None, referer: str = "", temperature: float = 0.3, max_tokens: int = 900, ) -> tuple[str, str]: """Возвращает (text, used_model).""" payload: dict[str, Any] = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } fallbacks = [m for m in (models or []) if m and m != model] if fallbacks: payload["models"] = [model, *fallbacks] payload["route"] = "fallback" async with httpx.AsyncClient(timeout=90.0) as client: resp = await client.post( f"{OPENROUTER_BASE}/chat/completions", headers=_headers(api_key, referer=referer), json=payload, ) if resp.status_code >= 400: detail = resp.text[:500] raise RuntimeError(f"OpenRouter HTTP {resp.status_code}: {detail}") data = resp.json() used = str(data.get("model") or model) choices = data.get("choices") or [] if not choices: raise RuntimeError("OpenRouter: пустой ответ") message = choices[0].get("message") or {} content = message.get("content") if isinstance(content, list): parts = [] for part in content: if isinstance(part, dict) and part.get("type") == "text": parts.append(str(part.get("text") or "")) elif isinstance(part, str): parts.append(part) text = "\n".join(parts).strip() else: text = str(content or "").strip() if not text: raise RuntimeError("OpenRouter: пустой content") return text, used _JSON_RE = re.compile(r"\{[\s\S]*\}") def parse_ai_decision(raw: str, *, used_model: str = "") -> AiDecision: text = (raw or "").strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text).strip() payload: dict[str, Any] | None = None try: payload = json.loads(text) except json.JSONDecodeError: match = _JSON_RE.search(text) if match: try: payload = json.loads(match.group(0)) except json.JSONDecodeError: payload = None if isinstance(payload, dict): action = str(payload.get("action") or "").strip().lower() reply = str(payload.get("reply") or payload.get("message") or "").strip() reason = str(payload.get("reason") or "").strip() if action == "escalate": return AiDecision( action="escalate", reason=reason or reply or "Нужен человек", model=used_model, raw=raw, ) if action == "answer" and reply: return AiDecision(action="answer", reply=reply, model=used_model, raw=raw) if reply and not action: return AiDecision(action="answer", reply=reply, model=used_model, raw=raw) # Fallback: если модель ответила обычным текстом — считаем ответом if len(text) >= 8 and "{" not in text[:20]: return AiDecision(action="answer", reply=text, model=used_model, raw=raw) return AiDecision( action="escalate", reason="ИИ вернул неразборчивый ответ", model=used_model, raw=raw, )