Use Portainer env for bot stack

BotBoard/.dockerignore

@@ -0,0 +1,8 @@
+.env
+.env.*
+.git
+.pytest_cache
+.mypy_cache
+.ruff_cache
+__pycache__
+*.pyc

BotBoard/.env.example

@@ -0,0 +1,6 @@
+TELEGRAM_BOT_TOKEN=put_telegram_bot_token_here
+BORDER_STATUS_URL=https://gpk.gov.by/situation-at-the-border/
+AI_BASE_URL=https://uat1-www.a1.by:7443/aisearch
+AI_API_KEY=put_ai_api_key_here
+AI_MODEL=gpt-5.5
+REQUEST_TIMEOUT_SECONDS=20

BotBoard/.gitignore

@@ -0,0 +1,11 @@
+__pycache__/
+*.py[cod]
+.Python
+.venv/
+venv/
+.env
+.env.*
+!.env.example
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/

BotBoard/Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app ./app
+
+CMD ["python", "-m", "app.main"]

BotBoard/README.md

@@ -0,0 +1,61 @@
+# BotBoard Travel Border Bot
+
+Telegram-бот `@vacation_sml_bot` для помощи в поездке через границу Беларусь-Польша.
+
+## Возможности
+
+- Показывает очереди по польскому направлению с официальной страницы ГПК.
+- Выделяет самые свободные пункты пропуска по текущим числовым данным.
+- Отвечает на вопросы через AI с учетом свежих данных очередей.
+- Работает через long polling, поэтому не требует публичного webhook URL.
+
+## Локальный запуск
+
+1. Создай `.env` рядом с `.env.example`.
+2. Заполни переменные окружения:
+
+```env
+TELEGRAM_BOT_TOKEN=...
+BORDER_STATUS_URL=https://gpk.gov.by/situation-at-the-border/
+AI_BASE_URL=https://uat1-www.a1.by:7443/aisearch
+AI_API_KEY=...
+AI_MODEL=gpt-5.5
+REQUEST_TIMEOUT_SECONDS=20
+```
+
+3. Установи зависимости и запусти:
+
+```bash
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+python -m app.main
+```
+
+## Docker
+
+```bash
+docker compose up -d --build
+docker compose logs -f botboard
+```
+
+## Portainer
+
+1. Открой Portainer.
+2. Создай новый Stack.
+3. Используй содержимое `docker-compose.yml`.
+4. В рабочую директорию/репозиторий стека добавь `.env` с реальными секретами.
+5. Запусти Deploy the stack.
+
+Секреты не коммитятся: `.env` исключен через `.gitignore` и `.dockerignore`.
+
+## Команды бота
+
+- `/start` - стартовое меню.
+- `/poland` - очереди по Польше.
+- `/best` - лучшие пункты по минимальной очереди.
+- `/ai вопрос` - вопрос AI с учетом очередей.
+
+## Важно
+
+Данные берутся из открытого официального источника ГПК. Бот помогает ориентироваться, но не заменяет официальные правила, требования к документам, визам и решения пограничных служб.

