RSS към канал в Telegram с одобрение и тихи часове

Канал в Telegram, който повтаря блог, новинарски сайт или новите продукти на магазин, е едно от най-простите неща за автоматизиране и едно от най-лесните за объркване. Обичайната първа версия публикува всичко от емисията в мига, в който ботът тръгне, публикува част от него повторно след рестарт, буди абонатите в три през нощта, губи края на дългите надписи и спира за известно време, когато Telegram отговори „Too Many Requests“. Тук изграждаме частите, които избягват всяко от тези неща: учтиво изтегляне, което помни какво е публикувано, съобщения в границите на Telegram, одобрение от екипа, тихи часове и ограничение на час. Примерите са на чист Python 3.12 без пакети и са пуснати точно както са отпечатани.

#Как е устроен ботът

Създайте бот с BotFather и го добавете към канала като администратор с право да публикува. След това ботът върти цикъл: за всяка емисия я изтегля, когато ѝ дойде интервалът, намира елементите, които още не е публикувал, превръща всеки в съобщение, праща го в чат за преглед, ако емисията иска одобрение, и го публикува в канала в първия момент, който правилата на канала позволяват. Всичко, което трябва да помни, отива в един SQLite файл, така че рестартът не променя нищо.

#Изтегляйте учтиво, публикувайте веднъж

Емисия, проверявана на всеки петнайсет минути, се тегли почти сто пъти на ден, а се променя няколко пъти. HTTP има евтин отговор за това: върнете стойностите ETag и Last-Modified от последния отговор като If-None-Match и If-Modified-Since, и непроменена емисия отговаря с 304 Not Modified без тяло.

import sqlite3
import urllib.error
import urllib.request

db = sqlite3.connect("razglaska-demo.sqlite")
db.execute("CREATE TABLE IF NOT EXISTS feed (url TEXT PRIMARY KEY, etag TEXT, modified TEXT)")
db.execute("CREATE TABLE IF NOT EXISTS seen (feed TEXT, key TEXT, PRIMARY KEY (feed, key))")


def fetch(url: str) -> bytes | None:
    """The feed body, or None when the server says it has not changed since the last fetch."""
    row = db.execute("SELECT etag, modified FROM feed WHERE url = ?", (url,)).fetchone()
    req = urllib.request.Request(url, headers={"User-Agent": "feed-to-channel/1.0"})
    if row and row[0]:
        req.add_header("If-None-Match", row[0])
    if row and row[1]:
        req.add_header("If-Modified-Since", row[1])
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            db.execute("INSERT OR REPLACE INTO feed VALUES (?, ?, ?)",
                       (url, r.headers.get("ETag"), r.headers.get("Last-Modified")))
            db.commit()
            return r.read()
    except urllib.error.HTTPError as e:
        if e.code == 304:
            return None
        raise


def is_new(feed: str, key: str) -> bool:
    """True the first time an item key (its guid or id, else its link) is seen for a feed."""
    cur = db.execute("INSERT OR IGNORE INTO seen VALUES (?, ?)", (feed, key))
    db.commit()
    return cur.rowcount == 1

Срещу малък локален сървър, който праща ETag, първото извикване връща тялото, а второто връща None. Всеки елемент има нужда от ключ, който не се променя: guid в RSS, id в Atom или JSON Feed и само когато няма такъв, връзката. is_new() оставя SQLite да реши с една заявка: записът минава само първия път, когато ключът се види за тази емисия, така че един и същ ключ в две емисии са две публикации, а един и същ ключ два пъти в една емисия е една.

Първата проверка на нова емисия има нужда от собствено правило. Емисията обикновено държи последните 10 до 50 елемента и публикуването на всички тях залива канала. Отбележете ги всички като видени и не публикувайте нищо, или само най-новите един-два, а след това публикувайте от най-стария към най-новия.

#Съобразете се с границите на Telegram

Надписът на снимка може да е до 1024 знака след обработката на форматирането, а текстовото съобщение до 4096 (Bot API, sendPhoto и sendMessage). „След обработката“ значи, че таговете на HTML режима не се броят, но текстът в тях се брои. Telegram мери форматирането в UTF-16 единици, в които емоджито заема две, така че броенето в UTF-16 ви държи на сигурната страна. В HTML режим в текста трябва да се екранират само &, < и >.

import html
import re

CAPTION_LIMIT = 1024   # photo captions; a text message may have 4096


def visible_length(message_html: str) -> int:
    """Length as Telegram counts it: the text after the HTML tags are parsed, in UTF-16 code units."""
    text = html.unescape(re.sub(r"<[^>]+>", "", message_html))
    return len(text.encode("utf-16-le")) // 2


def as_post(title: str, summary: str, link: str, photo: str | None) -> tuple[str, dict]:
    e = lambda s: html.escape(s, quote=False)
    body = f"<b>{e(title)}</b>\n\n{e(summary)}\n\n{e(link)}"
    if photo and visible_length(body) <= CAPTION_LIMIT:
        return "sendPhoto", {"photo": photo, "caption": body, "parse_mode": "HTML"}
    return "sendMessage", {"text": body, "parse_mode": "HTML"}

Със снимка и резюме от 900 знака публикацията излиза като снимка с надпис. С резюме от 1000 знака цялото съобщение е 1037 знака, твърде дълго за надпис, затова става текстово съобщение, в което прегледът на връзката пак може да покаже снимката. Това е по-добре от резюме, срязано по средата на дума, и много по-добре от отказ от Telegram. visible_length("<b>Нов брой</b> &amp; 😀") е 13: тагът изчезва, символът е един знак, а емоджито е два.

#Одобрение от екипа

