← Все гайды AI Release
EN Подписаться на @ai_release1

Как мы убили дубли в Telegram-канале: идемпотентная отправка, проверка по t.me и лимит частоты

Обновлено: 30.10.2026 · AI Release · @ai_release1
Как мы убили дубли в Telegram-канале: идемпотентная отправка, проверка по t.me и лимит частоты

TL;DR

Требования и подготовка

Иллюстрация к разделу «Требования и подготовка»
Иллюстрация к разделу «Требования и подготовка»

Идемпотентность — свойство операции, при котором повторный вызов с теми же входными данными не создаёт нового результата. Здесь результат — сообщение в канале, входные данные — текст, канал и слот публикации.

Дальше — важная часть: одна и та же логика используется и для первого запуска, и для повторного. Поэтому схема БД создаётся до первой отправки. Если вы только начинаете с ботами, базовый каркас проекта разобран в гайде Telegram-бот с ИИ на Python: от нуля до деплоя.

Что сделать: создайте файл dedup.py со схемой таблицы outbox и функцией генерации ключа идемпотентности.

Зачем: ключ — это «отпечаток» поста. Одинаковый ключ даёт конфликт первичного ключа, и второй INSERT не проходит. На этом конфликте и строится защита от дубля.

# dedup.py
import sqlite3, hashlib, json, time

DB = "outbox.db"