BotBoard/app/ai_client.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from urllib.parse import urljoin
+
+import httpx
+
+
+SYSTEM_PROMPT = """Ты интерактивный Telegram-помощник для поездок и пересечения границы Беларусь-Польша.
+Отвечай кратко, по-русски, практично. Используй текущие данные очередей из контекста.
+Не выдавай юридические гарантии: если вопрос о правилах, визах, детях, документах или запретах,
+советуй проверить официальный источник и указывай, что правила могут измениться.
+""".strip()
+
+
+class AiAssistant:
+    def __init__(self, base_url: str | None, api_key: str | None, model: str, timeout: float) -> None:
+        self.base_url = base_url.rstrip("/") if base_url else None
+        self.api_key = api_key
+        self.model = model
+        self.timeout = timeout
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.base_url and self.api_key)
+
+    async def ask(self, question: str, border_context: str) -> str:
+        if not self.enabled:
+            return "AI сейчас не настроен. Добавь AI_BASE_URL и AI_API_KEY в переменные окружения."
+
+        payload = {
+            "model": self.model,
+            "messages": [
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": f"Текущие данные:\n{border_context}\n\nВопрос пользователя:\n{question}"},
+            ],
+            "temperature": 0.2,
+        }
+        headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
+
+        async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
+            response = await self._post_chat_completion(client, payload, headers)
+            response.raise_for_status()
+
+        data = response.json()
+        return _extract_answer(data)
+
+    async def _post_chat_completion(
+        self,
+        client: httpx.AsyncClient,
+        payload: dict,
+        headers: dict[str, str],
+    ) -> httpx.Response:
+        assert self.base_url is not None
+        candidates = [self.base_url]
+        if not self.base_url.endswith("/v1/chat/completions"):
+            candidates.append(urljoin(f"{self.base_url}/", "v1/chat/completions"))
+
+        last_response: httpx.Response | None = None
+        for url in candidates:
+            response = await client.post(url, json=payload, headers=headers)
+            if response.status_code not in {404, 405}:
+                return response
+            last_response = response
+        assert last_response is not None
+        return last_response
+
+
+def _extract_answer(data: dict) -> str:
+    choices = data.get("choices") or []
+    if choices:
+        message = choices[0].get("message") or {}
+        content = message.get("content")
+        if isinstance(content, str) and content.strip():
+            return content.strip()
+
+    for key in ("answer", "response", "content", "text"):
+        value = data.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+
+    return "AI вернул ответ в неожиданном формате. Попробуй переформулировать вопрос."

