138 lines
4.8 KiB
Python
138 lines
4.8 KiB
Python
from __future__ import annotations
|
||
|
||
import html
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
from urllib.parse import urlencode
|
||
|
||
import config
|
||
from telebot.types import InlineKeyboardButton, InlineKeyboardMarkup, Message, WebAppInfo
|
||
|
||
from models import logger
|
||
from models.bot import bot, send
|
||
from models.navigator import (
|
||
RouteRequest,
|
||
calculate_route,
|
||
encode_route,
|
||
parse_route_request,
|
||
)
|
||
from models.navigator_request_log import append_navigator_request
|
||
|
||
|
||
_SEEN_REQUESTS: dict[tuple[int, str], float] = {}
|
||
_SEEN_TTL = 300.0
|
||
_ROOMS_BY_FLOOR: dict[str, set[str]] | None = None
|
||
|
||
|
||
def _map_rooms() -> dict[str, set[str]]:
|
||
"""Read the room IDs from the same static SVGs used by the WebApp."""
|
||
|
||
global _ROOMS_BY_FLOOR
|
||
if _ROOMS_BY_FLOOR is not None:
|
||
return _ROOMS_BY_FLOOR
|
||
rooms: dict[str, set[str]] = {}
|
||
pattern = re.compile(r'serif:id="([^"]+)"')
|
||
root = Path(__file__).resolve().parents[1] / "web" / "map" / "images"
|
||
for floor in range(1, 7):
|
||
path = root / f"{floor}.svg"
|
||
try:
|
||
rooms[str(floor)] = set(pattern.findall(path.read_text(encoding="utf-8")))
|
||
except OSError:
|
||
rooms[str(floor)] = set()
|
||
_ROOMS_BY_FLOOR = rooms
|
||
return rooms
|
||
|
||
|
||
def _validate_map_rooms(request: RouteRequest) -> bool:
|
||
rooms = _map_rooms()
|
||
# A missing bundle should not make the bot unusable during deployment;
|
||
# the actual navigator adapter will perform its own manifest validation.
|
||
if not any(rooms.values()):
|
||
return True
|
||
for endpoint in (request.start, request.end):
|
||
if endpoint.floor and endpoint.room not in rooms.get(endpoint.floor, set()):
|
||
return False
|
||
if not endpoint.floor and not any(endpoint.room in values for values in rooms.values()):
|
||
return False
|
||
return True
|
||
|
||
|
||
def _remember_request(user_id: int, request_id: str) -> bool:
|
||
now = time.monotonic()
|
||
expired = [key for key, timestamp in _SEEN_REQUESTS.items() if now - timestamp > _SEEN_TTL]
|
||
for key in expired:
|
||
_SEEN_REQUESTS.pop(key, None)
|
||
key = (user_id, request_id)
|
||
if key in _SEEN_REQUESTS:
|
||
return False
|
||
_SEEN_REQUESTS[key] = now
|
||
return True
|
||
|
||
|
||
def _map_url(request: RouteRequest, route: dict | None) -> str:
|
||
base = f"{config.WEB_BASE_URL.rstrip('/')}/map/"
|
||
params = {
|
||
"v": "15",
|
||
"mode": "navigator",
|
||
"start": request.start.room,
|
||
"end": request.end.room,
|
||
"status": "built" if route else "failed",
|
||
}
|
||
if request.start.floor:
|
||
params["sf"] = request.start.floor
|
||
if request.end.floor:
|
||
params["ef"] = request.end.floor
|
||
if route:
|
||
params["route"] = encode_route(route)
|
||
return f"{base}?{urlencode(params)}"
|
||
|
||
|
||
def _result_markup(url: str) -> InlineKeyboardMarkup:
|
||
markup = InlineKeyboardMarkup()
|
||
markup.add(InlineKeyboardButton("Показать маршрут", web_app=WebAppInfo(url)))
|
||
return markup
|
||
|
||
|
||
def _send_route_result(message: Message, request: RouteRequest, route: dict | None):
|
||
start = html.escape(request.start.room)
|
||
end = html.escape(request.end.room)
|
||
if route:
|
||
text = f"<b>Маршрут построен</b>\n\n{start} - {end}"
|
||
else:
|
||
text = (
|
||
f"<b>Маршрут не построен</b>\n\n{start} - {end}\n"
|
||
"Не удалось найти путь между выбранными аудиториями."
|
||
)
|
||
markup = _result_markup(_map_url(request, route)) if route else None
|
||
return send(message, text, reply_markup=markup)
|
||
|
||
|
||
@bot.message_handler(content_types=["web_app_data"])
|
||
def route_request_from_webapp(message: Message):
|
||
"""Receive ``Telegram.WebApp.sendData`` from the navigator tab."""
|
||
|
||
data = getattr(getattr(message, "web_app_data", None), "data", "")
|
||
try:
|
||
request = parse_route_request(data)
|
||
except (TypeError, ValueError, UnicodeError) as error:
|
||
logger.error("Navigator", f"[{message.from_user.id}] invalid request: {error}")
|
||
return send(message, "Не удалось распознать запрос маршрута. Открой карту заново.")
|
||
|
||
if not _remember_request(message.from_user.id, request.request_id):
|
||
return
|
||
|
||
if not _validate_map_rooms(request):
|
||
return send(message, "Одной из аудиторий нет на карте. Выбери аудитории из подсказок.")
|
||
|
||
try:
|
||
append_navigator_request(request)
|
||
except (OSError, ValueError) as error:
|
||
logger.error("Navigator", f"[{message.from_user.id}] failed to save request: {error}")
|
||
|
||
try:
|
||
route = calculate_route(request)
|
||
except Exception as error: # keep a future adapter from breaking polling
|
||
logger.error("Navigator", f"[{message.from_user.id}] route calculation failed: {error}")
|
||
route = None
|
||
return _send_route_result(message, request, route)
|