def connect():
    conn = sqlite3.connect(DB, timeout=30)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=5000")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS outbox (
            idempotency_key TEXT PRIMARY KEY,
            channel         TEXT NOT NULL,
            payload_hash    TEXT NOT NULL,
            status          TEXT NOT NULL,   -- sending | sent | failed
            message_id      INTEGER,
            created_at      INTEGER NOT NULL
        )
    """)
    conn.commit()
    return conn

def make_key(channel: str, text: str, slot: str) -> str:
    raw = json.dumps({"channel": channel, "text": text, "slot": slot},
                     ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

slot — окно публикации, например 2026-02-11T10:00. Повторный запуск в том же окне даёт тот же ключ, а запуск в 11:00 — уже новый.

Как проверить: создайте БД и убедитесь, что таблица появилась.

cd /opt/tg-publisher
python3 -c "import dedup;c=dedup.connect();print(c.execute('PRAGMA journal_mode').fetchone())"
python3 -c "import sqlite3;print(sqlite3.connect('outbox.db').execute(\"select name from sqlite_master where type='table'\").fetchall())"

Ожидаемый вывод:

('wal',)
[('outbox',)]

Проверьте доступ бота к каналу — до всей логики:

curl -s "https://api.telegram.org/bot<BOT_TOKEN>/getMe"
curl -s "https://api.telegram.org/bot<BOT_TOKEN>/getChat?chat_id=@my_channel"

Первая команда возвращает {"ok":true,"result":{"id":...,"is_bot":true,...}}, вторая — объект чата с "type":"channel". Если вместо этого {"ok":false,"error_code":403,...} — бот не админ, дальше идти смысла нет.

Шаг 2. Идемпотентная отправка и ограничитель частоты

Иллюстрация к разделу «Шаг 2. Идемпотентная отправка и ограничитель частоты»
Иллюстрация к разделу «Шаг 2. Идемпотентная отправка и ограничитель частоты»

Что сделать: напишите publish.py, который занимает ключ в БД, отправляет сообщение и записывает message_id. Плюс поставьте поверх отправки token bucket — ограничитель частоты (механизм, который выдаёт не больше N запросов за интервал).

Зачем: два независимых риска. Первый — повторный запуск (cron сработал дважды, оператор нажал «ещё раз»). Второй — flood-лимит Telegram: слишком частые запросы бот получает как 429. Ключ лечит первый риск, ограничитель — второй.

# publish.py
import argparse, time, requests
from dedup import connect, make_key
from dedup import hashlib  # см. импорт ниже

BOT_TOKEN = "<BOT_TOKEN>"
API = f"https://api.telegram.org/bot{BOT_TOKEN}"

class TokenBucket:
    def __init__(self, rate_per_sec: float, burst: int):
        self.rate, self.capacity = rate_per_sec, burst
        self.tokens, self.ts = burst, time.monotonic()

    def take(self):
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.ts) * self.rate)
            self.ts = now
            if self.tokens >= 1:
                self.tokens -= 1
                return
            time.sleep((1 - self.tokens) / self.rate)

bucket = TokenBucket(rate_per_sec=0.33, burst=1)  # ~20 сообщений/минуту

def send_with_retry(payload, attempts=5):
    for i in range(attempts):
        bucket.take()
        r = requests.post(f"{API}/sendMessage", json=payload, timeout=15)
        data = r.json()
        if data.get("ok"):
            return data["result"]
        if data.get("error_code") == 429:
            wait = data.get("parameters", {}).get("retry_after", 2 ** i)
            print(f"429: sleep {wait}s")
            time.sleep(wait + 1)
            continue
        raise RuntimeError(data.get("description"))
    raise RuntimeError("rate limit retries exhausted")

def publish(channel: str, text: str, slot: str):
    conn = connect()
    key = make_key(channel, text, slot)
    cur = conn.execute(
        "INSERT OR IGNORE INTO outbox(idempotency_key, channel, payload_hash, status, created_at)"
        " VALUES (?,?,?,?,?)",
        (key, channel, hashlib.sha256(text.encode("utf-8")).hexdigest(),
         "sending", int(time.time())))
    conn.commit()
    if cur.rowcount == 0:
        print(f"skip: duplicate key {key[:12]}")
        return None

    try:
        result = send_with_retry({"chat_id": channel, "text": text,
                                  "disable_web_page_preview": False})
    except Exception as e:
        conn.execute("UPDATE outbox SET status='failed' WHERE idempotency_key=?", (key,))
        conn.commit()
        raise

    mid = result["message_id"]
    conn.execute("UPDATE outbox SET status='sent', message_id=? WHERE idempotency_key=?",
                 (mid, key))
    conn.commit()
    print(f"sent message_id={mid}")
    return mid

if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--slot", required=True)
    p.add_argument("--channel", default="@my_channel")
    p.add_argument("--text-file", default="post.txt")
    a = p.parse_args()
    publish(a.channel, open(a.text_file, encoding="utf-8").read(), a.slot)

Строку from dedup import hashlib замените на обычный import hashlib в начале файла — модуль стандартный.

Почему это работает: INSERT OR IGNORE при конфликте по первичному ключу возвращает rowcount == 0, и до сетевого вызова дело не доходит. status='sending' фиксируется до запроса — значит, факт «начали отправлять» не потеряется при падении процесса.

Как проверить: запустите публикацию дважды с одним и тем же слотом.

echo "Тестовый пост: проверка дедупликации" > post.txt
python3 publish.py --slot 2026-02-11T10:00
python3 publish.py --slot 2026-02-11T10:00

Ожидаемый экран (значения message_id зависят от канала):

sent message_id=4127
skip: duplicate key 9f2c1a7b4d10

В канале — одно сообщение. Проверьте содержимое БД:

sqlite3 outbox.db "select substr(idempotency_key,1,12), status, message_id from outbox;"
9f2c1a7b4d10|sent|4127

Шаг 3. Проверка по t.me и закрытие «шва»

Иллюстрация к разделу «Шаг 3. Проверка по t.me и закрытие шва»
Иллюстрация к разделу «Шаг 3. Проверка по t.me и закрытие шва»

Что сделать: добавьте reconcile.py — сверку зависших записей со статусом sending через публичную страницу https://t.me/<канал>/<message_id>.

Зачем: остаётся узкое место — процесс упал между sendMessage и UPDATE status='sent'. Пост в канале есть, а в БД sending. При следующем запуске появится дубль. Проверка по t.me отвечает на вопрос «пост реально существует?» без риска повторной отправки.

# reconcile.py
import re, time, requests
from dedup import connect

def exists_on_tme(username: str, message_id: int) -> bool:
    url = f"https://t.me/{username.lstrip('@')}/{message_id}"
    html = requests.get(url, timeout=10,
                        headers={"User-Agent": "Mozilla/5.0"}).text
    if "Post not found" in html or "tgme_widget_message_error" in html:
        return False
    return "tgme_widget_message" in html

def find_message_id(username: str, text_head: str):
    # запасной путь: последние посты канала через веб-превью
    html = requests.get(f"https://t.me/s/{username.lstrip('@')}", timeout=10,
                        headers={"User-Agent": "Mozilla/5.0"}).text
    ids = re.findall(r'data-post="[^/]+/(\d+)"', html)
    if text_head[:40] in html and ids:
        return int(ids[-1])
    return None

conn = connect()
rows = conn.execute(
    "SELECT idempotency_key, channel FROM outbox"
    " WHERE status='sending' AND created_at < ?", (int(time.time()) - 300,)).fetchall()

for key, channel in rows:
    mid = find_message_id(channel, "")
    if mid and exists_on_tme(channel, mid):
        conn.execute("UPDATE outbox SET status='sent', message_id=? WHERE idempotency_key=?",
                     (mid, key))
        print(f"orphan key={key[:12]} -> found on t.me/{channel.lstrip('@')}/{mid}, marked sent")
    else:
        conn.execute("UPDATE outbox SET status='failed' WHERE idempotency_key=?", (key,))
        print(f"orphan key={key[:12]} -> not found, marked failed (safe to resend)")
conn.commit()

Разметка веб-превью t.me не является публичным API и может меняться — после обновления Telegram сверьтесь с документацией и проверьте вручную: откройте https://t.me/my_channel/4127 в браузере. Если пост удалён, страница содержит Post not found.

Как проверить: сымитируйте сбой — вручную поставьте статус sending и запустите сверку.

sqlite3 outbox.db "update outbox set status='sending', message_id=NULL where message_id=4127;"
python3 reconcile.py
sqlite3 outbox.db "select status, message_id from outbox;"

Ожидаемый вывод:

orphan key=9f2c1a7b4d10 -> found on t.me/my_channel/4127, marked sent
sent|4127

Сверьте итог с каналом: откройте https://t.me/my_channel — на экране должен быть ровно один пост с текстом Тестовый пост: проверка дедупликации. Дальше поставьте publish.py в cron с шагом слота (например, 0 * * * *) и запускайте reconcile.py раз в 15 минут. Если публикацию собирает агент, который сам решает, когда и что постить, схему «агент → ключ → отправка» разберите по гайду ИИ-агенты для автоматизации задач: гид по инструментам.

Лимиты, которые заложены в примере (сверьтесь с актуальной версией документации Telegram Bot API, значения могут меняться):

ОграничениеОриентирЧто делает код
Сообщений в минуту в один канал~20`TokenBucket(0.33, 1)` — пауза ~3 с
Общая пропускная способность бота30 сообщений/сектот же bucket на все вызовы
Ответ 429`parameters.retry_after` в секундахсон `retry_after + 1` и повтор

Возможные проблемы и решения

Проблема. В логах 429, публикация пачки постов растягивается на минуты

Проблема. Дубль в канале после падения контейнера

Проблема. Второй воркер падает на запись в БД

Резюме и следующие шаги

Три механизма закрывают три разных источника дублей: первичный ключ по хешу поста — повторный запуск, TokenBucket с обработкой retry_after — flood-лимит, сверка по t.me — падение процесса на середине отправки. Дальше вынесите BOT_TOKEN и @my_channel в переменные окружения, добавьте reconcile.py в cron и прогоните тест с намеренно убитым процессом во время отправки.

FAQ

Почему пост дублируется, если скрипт запускается по cron каждую минуту?

Потому что ключ идемпотентности не включает слот или включает время запуска. Ключ должен строиться из канала, текста и окна публикации (make_key(channel, text, slot)), тогда все запуски внутри окна схлопываются в один.

Зачем проверять t.me, если в ответе API уже есть message_id?

message_id появляется только после успешного ответа. Если процесс умер до записи в БД, message_id потерян, а пост в канале уже есть — веб-превью t.me/<канал>/<id> позволяет найти его без повторной отправки.

Какой лимит отправки в Telegram-канал?

Ориентир — около 20 сообщений в минуту в один чат и 30 сообщений в секунду суммарно для бота; точные значения и формат ответа 429 сверьте с документацией Telegram Bot API. В коде лимит задан параметрами TokenBucket(0.33, 1).

Что делать, если getChat возвращает 403?

Добавьте бота в администраторы канала с правом публикации сообщений и проверьте, что chat_id указан как @my_channel, а не как числовой id без префикса -100. После правки повторите curl -s "https://api.telegram.org/bot<BOT_TOKEN>/getChat?chat_id=@my_channel" — должен вернуться объект с "type":"channel".

🤖 Кратко для ИИ

📚 Связанные гайды

ИИ-помощники для изучения английского: лучшие боты и инструменты в Telegram

Обзор эффективных способов учить английский с помощью ИИ прямо в Telegram: боты, диалоговые тренажёры, генерация упражне

Telegram-бот с ИИ на Python: от нуля до деплоя

Пишем Telegram-бота с подключением LLM на Python: получение токена, обработка сообщений, память диалога и запуск на серв

Лучшие нейросети и ИИ-модели 2026: что умеют и где попробовать

Обзор главных ИИ-моделей года — GPT, Claude, Gemini, DeepSeek и другие: возможности, тесты и ссылки на наши обзоры.