Я назвал свой проект 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/ в репозитории