import os import re import logging import json import asyncio import httpx logger = logging.getLogger("CyberAgentCritic") SUPABASE_URL = os.environ.get("SUPABASE_URL") SUPABASE_KEY = os.environ.get("SUPABASE_KEY") async def fetch_blind_spots() -> list: """Асинхронно получает список нерешенных задач и ошибок бота из Supabase.""" if not SUPABASE_URL or not SUPABASE_KEY: logger.error("❌ Критик: Переменные среды Supabase не настроены.") return [] url = f"{SUPABASE_URL}/rest/v1/agent_blind_spots?resolved=eq.false&select=user_text,reason&limit=20" headers = {"apikey": SUPABASE_KEY, "Authorization": f"Bearer {SUPABASE_KEY}"} async with httpx.AsyncClient(timeout=10.0) as client: try: res = await client.get(url, headers=headers) return res.json() if res.status_code == 200 else [] except Exception as e: logger.error(f"⚠️ Критик: Ошибка запроса к слепым зонам: {e}") return [] async def mark_spots_resolved(): """Асинхронно помечает обработанные ошибки в базе данных как решенные.""" if not SUPABASE_URL or not SUPABASE_KEY: return url = f"{SUPABASE_URL}/rest/v1/agent_blind_spots?resolved=eq.false" headers = {"apikey": SUPABASE_KEY, "Authorization": f"Bearer {SUPABASE_KEY}"} async with httpx.AsyncClient(timeout=10.0) as client: try: await client.patch(url, json={"resolved": True}, headers=headers) logger.info("✅ Критик: Все текущие слепые зоны успешно переведены в статус Resolved.") except Exception as e: logger.error(f"⚠️ Критик: Не удалось обновить статус слепых зон: {e}") def critique_code(code: str) -> tuple[bool, str]: """ Статический цензор кода. Проверяет сгенерированный Python-код на соответствие жестким стандартам безопасности Web3CyberServices. """ # Попытка загрузить кастомные правила rules = {} if os.path.exists('api/coding_rules.json'): try: with open('api/coding_rules.json', 'r', encoding='utf-8') as f: rules = json.load(f) except Exception as e: logger.warning(f"Не удалось прочесть coding_rules.json: {e}") lines = code.split('\n') # Правило 1: Локальная архитектура импортов внутри функций for line in lines: stripped = line.strip() if (stripped.startswith('import ') or stripped.startswith('from ')) and not line.startswith(' '): # Исключаем базовые системные импорты, если они явно разрешены конфигурацией if not any(pkg in stripped for pkg in ["os", "json", "asyncio", "httpx", "time", "logging", "traceback"]): return False, "Глобальные импорты запрещены. Модули должны импортироваться локально внутри функций." # Правило 2: Запрет сторонних неодобренных SDK (взаимодействие только через raw HTTP/REST) for line in lines: if 'groq' in line.lower() and 'import' in line.lower(): return False, "Использование сторонних SDK (например, библиотеки groq) запрещено. Используйте httpx API." # Правило 3: Наличие блоков отказоустойчивости if 'try:' not in code or 'except' not in code: return False, "Нарушение стандартов надежности: В модифицированном фрагменте отсутствует обработка исключений (try/except)." return True, "OK" async def run_autonomous_evolution(target_chat_id: int) -> str: """ ЯДРО АВТОНОМНОЙ РЕФЛЕКСИИ Анализирует сбои системы, формирует ТЗ и передает его в неблокирующий контур мутации. """ spots = await fetch_blind_spots() if not spots: logger.info("⚡ Критик: Слепых зон не обнаружено. Контур работает в оптимальном режиме.") return "Контур стабилен, улучшений не требуется." # Формируем аналитический отчет для ИИ-Критика logs_summary = "\n".join([f"- Запрос: {s['user_text']} | Причина сбоя: {s['reason']}" for s in spots]) critic_prompt = ( f"Ты — автономный системный аналитик ИБ-агента школы Web3CyberServices.\n" f"Изучи список логов и невыполненных ботом команд за последнее время:\n\n" f"{logs_summary}\n\n" f"Определи ОДНУ самую критическую уязвимость, баг или недостающую OSINT-функцию.\n" f"Сформулируй ОДНУ четкую лаконичную инструкцию для разработчика, как расширить файл `custom_logic.py`.\n" f"Инструкция должна начинаться строго со слов 'улучши себя:' и содержать конкретное ТЗ на доработку кода.\n" f"Пример ответа: улучши себя: добавь парсинг WHOIS для доменов.\n" f"Выдай ТОЛЬКО эту строку и ничего больше." ) directive = "" groq_api_key = os.environ.get("GROQ_API_KEY", "") keys_pool = [k.strip() for k in groq_api_key.split(",") if k.strip()] if not keys_pool: return "❌ Критик: Пул ключей GROQ_API_KEY пуст. Анализ невозможен." # Запрашиваем инструкцию у пула моделей Groq через асинхронный клиент async with httpx.AsyncClient(timeout=30.0) as client: for key in keys_pool: try: res = await client.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, json={ "model": "llama-3.3-70b-versatile", "messages": [{"role": "user", "content": critic_prompt}], "temperature": 0.2 } ) if res.status_code == 200: directive = res.json()["choices"][0]["message"]["content"].strip() break except Exception as e: logger.warning(f"Критик: Сбой обращения к Groq на ключе: {e}") continue if directive and directive.lower().startswith("улучши себя:"): # Извлекаем чистое ТЗ (убираем префикс "улучши себя:") clean_instruction = directive[12:].strip() if not clean_instruction: return "❌ Критик: Сгенерированная директива оказалась пустой." logger.info(f"🚀 Критик зафиксировал директиву авто-модификации: {clean_instruction}") # Переводим логи в статус решенных, чтобы избежать зацикливания при генерации await mark_spots_resolved() # ДИНАМИЧЕСКИЙ ИМПОРТ ДЛЯ ИСКЛЮЧЕНИЯ CIRCULAR DEPENDENCY try: index_module = importlib.import_module("api.index") background_evolution_worker = getattr(index_module, "background_evolution_worker") except (ImportError, AttributeError) as e: logger.critical(f"❌ Критик: Не удалось динамически подключить ядро эволюции: {e}") return f"❌ Критический сбой инфраструктуры импортов: {str(e)}" # Запускаем асинхронный воркер мутации ядра через await, так как мы находимся в async-контексте await background_evolution_worker(clean_instruction, target_chat_id, "api/custom_logic.py") return f"🚀 Запущена автономная эволюция по ТЗ Критика: {clean_instruction}" return "⚠️ Не удалось сформировать валидную директиву автоэволюции."