#!/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-flash"
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"
"1. Суть ситуации простыми словами (1-2 абзаца).\n"
"2. Пошаговый порядок действий (Шаг 1, Шаг 2, Шаг 3).\n"
"3. Кто несет ответственность (УК, застройщик, поставщик ресурсов, совет дома).\n"
"4. Образец краткого заявления при необходимости."
)
else:
return (
"СТИЛЬ ОТВЕТА: Сугубо юридический, строгий, аргументированный.\n"
"ТРЕБОВАНИЯ:\n"
"1. Правовая квалификация ситуации со ссылками на статьи нормативных актов РФ (ЖК РФ, ГК РФ, 59-ФЗ, ЗоЗПП, ПП РФ № 354, ПП РФ № 491, Госстрой № 170, СанПиН).\n"
"2. Оценка рисков, перспектив в надзорных органах и суде.\n"
"3. Четкий алгоритм действий (досудебная претензия, жалоба в ГЖИ / прокуратуру, иск).\n"
"4. Готовый образец мотивировочной части претензии или заявления при необходимости."
)
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 transcribe_audio(self, audio_path: str) -> str:
"""Транскрибирует аудиосообщение жильца в текст через Gemini Multimodal Audio (Voice-to-Text)."""
if not self.api_key or not audio_path or not os.path.exists(audio_path):
return ""
prompt = (
"Дословно расшифруй аудиосообщение на русском языке. "
"Не добавляй пояснений, не отвечай на вопрос и не искажай слова жильца. "
"Выведи ТОЛЬКО расшифрованный текст."
)
system_instruction = "Ты — высокоточный модуль транскрибации речи (Speech-to-Text). Верни только произнесенный текст."
for model_to_use in [self.model_name, "gemini-flash-latest", "gemini-2.5-flash"]:
try:
transcription = await self._query_rest_api(
prompt=prompt,
system_instruction=system_instruction,
model=model_to_use,
audio_path=audio_path
)
if transcription and transcription.strip():
return transcription.strip()
except Exception as e:
log.warning(f"Voice transcription attempt failed on {model_to_use}: {e}")
await asyncio.sleep(0.5)
return ""
async def query_notebooklm(self, prompt: str) -> dict:
"""Прямой запрос к блокноту NotebookLM «ЛенКом37 Юрист» (RAG по документам дома № 37)."""
import shutil
notebook_id = os.getenv("NOTEBOOKLM_ID", "59c41b23-5858-4a74-bf60-4ce17e0ca710")
candidates = [
"/usr/local/bin/notebooklm",
shutil.which("notebooklm"),
"/home/matrixhasyou/swarm-services/domovoy-bot/venv-ai/bin/notebooklm",
"notebooklm"
]
nb_bin = None
for b in candidates:
if b and os.path.exists(b) and os.access(b, os.X_OK):
try:
with open(b, "rb") as f:
first_line = f.readline()
if first_line.startswith(b"#!"):
interp = first_line[2:].strip().decode("utf-8", errors="ignore")
if not os.path.exists(interp):
continue
except Exception:
pass
nb_bin = b
break
if not nb_bin:
nb_bin = "/usr/local/bin/notebooklm" if os.path.exists("/usr/local/bin/notebooklm") else "notebooklm"
cmd = [
nb_bin, "ask",
"-n", notebook_id,
prompt
]
env = os.environ.copy()
env["HOME"] = "/root" if os.path.exists("/root/.notebooklm") else os.environ.get("HOME", "/root")
env["SSL_CERT_FILE"] = "/etc/ssl/certs/ca-certificates.crt"
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=env
)
try:
# Увеличенный лимит ожидания (до 300 сек для глубокого RAG-анализа смет и решений)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300.0)
except asyncio.TimeoutError:
try: proc.kill()
except: pass
return {
"success": False,
"error": "Таймаут обработки запроса к блокноту документов дома (более 300 сек)",
"source": "notebooklm"
}
out_str = stdout.decode('utf-8', errors='ignore').strip()
err_str = stderr.decode('utf-8', errors='ignore').strip()
if proc.returncode == 0 and out_str and "Authentication Error" not in out_str and "Not logged in" not in out_str:
lines = [
l for l in out_str.splitlines()
if not l.startswith("Continuing conversation")
and not l.startswith("Resumed conversation")
]
cleaned = "\n".join(lines).strip()
if cleaned.startswith("Answer:"):
cleaned = cleaned[len("Answer:"):].strip()
return {"success": True, "text": cleaned, "source": "notebooklm"}
else:
log.warning(f"NotebookLM non-zero or auth err: code={proc.returncode}, out={out_str[:250]}, err={err_str[:250]}")
return {"success": False, "error": err_str or out_str, "source": "notebooklm"}
except Exception as e:
log.error(f"NotebookLM exec exception: {e}")
return {"success": False, "error": str(e), "source": "notebooklm"}
async def query(self, prompt: str, role: str = "resident", audio_path: str = None) -> dict:
"""
Канонический метод запроса юридического помощника МКД «Домовой 2.0»:
- Все ответы формируются СТРОГО на основе документов дома № 37 из блокнота NotebookLM.
- Никаких общих абстрактных ответов или галлюцинаций Gemini!
- При отправке аудиосообщения: Gemini выполняет ИСКЛЮЧИТЕЛЬНО распознавание голоса в текст (STT),
после чего текст вопроса направляется в NotebookLM.
"""
start_time = datetime.now()
transcribed_text = None
# 1. Если передано аудиосообщение — транскрибируем его через Gemini Multimodal в текст
if audio_path:
log.info(f"Received audio voice message ({audio_path}), transcribing via Gemini STT...")
transcribed_text = await self.transcribe_audio(audio_path)
if not transcribed_text:
return {
"success": False,
"text": "❌ Не удалось распознать голосовое сообщение. Пожалуйста, напишите ваш вопрос текстом.",
"role": role,
"latency_sec": round((datetime.now() - start_time).total_seconds(), 2)
}
log.info(f"Audio voice successfully transcribed: '{transcribed_text[:120]}...'")
prompt = transcribed_text
if not prompt or not prompt.strip():
return {
"success": False,
"text": "❌ Вопрос пустой. Пожалуйста, сформулируйте юридический вопрос.",
"role": role,
"latency_sec": 0.0
}
# 2. Обращение СТРОГО к блокноту NotebookLM (RAG по сметам, договорам и протоколам ОСС)
nlm_res = await self.query_notebooklm(prompt)
latency = round((datetime.now() - start_time).total_seconds(), 2)
if nlm_res.get("success"):
return {
"success": True,
"text": nlm_res["text"],
"answer": nlm_res["text"],
"role": role,
"latency_sec": latency,
"model": "NotebookLM: ЛенКом37 Юрист (Документы дома № 37)",
"transcribed_question": transcribed_text
}
else:
err_msg = nlm_res.get("error", "Неизвестная ошибка")
log.error(f"NotebookLM legal query failed: {err_msg}")
# Никакого fallback на обычный Gemini! Выдаем честное уведомление об обновлении базы
return {
"success": False,
"text": (
"❌ База документов дома № 37 временно обновляет сессию.\n\n"
"Юридический помощник не выдает выдуманных общих ответов нейросетей, "
"а обращается строго к протоколам ОСС, сметам и договорам нашего дома в NotebookLM. "
"Пожалуйста, повторите вопрос через минуту."
),
"role": role,
"latency_sec": latency,
"error_detail": err_msg
}
# Глобальный синглтон юридического ядра для роутеров бота
legal_engine = AILegalEngine()
if __name__ == "__main__":
async def test():
engine = AILegalEngine()
res = await engine.query("Сравни за что мы голосуем на общих собраниях - за обрезку или глубокую обрезку деревьев во дворе. А в отчетах - как списывают?")
print("Success:", res["success"])
print("Latency:", res["latency_sec"])
print("Model:", res.get("model"))
print("Text snippet:\n", res["text"][:300])
asyncio.run(test())