Мой первый пет‑проект: как я создавал ядро для Telegram‑ботов

Я назвал свой проект Coreness — это современное ядро для Telegram‑ботов, построенное вокруг идеи полного контроля: вся логика описывается в YAML, плагины подключаются декларативно, инфраструктура остаётся у вас. Получается не «бот на вечер», а платформа, которую легко развивать, сопровождать и переносить между средами.

— on‑premise без лишней магии, — чёткая архитектура и быстрая отладка, — масштабирование по мере роста.

TL;DR

• On‑prem ядро Telegram‑ботов: сценарии в YAML, расширение через плагины, полный контроль данных

• Архитектура: Event‑Driven + Database Queue, батчи (50/0.1 с), один терминальный UPDATE на действие

• Предсказуемые цепочки (chain, chain_drop) без гонок; отдельный «разблокировщик»

• Плейсхолдеры с модификаторами и валидатор как «прослойка логики»

• Отправка сообщений, запросы, подключение AI-моделей, параллельные задачи, кэш

Демо: Coreness Bot • Группа: Coreness • Репозиторий: GitHub

Что получилось (коротко)

• Database Queue вместо брокера для старта: проще деплой, достаточно для типичных нагрузок

• Батч‑чтение и 1 терминальный UPDATE резко снижают нагрузку на БД

• Цепочки действий управляют потоком статусов и дают предсказуемость

• Конфиги и сценарии — в репозитории, «без магии», удобно настраивать

Быстрый пример конфигурации

# config/triggers.yaml text:   exact:     "/start": "menu_start"  # config/scenarios/menu.yaml menu_start:   actions:     - type: send       text: "Выберите раздел"       inline:         - ["Инструкции", "Контакты"]     - type: send       text: "Спасибо!"       chain: completed

Пример

Зачем и с чего начинал

Хотелось конструктора, где сценарии прозрачно управляют поведением, а код ядра остаётся минимальным и модульным. Большинство готовых решений либо слишком «чёрные ящики», либо тянут за собой лишнюю инфраструктуру. Поэтому начался путь собственного ядра — от простых реализаций до текущей стабильной архитектуры.

Эволюция архитектуры (4 итерации)

• V1: «монолит» вокруг фреймворка. Быстрый старт, но сложный рост и слабая тестируемость.

• V2: попытка «сервисности» без чётких границ. Много связей, тяжёлая координация.

• V3: события и очереди, но ещё неоптимальные цепочки и сильные зависимости.

• V4 (нынешнее ядро): микросервисы на базе плагинов, Event‑Driven, Database Queue, DI, YAML‑конфигурации. Чёткие вертикальные срезы и слабые связи между сервисами.

Подробнее:

• V1: логика вплетена в обработчики фреймворка. Плюсы — скорость старта. Минусы — «раздувание» кода, повторение шаблонов, сложная изоляция и тестирование.

• V2: выделение сервисов без строгих контрактов. Появились дубли зависимостей, «ползущие» импорт‑связи и взаимные знания слоёв.

• V3: события/очереди упростили композицию, но цепочки ещё требовали ручных «склеек», а апдейтов в БД было слишком много.

• V4: плагинная модель + DI + DB Queue. Чёткие границы, сценарии в YAML, один терминальный апдейт, отдельный разблокировщик, предсказуемая производительность.

Главный урок: изоляция и простые контракты важнее универсальности. Лучше узкие, хорошо определённые сервисы и декларативные сценарии поверх них.

Архитектура и возможности

• Микросервисы как плагины (Utilities и Services)

• Event‑Driven обработка: всё — через очередь действий в БД

• Database Queue вместо внешнего брокера (простота, прозрачность, локальный запуск)

• Vertical Slice: каждый сервис решает свой чёткий кусок домена

• DI‑контейнер: зависимости подтягиваются автоматически

• YAML‑конфигурации: сценарии, триггеры, настройки — всё в репозитории