BotBoard/app/border.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import datetime
+
+import httpx
+from bs4 import BeautifulSoup
+
+
+POLAND_CHECKPOINTS = (
+    "Брузги",
+    "Берестовица",
+    "Песчатка",
+    "Козловичи",
+    "Брест",
+    "Домачево",
+)
+
+SECTION_NAMES = ("Легковые", "Грузовые", "Автобусы")
+
+
+@dataclass(frozen=True)
+class QueueInfo:
+    checkpoint: str
+    queue_text: str
+    queue_value: int | None
+    electronic_queue: bool
+
+
+@dataclass(frozen=True)
+class BorderSection:
+    name: str
+    checkpoints: tuple[QueueInfo, ...]
+
+
+@dataclass(frozen=True)
+class BorderSnapshot:
+    updated_at: str | None
+    fetched_at: datetime
+    sections: tuple[BorderSection, ...]
+    source_url: str
+
+
+class BorderParseError(RuntimeError):
+    pass
+
+
+async def fetch_border_snapshot(url: str, timeout: float = 20.0) -> BorderSnapshot:
+    async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
+        response = await client.get(url)
+        response.raise_for_status()
+    return parse_border_page(response.text, url)
+
+
+def parse_border_page(html: str, source_url: str) -> BorderSnapshot:
+    soup = BeautifulSoup(html, "html.parser")
+    for tag in soup(["script", "style", "noscript"]):
+        tag.decompose()
+
+    lines = [line.strip() for line in soup.get_text("\n").splitlines() if line.strip()]
+    updated_at = _extract_updated_at(lines)
+    poland_blocks = _extract_poland_blocks(lines)
+
+    if not poland_blocks:
+        raise BorderParseError("Не удалось найти блок очередей по Польше на странице ГПК")
+
+    sections = tuple(
+        BorderSection(SECTION_NAMES[index] if index < len(SECTION_NAMES) else f"Таблица {index + 1}", block)
+        for index, block in enumerate(poland_blocks[:3])
+    )
+    return BorderSnapshot(
+        updated_at=updated_at,
+        fetched_at=datetime.now().astimezone(),
+        sections=sections,
+        source_url=source_url,
+    )
+
+
+def format_snapshot(snapshot: BorderSnapshot) -> str:
+    updated = snapshot.updated_at or "не указано"
+    parts = [f"Очереди на границе Беларуси с Польшей", f"Обновлено ГПК: {updated}"]
+
+    for section in snapshot.sections:
+        parts.append("")
+        parts.append(f"{section.name}:")
+        for item in section.checkpoints:
+            suffix = " ЭО" if item.electronic_queue else ""
+            parts.append(f"- {item.checkpoint}: {item.queue_text}{suffix}")
+
+    parts.append("")
+    parts.append("ЭО - электронная очередь. '-' означает, что значение не опубликовано или пункт не работает.")
+    return "\n".join(parts)
+
+
+def format_best_options(snapshot: BorderSnapshot) -> str:
+    lines = ["Самые свободные варианты по текущим данным:"]
+    for section in snapshot.sections:
+        numeric = [item for item in section.checkpoints if item.queue_value is not None]
+        if not numeric:
+            lines.append(f"{section.name}: нет числовых данных")
+            continue
+        best_value = min(item.queue_value for item in numeric)
+        best_names = ", ".join(item.checkpoint for item in numeric if item.queue_value == best_value)
+        lines.append(f"{section.name}: {best_names} ({best_value})")
+    lines.append("")
+    lines.append("Проверяй официальную страницу перед выездом: ситуация может измениться за минуты.")
+    return "\n".join(lines)
+
+
+def snapshot_context(snapshot: BorderSnapshot) -> str:
+    return format_snapshot(snapshot)
+
+
+def _extract_updated_at(lines: list[str]) -> str | None:
+    date_pattern = re.compile(r"\b\d{2}\.\d{2}\.\d{4}\s+\d{2}:\d{2}\b")
+    for line in lines:
+        match = date_pattern.search(line)
+        if match:
+            return match.group(0)
+    return None
+
+
+def _extract_poland_blocks(lines: list[str]) -> list[tuple[QueueInfo, ...]]:
+    blocks: list[tuple[QueueInfo, ...]] = []
+    for index, line in enumerate(lines):
+        if line != "Польша":
+            continue
+
+        block_lines: list[str] = []
+        for next_line in lines[index + 1 :]:
+            if next_line == "Украина":
+                break
+            block_lines.append(next_line)
+
+        parsed = _parse_checkpoint_block(block_lines)
+        if parsed:
+            blocks.append(tuple(parsed))
+    return blocks
+
+
+def _parse_checkpoint_block(lines: list[str]) -> list[QueueInfo]:
+    result: list[QueueInfo] = []
+    for index, line in enumerate(lines):
+        if line not in POLAND_CHECKPOINTS:
+            continue
+
+        electronic = False
+        value_line = None
+        cursor = index + 1
+        while cursor < len(lines):
+            candidate = lines[cursor]
+            if candidate in POLAND_CHECKPOINTS:
+                break
+            if "Электронная очередь" in candidate:
+                electronic = True
+            elif _looks_like_queue_value(candidate):
+                value_line = candidate
+                break
+            cursor += 1
+
+        if value_line is None:
+            continue
+        result.append(
+            QueueInfo(
+                checkpoint=line,
+                queue_text=_clean_queue_text(value_line),
+                queue_value=_queue_value(value_line),
+                electronic_queue=electronic,
+            )
+        )
+    return result
+
+
+def _looks_like_queue_value(value: str) -> bool:
+    return bool(re.fullmatch(r"\(?\d+\)?\*?|[-—]", value.strip()))
+
+
+def _clean_queue_text(value: str) -> str:
+    cleaned = value.strip().replace("*", "").strip()
+    if cleaned.startswith("(") and cleaned.endswith(")"):
+        cleaned = cleaned[1:-1]
+    return cleaned
+
+
+def _queue_value(value: str) -> int | None:
+    match = re.search(r"\d+", value)
+    if not match:
+        return None
+    return int(match.group(0))

BotBoard/app/config.py

