import logging import asyncio import httpx from typing import Optional # Настройка локального логгера для слоя базы данных logger = logging.getLogger("CyberAgentDB") class SupabaseManager: def __init__(self, url: str, key: str): """ Асинхронный менеджер для прямой работы с REST API Supabase (PostgREST) с ленивой инициализацией сетевого пула. """ self.url = url or "" self.key = key or "" # Базовые заголовки авторизации в инфраструктуре Supabase self.headers = { "apikey": self.key, "Authorization": f"Bearer {self.key}", "Content-Type": "application/json" } # Внутренний объект клиента, инициализируемый лениво self._client: Optional[httpx.AsyncClient] = None def _get_client(self) -> httpx.AsyncClient: """ Ленивая (Lazy) инициализация HTTP-клиента. Гарантирует привязку пула соединений к правильному running event loop приложения FastAPI. """ if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient(headers=self.headers, timeout=10.0) return self._client async def close(self): """Корректное закрытие пула сетевых соединений при остановке агента""" if self._client and not self._client.is_closed: await self._client.aclose() logger.info("Пул соединений с Supabase успешно закрыт.") async def log_blind_spot(self, chat_id: int, user_text: str, reason: str): """Регистрация 'слепых зон' инференса в удаленный лог безопасности""" if not self.url or not self.key: return url = f"{self.url}/rest/v1/agent_blind_spots" payload = { "chat_id": chat_id, "user_text": user_text, "reason": reason, "resolved": False } custom_headers = {"Prefer": "return=minimal"} client = self._get_client() # Для логов безопасности делаем 2 быстрые попытки, чтобы не тормозить вебхук диспетчера for attempt in range(2): try: resp = await client.post(url, json=payload, headers=custom_headers) if resp.status_code < 300: return logger.warning(f"[Попытка {attempt+1}/2] Ошибка логирования слепой зоны: HTTP {resp.status_code}") except Exception as e: logger.warning(f"[Попытка {attempt+1}/2] Сетевое исключение при записи слепой зоны: {e}") if attempt < 1: await asyncio.sleep(0.3) async def get_bot_memory(self, chat_id: int) -> str: """Извлечение долгосрочного контекста (памяти) чата с защитой от сетевых сбоев""" if not self.url or not self.key: return "" url = f"{self.url}/rest/v1/bot_memory" params = { "chat_id": f"eq.{chat_id}", "select": "context" } client = self._get_client() for attempt in range(3): try: resp = await client.get(url, params=params) if resp.status_code == 200: data = resp.json() if isinstance(data, list) and data: return data[0].get("context", "") return "" logger.warning(f"[Попытка {attempt+1}/3] Ошибка чтения памяти Supabase: HTTP {resp.status_code}") except Exception as e: logger.warning(f"[Попытка {attempt+1}/3] Ошибка сети при чтении памяти: {e}") # Экспоненциальный бэкбофф перед повторным запросом if attempt < 2: await asyncio.sleep(0.5 * (attempt + 1)) logger.error(f"Не удалось извлечь память чата {chat_id} после 3 попыток.") return "" async def save_bot_memory(self, chat_id: int, new_context: str): """ Сохранение или обновление контекста чата. Использует атомарную операцию UPSERT на стороне PostgreSQL с защитой от сбоев. """ if not self.url or not self.key: logger.error("Критическая ошибка: Параметры Supabase URL или Key не инициализированы.") return url = f"{self.url}/rest/v1/bot_memory" payload = { "chat_id": chat_id, "context": new_context } upsert_headers = { "Prefer": "resolution=merge-duplicates, return=minimal" } client = self._get_client() for attempt in range(3): try: resp = await client.post(url, json=payload, headers=upsert_headers) if resp.status_code < 300: logger.info(f"Данные памяти чата {chat_id} успешно синхронизированы с Supabase.") return logger.warning(f"[Попытка {attempt+1}/3] Ошибка UPSERT памяти в Supabase: HTTP {resp.status_code}") except Exception as e: logger.warning(f"[Попытка {attempt+1}/3] Исключение при записи памяти чата: {e}") if attempt < 2: await asyncio.sleep(0.5 * (attempt + 1)) logger.error(f"Критический сбой: Не удалось выполнить UPSERT памяти чата {chat_id} после 3 попыток.")