Бот для сбора лидов из Telegram-каналов и RSS Бот для сбора лидов из Telegram-каналов и RSS

Задача

Источники заказов приходят в разные моменты: каналы — событием, RSS — раз в минуту. Нужен один поток лидов в личный чат и настройка фильтров без правки кода и передеплоя.

Как делал

Главная сложность в том, что два телеграм-транспорта умеют разное. Пришлось запустить оба в одном процессе, иначе либо команды, либо чтение каналов не работало бы:

self._tg_client = TelegramClient(
    StringSession(self.user_string_session),   # сессия в одну строку, без файлов
    self.api_id, self.api_hash,
)
self._tg_client.add_event_handler(self._on_new_message, events.NewMessage)
...
params: dict = {"timeout": 55, "allowed_updates": ["message"]}
if self._update_offset is not None:
    params["offset"] = self._update_offset
data = await self._api("getUpdates", params)

offset лежит в SQLite, а не в памяти. Без этого после каждого редеплоя на Render long-poll откатывался бы назад и повторно обрабатывал старые команды.

Фильтр — не «есть слово», а вес: каждое ключевое слово кладёт в score своё значение, и текст проходит только когда сумма перевалила min_score. Стоп-слова проверяются первыми и отменяют всё, даже набранный score — один «букмекер» перекрывает десять совпадений. Нормализация приводит ё к е и схлопывает пробелы, иначе часть заказов отсеивалась бы из-за регистра.

rejected_by = find_terms(text, tuple(stop_words))
if rejected_by:
    return MatchResult(False, 0, (), rejected_by)   # score не смотрим вообще

for keyword, weight in keywords.items():
    if normalize(keyword) in normalized:
        matched.append(keyword)
        score += weight

accepted = score >= min_score

Отдельный слой отвечает за то, что делать с самим текстом заказа. Из лида вытаскиваются контакты, бюджет и дата — их потом видно в сообщении, не открывая канал. Контакты ищутся тремя шаблонами: @username, почта и любая ссылка. Дубли отбрасываются, и в лид попадает максимум восемь находок, иначе длинный пост засорял бы сообщение. Бюджет ловится по ключевым словам с двоеточием или дефисом: «бюджет:», «цена —», «ставка -».

CONTACT_RE = re.compile(
    r"(?P<username>@[A-Za-z0-9_]{5,32})"
    r"|(?P<email>[\w.+-]+@[\w-]+\.[\w.-]+)"
    r"|(?P<url>https?://\S+)"
)
BUDGET_RE = re.compile(
    r"(?:бюджет|цена|стоимость|оплата|зарплата|ставка)\s*[:\-]",
    re.IGNORECASE,
)

Самое полезное решение — доставка через SQLite как очередь с подтверждением. Запись о лиде создаётся до отправки с notified_at = NULL, и поле заполняется только после успешного sendMessage. Если отправка упала, запись осталась неподтверждённой, и следующий цикл отдаст её снова.

def record_or_should_retry(self, lead: LeadRecord) -> bool:
    existing = self._conn.execute(
        "SELECT notified_at FROM leads WHERE source = ? AND message_id = ?",
        (lead.source, lead.message_id),
    ).fetchone()
    if existing:
        return existing["notified_at"] is None   # True, только если не доставили
    self._conn.execute("INSERT INTO leads(...) VALUES(...)", (...))
    return True

UNIQUE(source, message_id) в схеме закрывает дубли на уровне базы, а не приложения. Ключевые слова, стоп-слова и источники лежат там же, отдельными таблицами — поэтому фильтры меняются командами, а деплой для этого не нужен.

У RSS нет числового ID сообщения, а колонка объявлена INTEGER. Поэтому guid ленты хешается, и первые восемь hex-символов становятся ID. feedparser при этом синхронный, поэтому разбор уходит в executor и не встаёт event loop.

msg_id = int(hashlib.md5(item.guid.encode()).hexdigest()[:8], 16)

Остановка сделана через add_signal_handler на SIGINT и SIGTERM: деплой на Render — это обычный SIGTERM, и без обработчика база закрывалась бы в середине записи.

for sig in (signal.SIGINT, signal.SIGTERM):
    loop.add_signal_handler(sig, stop_event.set)
...
tg_task = asyncio.create_task(self._tg_polling(stop_event))
rss_task = asyncio.create_task(self._rss_loop(stop_event))
try:
    await stop_event.wait()
finally:
    tg_task.cancel(); rss_task.cancel()
    await self._http.close()
    await self._tg_client.disconnect()
    self.storage.close()

Итог

1221 строка, четыре зависимости, три среды. На старте бот добирает по пять последних сообщений каждого канала, иначе первый прогон молчал бы до первого нового поста. Фильтры, стоп-слова и источники правятся из чата.

Развёрнут на Render как Web Service через render.yaml, доступ к данным только через переменные окружения. Встроенный HTTP-сервер отдаёт /health на трёх путях, чтобы не угадывать, какой именно проверяет платформа. Пока инстанс не спит, cron в GitHub Actions пингует его каждые 10 минут.

Логика разложена по транспортам, а не по слоям: транспортный код в app.py, фильтры отдельно, разбор лент отдельно, база отдельно. Конфиг читается из переменных окружения прямо в app.py — свой модуль для шести os.getenv не окупался.