333 lines
16 KiB
Python
333 lines
16 KiB
Python
#!/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("<br>", "\n").replace("<br/>", "\n")
|
||
safe = html.escape(clean)
|
||
# **жирный** -> <b>жирный</b>
|
||
safe = re.sub(r'\*\*(.*?)\*\*', r'<b>\1</b>', safe, flags=re.DOTALL)
|
||
# *курсив* -> <i>курсив</i>
|
||
safe = re.sub(r'\*(.*?)\*', r'<i>\1</i>', safe, flags=re.DOTALL)
|
||
# [ссылка](url) -> <a href="url">ссылка</a>
|
||
safe = re.sub(r'\[(.*?)\]\((https?://.*?)\)', r'<a href="\2">\1</a>', safe, flags=re.DOTALL)
|
||
return safe
|
||
|
||
async def send_split_message(message: Message, header: str, body: str):
|
||
"""Отправляет длинный ответ аккуратными частями с защитой от разрыва тегов."""
|
||
# Разбиваем исходный текст по параграфам
|
||
paragraphs = body.split("\n\n")
|
||
chunks = []
|
||
current_chunk = []
|
||
current_len = 0
|
||
|
||
for p in paragraphs:
|
||
p_len = len(p) + 2
|
||
if current_len + p_len > 3200 and current_chunk:
|
||
chunks.append("\n\n".join(current_chunk))
|
||
current_chunk = [p]
|
||
current_len = p_len
|
||
else:
|
||
current_chunk.append(p)
|
||
current_len += p_len
|
||
|
||
if current_chunk:
|
||
chunks.append("\n\n".join(current_chunk))
|
||
|
||
if not chunks:
|
||
chunks = [body]
|
||
|
||
for idx, chunk in enumerate(chunks, 1):
|
||
if idx == 1:
|
||
part_header = f"{header}\n\n" if len(chunks) == 1 else f"{header} (Часть 1/{len(chunks)}):\n\n"
|
||
else:
|
||
part_header = f"<i>(Продолжение {idx}/{len(chunks)}):</i>\n\n"
|
||
|
||
part_text = f"{part_header}{format_telegram_html(chunk)}"
|
||
# Прикрепляем инлайн-кнопки только к последней части ответа
|
||
reply_markup = get_legal_nav_keyboard() if idx == len(chunks) else None
|
||
try:
|
||
await message.answer(part_text, parse_mode='HTML', reply_markup=reply_markup)
|
||
except Exception as e:
|
||
log.warning(f"HTML send failed ({e}), falling back to plain text")
|
||
plain_header = header if idx == 1 else f"(Продолжение {idx}/{len(chunks)}):"
|
||
await message.answer(f"{plain_header}\n\n{chunk}", reply_markup=reply_markup)
|
||
|
||
def get_legal_nav_keyboard() -> InlineKeyboardMarkup:
|
||
"""Инлайн-кнопки навигации после ответа юриста."""
|
||
kb = [
|
||
[
|
||
InlineKeyboardButton(text="📝 Задать следующий вопрос", callback_data="jurist_ask_next"),
|
||
InlineKeyboardButton(text="🚪 Выход из режима юриста", callback_data="jurist_exit")
|
||
]
|
||
]
|
||
return InlineKeyboardMarkup(inline_keyboard=kb)
|
||
|
||
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.casefold().in_(["/юрист", "/jurist", "/вопрос", "юрист", "⚖️ юрист (жк рф)"]))
|
||
@router.message(F.text.casefold().startswith(("/юрист", "/jurist")))
|
||
async def cmd_legal_advisor(message: Message, state: FSMContext, bot: Bot):
|
||
role = await determine_user_role(message.from_user.id)
|
||
if role not in ("admin", "lawyer"):
|
||
await message.answer(
|
||
"🔒 <b>Модуль юридической аналитики находится в режиме калибровки Советом Дома.</b>\n\n"
|
||
"Доступ открывается индивидуально администратором (@ktybdsq).\n"
|
||
"Если вам требуется консультация по протечкам, тарифам или судебным спорам — напишите председателю Совета Дома.",
|
||
parse_mode='HTML'
|
||
)
|
||
return
|
||
|
||
# Проверяем, не задал ли пользователь вопрос прямо в команде: /юрист Какой срок ответа?
|
||
parts = (message.text or "").split(maxsplit=1)
|
||
if len(parts) > 1 and parts[0].casefold() in ("/юрист", "/jurist", "/вопрос"):
|
||
inline_question = parts[1].strip()
|
||
if len(inline_question) > 3:
|
||
status_msg = await message.answer("⏳ <i>Анализирую нормы законодательства РФ...</i>", parse_mode='HTML')
|
||
result = await legal_engine.query(prompt=inline_question, role=role)
|
||
await status_msg.delete()
|
||
if result.get("success"):
|
||
header = "⚖️ <b>Юридический ответ:</b>"
|
||
await send_split_message(message, header, result["text"])
|
||
await state.set_state(LegalQuestionState.waiting_for_question)
|
||
else:
|
||
await message.answer(f"❌ <b>Ошибка:</b> {result.get('text')}", parse_mode='HTML')
|
||
return
|
||
|
||
text = (
|
||
"⚖️ <b>Юридический помощник (ЖК РФ)</b>\n\n"
|
||
"Напишите ваш вопрос (текстом или отправьте голосовое сообщение).\n"
|
||
"<i>Ответ формируется по нормам законодательства РФ и судебной практике.</i>"
|
||
)
|
||
await message.answer(text, parse_mode='HTML')
|
||
await state.set_state(LegalQuestionState.waiting_for_question)
|
||
|
||
@router.callback_query(F.data.in_(["admin_jurist", "jurist_ai", "main_jurist"]))
|
||
async def cb_legal_advisor(callback: CallbackQuery, state: FSMContext):
|
||
await callback.answer()
|
||
role = await determine_user_role(callback.from_user.id)
|
||
if role not in ("admin", "lawyer"):
|
||
await callback.message.answer(
|
||
"🔒 <b>Модуль юридической аналитики находится в режиме калибровки Советом Дома.</b>\n\n"
|
||
"Доступ открывается индивидуально администратором (@ktybdsq).\n"
|
||
"Если вам требуется консультация по протечкам, тарифам или судебным спорам — напишите председателю Совета Дома.",
|
||
parse_mode='HTML'
|
||
)
|
||
return
|
||
|
||
text = (
|
||
"⚖️ <b>Юридический помощник (ЖК РФ)</b>\n\n"
|
||
"Напишите ваш вопрос (текстом или отправьте голосовое сообщение).\n"
|
||
"<i>Ответ формируется по нормам законодательства РФ и судебной практике.</i>"
|
||
)
|
||
await callback.message.answer(text, parse_mode='HTML')
|
||
await state.set_state(LegalQuestionState.waiting_for_question)
|
||
|
||
# === КНОПКИ НАВИГАЦИИ ПОСЛЕ ОТВЕТА ЮРИСТА ===
|
||
|
||
@router.callback_query(F.data == "jurist_ask_next")
|
||
async def cb_jurist_ask_next(callback: CallbackQuery, state: FSMContext):
|
||
"""Пользователь хочет задать следующий вопрос юристу"""
|
||
await callback.answer()
|
||
role = await determine_user_role(callback.from_user.id)
|
||
if role not in ("admin", "lawyer"):
|
||
await callback.message.answer("🔒 <b>Доступ ограничен.</b>", parse_mode='HTML')
|
||
await state.clear()
|
||
return
|
||
|
||
await state.set_state(LegalQuestionState.waiting_for_question)
|
||
await callback.message.answer(
|
||
"⚖️ <b>Режим юриста активен.</b>\n\n"
|
||
"Напишите ваш следующий вопрос или отправьте голосовое сообщение:",
|
||
parse_mode='HTML'
|
||
)
|
||
|
||
@router.callback_query(F.data == "jurist_exit")
|
||
async def cb_jurist_exit(callback: CallbackQuery, state: FSMContext):
|
||
"""Выход из режима юриста с возвратом в главное меню"""
|
||
await callback.answer("Режим юриста завершён")
|
||
await state.clear()
|
||
|
||
user_id = callback.from_user.id
|
||
try:
|
||
from keyboards.inline import get_main_menu
|
||
from database.models import InitiativeGroup
|
||
import config
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
stmt = select(User).where(User.user_id == user_id)
|
||
user = (await session.execute(stmt)).scalar_one_or_none()
|
||
ig_stmt = select(InitiativeGroup).where(InitiativeGroup.user_id == user_id, InitiativeGroup.is_active == True)
|
||
is_ig = (await session.execute(ig_stmt)).scalar_one_or_none() is not None
|
||
|
||
is_admin = (user.is_admin if user else False) or (user_id == config.ADMIN_USER_ID)
|
||
is_verified = user.verified if user else False
|
||
|
||
await callback.message.answer(
|
||
"🚪 <b>Вы вышли из режима юриста.</b>\n\nГлавное меню:",
|
||
reply_markup=get_main_menu(
|
||
verified=is_verified,
|
||
is_ig=is_ig,
|
||
is_admin=is_admin,
|
||
is_superadmin=(user_id == config.ADMIN_USER_ID)
|
||
),
|
||
parse_mode='HTML'
|
||
)
|
||
except Exception as e:
|
||
log.warning(f"Failed to show main menu on jurist exit: {e}")
|
||
await callback.message.answer("🚪 <b>Вы вышли из режима юриста.</b>", parse_mode='HTML')
|
||
|
||
# === ОБРАБОТКА ТЕКСТОВЫХ ВОПРОСОВ ===
|
||
|
||
@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)
|
||
if role not in ("admin", "lawyer"):
|
||
await message.answer(
|
||
"🔒 <b>Модуль юридической аналитики находится в режиме калибровки Советом Дома.</b>\n\n"
|
||
"Доступ открывается индивидуально администратором (@ktybdsq).",
|
||
parse_mode='HTML'
|
||
)
|
||
await state.clear()
|
||
return
|
||
|
||
question = message.text
|
||
|
||
status_msg = await message.answer(
|
||
"⏳ <i>Анализирую нормы законодательства РФ...</i>",
|
||
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"❌ <b>Ошибка:</b> {result.get('text')}", parse_mode='HTML')
|
||
return
|
||
|
||
# Защита от спама: если вопрос задан в групповом чате
|
||
if message.chat.type in ("group", "supergroup"):
|
||
# Ответ отправляется жильцу в ЛИЧКУ
|
||
try:
|
||
await bot.send_message(
|
||
chat_id=user_id,
|
||
text=f"⚖️ <b>Юридический ответ:</b>\n\n{format_telegram_html(result['text'])}",
|
||
parse_mode='HTML',
|
||
reply_markup=get_legal_nav_keyboard()
|
||
)
|
||
await message.reply(
|
||
"💬 <b>Ответ сформирован и отправлен вам в личные сообщения с ботом</b>, чтобы не перегружать общий чат дома.",
|
||
parse_mode='HTML'
|
||
)
|
||
except Exception:
|
||
await message.reply(
|
||
"💬 Пожалуйста, откройте личные сообщения с ботом (@lkmthreeseven_bot) и нажмите /start, чтобы получить ответ конфиденциально.",
|
||
parse_mode='HTML'
|
||
)
|
||
else:
|
||
# В личном диалоге отвечаем напрямую
|
||
header = "⚖️ <b>Юридический ответ:</b>"
|
||
await send_split_message(message, header, result["text"])
|
||
|
||
# Оставляем состояние активным, чтобы жилец мог сразу задать следующий вопрос без клика
|
||
await state.set_state(LegalQuestionState.waiting_for_question)
|
||
|
||
# === ОБРАБОТКА ГОЛОСОВЫХ ВОПРОСОВ (НАТИВНОЕ МУЛЬТИМОДАЛЬНОЕ РАСПОЗНАВАНИЕ) ===
|
||
|
||
@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)
|
||
if role not in ("admin", "lawyer"):
|
||
await message.answer(
|
||
"🔒 <b>Модуль юридической аналитики находится в режиме калибровки Советом Дома.</b>\n\n"
|
||
"Доступ открывается индивидуально администратором (@ktybdsq).",
|
||
parse_mode='HTML'
|
||
)
|
||
await state.clear()
|
||
return
|
||
|
||
status_msg = await message.answer(
|
||
"🎙 <i>Анализирую голосовое обращение по нормам законодательства...</i>",
|
||
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"❌ <b>Ошибка:</b> {result.get('text')}", parse_mode='HTML')
|
||
return
|
||
|
||
header = "⚖️ <b>Юридический ответ:</b>"
|
||
await send_split_message(message, header, result["text"])
|
||
await state.set_state(LegalQuestionState.waiting_for_question)
|
||
|
||
except Exception as e:
|
||
log.error(f"Voice processing error: {e}")
|
||
await status_msg.edit_text(f"❌ <b>Ошибка обработки голосового сообщения:</b> {e}")
|
||
finally:
|
||
# Всегда зачищаем временный аудиофайл
|
||
if voice_file_path and os.path.exists(voice_file_path):
|
||
try: os.unlink(voice_file_path)
|
||
except: pass
|