#!/usr/bin/env python3 """ Интеллектуальное юридическое ядро МКД «Домовой 2.0» (AI Legal Engine) Работает на базе Gemini 2.5 Flash / Pro: - Нативный слух (Gemini Multimodal Audio) без Whisper - Прямой асинхронный REST API (высокая отказоустойчивость без конфликтов pydantic) - Разграничение ролей: resident (соседский помощник), lawyer (претензионно-судебный), admin (глубокий аудит) """ import os import sys import logging import asyncio import aiohttp import base64 from datetime import datetime from pathlib import Path log = logging.getLogger("ai_legal_engine") class AILegalEngine: def __init__(self): raw_key = os.getenv("GOOGLE_API_KEY", "AIzaSyAJMfQ_mRZ1picRXVFNaeFJ7fpAkZxUKyk") self.api_key = raw_key.strip("\"'") if raw_key else "" self.client = None self.model_name = "gemini-2.5-flash" self.deep_model_name = "gemini-2.5-pro" self._init_sdk_client() def _init_sdk_client(self): if not self.api_key: log.warning("Gemini API Key missing! AI Legal Engine will run in offline mode.") return try: from google import genai self.client = genai.Client(api_key=self.api_key) log.info("Gemini SDK Client initialized.") except Exception as e: log.info(f"Using high-performance direct REST engine (SDK not active: {e})") def get_role_instructions(self, role: str) -> str: if role == "resident": return ( "РОЛЬ ДЛЯ ОТВЕТА: 'Соседский помощник для жильца'.\n" "СТИЛЬ: Спокойный, уважительный, без сложного бюрократического языка и канцелярита.\n" "ТРЕБОВАНИЯ:\n" "1. Объясни ситуацию простыми словами за 2-3 коротких абзаца.\n" "2. Дай пошаговый житейский совет (Шаг 1, Шаг 2, Шаг 3).\n" "3. Укажи, кто несет ответственность (УК, застройщик, водоканал, совет дома).\n" "4. Если нужно подать заявление — приведи краткий текст (3-4 строки), который жилец может скопировать и отправить в УК." ) elif role == "lawyer": return ( "РОЛЬ ДЛЯ ОТВЕТА: 'Профессиональный юрист по жилищному праву и защите прав потребителей'.\n" "СТИЛЬ: Строгий, юридически выверенный, аргументированный.\n" "ТРЕБОВАНИЯ:\n" "1. Четкая правовая квалификация действий/бездействия лиц.\n" "2. Прямые ссылки на статьи законов (ЖК РФ, ГК РФ, 59-ФЗ, ПП РФ № 354, ПП РФ № 491, Госстрой № 170, СанПиН 2.1.3684-21).\n" "3. Оценка перспектив в суде или надзорных органах (ГЖИ, Прокуратура, Роспотребнадзор).\n" "4. Готовый проект претензии / жалобы с правовой мотивировкой." ) elif role == "admin": return ( "РОЛЬ ДЛЯ ОТВЕТА: 'Главный аналитик Совета Дома и стратег судебных процессов'.\n" "СТИЛЬ: Стратегический, системный, аналитический.\n" "ТРЕБОВАНИЯ:\n" "1. Полный расклад с точки зрения системной выгоды для всего дома.\n" "2. Выявление уловок УК, скрытых манипуляций со сметами, протоколами ОСС или объемами коммунальных ресурсов.\n" "3. Сравнение с прецедентами и судебной практикой Ульяновского областного суда и Верховного Суда РФ.\n" "4. Стратегия принуждения госорганов к исполнению служебных обязанностей по 59-ФЗ и Приказу ГП № 45." ) return "Отвечай вежливо и по делу." async def _query_rest_api(self, prompt: str, system_instruction: str, model: str, audio_path: str = None) -> str: """Прямой асинхронный REST запрос к Google Gemini API.""" url = f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={self.api_key}" parts = [] # Аудио вложение if audio_path and os.path.exists(audio_path): with open(audio_path, "rb") as af: audio_b64 = base64.b64encode(af.read()).decode("utf-8") parts.append({ "inlineData": { "mimeType": "audio/ogg", "data": audio_b64 } }) if not prompt: prompt = "Внимательно прослушай аудиосообщение жильца дома. Точно выдели юридическую суть и дай обоснованный ответ." parts.append({"text": prompt}) payload = { "system_instruction": { "parts": [{"text": system_instruction}] }, "contents": [ {"parts": parts} ], "generationConfig": { "temperature": 0.2 if model == self.deep_model_name else 0.4 } } async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=45)) as resp: if resp.status != 200: err_text = await resp.text() raise RuntimeError(f"Gemini API HTTP {resp.status}: {err_text[:300]}") data = await resp.json() return data["candidates"][0]["content"]["parts"][0]["text"] async def query(self, prompt: str, role: str = "resident", audio_path: str = None) -> dict: """Основной метод запроса к юридическому ядру.""" if not self.api_key: return { "success": False, "text": "ИИ-модуль временно недоступен. Проверьте API ключ.", "role": role, "latency_sec": 0.0 } start_time = datetime.now() system_instruction = ( "Ты — интеллектуальный юридический щит многоквартирного дома по адресу г. Ульяновск, пр-кт Генерала Тюленева / Ленком 37 (10 подъездов).\n" "Дом ведет системную юридическую защиту прав собственников против нарушений управляющей компании, бездействия надзорных органов и фальсификаций.\n" f"{self.get_role_instructions(role)}" ) model_to_use = self.deep_model_name if role in ("lawyer", "admin") else self.model_name try: # Используем прямой REST (наивысшая надежность без конфликтов библиотек) reply_text = await self._query_rest_api( prompt=prompt, system_instruction=system_instruction, model=model_to_use, audio_path=audio_path ) latency = (datetime.now() - start_time).total_seconds() return { "success": True, "text": reply_text.strip(), "answer": reply_text.strip(), "role": role, "latency_sec": round(latency, 2), "model": model_to_use } except Exception as e: latency = (datetime.now() - start_time).total_seconds() log.error(f"Gemini Error: {e}") return { "success": False, "text": f"Ошибка обработки запроса ИИ: {e}", "role": role, "latency_sec": round(latency, 2) } # Глобальный синглтон юридического ядра для роутеров бота legal_engine = AILegalEngine() if __name__ == "__main__": async def test(): engine = AILegalEngine() res = await engine.query("Какие сроки ответа прокуратуры по закону 59-ФЗ?", role="resident") print("Success:", res["success"]) print("Latency:", res["latency_sec"]) print("Model:", res.get("model")) print("Text snippet:\n", res["text"][:300]) asyncio.run(test())