morozovka-portal/app/assistant.py
2026-10-20 03:27:14 +03:00

112 lines
5.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import re
from datetime import datetime, timezone, timedelta
import pytz
from sqlmodel import Session, select
from .models import Event, GlobalStatus
from .db import engine
MSK = pytz.timezone("Europe/Moscow")
def _msk(dt):
if dt is None: return None
if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(MSK)
def get_response(text: str) -> str:
"""Принимает вопрос, определяет интент и возвращает ответ."""
text = text.lower().strip()
with Session(engine) as db_session:
# Получаем текущий статус
current_status = db_session.exec(select(GlobalStatus)).first()
# --- ИНТЕНТ: ГДЕ СЕЙЧАС / СТАТУС ---
if re.search(r'где|статус|сейчас', text):
if current_status.status == 'on_place':
return "Никита Сергеевич сейчас на месте (в аудитории)."
elif current_status.status == 'delayed':
return f"Никита Сергеевич опаздывает на {current_status.delay_minutes} мин. Причина: {current_status.message or 'не указана'}."
elif current_status.status == 'away':
return "Никита Сергеевич вне вуза."
elif current_status.status == 'cancelled':
return f"Занятия отменены! {current_status.message or ''}"
else:
return "Никита Сергеевич сейчас свободен / вне расписания."
# --- ИНТЕНТ: ОПАЗДЫВАЕТ? ---
if re.search(r'опаздыва|задержива', text):
if current_status.status == 'delayed':
return f"Да, опаздывает на {current_status.delay_minutes} мин. Причина: {current_status.message or 'не указана'}."
else:
return "По текущему статусу, он не опаздывает."
# --- ИНТЕНТ: КОГДА СЛЕДУЮЩАЯ ПАРА / БЛИЖАЙШЕЕ ---
if re.search(r'когда|следующ|ближайш', text):
now = datetime.now(timezone.utc)
next_event = db_session.exec(
select(Event)
.where(Event.is_public == True)
.where(Event.start_time >= now)
.order_by(Event.start_time)
.limit(1)
).first()
if next_event:
local_time = _msk(next_event.start_time).strftime('%d.%m в %H:%M')
return f"Ближайшее событие: {next_event.title} ({local_time}). Место: {next_event.location or 'не указано'}."
else:
return "В ближайшее время событий в расписании нет."
# --- ИНТЕНТ: РАСПИСАНИЕ НА СЕГОДНЯ ---
if re.search(r'сегодня|седня', text):
# Берем начало сегодняшнего дня и конец сегодняшнего дня
now_msk = datetime.now(MSK)
start_of_day_msk = now_msk.replace(hour=0, minute=0, second=0, microsecond=0)
start_of_day = start_of_day_msk.astimezone(timezone.utc)
end_of_day = start_of_day + timedelta(days=1)
events = db_session.exec(
select(Event)
.where(Event.is_public == True)
.where(Event.start_time >= start_of_day)
.where(Event.start_time < end_of_day)
.order_by(Event.start_time)
).all()
if not events:
return "На сегодня событий в расписании нет."
res = "Расписание на сегодня:\n"
for e in events:
res += f"• {_msk(e.start_time).strftime('%H:%M')} - {e.title} ({e.location or 'без места'})\n"
return res
# --- ИНТЕНТ: РАСПИСАНИЕ НА НЕДЕЛЮ ---
if re.search(r'недел|7 дней', text):
now = datetime.now(timezone.utc)
end_of_week = now + timedelta(days=7)
events = db_session.exec(
select(Event)
.where(Event.is_public == True)
.where(Event.start_time >= now)
.where(Event.start_time <= end_of_week)
.order_by(Event.start_time)
).all()
if not events:
return "На ближайшую неделю событий нет."
res = "Расписание на неделю:\n"
for e in events[:5]: # Показываем первые 5, чтобы не спамить
res += f"• {_msk(e.start_time).strftime('%d.%m %H:%M')} - {e.title}\n"
if len(events) > 5:
res += "...и другие события (смотрите на странице)."
return res
# --- ИНТЕНТ: КОНТАКТЫ ---
if re.search(r'связаться|контакт|телеграм|почта|кабинет', text):
return "Связаться с Никитой Сергеевичем можно по почте: freezer.nik@gmail.com (или в Telegram)."
# --- ЕСЛИ НИЧЕГО НЕ ПОНЯЛ ---
return "Извините, я пока не понимаю этот вопрос. Попробуйте спросить: 'Где сейчас?', 'Когда следующая пара?', 'Что сегодня?' или 'Как связаться?'."