@@ -0,0 +1,23 @@
+from functools import lru_cache
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
+
+    telegram_bot_token: str = Field(alias="TELEGRAM_BOT_TOKEN")
+    border_status_url: str = Field(
+        default="https://gpk.gov.by/situation-at-the-border/",
+        alias="BORDER_STATUS_URL",
+    )
+    ai_base_url: str | None = Field(default=None, alias="AI_BASE_URL")
+    ai_api_key: str | None = Field(default=None, alias="AI_API_KEY")
+    ai_model: str = Field(default="gpt-5.5", alias="AI_MODEL")
+    request_timeout_seconds: float = Field(default=20.0, alias="REQUEST_TIMEOUT_SECONDS")
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

BotBoard/app/main.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from aiogram import Bot, Dispatcher, F, Router
+from aiogram.client.default import DefaultBotProperties
+from aiogram.enums import ParseMode
+from aiogram.filters import Command, CommandStart
+from aiogram.fsm.context import FSMContext
+from aiogram.fsm.state import State, StatesGroup
+from aiogram.types import KeyboardButton, Message, ReplyKeyboardMarkup
+from aiogram.utils.chat_action import ChatActionSender
+
+from app.ai_client import AiAssistant
+from app.border import fetch_border_snapshot, format_best_options, format_snapshot, snapshot_context
+from app.config import get_settings
+
+
+class DialogState(StatesGroup):
+    waiting_for_ai_question = State()
+
+
+router = Router()
+
+
+def main_keyboard() -> ReplyKeyboardMarkup:
+    return ReplyKeyboardMarkup(
+        keyboard=[
+            [KeyboardButton(text="Очереди Польша"), KeyboardButton(text="Лучший пункт")],
+            [KeyboardButton(text="Спросить AI"), KeyboardButton(text="Источник")],
+        ],
+        resize_keyboard=True,
+        input_field_placeholder="Выбери действие или задай вопрос",
+    )
+
+
+@router.message(CommandStart())
+async def start(message: Message, state: FSMContext) -> None:
+    await state.clear()
+    await message.answer(
+        "Я помощник по поездке и пересечению границы Беларусь-Польша. "
+        "Показываю очереди с сайта ГПК и могу подсказать маршрутные решения через AI.",
+        reply_markup=main_keyboard(),
+    )
+
+
+@router.message(Command("help"))
+async def help_command(message: Message, state: FSMContext) -> None:
+    await state.clear()
+    await message.answer(
+        "Команды:\n"
+        "/poland - текущие очереди по Польше\n"
+        "/best - самые свободные пункты\n"
+        "/ai вопрос - спросить AI с учетом текущих очередей\n\n"
+        "Можно просто нажимать кнопки меню.",
+        reply_markup=main_keyboard(),
+    )
+
+
+@router.message(Command("poland"))
+@router.message(F.text.casefold() == "очереди польша")
+async def poland_queues(message: Message, state: FSMContext) -> None:
+    await state.clear()
+    settings = get_settings()
+    try:
+        async with ChatActionSender.typing(bot=message.bot, chat_id=message.chat.id):
+            snapshot = await fetch_border_snapshot(settings.border_status_url, settings.request_timeout_seconds)
+    except Exception:
+        logging.exception("Failed to fetch border queues")
+        await message.answer("Не удалось получить данные ГПК. Попробуй еще раз через несколько минут.", reply_markup=main_keyboard())
+        return
+    await message.answer(format_snapshot(snapshot), reply_markup=main_keyboard())
+
+
+@router.message(Command("best"))
+@router.message(F.text.casefold() == "лучший пункт")
+async def best_checkpoint(message: Message, state: FSMContext) -> None:
+    await state.clear()
+    settings = get_settings()
+    try:
+        async with ChatActionSender.typing(bot=message.bot, chat_id=message.chat.id):
+            snapshot = await fetch_border_snapshot(settings.border_status_url, settings.request_timeout_seconds)
+    except Exception:
+        logging.exception("Failed to fetch border queues")
+        await message.answer("Не удалось получить данные ГПК. Попробуй еще раз через несколько минут.", reply_markup=main_keyboard())
+        return
+    await message.answer(format_best_options(snapshot), reply_markup=main_keyboard())
+
+
+@router.message(F.text.casefold() == "источник")
+async def source(message: Message, state: FSMContext) -> None:
+    await state.clear()
+    settings = get_settings()
+    await message.answer(
+        f"Официальный источник данных: {settings.border_status_url}\n"
+        "Информация обновляется на стороне ГПК, обычно каждый четный час.",
+        reply_markup=main_keyboard(),
+    )
+
+
+@router.message(F.text.casefold() == "спросить ai")
+async def ask_ai_prompt(message: Message, state: FSMContext) -> None:
+    await state.set_state(DialogState.waiting_for_ai_question)
+    await message.answer("Напиши вопрос. Например: 'Через какой пункт лучше ехать сегодня вечером?'", reply_markup=main_keyboard())
+
+
+@router.message(Command("ai"))
+async def ask_ai_command(message: Message, state: FSMContext) -> None:
+    question = (message.text or "").partition(" ")[2].strip()
+    if not question:
+        await ask_ai_prompt(message, state)
+        return
+    await answer_ai(message, question, state)
+
+
+@router.message(DialogState.waiting_for_ai_question)
+async def ask_ai_state(message: Message, state: FSMContext) -> None:
+    question = (message.text or "").strip()
+    if not question:
+        await message.answer("Напиши вопрос текстом.")
+        return
+    await answer_ai(message, question, state)
+
+
+@router.message(F.text)
+async def fallback_question(message: Message, state: FSMContext) -> None:
+    question = (message.text or "").strip()
+    if len(question) < 4:
+        await message.answer("Выбери действие в меню или задай вопрос по поездке.", reply_markup=main_keyboard())
+        return
+    await answer_ai(message, question, state)
+
+
+async def answer_ai(message: Message, question: str, state: FSMContext) -> None:
+    await state.clear()
+    settings = get_settings()
+    assistant = AiAssistant(
+        settings.ai_base_url,
+        settings.ai_api_key,
+        settings.ai_model,
+        settings.request_timeout_seconds,
+    )
+    try:
+        async with ChatActionSender.typing(bot=message.bot, chat_id=message.chat.id):
+            snapshot = await fetch_border_snapshot(settings.border_status_url, settings.request_timeout_seconds)
+            answer = await assistant.ask(question, snapshot_context(snapshot))
+    except Exception:
+        logging.exception("Failed to answer AI question")
+        await message.answer(
+            "Не удалось получить ответ сейчас: проблема с источником данных или AI. Попробуй позже.",
+            reply_markup=main_keyboard(),
+        )
+        return
+    await message.answer(answer, reply_markup=main_keyboard())
+
+
+async def run() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    settings = get_settings()
+    bot = Bot(settings.telegram_bot_token, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
+    dispatcher = Dispatcher()
+    dispatcher.include_router(router)
+    await dispatcher.start_polling(bot)
+
+
+if __name__ == "__main__":
+    asyncio.run(run())

BotBoard/docker-compose.yml

@@ -0,0 +1,12 @@
+services:
+  botboard:
+    build: .
+    container_name: botboard-travel-border-bot
+    restart: unless-stopped
+    environment:
+      TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN}
+      BORDER_STATUS_URL: ${BORDER_STATUS_URL:-https://gpk.gov.by/situation-at-the-border/}
+      AI_BASE_URL: ${AI_BASE_URL:-}
+      AI_API_KEY: ${AI_API_KEY:-}
+      AI_MODEL: ${AI_MODEL:-gpt-5.5}
+      REQUEST_TIMEOUT_SECONDS: ${REQUEST_TIMEOUT_SECONDS:-20}

BotBoard/requirements.txt

@@ -0,0 +1,4 @@
+aiogram==3.13.1
+beautifulsoup4==4.12.3
+httpx==0.27.2
+pydantic-settings==2.6.1