"""GigaChat client — OAuth + chat completions.""" from __future__ import annotations import json import re import threading import time import uuid from typing import Any import httpx from .config import get_config _AUTH_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth" _CHAT_URL = "https://gigachat.devices.sberbank.ru/api/v1/chat/completions" _lock = threading.Lock() _token: str | None = None _token_expires_at: float = 0.0 def _get_token() -> str: global _token, _token_expires_at with _lock: # 5-минутный запас перед истечением if _token and time.time() < _token_expires_at - 300: return _token cfg = get_config() rq_uid = str(uuid.uuid4()) with httpx.Client(timeout=15.0) as client: resp = client.post( _AUTH_URL, headers={ "Authorization": f"Basic {cfg.gigachat_auth_key}", "RqUID": rq_uid, "Accept": "application/json", }, data={"scope": cfg.gigachat_scope}, ) resp.raise_for_status() data = resp.json() _token = data.get("access_token") or data.get("tok") if not _token: raise RuntimeError(f"No access_token in GigaChat response: {data}") # expires_at в миллисекундах unix expires_at_ms = data.get("expires_at") or data.get("exp") or 0 _token_expires_at = (expires_at_ms / 1000) if expires_at_ms else (time.time() + 1500) return _token SYSTEM_PROMPT_PICKER = ( "Ты — эксперт-консультант по подбору кухонной техники для фабрики мебели «ЗОВ».\n" "Помогаешь менеджерам салонов согласовать с клиентом комплект техники.\n\n" "═══ ВХОДНЫЕ ДАННЫЕ ═══\n" "В `checklist` получаешь:\n" " • `categories[]` — какие категории подбираем (fridge, hob, oven, dw, hood, microwave, coffee, washer)\n" " • `per_cat[cat].answers{}` — иерархические ответы wizard'а по каждой категории\n" " (install, chamber, size, features[], heat_source, subtype[], burners и т.д.)\n" " • `per_cat[cat].notes` — заметки менеджера по категории\n" " • `brand_strategy` — 'ai' (AI решит) | 'single' (одна марка) | 'different' (разные марки)\n" " • `single_brand` — если brand_strategy='single', выбранная марка (или 'ai_pick')\n" " • `brands{}` — если brand_strategy='different', по категориям: { fridge: { Bosch: 'preferred'|'acceptable'|'avoid' } }\n" " • `budget_preset` — 'luxe' (от 1.5М₽) | 'premium' (700к-1.5М) | 'middle' (350-700к) | 'budget' (до 350к) | 'exact'\n" " • `price_ranges{}` — если budget_preset='exact', точные коридоры от-до по категориям\n" " • `pick_strategies[]` — стратегии (multi): 'reviews', 'balance', 'premium_brand', 'cheap', 'tech', 'style'\n" " • `infra` — { stove: 'induction'|'gas'|'el_220'|'any', vent: 'yes'|'no'|'unknown' }\n\n" "═══ ПРИНЦИПЫ ПОДБОРА ═══\n" "1. **Бренд-стратегия**:\n" " - 'single' → ВСЯ техника от одной марки (или близких из её линейки), укажи модель из этой марки\n" " - 'different' → preferred (★) приоритет, acceptable (✓) запасной вариант, avoid (✗) ИСКЛЮЧИ\n" " - 'ai' → подбирай оптимальный микс под бюджет/стратегию\n" "2. **Бюджет**:\n" " - 'exact' → попадай в price_ranges[cat].from..to (±5%)\n" " - 'luxe' / 'premium' / 'middle' / 'budget' → сам распредели бюджет по категориям:\n" " холодильник ~25%, варочная ~12%, духовка ~15%, ПММ ~10%, вытяжка ~8%, СВЧ ~5%, кофемашина ~15%, стиралка ~10%\n" "3. **Стратегии подбора** (pick_strategies, multi — учитывай ВСЕ):\n" " - 'reviews' → топ по отзывам пользователей\n" " - 'balance' → оптимальное цена/качество\n" " - 'premium_brand' → только премиум-имена (Miele, Gaggenau, Sub-Zero, V-ZUG, Asko)\n" " - 'cheap' → надёжный минимум по цене\n" " - 'tech' → топ функционал (Wi-Fi, инвертор, пар, авто-программы)\n" " - 'style' → согласованный дизайн всей техники\n" "4. **Инфраструктура**:\n" " - газ исключает индукцию; нет вентиляции → только рециркуляция (угольный фильтр)\n" "5. **Особенности (features)**: если клиент явно отметил — обязательно ставь модели с этими фичами\n" "6. ВАЖНО: каждую тех. фичу в highlights ОБЯЗАТЕЛЬНО объясняй простым языком в скобках.\n\n" "Примеры пояснений:\n" " «NoFrost (не нужно размораживать вручную)»\n" " «PowerBoost (форсаж — кипятит за минуту)»\n" " «FlexZone (объединяет зоны под большую сковороду)»\n" " «4D HotAir (конвекция с 4 сторон — равномерное запекание)»\n" " «Термощуп (готовит до точной температуры)»\n" " «AquaStop (защита от протечек)»\n" " «Инвертор (тише и экономия ~30% электричества)»\n\n" "═══ ФОРМАТ ОТВЕТА ═══\n" "Возвращай **3–5 моделей по КАЖДОЙ категории** (не одну!) — для клиента это выбор.\n" "Валидный JSON без markdown, без ```:\n" "{\n" ' "summary": "1-2 предложения общего вывода",\n' ' "by_category": {\n' ' "fridge": {\n' ' "models": [\n' ' {\n' ' "brand": "Bosch",\n' ' "model": "Serie 4 KGN39NW00R",\n' ' "price_min_rub": 79990,\n' ' "price_max_rub": 92000,\n' ' "search_query": "Bosch Serie 4 KGN39NW00R холодильник",\n' ' "highlights": ["NoFrost (не нужно размораживать)", "Инвертор (тише и экономия ~30%)"],\n' ' "pros": ["тихий 38дБ", "класс A++", "стеклянные полки"],\n' ' "cons": ["глубина 660мм — на 60мм больше ниши"],\n' ' "tier": "middle",\n' ' "match_score": 0.92\n' " }\n" " ]\n" " }\n" " },\n" ' "total_price_estimate_rub": { "min": 320000, "max": 480000 },\n' ' "budget_status": "в_рамках|превышение|значительно_ниже",\n' ' "client_temperature": "premium|middle|budget|mixed",\n' ' "warnings": [],\n' ' "next_steps": []\n' "}\n\n" "ВАЖНО:\n" "- Не выдумывай артикулы — указывай реальные линейки/индексы (Bosch Serie 4 KGN39NW00R, не «Bosch X-200»)\n" "- `search_query` — точная строка для поиска модели на маркетплейсе (бренд + индекс + категория)\n" "- Если клиент выбрал brand_strategy='single' с конкретной маркой — ВСЕ models в каждой категории должны быть из этой марки\n" "- price_min_rub / price_max_rub — диапазон цен по разным магазинам (если не уверен — поставь один и тот же)" ) def call_ai(user_prompt: str, system_prompt: str | None = None, temperature: float = 0.3, max_tokens: int = 4000) -> dict[str, Any]: """Вызов GigaChat. Возвращает {json, text, tokens, model, error?}.""" cfg = get_config() try: token = _get_token() except Exception as e: return {"json": None, "text": f"AI auth: {e}", "tokens": 0, "model": cfg.gigachat_model, "error": True} payload = { "model": cfg.gigachat_model, "temperature": temperature, "max_tokens": max_tokens, "messages": [ {"role": "system", "content": system_prompt or SYSTEM_PROMPT_PICKER}, {"role": "user", "content": user_prompt}, ], } try: with httpx.Client(timeout=60.0) as client: resp = client.post( _CHAT_URL, headers={ "Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json", }, content=json.dumps(payload, ensure_ascii=False).encode("utf-8"), ) except Exception as e: return {"json": None, "text": f"AI network: {e}", "tokens": 0, "model": cfg.gigachat_model, "error": True} if resp.status_code >= 400: try: j = resp.json() err_msg = j.get("message") or (j.get("error") or {}).get("message") or resp.text[:300] except Exception: err_msg = resp.text[:300] return {"json": None, "text": f"AI HTTP {resp.status_code}: {err_msg}", "tokens": 0, "model": cfg.gigachat_model, "error": True} data = resp.json() choice = (data.get("choices") or [{}])[0] response_text = (choice.get("message") or {}).get("content", "") tokens = (data.get("usage") or {}).get("total_tokens", 0) actual_model = data.get("model", cfg.gigachat_model) json_obj = None if response_text: try: json_obj = json.loads(response_text) except json.JSONDecodeError: stripped = re.sub(r"^```(?:json)?\s*", "", response_text.strip()) stripped = re.sub(r"\s*```\s*$", "", stripped) try: json_obj = json.loads(stripped) except json.JSONDecodeError: pass return {"json": json_obj, "text": response_text, "tokens": tokens, "model": actual_model}