Telegram давно стал не только мессенджером, но и большой средой для сообществ: локальные чаты, профессиональные группы, каналы с комментариями, чаты по аренде, работе, продаже вещей, услугам и так далее.
В какой-то момент у нас появилась техническая задача: сделать систему, которая умеет читать сообщения из Telegram-групп, проверять их по пользовательским правилам и отправлять уведомления, если найдено совпадение.
Например, один пользователь хочет получать сообщения, где есть слова:
аренда байка снять квартиру обмен валюты
Другой пользователь следит за теми же группами, но ищет совсем другие фразы:
ищу разработчика python удаленная работа
При этом группы у пользователей могут пересекаться. Одну и ту же группу могут добавить 10, 100 или 1000 пользователей, но у каждого будут свои ключевые слова, минус-слова и настройки.
В этой статье расскажем, как мы подошли к проектированию такой системы на Python и Telethon, какие сущности заложили в базу, почему важно группировать правила по chat_id, какие компромиссы появились уже на этапе MVP и как мы планируем развивать систему дальше.
Почему не Bot API
Первый очевидный вариант — обычный Telegram-бот через BotFather.
У него есть плюсы:
простая авторизация;
хорошая документация;
удобно отправлять сообщения пользователям;
легко строить интерфейс через кнопки;
не нужно хранить пользовательскую сессию Telegram.
Но для задачи мониторинга групп у Bot API есть ограничения.
Обычный бот не может просто так читать произвольные публичные или приватные группы. Его нужно добавить в чат. В некоторых случаях ему нужно выдать права. Кроме того, поведение бота зависит от privacy mode, типа чата и настроек группы.
Мы рассматривали два варианта:
Вариант 1: Bot API - проще - безопаснее с точки зрения авторизации - удобно для интерфейса - но ограничен при чтении чужих групп Вариант 2: userbot через Telethon - сложнее в эксплуатации - требует аккуратной работы с аккаунтом - но позволяет читать те чаты, к которым аккаунт уже имеет доступ
Что выбрали мы
Для чтения сообщений мы выбрали Telethon и userbot-подход, потому что для нашей задачи важно было работать с группами, где обычный BotFather-бот не всегда может получать все сообщения.
При этом для пользовательского интерфейса мы оставили обычного Telegram-бота: через него удобнее настраивать группы, ключевые слова, минус-слова и подписку.
То есть архитектурно мы разделили роли:
Userbot / Telethon — читает сообщения BotFather-бот — дает пользователю интерфейс управления Отдельный бот или чат — получает результаты
Важно: такой подход не должен использоваться для обхода приватности, спама или действий, нарушающих правила площадки. Система должна работать только с теми чатами, к которым у аккаунта есть легитимный доступ.
Базовая идея архитектуры
На верхнем уровне система выглядит так:
Telegram groups/channels ↓ Telethon client ↓ NewMessage handler ↓ Rule matching service ↓ Database ↓ Notification sender ↓ Telegram bot / result chat
Основной поток такой:
Telethon получает новое входящее сообщение.
Система определяет
chat_id.По
chat_idнаходятся все правила, которые относятся к этой группе.Текст сообщения проверяется по ключевым словам и минус-словам.
Если есть совпадение, создается событие результата.
Пользователю отправляется уведомление.
Главная идея: одно сообщение из Telegram должно читаться один раз, даже если эту группу добавили много пользователей.
Неправильный подход: читать группу под каждого пользователя
Наивная реализация могла бы выглядеть так:
Пользователь 1 → читает группу A Пользователь 2 → читает группу A Пользователь 3 → читает группу A
Если одна и та же группа добавлена у 100 пользователей, появляется риск, что система будет пытаться обрабатывать один и тот же источник 100 раз.
Это плохо по нескольким причинам:
лишняя нагрузка на Telegram-клиент;
сложнее контролировать лимиты;
выше риск дублирования;
сложнее отлаживать поведение;
хуже масштабируемость.
Мы рассматривали два подхода:
Вариант 1: отдельная обработка групп под каждого пользователя Вариант 2: единый поток сообщений и проверка по правилам всех пользователей
Что выбрали мы
Мы выбрали второй вариант: группа читается один раз, а затем одно входящее сообщение прогоняется по правилам всех пользователей, у которых эта группа добавлена.
Это важное архитектурное решение.
Telegram-сообщение приходит в сервис: 1 раз Проверка по правилам пользователей: N раз
Так мы уменьшаем нагрузку на получение данных из Telegram и переносим масштабирование внутрь нашей системы, где им проще управлять.
Группировка правил по chat_id
Внутри приложения удобно держать структуру вида:
rules_by_chat = { -1001234567890: [ rule_1, rule_2, rule_3, ], -1009876543210: [ rule_4, rule_5, ], }
Где ключ — это chat_id, а значение — список правил пользователей, которые относятся к этому чату.
Обработчик нового сообщения может выглядеть так:
from telethon import events @client.on(events.NewMessage(incoming=True)) async def handle_new_message(event): chat_id = event.chat_id text = event.raw_text or "" rules = rules_by_chat.get(chat_id, []) for rule in rules: if is_match(text, rule): await save_match(rule, event) await send_notification(rule, event)
Мы рассматривали три варианта поиска правил:
Вариант 1: каждый раз читать правила из базы Вариант 2: держать все активные правила в памяти Вариант 3: держать правила в памяти, но обновлять их при изменениях
Что выбрали мы
Мы выбрали третий вариант: кеш правил в памяти с обновлением при изменениях.
Почему так:
не нужно ходить в базу на каждое сообщение;
можно быстро получать правила по
chat_id;при изменении настроек можно перезагрузить только конкретный чат;
это достаточно просто для MVP и не требует отдельной сложной инфраструктуры.
Что такое правило поиска
Минимальное правило состоит из таких полей:
id user_id chat_id include_keywords exclude_keywords is_active created_at updated_at
Например:
{ "user_id": 12345, "chat_id": -1001234567890, "include_keywords": ["аренда", "байк"], "exclude_keywords": ["продано", "не актуально"], "is_active": true }
Логика простая:
Если сообщение не содержит ни одного ключевого слова — совпадения нет.
Если содержит минус-слово — совпадения нет.
Иначе создаем результат.
Пример функции:
def normalize_text(text: str) -> str: return text.lower().replace("ё", "е") def is_match(text: str, rule) -> bool: normalized = normalize_text(text) include_keywords = [ normalize_text(word) for word in rule.include_keywords ] exclude_keywords = [ normalize_text(word) for word in rule.exclude_keywords ] has_include = any( keyword in normalized for keyword in include_keywords ) has_exclude = any( keyword in normalized for keyword in exclude_keywords ) return has_include and not has_exclude
Мы рассматривали несколько уровней сложности:
Вариант 1: поиск по подстроке Вариант 2: регулярные выражения Вариант 3: морфология Вариант 4: fuzzy matching Вариант 5: ML/LLM-классификация
Что выбрали мы
Для MVP мы выбрали поиск по подстроке с нормализацией текста.
Почему:
это быстро реализуется;
легко объяснить пользователю, почему правило сработало;
просто отлаживать;
достаточно для первой версии;
не требует дополнительных ML-моделей и сложной инфраструктуры.
Более сложные варианты мы оставили на следующие этапы.
Ограничения простого поиска
Поиск через keyword in text работает быстро и понятно, но он не всегда корректен.
Например, пользователь ищет слово:
кот
А система может найти его в словах:
котел котировка наркотик
Иногда это приемлемо, иногда нет.
Для более точного поиска можно использовать регулярные выражения:
import re def contains_word(text: str, word: str) -> bool: pattern = rf"\b{re.escape(word)}\b" return re.search(pattern, text, flags=re.IGNORECASE) is not None
Но для русского языка \b не всегда дает идеальное поведение, особенно если в тексте есть эмодзи, пунктуация, смешанные языки, сленг, опечатки и транслитерация.
Что выбрали мы
Мы начали с простого поиска по подстроке, но заложили архитектуру так, чтобы позже заменить matching-логику без переписывания всей системы.
То есть проверка правила вынесена в отдельный модуль:
matching/ matcher.py normalizer.py
Так мы можем сначала использовать простую функцию is_match, а потом заменить ее на более сложную реализацию: регулярки, морфологию, fuzzy matching или ИИ-анализ.
Сущности в базе данных
Для такой системы мы не стали хранить настройки пользователей одним большим JSON-полем. На старте это кажется простым, но потом усложняет фильтрацию, аудит и поддержку.
Мы рассматривали два варианта:
Вариант 1: хранить настройки пользователя в JSON Вариант 2: сделать нормализованную структуру таблиц
Что выбрали мы
Мы выбрали нормализованную структуру таблиц.
Примерная схема:
users ----- id telegram_user_id created_at status chats ----- id telegram_chat_id title username type created_at updated_at user_chats ---------- id user_id chat_id is_active created_at rules ----- id user_id chat_id name is_active created_at updated_at rule_keywords ------------- id rule_id type value created_at matches ------- id rule_id user_id chat_id telegram_message_id message_text message_date created_at subscription_periods -------------------- id user_id started_at ended_at status source created_at
Почему мы выбрали этот вариант:
одна группа может быть связана с разными пользователями;
у одного пользователя может быть много групп;
у одной группы может быть много правил;
ключевые и минус-слова можно хранить отдельно;
проще строить аналитику;
проще делать аудит изменений;
проще отключать отдельные правила;
проще объяснять пользователю, почему сработало конкретное правило.
JSON-подход быстрее на старте, но он хуже масштабируется с точки зрения поддержки.
Связь many-to-many между пользователями и группами
Одна из важных частей модели — связь пользователей и групп.
Плохой вариант:
user.groups = ["chat1", "chat2", "chat3"]
Более правильный вариант — отдельная таблица:
user_chats ---------- user_id chat_id is_active
Потому что одна и та же группа может быть добавлена у многих пользователей.
Пример:
Группа "Работа в IT" ↓ Пользователь 1: ищет "python", "backend" Пользователь 2: ищет "аналитик", "удаленка" Пользователь 3: ищет "devops", "kubernetes"
Что выбрали мы
Мы выбрали отдельную таблицу user_chats, потому что это явно отражает связь many-to-many.
Это позволяет не дублировать группы в базе, а хранить одну запись о Telegram-чате и много связей с пользователями.
Отдельный бот для управления и отдельный контур результатов
На этапе проектирования появился вопрос: куда отправлять найденные сообщения?
Первый вариант — отправлять результаты в тот же бот, через который пользователь настраивает поиск.
Это просто, но есть минус: бот управления быстро превращается в поток уведомлений. Пользователю сложнее менять настройки, если чат забит результатами.
Мы рассматривали три варианта:
Вариант 1: один бот для настроек и результатов Вариант 2: один бот, но разные режимы внутри него Вариант 3: отдельный бот или отдельный чат для результатов
Что выбрали мы
Мы выбрали разделение контуров:
Settings bot ------------ - добавление групп - настройка ключевых слов - настройка минус-слов - просмотр активных правил - управление подпиской Result bot / result chat ------------------------ - только найденные совпадения - ссылки на сообщения - краткий контекст
Почему так:
пользовательский интерфейс не смешивается с потоком уведомлений;
проще отлаживать отправку результатов;
проще масштабировать уведомления;
в будущем можно дать пользователю выбор: получать результаты в личку, отдельный бот или отдельный чат.
Очередь между поиском и отправкой уведомлений
В MVP можно отправлять уведомление прямо из обработчика события:
@client.on(events.NewMessage(incoming=True)) async def handle_new_message(event): ... if is_match(text, rule): await bot.send_message(user_id, result_text)
Но у этого подхода есть проблема: обработчик начинает зависеть от скорости отправки сообщений.
Если Telegram временно отвечает медленно, если превышены лимиты или если отправка падает с ошибкой, это влияет на обработку новых входящих сообщений.
Мы рассматривали варианты:
Вариант 1: отправлять уведомления прямо из обработчика Вариант 2: сохранять совпадение в базу и отправлять позже Вариант 3: использовать отдельную очередь: Redis, RabbitMQ, Kafka
Что выбрали мы
Для MVP мы выбрали промежуточный вариант: сохранять совпадения и задания на отправку в базу, а отправку делать отдельным воркером.
Схема:
NewMessage handler ↓ match detection ↓ save to database ↓ create notification job ↓ notification worker ↓ send message
Пример таблицы:
notification_jobs ----------------- id user_id match_id status attempts next_retry_at created_at sent_at error
Почему мы выбрали этот вариант:
не нужен отдельный брокер сообщений на старте;
можно повторять неуспешные отправки;
основной обработчик не блокируется;
проще анализировать ошибки;
PostgreSQL уже есть в системе.
В дальнейшем такую таблицу можно заменить или дополнить Redis/RabbitMQ/Kafka, если нагрузки станет больше.
Дедупликация результатов
Еще одна проблема — дубли.
Они могут появиться из-за:
повторной обработки сообщения после рестарта;
нескольких похожих правил у одного пользователя;
пересечения ключевых слов;
повторной отправки после ошибки;
изменения настроек в момент обработки.
Мы рассматривали два варианта дедупликации:
Вариант 1: дедупликация на уровне приложения Вариант 2: уникальные индексы на уровне базы данных
Что выбрали мы
Мы выбрали уникальные индексы в базе как основной механизм защиты.
Например:
CREATE UNIQUE INDEX uniq_match ON matches (user_id, rule_id, chat_id, telegram_message_id);
Так база не даст создать дубль, даже если приложение ошибется или обработчик запустится повторно.
При этом есть продуктовый нюанс.
Можно дедуплицировать так:
user_id + rule_id + chat_id + telegram_message_id
Тогда пользователь увидит, какое именно правило сработало.
А можно так:
user_id + chat_id + telegram_message_id
Тогда пользователь получит только одно уведомление на одно сообщение, даже если совпали несколько правил.
Что выбрали мы
На старте мы выбрали вариант:
user_id + rule_id + chat_id + telegram_message_id
Почему:
важно понимать, какое правило сработало;
проще отлаживать;
пользователь может видеть причину совпадения;
можно строить статистику по эффективности правил.
Позже можно добавить настройку: объединять несколько срабатываний по одному сообщению в одно уведомление.
Обновление правил без перезапуска сервиса
Если правила хранятся в памяти в виде rules_by_chat, возникает вопрос: что делать, когда пользователь изменил настройки?
Мы рассматривали три варианта.
Вариант 1. Читать правила из базы на каждое сообщение
rules = await load_rules_from_db(chat_id)
Плюс: всегда актуальные данные.
Минус: при большом количестве сообщений база быстро станет узким местом.
Вариант 2. Загружать правила один раз при старте
rules_by_chat = await load_all_active_rules()
Плюс: быстро работает при обработке сообщений.
Минус: после изменения настроек нужен перезапуск или ручная синхронизация.
Вариант 3. Кешировать правила и обновлять их точечно
async def reload_rules_for_chat(chat_id: int): rules_by_chat[chat_id] = await load_active_rules(chat_id)
Плюс: быстро и достаточно актуально.
Минус: нужно реализовать механизм инвалидации кеша.
Что выбрали мы
Мы выбрали третий вариант: кеш правил в памяти + точечное обновление по chat_id.
Когда пользователь меняет ключевые слова, минус-слова или список групп, сервис управления создает внутреннее событие:
RuleUpdated(chat_id=-1001234567890)
После этого monitoring-service перезагружает правила только для нужного чата.
Почему мы выбрали этот вариант:
не перегружаем базу;
не перезапускаем сервис после каждого изменения;
обновляем только нужный участок кеша;
сохраняем простую архитектуру без сложного брокера событий на старте.
Почему важны статусы
Во многих MVP статусы сначала игнорируют. Например, данные просто создаются и удаляются.
Но для мониторинга это быстро становится проблемой. Нужно понимать:
активна ли группа;
активно ли правило;
истекла ли подписка;
отправлено ли уведомление;
была ли ошибка;
можно ли повторить отправку.
Мы рассматривали два подхода:
Вариант 1: физически удалять неактуальные данные Вариант 2: использовать статусы и soft delete
Что выбрали мы
Мы выбрали статусы и soft delete.
Например, у правила может быть статус:
active paused deleted error
У группы:
active unavailable deleted permission_lost
У уведомления:
pending sent failed retry cancelled
У подписки:
trial active expired cancelled blocked
Почему мы выбрали этот вариант:
сохраняется история;
проще делать аудит;
можно объяснить поведение системы;
можно восстановить правило или группу;
проще расследовать ошибки;
меньше риск случайной потери данных.
Например, если пользователь временно отключил группу, мы не удаляем связь, а ставим:
is_active = false
Так мы сохраняем историю и можем быстро вернуть настройку обратно.
Что делать с большим количеством правил
Если в одной группе 5 правил — все просто.
Если 5000 правил — уже интереснее.
Наивный цикл:
for rule in rules: if is_match(text, rule): ...
Работает за O(N * K), где:
N— количество правил;K— количество ключевых слов в правиле.
Мы рассматривали несколько вариантов оптимизации:
Вариант 1: простой перебор всех правил Вариант 2: обратный индекс keyword → rule_id Вариант 3: полнотекстовый поиск Вариант 4: специализированный поисковый движок
Что выбрали мы
Для MVP мы выбрали простой перебор правил внутри конкретного chat_id.
Почему:
сначала важнее проверить продуктовую гипотезу;
реализация проще;
поведение легко объяснить;
нагрузка на старте прогнозируемая;
оптимизацию можно добавить позже.
Но архитектурно мы уже видим следующий шаг: построить обратный индекс.
Например:
keyword → list[rule_id]
Тогда при новом сообщении можно сначала понять, какие ключевые слова вообще встретились, а уже потом проверять связанные с ними правила.
Пример:
keyword_index = { "python": [rule_1, rule_5, rule_9], "аренда": [rule_2, rule_7], "байк": [rule_2, rule_8], }
Такой подход позволит не прогонять каждое сообщение через все правила подряд.
Минус-слова
Минус-слова нужны, чтобы снижать шум.
Например, пользователь ищет:
байк
Но не хочет получать сообщения:
продам байк байк уже сдан неактуально
Тогда правило может выглядеть так:
Ключевые слова: - байк - аренда байка Минус-слова: - продам - продано - неактуально
Мы рассматривали два порядка проверки:
Вариант 1: сначала проверять минус-слова, потом ключевые Вариант 2: сначала проверять ключевые, потом минус-слова
Что выбрали мы
Мы выбрали второй вариант: сначала ключевые слова, потом минус-слова.
def is_match(text: str, rule) -> bool: normalized = normalize_text(text) if not has_include_keyword(normalized, rule): return False if has_exclude_keyword(normalized, rule): return False return True
Почему:
сначала проверяем, есть ли вообще интерес к сообщению;
минус-слова применяются только к потенциально релевантным сообщениям;
проще отлаживать причину отказа;
проще объяснить пользователю: сообщение подошло по ключевому слову, но было исключено по минус-слову.
Логирование и аудит
Обычных технических логов недостаточно.
Для такой системы полезно разделять:
Технические логи ---------------- - сервис запущен - ошибка Telegram API - ошибка базы - ошибка отправки уведомления Бизнес-события -------------- - пользователь добавил группу - пользователь удалил ключевое слово - правило сработало - уведомление отправлено - уведомление не доставлено Аудит изменений --------------- - кто изменил правило - когда изменил - старое значение - новое значение
Мы рассматривали два варианта:
Вариант 1: писать только технические логи Вариант 2: отдельно хранить аудит пользовательских изменений
Что выбрали мы
Мы выбрали второй вариант: технические логи отдельно, аудит изменений отдельно.
Пример таблицы аудита:
audit_log --------- id user_id entity_type entity_id action old_value new_value created_at
Почему:
можно понять, почему пользователь получил конкретное сообщение;
можно понять, почему пользователь не получил другое сообщение;
можно восстановить историю изменения правил;
проще разбирать спорные ситуации;
проще поддерживать пользователей.
Хранение исходного сообщения
Для совпадений полезно сохранять не только факт срабатывания, но и часть исходных данных:
chat_id telegram_message_id message_text message_date sender_id matched_keyword rule_id
Но здесь есть компромисс: чем больше текста хранится, тем выше требования к приватности и безопасности.
Мы рассматривали варианты:
Вариант 1: хранить полный текст сообщения Вариант 2: хранить только фрагмент вокруг совпадения Вариант 3: хранить только ссылку на сообщение Вариант 4: хранить только факт совпадения
Что выбрали мы
Для MVP мы выбрали хранение текста найденного сообщения, потому что это сильно упрощает отладку и позволяет показывать пользователю контекст.
Но мы сразу закладываем возможность ограничивать срок хранения и в будущем перейти к более аккуратной модели:
- хранить только фрагмент; - хранить только ссылку; - удалять старые сообщения через N дней; - не хранить лишние персональные данные.
Обработка ошибок Telegram
При работе с Telegram-клиентом нужно быть готовым к ошибкам:
FloodWait SessionPasswordNeeded Unauthorized ChatWriteForbidden UserPrivacyRestricted ChannelPrivate
Даже если система только читает сообщения и отправляет уведомления, ошибки все равно будут:
аккаунт потерял доступ к группе;
группу переименовали;
пользователь заблокировал бота;
превышен лимит отправки;
сессия слетела;
Telegram попросил повторную авторизацию.
Мы рассматривали два варианта обработки ошибок:
Вариант 1: обрабатывать ошибки прямо в месте вызова Вариант 2: централизовать обработку и сохранять ошибки в статусах/логах
Что выбрали мы
Мы выбрали второй вариант: централизованная обработка ошибок + статусы + повторные попытки.
Например, если отправка уведомления не удалась, мы не теряем событие, а переводим задачу в статус:
failed
И сохраняем причину ошибки.
Для повторной отправки используем:
attempts next_retry_at error
Так воркер может повторить отправку позже, не блокируя основной поток обработки сообщений.
Разделение сервисов
На раннем этапе все можно держать в одном приложении:
app.py
Но логически внутри уже стоит разделять модули:
telegram_client/ listener.py matching/ matcher.py normalizer.py notifications/ sender.py templates.py storage/ repositories.py models.py settings_bot/ handlers.py keyboards.py
Мы рассматривали три варианта:
Вариант 1: монолитный скрипт Вариант 2: один проект, но разделенный на модули Вариант 3: несколько отдельных сервисов
Что выбрали мы
Для MVP мы выбрали вариант между вторым и третьим: один проект, но несколько отдельных процессов.
settings-bot monitoring-service notification-worker database
Пример docker-compose:
services: settings-bot: build: . command: python -m app.settings_bot monitoring-service: build: . command: python -m app.monitoring notification-worker: build: . command: python -m app.notification_worker database: image: postgres:16 environment: POSTGRES_DB: monitor POSTGRES_USER: monitor POSTGRES_PASSWORD: example
Почему мы выбрали этот вариант:
код остается в одном репозитории;
процессы можно перезапускать независимо;
проще разносить ответственность;
проще смотреть логи;
проще масштабировать отдельные части позже.
Планируемое развитие: ИИ-анализ сообщений
После базового поиска по ключевым словам мы планируем добавить более умный слой обработки: ИИ-анализ уже отобранных сообщений.
Важно: мы не хотим отправлять в ИИ весь поток сообщений из всех групп. Это дорого, медленно и избыточно.
Мы рассматриваем несколько вариантов:
Вариант 1: прогонять через ИИ все сообщения подряд Вариант 2: сначала фильтровать по правилам, потом анализировать только кандидатов Вариант 3: использовать ИИ только по запросу пользователя
Что выбрали мы
Мы выбрали второй вариант: сначала обычный быстрый фильтр, потом ИИ-анализ только отобранных сообщений.
Схема:
Новое сообщение ↓ Быстрый поиск по ключевым словам ↓ Потенциальное совпадение ↓ ИИ-анализ релевантности ↓ Классификация / краткое резюме / черновик ответа ↓ Уведомление пользователю
Почему так:
не тратим ресурсы на весь поток сообщений;
ИИ работает только с потенциально релевантными сообщениями;
снижаем стоимость обработки;
уменьшаем задержку;
сохраняем объяснимость первого уровня фильтрации.
Что именно может делать ИИ:
1. Оценивать релевантность сообщения 2. Классифицировать тип запроса 3. Делать краткое резюме 4. Выделять важные детали 5. Генерировать черновик ответа
Например, пользователь ищет клиентов на аренду байка. Система нашла сообщение:
Привет, подскажите, где можно взять байк на неделю в Нячанге?
Базовый фильтр найдет его по словам байк и неделя.
ИИ-слой сможет дополнительно определить:
Тип: потенциальный клиент Интент: аренда байка Срочность: средняя Город: Нячанг Нужен ответ: да
И сгенерировать черновик ответа:
Здравствуйте! Можем подсказать варианты аренды байка на неделю в Нячанге. Напишите, пожалуйста, какие даты и какой тип байка вам нужен.
При этом мы не планируем делать полностью автоматическую отправку ответов без контроля пользователя. Более безопасный вариант — генерировать именно черновик, который человек сможет проверить и отправить сам.
Возможная финальная схема
В итоге архитектура может выглядеть так:
┌────────────────────┐ │ Telegram groups │ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ Telethon listener │ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ Message normalizer │ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ Rules cache │ │ rules_by_chat │ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ Matching engine │ └─────────┬──────────┘ │ ┌───────────┴───────────┐ ▼ ▼ ┌──────────────────┐ ┌────────────────────┐ │ PostgreSQL │ │ Notification jobs │ │ matches/audit │ │ pending/retry │ └──────────────────┘ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ AI analysis layer │ │ optional │ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ Notification worker │ └─────────┬──────────┘ │ ▼ ┌────────────────────┐ │ Telegram bot/user │ └────────────────────┘
Что мы бы заложили сразу
После проектирования MVP мы пришли к выводу, что несколько вещей лучше закладывать сразу.
1. Нормальную модель данных
Мы выбрали нормализованные таблицы, а не один JSON с настройками пользователя.
Это сложнее на старте, но проще в развитии.
2. Аудит изменений
Мы выбрали отдельную таблицу аудита, потому что без нее сложно объяснять поведение системы.
3. Отдельную отправку уведомлений
Мы выбрали notification worker, чтобы не блокировать обработчик входящих сообщений.
4. Дедупликацию на уровне базы
Мы выбрали уникальные индексы, потому что они надежнее, чем проверка только в коде.
5. Кеш правил
Мы выбрали rules_by_chat в памяти с точечным обновлением, чтобы не ходить в базу на каждое сообщение.
6. Статусы вместо удаления
Мы выбрали soft delete и статусы, чтобы сохранять историю и упростить поддержку.
7. Возможность подключить ИИ позже
Мы не стали делать ИИ-анализ первым этапом. Сначала нужен быстрый и понятный фильтр. Но архитектуру matching-слоя мы выделили отдельно, чтобы позже добавить классификацию, резюме и генерацию черновиков ответов.
Вывод
На первый взгляд мониторинг Telegram-групп кажется простой задачей: получил сообщение, проверил наличие слова, отправил уведомление.
Но уже на уровне MVP появляются архитектурные вопросы:
как не читать одну и ту же группу много раз;
как хранить правила разных пользователей;
как обрабатывать пересечения групп;
как избежать дублей;
как не блокировать обработчик отправкой уведомлений;
как обновлять правила без перезапуска;
как объяснять пользователю, почему сработало или не сработало правило;
как подготовить систему к будущему ИИ-анализу.
Главная идея, к которой мы пришли: источник сообщений нужно читать один раз, а пользовательские правила группировать вокруг chat_id.
То есть:
Telegram-сообщение приходит один раз. Дальше система проверяет его по правилам всех пользователей, которым интересен этот чат.
Для первого MVP мы выбрали простую и объяснимую архитектуру:
Telethon для чтения сообщений PostgreSQL для хранения пользователей, групп, правил и совпадений rules_by_chat для быстрого поиска правил notification worker для отправки результатов статусы и аудит вместо физического удаления данных
А следующий этап развития — не просто искать сообщения по словам, а анализировать уже найденные кандидаты с помощью ИИ: определять интент, релевантность, срочность и готовить черновики ответов для пользователя.
Такой подход позволяет начать с понятного MVP, но не загоняет архитектуру в тупик при дальнейшем росте.