Някои емисии не бива да излизат непрочетени: емисия за медии с по някой несвързан материал или емисия на магазин по време на промяна на цените. Пращайте тези елементи в затворена група на екипа, като същото съобщение с бутони под него: „Публикувай“, „В опашката“ и „Пропусни“. Натискането на бутон пристига като callback_query; проверете, че натисналият е един от администраторите ви, отговорете с answerCallbackQuery, изпълнете действието и редактирайте картата, за да вижда групата кой е решил. Пазете идентификатора на елемента в callback_data на бутона, което побира най-много 64 байта, а останалото в базата данни.

#Тихи часове и ограничение на час

Канал, който публикува през нощта, учи абонатите си да го заглушат. Задържайте елементите от емисиите в тихите часове и ги разреждайте с максимум на час:

from datetime import datetime, time, timedelta, timezone
from zoneinfo import ZoneInfo

TZ = ZoneInfo("Europe/Sofia")


def in_quiet(moment: datetime, start: time, end: time) -> bool:
    t = moment.astimezone(TZ).time()
    return start <= t < end if start < end else (t >= start or t < end)


def next_allowed(moment: datetime, start: time, end: time, posted: list[datetime], per_hour: int) -> datetime:
    """The earliest moment from `moment` on that is outside quiet hours and under the hourly limit.
    Arithmetic is done in UTC, so an hour is always sixty minutes, also on the night the clocks change."""
    t = moment.astimezone(timezone.utc)
    posted = [p.astimezone(timezone.utc) for p in posted]
    while True:
        if in_quiet(t, start, end):
            local = t.astimezone(TZ)
            day = local.date() if local.time() < end else local.date() + timedelta(days=1)
            t = datetime.combine(day, end, TZ).astimezone(timezone.utc)
            continue
        recent = sorted(p for p in posted if t - timedelta(hours=1) < p <= t)
        if len(recent) >= per_hour:
            t = recent[-per_hour] + timedelta(hours=1)
            continue
        return t.astimezone(TZ)

С тихи часове от 22:30 до 07:30 елемент, намерен в 23:10, чака до 07:30 на следващата сутрин, а намерен в 07:29 чака минута. С ограничение от четири на час и публикации в 09:00, 09:10, 09:20 и 09:40 елемент в 09:45 чака до 10:00, когато публикацията от 09:00 излиза от прозореца. Сметката е в UTC, защото часовникът се мести: на 28 март 2027 г. часовникът в София скача от 03:00 на 04:00, така че тихи часове, които свършват в 03:30, в тази нощ свършват в 04:30, с час по-късно по стенния часовник, но в правилния момент в реалното време. За публикации, които могат да излязат по всяко време, но не бива да звънят, Bot API има disable_notification.

#Ограничения и грешки

Въпросите и отговорите за ботове на Telegram молят да не пращате повече от едно съобщение в секунда в един чат и казват, че ботовете не могат да пращат повече от 20 съобщения в минута в група. Над това получавате HTTP 429 с parameters.retry_after, броя секунди за изчакване: изчакайте толкова и опитайте пак, вместо да повтаряте веднага. Отговорите 5xx и мрежовите грешки са временни: опитайте няколко пъти с растяща пауза. Някои грешки са постоянни, например когато ботът вече не е администратор на канала; отбележете публикацията като неуспешна и кажете на администраторите веднъж, вместо да опитвате всяка минута.

Вашите собствени публикации са в същата опашка. Ако администратор прати или препрати съобщение на бота в личен чат, copyMessage го публикува в канала такова, каквото е, с форматирането и медията и без реда „Препратено от“.

#Бот, който прави всичко това

Направихме Razglaska, бот за Telegram, който пускате на свой сървър и който води каналите ви, на български и английски. Публикува новите елементи от RSS, Atom и JSON емисии със снимките им, от най-стария към най-новия и никога два пъти, с изтегляне чрез ETag и Last-Modified, шаблон за всяка емисия, филтри и безопасно първо пускане. Елементите могат да чакат одобрение в чат за преглед с „Публикувай“, „В опашката“ и „Пропусни“; тихите часове, ограничението на час, тихите публикации и часовете за публикуване пазят канала спокоен. Можете да му пращате и свои публикации и да изберете „Публикувай сега“, следващия свободен час или „Ще напиша час“ с час като „утре 9:00“ или „пт 18:00“. Всичко се настройва в един TOML файл, иска Python 3.11 или по-нов и никакви други пакети и е тестван с 45 теста, включително пълни изпълнения срещу фалшив Telegram Bot API и фалшив сървър за емисии. По време на тестовете не е пускан срещу истинската услуга на Telegram и е обявен като „Очаквайте скоро“.

#Често задавани въпроси

Защо ботът ми публикува един и същ елемент от RSS два пъти?

Обикновено защото помни публикуваното само в паметта или го разпознава по заглавие или дата, които могат да се променят. Пазете ключ от guid или id на елемента в база данни и го записвайте, преди да публикувате.

Колко дълъг може да е надписът на снимка в Telegram?

До 1024 знака след обработката на форматирането, а текстовото съобщение до 4096. Пращайте дълъг елемент като текстово съобщение с преглед на връзката, вместо да го режете.

Какво означава грешка 429 от Bot API?

Пратили сте твърде много заявки. Отговорът съдържа retry_after, секундите за изчакване преди нов опит; спазвайте го и се дръжте около едно съобщение в секунда за чат.

Мога ли да публикувам в канал, без да известявам абонатите?

Да. Пратете съобщението с disable_notification, зададено на true, и абонатите ще го получат без звук.

#Източници

• Telegram Bot API
• Telegram, въпроси и отговори за ботове: изпращане и ограничения
• RFC 9110: HTTP Semantics (условни заявки, 304 Not Modified)
• Спецификация на RSS 2.0