DI и загрузка плагинов

• Плагины сканируются в plugins/, читается config.yaml.

• Строится граф зависимостей, циклы исключаются, порядок инициализации — топологический.

• Foundation‑утилиты независимы; Level‑слои опираются на предыдущие; Core использует Foundation/Level; Services зависят только от утилит.

Foundation (logger, plugins_manager)     ↑ Level 0 (базовые утилиты)     ↑ Level 1 (промежуточные утилиты)     ↑  ...     ↑ Level N (вспомогательные слои)     ↑ ┌────────┐ │  Core  │ │  ───>  │ └────────┘     ↑ Services

name: "trigger_manager" dependencies:   utilities:     - "logger"     - "settings_manager"     - "database_service"

Триггеры и маршрутизация

Триггеры сопоставляют входные апдейты сценариям: exact, starts_with, contains, regex, state.

text:   exact:     "/start": "menu.start"   contains:     "справка": "menu.help"   regex:     "^код\\s+(\\d{4})$": "code.capture"

Сценарии на YAML

Сценарий — последовательность действий с цепочками (chain, chain_drop).
Поддерживаются типы: send, scenario, restrict, invite_link, request, request_management, to_speech, from_speech, user, validator.

menu_start:   actions:     - type: send       text: "Выберите раздел"       inline:         - ["Инструкции", "Контакты"]     - type: scenario       name: "house.instructions_menu"       chain: completed

Как работает очередь действий

Событие из Telegram преобразуется в набор действий и складывается в таблицу actions. Сервисы читают только свои типы действий батчами, обрабатывают и записывают результат. Следующие шаги по цепочке автоматически разблокируются.

Что важно: мы делаем INSERT при создании, читаем батчами (по умолчанию 50), и лишь один UPDATE — при установке терминального статуса (completed / failed / drop). Это резко снижает нагрузку на БД.

Цепочки действий: простой контроль потока

Следующий шаг сценария ждёт завершения предыдущего. По умолчанию разблокировка идёт на статусе completed, но можно задать chain: true (любой предыдущий статус) и chain_drop — для ветвления и принудительной остановки ветки.

chain_example:   actions:     - type: send       text: "Старт"     - type: validator       rules:         user_id: [{ rule: not_empty }]       chain: completed     - type: send       text: "Ошибка валидации"       chain: failed       chain_drop: completed     - type: send       text: "Ок"       chain: true

Разблокировщик: минимум апдейтов, максимум простоты

Отдельный сервис читает завершённые действия и аккуратно разблокирует связанные hold‑шаги. Он делает это пакетно и только один раз на элемент (служебный флаг защищает от повторов). Данные «накапливаются» вдоль цепочки и передаются дальше.

loop every queue_read_interval:   completed_batch = get_for_unlocker(limit=batch, statuses=[completed, failed, drop])   for completed in completed_batch:     waiting = get_by_prev_action_id(prev_action_id=completed.id, status=hold)     if waiting is empty:       mark_unlocker_checked(completed.id)       continue      merged = merge(prev_data=completed.prev_data, response_data=completed.response_data)      for w in waiting:       if completed.status in w.chain_drop_status: drop_chain_from(w.id)       elif completed.status in w.unlock_statuses(default=[completed]):         update_action(w.id, status=pending, prev_data=merged)       else:         update_action(w.id, status=drop, prev_data=merged)      mark_unlocker_checked(completed.id)

По умолчанию у разблокировщика интервал опроса 0.05 с — чуть быстрее, чем у остальных (0.1 с), потому что он отвечает за «дыхание» всей очереди.

Экономия апдейтов в БД

• На каждое действие: INSERT при создании, чтение батчами, и один UPDATE — только при установке терминального статуса (completed/failed/drop).

• Разблокировщик помечает исходное действие флагом «проверено» (is_unlocker_checked), чтобы не возвращаться к нему повторно.

