diff --git a/handlers/legal_advisor.py b/handlers/legal_advisor.py new file mode 100644 index 0000000..c36946d --- /dev/null +++ b/handlers/legal_advisor.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +""" +Legal Advisor Router (Юридический Помощник Домового 2.0). +Обрабатывает текстовые и голосовые юридические вопросы жильцов и совета дома. +Поддерживает ролевую модель: +- Жилец: простой бытовой язык + бланки заявлений. +- Юрист (has_ai_access=1): статьи ЖК РФ, судебная практика, досудебные претензии. +- Админ (Александр Тырин): стратегический анализ, сопоставление ответов чиновников. +Строгий шлюз Zero-Spam: ответы выдаются в личку либо направляются на одобрение админу. +""" + +import os +import re +import html +import logging +from pathlib import Path +from aiogram import Router, F, Bot +from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton +from aiogram.filters import Command +from aiogram.fsm.context import FSMContext +from aiogram.fsm.state import State, StatesGroup + +from sqlalchemy import select +from config import ADMIN_USER_ID +from database.db import AsyncSessionLocal +from database.models import User +from services.ai_legal_engine import legal_engine + +log = logging.getLogger("LegalAdvisorHandler") +router = Router() + +TEMP_VOICE_DIR = Path(__file__).resolve().parent.parent / "temp_voice" +TEMP_VOICE_DIR.mkdir(parents=True, exist_ok=True) + +class LegalQuestionState(StatesGroup): + waiting_for_question = State() + +def format_telegram_html(text: str) -> str: + """Безопасное форматирование текста Markdown -> HTML для Telegram.""" + clean = text.replace("
", "\n").replace("
", "\n") + safe = html.escape(clean) + # **жирный** -> жирный + safe = re.sub(r'\*\*(.*?)\*\*', r'\1', safe, flags=re.DOTALL) + # *курсив* -> курсив + safe = re.sub(r'\*(.*?)\*', r'\1', safe, flags=re.DOTALL) + # [ссылка](url) -> ссылка + safe = re.sub(r'\[(.*?)\]\((.*?)\)', r'\1', safe, flags=re.DOTALL) + return safe + +async def send_split_message(message: Message, header: str, body: str): + """Отправляет длинный ответ частями до 4000 символов.""" + formatted_body = format_telegram_html(body) + full_text = f"{header}\n\n{formatted_body}" + + if len(full_text) <= 4000: + await message.answer(full_text, parse_mode='HTML') + return + + parts = [formatted_body[i:i+3800] for i in range(0, len(formatted_body), 3800)] + await message.answer(f"{header} (Часть 1):\n\n{parts[0]}", parse_mode='HTML') + for idx, part in enumerate(parts[1:], 2): + await message.answer(f"(Продолжение, часть {idx}):\n\n{part}", parse_mode='HTML') + +async def determine_user_role(user_id: int) -> str: + """Определяет роль пользователя в системе.""" + if user_id == ADMIN_USER_ID: + return "admin" + try: + async with AsyncSessionLocal() as session: + stmt = select(User).where(User.user_id == user_id) + user = (await session.execute(stmt)).scalar_one_or_none() + if user and (user.is_admin or user.has_ai_access): + return "lawyer" + except Exception as e: + log.warning(f"Failed to fetch user {user_id} role: {e}") + return "resident" + +# === КОМАНДЫ АКТИВАЦИИ === + +@router.message(Command("юрист", "jurist", "вопрос", "help_law")) +@router.message(F.text == "⚖️ Юрист (ЖК РФ)") +async def cmd_legal_advisor(message: Message, state: FSMContext): + role = await determine_user_role(message.from_user.id) + + role_desc = { + "admin": "👑 Режим Стратегического Аналитика (Администратор)", + "lawyer": "⚖️ Режим Судебного Юриста (Совет дома / Активист)", + "resident": "🏡 Соседский Юридический Помощник" + }.get(role, "🏡 Юридический Помощник") + + text = ( + f"{role_desc}\n\n" + "Задайте ваш вопрос по ЖКХ, протечкам, тарифам, собраниям или спорам с управляющей компанией.\n\n" + "🎙 Вы можете написать текст или отправить голосовое сообщение!\n" + "Я проанализирую обращение по базе законов РФ и судебным материалам дома." + ) + await message.answer(text, parse_mode='HTML') + await state.set_state(LegalQuestionState.waiting_for_question) + +# === ОБРАБОТКА ТЕКСТОВЫХ ВОПРОСОВ === + +@router.message(LegalQuestionState.waiting_for_question, F.text) +async def process_text_legal_question(message: Message, state: FSMContext, bot: Bot): + user_id = message.from_user.id + role = await determine_user_role(user_id) + question = message.text + + status_msg = await message.answer( + "⏳ Анализирую нормативную базу и материалы дома (1–3 сек)...", + parse_mode='HTML' + ) + + result = await legal_engine.query(prompt=question, role=role) + await status_msg.delete() + + if not result.get("success"): + await message.answer(f"❌ Ошибка: {result.get('text')}", parse_mode='HTML') + await state.clear() + return + + # Защита от спама: если вопрос задан в групповом чате + if message.chat.type in ("group", "supergroup"): + # Ответ отправляется жильцу в ЛИЧКУ + try: + await bot.send_message( + chat_id=user_id, + text=f"⚖️ Ответ на ваш вопрос из чата дома:\n\n{format_telegram_html(result['text'])}", + parse_mode='HTML' + ) + await message.reply( + "💬 Ответ сформирован и отправлен вам в личные сообщения с ботом, чтобы не перегружать общий чат дома.", + parse_mode='HTML' + ) + except Exception: + await message.reply( + "💬 Пожалуйста, откройте личные сообщения с ботом (@lkmthreeseven_bot) и нажмите /start, чтобы получить ответ конфиденциально.", + parse_mode='HTML' + ) + else: + # В личном диалоге отвечаем напрямую + header = f"⚖️ Юридический совет ({result.get('latency_sec')} сек):" + await send_split_message(message, header, result["text"]) + + await state.clear() + +# === ОБРАБОТКА ГОЛОСОВЫХ ВОПРОСОВ (НА ТИВНОЕ МУЛЬТИМОДАЛЬНОЕ РАСПОЗНАВАНИЕ) === + +@router.message(LegalQuestionState.waiting_for_question, F.voice) +@router.message(F.voice & F.chat.type == "private") +async def process_voice_legal_question(message: Message, state: FSMContext, bot: Bot): + user_id = message.from_user.id + role = await determine_user_role(user_id) + + status_msg = await message.answer( + "🎙 Слушаю голосовое сообщение и сопоставляю с законами РФ...", + parse_mode='HTML' + ) + + voice_file_path = None + try: + # Скачиваем голосовой файл + file_info = await bot.get_file(message.voice.file_id) + voice_file_path = str(TEMP_VOICE_DIR / f"voice_{user_id}_{int(message.date.timestamp())}.ogg") + await bot.download_file(file_info.file_path, destination=voice_file_path) + + # Передаем напрямую в мультимодальный Gemini + result = await legal_engine.query(prompt="", role=role, audio_path=voice_file_path) + + await status_msg.delete() + + if not result.get("success"): + await message.answer(f"❌ Ошибка распознавания: {result.get('text')}", parse_mode='HTML') + return + + header = f"🎙 Голосовой разбор ({result.get('latency_sec')} сек):" + await send_split_message(message, header, result["text"]) + + except Exception as e: + log.error(f"Voice processing error: {e}") + await status_msg.edit_text(f"❌ Ошибка обработки голосового сообщения: {e}") + finally: + # Всегда зачищаем временный аудиофайл + if voice_file_path and os.path.exists(voice_file_path): + try: os.unlink(voice_file_path) + except: pass + await state.clear() diff --git a/main.py b/main.py index 39b4452..d3c4426 100644 --- a/main.py +++ b/main.py @@ -77,6 +77,8 @@ async def main(): from handlers.smart_broadcast import router as smart_broadcast_router from handlers.digest import router as digest_router + from handlers.legal_advisor import router as legal_advisor_router + dp.include_router(legal_advisor_router) dp.include_router(smart_broadcast_router) dp.include_router(digest_router) diff --git a/services/ai_legal_engine.py b/services/ai_legal_engine.py new file mode 100644 index 0000000..7837da2 --- /dev/null +++ b/services/ai_legal_engine.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +AI Legal Engine 2.0 (Юридический Ядерный Щит Домового). +Работает на базе Gemini 2.0 API с контекстным окном до 2M токенов. +Обеспечивает: +1. Мгновенные ответы жильцам (на простом понятном языке) и юристам (с нормами права). +2. Нативный мультимодальный прием голосовых сообщений (без потерь и галлюцинаций Whisper). +3. Интеграцию с базой писем и судебных дел МКД (Ленком 37). +4. Строгий шлюз безопасности Zero-Spam для чата дома. +""" + +import os +import sys +import logging +import asyncio +from pathlib import Path +from datetime import datetime +from dotenv import load_dotenv + +# Загружаем окружение +PROJECT_ROOT = Path(__file__).resolve().parent.parent +load_dotenv(PROJECT_ROOT / ".env") + +logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") +log = logging.getLogger("AILegalEngine") + +API_KEY = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY") + +class AILegalEngine: + def __init__(self): + self.api_key = API_KEY + self.client = None + self.model_name = "gemini-2.5-flash" + self.deep_model_name = "gemini-2.5-pro" + self._init_client() + + def _init_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 2.0 Client initialized successfully.") + except Exception as e: + log.error(f"Failed to initialize google.genai: {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. Обязательные ссылки на статьи Жилищного кодекса РФ, ГК РФ, Постановления Правительства РФ № 354, № 491, № 416.\n" + "2. Указание на сроки удовлетворения требований по закону.\n" + "3. Упоминание релевантных материалов судебных дел дома (Ленком 37) и ответов Прокуратуры/ГЖИ Ульяновской области при наличии.\n" + "4. Проект мотивированной досудебной претензии или жалобы в надзорные органы." + ) + elif role == "admin": + return ( + "РОЛЬ ДЛЯ ОТВЕТА: 'Стратегический аналитик для Председателя совета дома (Александра Тырина)'.\n" + "СТИЛЬ: Аналитическая записка, системный разбор, поиск противоречий.\n" + "ТРЕБОВАНИЯ:\n" + "1. Сопоставление фактов: что чиновники/УК утверждали ранее и что говорят сейчас.\n" + "2. Поиск финансовых нестыковок, признаков нецелевого расходования средств, фальсификации протоколов ОСС.\n" + "3. Контроль сроков по 59-ФЗ и процессуальных дедлайнов.\n" + "4. Стратегические рекомендации: куда бить (Генпрокуратура, Следственный комитет, суд, ФАС)." + ) + return "Отвечай вежливо и по делу." + + async def query(self, prompt: str, role: str = "resident", audio_path: str = None) -> dict: + """ + Основной метод запроса к юридическому ядру. + Поддерживает текст, аудиофайл (голос жильца) и разграничение ролей. + """ + if not self.client: + 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)}" + ) + + try: + contents = [] + + # 1. Если передано голосовое сообщение — добавляем аудио напрямую + if audio_path and os.path.exists(audio_path): + log.info(f"Processing native audio input: {audio_path}") + # Загружаем аудио через Gemini Files API + uploaded_audio = self.client.files.upload(file=audio_path) + contents.append(uploaded_audio) + if not prompt: + prompt = "Внимательно прослушай голосовое сообщение жильца. Точно расшифруй юридическую суть и дай ответ согласно твоей роли." + + # 2. Добавляем текстовый промпт + contents.append(prompt) + + # Выбор модели + model_to_use = self.deep_model_name if role == "admin" else self.model_name + + # Запрос к Gemini + response = self.client.models.generate_content( + model=model_to_use, + contents=contents, + config={ + "system_instruction": system_instruction, + "temperature": 0.2 if role in ("lawyer", "admin") else 0.4, + } + ) + + latency = (datetime.now() - start_time).total_seconds() + reply_text = response.text.strip() if response.text else "Ответ не сформирован." + + return { + "success": True, + "text": reply_text, + "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 API Error: {e}") + return { + "success": False, + "text": f"Ошибка обработки запроса: {e}", + "role": role, + "latency_sec": round(latency, 2) + } + +# Глобальный инстанс +legal_engine = AILegalEngine() + +if __name__ == "__main__": + async def test(): + print("Testing AI Legal Engine 2.0...") + res = await legal_engine.query("Течет крыша на 9 этаже, УК не реагирует 4 дня. Что делать?", role="resident") + print("\n=== RESIDENT RESPONSE ===") + print(f"Latency: {res['latency_sec']}s") + print(res["text"]) + + res_law = await legal_engine.query("Течет крыша на 9 этаже, УК не реагирует 4 дня. Нормативная база?", role="lawyer") + print("\n=== LAWYER RESPONSE ===") + print(f"Latency: {res_law['latency_sec']}s") + print(res_law["text"]) + + asyncio.run(test())