• Разблокировка следующего шага — один точечный UPDATE (из hold в pending) с «переносом» нужных данных (prev_data).

• В сумме: минимум записи, предсказуемые чтения, индексы по статусам/времени/связям — и устойчивое поведение на нагрузке.

Плоские действия (ActionParser)

Сервисы получают «плоский» словарь с правильным приоритетом данных: prev_data > action_data > базовые поля. Это позволяет цепочкам «накапливать» контекст без дополнительных таблиц.

def parse_action(self, action: dict) -> dict:     merged = action.copy()     action_data = action.get('action_data', {})     if action_data:         merged.update(action_data)     prev_data = action.get('prev_data')     if prev_data:         merged.update(prev_data)     merged.pop('action_data', None)     merged.pop('prev_data', None)     self._process_time_attributes(merged)     return merged

Плейсхолдеры: динамические данные без кода

Плейсхолдеры подставляют значения и умеют модификаторы: арифметика, форматирование, текстовые операции, regex‑извлечение, fallback. Важные оптимизации под капотом: fast‑check, предкомпиляция шаблонов, кэширование, сохранение типов.

placeholder_demo:   actions:     - type: send       text: |         Пользователь: {username|fallback:Гость}         Цена: {price|*0.9}         Email: {event_text|regex:[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}}       placeholder: true

Оптимизации плейсхолдеров (коротко)

• Быстрая проверка без regex (fast‑check) для строк без {…}.

• Предкомпиляция регулярных выражений для плейсхолдеров и модификаторов.

• «Горячий путь»: ветка простой замены vs. сложной с модификаторами.

• Сохранение типов результата (bool/number/string), а не только строк.

• Кэширование промежуточных результатов и ограничение вложенности.

• Акуратный диспетчер модификаторов, включая арифметику (+ - * / %).

# Предкомпиляция self.placeholder_pattern = re.compile(r'\{([^}]+)\}') self.modifier_pattern = re.compile(r'([^|:]+)(?::([^|]+))?') self.max_nesting_depth = settings.get('max_nesting_depth', 3)  # Fast‑check def _has_placeholders_fast(self, text: str) -> bool:     return '{' in text and '}' in text  # Горячий путь def _process_string_optimized(self, text: str, values: Dict, depth: int = 0):     if self.enable_fast_check and not self._has_placeholders_fast(text):         return text     if self._is_simple_replacement(text):         return self._simple_replace(text, values)     return self._complex_replace(text, values, depth)  # Сохранение типов def _determine_result_type(self, value: str):     if value.lower() == 'true':         return True     if value.lower() == 'false':         return False     try:         return float(value) if '.' in value else int(value)     except ValueError:         return value  # Диспетчер модификаторов def _apply_modifier(self, value, modifier: str):     if modifier and modifier[0] in ['/', '+', '-', '*', '%']:         mod_name, mod_param = modifier[0], modifier[1:] or None     elif ':' in modifier:         mod_name, mod_param = modifier.split(':', 1)     else:         mod_name, mod_param = modifier, None     func = self.modifiers.get(mod_name)     return func(value, mod_param) if func else value

Валидатор: правила на лету

Правила проверяют данные до выполнения следующего шага. Базовые: equals, not_equals, not_empty, empty, contains, starts_with, regex, length_min, length_max, in_list, not_in_list.

user_validate:   actions:     - type: validator       rules:         username:           - { rule: not_empty }         phone:           - { rule: regex, value: "^\\+7\\d{10}$" }       chain: completed     - type: send       text: "Ошибка валидации"       chain: failed       chain_drop: completed     - type: send       text: "Ок"       chain: true

Запросы (request system)

Храним обращения пользователей вместе с метаданными и вложениями, фильтруем и показываем детально.

request_create:   actions:     - type: request       request_name: feedback       request_info: "Запрос от пользователя {username}"       placeholders: true     - type: send       text: "Спасибо! Ваше обращение сохранено."       chain: completed

Группы и модерация

• Пригласительные ссылки (генерация/выдача/продлеваемость)

• Ограничения пользователей (mute), приветствия, сервисные сообщения

• Пример: отправка ссылки на чат по запросу пользователя

chat_link:   actions:     - type: invite_link       member_limit: 1       expire: 1d     - type: send       text: |         Уникальная ссылка для вступления в группу:         {invite_link}       chain: completed       private_answer: true

Речь: параллельные задания (STT/TTS)

Сервис речи — единственный, кто обрабатывает действия параллельно (по умолчанию до 10 задач), чтобы не «висеть» на загрузках и внешнем API. Есть SSML, несколько форматов и языков.

speech_demo:   actions:     - type: to_speech       text_to_speech: "Добро пожаловать!"       voice: "Bys_24000"     - type: send       text: "Файл готов"       attachment: "{file_path}"       placeholder: true       chain: completed

Дополнительные функции

Деплой и обслуживание

• tools/core_updater.py — установка/обновления ядра

• tools/database_manager.py — работа с базой: миграции, пересоздание, индексы

• docs/SSL_CERTIFICATES_GUIDE.md — российские сертификаты для внешних API

Локальный запуск прост: .env + python main.py.
Все сценарии/настройки — в репозитории, что упрощает код‑ревью.

Отладка и логи

• Логи — человекочитаемые (по‑русски), уровнями, с минимальным шумом

• Прозрачные статусы действий и цепочек: легко понять, что и когда разблокировалось

• Ошибки модификаторов/плейсхолдеров не «роняют» цепочки, а логируются и позволяют продолжить

Гибкая настройка: всё через settings.yaml

Можно тонко тюнить латентность и пропускную способность без правок кода. Сервисы легко отключать целиком, если они не нужны.

action_unlocker:   queue_read_interval: 0.05  messenger:   queue_batch_size: 100   queue_read_interval: 0.05  speech_processor:   enabled: false

При высоких объёмах отправок для мессенджера уместно увеличить батч, уменьшить интервал, а при необходимости запустить несколько экземпляров сервиса.

Почему не брокер

• Для типичных ботов «узкое место» — сеть/Telegram API, а не очередь

• Database Queue проще развернуть и сопровождать; SQLite «из коробки», PostgreSQL — предсказуемый апгрейд

• Меньше инфраструктурной сложности (кластеры, ACL, мониторинг)

• Контракты сервисов уже «батчевые» — миграция на брокер возможна без переписывания бизнес‑логики

• Когда нужен брокер: сотни тысяч событий/мин, жёсткие SLA, сложные топологии подписок, сейчас такой потребности нет.

Итоги

• Чёткие вертикальные сервисы и слабые связи

• Прозрачная конфигурация в YAML

• Дешёвая и предсказуемая очередь в БД

• Высокая скорость на типовых нагрузках и готовность к росту

Если вы делаете Telegram‑ботов и хотите контролировать логику и инфраструктуру — этот подход может вам зайти. Репозиторий, демо и документация доступны, а сценарии можно адаптировать под любой домен.

Ошибки и уроки

• «Модульность» без правил зависимостей быстро превращается в связность.

• Shared/utils мало — нужны уровни утилит и формализованные интерфейсы.

• Очередь в БД отлично работает при правильных индексах и минимизации апдейтов.

• Конфиг как документация снижает порог входа и облегчает поддержку.

• Ранние оптимизации плейсхолдеров окупаются на реальной нагрузке.

Что дальше

• Опциональная миграция на PostgreSQL под высокие нагрузки.

• Больше готовых сценариев.

• Расширение набора плагинов и интеграций.

Ссылки

• Репозиторий: github.com/Vensus137/Coreness

• Демо‑бот: t.me/coreness_bot

• Группа проекта: t.me/coreness

• Документация: раздел docs/ в репозитории