Сегодня у многих есть свой Telegram-канал: личный блог, канал продукта, проектная рассылка или просто канал "для своих". Посты публикуются, идут реакции, подписчики иногда растут, иногда падают. Но до статистики большинство добирается редко: нужно отдельно открывать статистику от телеграм, которая не у всех то и доступна, смотреть графики, считать охваты и пытаться понять, какие посты зашли, а какие нет.
Очевидный путь - автоматизировать все самому: поднять VPS, поставить туда Python, написать скрипт, настроить crontab, следить за работой и молиться, что все будет работать без ошибок. Ради одного короткого задания раз в день это выглядит избыточно, приходится платить за целую виртуалку и тратить время на настройку.
В этой статье я покажу, как сделать проще: собрать небольшой Python-скрипт, который раз в день отправляет вам в Telegram краткий отчет по каналу: количество постов, просмотры, репосты, реакции, самые популярные реакции и топ-посты за указанный вами период. Без своего сервера и без настройки crontab: запуском по расписанию будет заниматься сервис Cron Jobs в Amvera Cloud, а вы будете платить только за фактическое время работы контейнера и буквально копейки за ожидание запуска.
В этой статье будет показан конкретный вариант реализации привязанный к нашему сервису. Но вы можете сделать по аналогии у любого другого провайдера, внеся минимальные изменения. Вы же можете дополнять код, улучшать и делать с ним все, к чему душа лежит, т.к. в конце будет доступна ссылка на исходники бота.
В статье разберем:
как с помощью клиента в Pyrogram получить статистику по постам канала (просмотры, репосты, реакции, прочее);
как сформировать простой отчет, который удобно читать прямо в Telegram;
как настроить и запусить это все в Amvera буквально за считанные минуты.
В итоге у вас получится простенький проект, который раз в определенный период будет отсылать настоящую статистику.
Что понадобится
Для повторения примера из статьи вам понадобится:
Аккаунт в Telegram и свой канал, в котором вы либо владелец, либо админ;
Аккаунт в Amvera Cloud - создадим чуть позже;
Созданный файл сессии Telegram - также создадим позже;
20 минут свободного времени.
Теоретическая часть
Прежде чем начать разработку самого бота, давайте разберемся, как вообще все будет работать логически.
По сути у нас один разовый рабочий процесс:
Скрипт стартует по расписанию;
Забирает данные о постах в канале за нужный период;
Собирает из этих данных сообщение-отчет;
Отправляет сообщение в указанный чат.
Все это выполняет userbot - бот, запущенный с аккаунта пользователя Telegram. Т.е. все действия выполняются именно от имени вашего аккаунта. Через него мы можем получить всю необходимую информацию о канале и счетчики просмотров, репостов, реакций и т.д.
Задача занимает буквально 5-10 секунд в зависимости от числа постов, что очень выгодно для нас.
Алгоритм работы бота
Кратко логика такая:
Прочитать переменные окружения (канал для статистики, чат для отчета, период в днях и т.д.).
Подключиться к Telegram через Pyrogram-сессию пользователя, файл которой мы создадим чуть позже.
Определить временной диапазон (
сейчас - N дней) и пройтись по постам канала за этот период.Для каждого поста собрать все доступные счетчики: просмотры, репосты, реакции, популярные реакции.
Посчитать суммарные и средние значения, выделить несколько топ-постов, которые позже добавим в сообщение отчета.
Прочитать прошлое количество подписчиков из
stats.json, посчитать прирост и обновить файл.Сформировать текстовый отчет и отправить его в указанный чат.
Подготовка, разработка и деплой
Теперь, когда у нас есть ясная картина работы бота, мы можем начать подготовку.
Создание файла сессии
Как я говорил ранее, нам понадобится файл сессии .session. Его мы создадим с помощью специального скрипта init_session.py:
# pip install pyrogram tgcrypto
from pyrogram import Client
api_id = int(input("api_id: "))
api_hash = input("api_hash: ")
session_name = input("session name: ")
with Client(session_name, api_id=api_id, api_hash=api_hash) as app:
print("Успешная авторизация как", app.me.id)
Фактически скрипт авторизовывается с помощью app_id и app_hash, который вы можете создать по ссылке, и создает файл сессии с введенным session name названием. После ввода session_name, начнется авторизация через номер телефона - просто следуйте инструкциям.
Если вы все сделали правильно, в директории запускаемого скрипта появится файл <session_name>.session - именно он нам понадобится вдальнейшем.
Знакомство с Amvera и регистрация
Что за сервис Amvera и зачем он нужен? Amvera Cloud - облачный сервис для быстрого и удобного деплоя IT-приложений различной сложности. Его особенность в простоте деплоя и обновлений любого проекта.
Из бонусов:
Можно сделать git push amvera master и сервис сам все соберет и настроит.
Встроенный сервис запуска Cron Jobs, что позволяет платить только за те минуты, пока выполняется наша крон-задача.
Стартовые 111 рублей после регистрации для тестов, которые вы получите без дополнительных условий;
А что интересно конкретно нам: вы платите лишь за работу самого приложения, запущенного с помощью Cron и копейки за время ожидания запуска.
На примере тарифа "Начальный Плюс CPU", вы платите 4 р./час за работающее приложение (а работать у нас проект будет не более 30 секунд в день).
Код main.py и файл зависимостей
Для деплоя приложения нам необходимо иметь 2 файла: main.py - сам код бота и requirements.txt - файл зависимостей в PIP.
Для нашего кода requirements.txt будет выглядеть так:
Pyrogram==2.0.106
TgCrypto==1.2.5
python-dotenv==1.2.1
Ранее мы уже описывали алгоритм работы бота, поэтому не вижу смысла расписывать каждую строчку кода.
import os
import json
import asyncio
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
from pyrogram import Client
STATE_FILE = os.getenv("TG_STATE_FILE", "stats.json")
def parse_chat_ref(text: str):
if not text:
return text
text = text.strip()
if text.startswith("http://") or text.startswith("https://"):
if "t.me/" in text:
text = text.split("t.me/", 1)[1]
text = text.split("?", 1)[0]
if text.startswith("@"):
text = text[1:]
if text.lstrip("-").isdigit():
return int(text)
return text
def format_int(n):
return f"{int(n):,}".replace(",", " ")
async def main():
load_dotenv()
api_id = os.getenv("TG_API_ID")
api_hash = os.getenv("TG_API_HASH")
session_name = os.getenv("TG_SESSION_NAME")
if not api_id or not api_hash or not session_name:
raise RuntimeError("Нужно прописать TG_API_ID, TG_API_HASH и TG_SESSION_NAME в переменные окружения")
api_id = int(api_id)
stats_chat_raw = os.getenv("TG_STATS_CHANNEL")
report_chat_raw = os.getenv("TG_REPORT_CHAT")
if not stats_chat_raw or not report_chat_raw:
raise RuntimeError("Нужно прописать TG_STATS_CHANNEL и TG_REPORT_CHAT в переменные окружения")
period_days = int(os.getenv("TG_PERIOD_DAYS", "1"))
top_posts = int(os.getenv("TG_TOP_POSTS", "3"))
stats_chat_ref = parse_chat_ref(stats_chat_raw)
report_chat_ref = parse_chat_ref(report_chat_raw)
async with Client(
name=session_name,
api_id=api_id,
api_hash=api_hash,
workdir=".",
) as app:
chat = await app.get_chat(stats_chat_ref)
now = datetime.now(timezone.utc)
since = now - timedelta(days=period_days)
posts = 0
views_sum = 0
forwards_sum = 0
reactions_sum = 0
emoji_stats = {}
top_list = []
async for msg in app.get_chat_history(chat.id, limit=0):
msg_date = msg.date
if msg_date.tzinfo is None:
msg_date = msg_date.replace(tzinfo=timezone.utc)
if msg_date < since:
break
if getattr(msg, "service", False):
continue
posts += 1
# Основные счетчики
views = msg.views or 0
forwards = msg.forwards or 0
views_sum += views
forwards_sum += forwards
if msg.reactions and getattr(msg.reactions, "reactions", None):
for r in msg.reactions.reactions:
count = r.count or 0
if count <= 0:
continue
reactions_sum += count
emoji = r.emoji or "⭐"
emoji_stats[emoji] = emoji_stats.get(emoji, 0) + count
score = views * 10 + forwards
top_list.append((score, views, forwards, msg.id))
top_list.sort(reverse=True)
if top_posts > 0:
top_list = top_list[:top_posts]
subs_now = chat.members_count or 0
state = {}
if os.path.exists(STATE_FILE):
try:
with open(STATE_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict):
state = data
except Exception:
state = {}
key = str(chat.id)
old_info = state.get(key, {})
old_subs = old_info.get("subscribers")
subs_diff = None
if isinstance(old_subs, int):
subs_diff = subs_now - old_subs
lines = []
title = chat.title or chat.username or str(chat.id)
start_str = since.strftime("%d.%m.%Y")
end_str = now.strftime("%d.%m.%Y")
lines.append(f"? Статистика канала «{title}»")
if period_days == 1:
lines.append(f"За последние 24 часа (с {start_str} по {end_str})")
else:
lines.append(f"За последние {period_days} дн. (с {start_str} по {end_str})")
lines.append("")
lines.append(f"Подписчики: {format_int(subs_now)}")
if subs_diff is None:
if subs_now:
lines.append("• Новые за период: первый запуск, пока не с чем сравнить")
else:
sign = "+" if subs_diff >= 0 else ""
lines.append(f"• Новые за период: {sign}{format_int(subs_diff)}")
lines.append("")
lines.append("Посты и охваты")
lines.append(f"• Постов: {posts}")
if posts > 0:
lines.append(f"• Средний охват на пост: {format_int(views_sum // posts)}")
else:
lines.append("• Средний охват на пост: 0")
lines.append(f"• Суммарные просмотры: {format_int(views_sum)}")
if forwards_sum > 0 and posts > 0:
lines.append(f"• Суммарные репосты: {format_int(forwards_sum)}")
lines.append(f"• Средние репосты на пост: {format_int(forwards_sum // posts)}")
else:
lines.append("• Репостов пока не было")
if reactions_sum > 0 and posts > 0:
lines.append(f"• Суммарные реакции: {format_int(reactions_sum)}")
lines.append(f"• Средние реакции на пост: {format_int(reactions_sum // posts)}")
top_emoji = sorted(
emoji_stats.items(),
key=lambda x: x[1],
reverse=True
)[:3]
if top_emoji:
emoji_text = ", ".join(
f"{emoji} - {format_int(count)}" for emoji, count in top_emoji
)
lines.append(f"• Чаще всего ставят: {emoji_text}")
else:
lines.append("• Реакций пока не было")
if top_list:
lines.append("")
lines.append("Топ постов за период:")
username = chat.username
for idx, (score, views, forwards, msg_id) in enumerate(top_list, start=1):
if username:
link = f"https://t.me/{username}/{msg\\_id}"
else:
link = f"ID сообщения: {msg_id}"
line = f"{idx}) {format_int(views)} просмотров"
if forwards > 0:
line += f", {format_int(forwards)} репостов"
line += f" - {link}"
lines.append(line)
text = "\n".join(lines)
await app.send_message(report_chat_ref, text)
state[key] = {
"subscribers": subs_now,
"last_run": now.isoformat()
}
tmp_file = STATE_FILE + ".tmp"
with open(tmp_file, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
os.replace(tmp_file, STATE_FILE)
if __name__ == "__main__":
asyncio.run(main())
Основную массу кода занимает формирование строки отчета.
Создание проекта и переменных окружения
Когда мы имеем все для запуска, мы можем начать создание проекта. Для этого откройте главную страницу проектов и жмите на плитку Cron Jobs -> Создайте первый Cron Job.
В открывшемся окне выбираем:
Название проекта: совершенно произвольное, это не играет роли;
Выберите тариф: для нашего проекта будет достаточно Начального Плюс CPU;
Подключить постоянное хранилище: жмем галочку;
Обьем: нам хватит и 1ГБ.
Жмём далее.
Теперь нужно настроить сам Cron Job.
Cron выражение: у вас тут будет свое значение - его можно взять из примеров в документации. Для теста я выберу
* * * * *, что будет запускать код каждую минуту.Максимальное время работы в секундах: можно поставить 600
Политика запуска: рекомендуется ставить
REPLACE
Все остальные этапы создания проекта просто пропускаем.
После создания проекта, в разделе Cron Jobs появится ваш проект - откройте его и перейдите во вкладку "Переменные".
Здесь вам нужно создать следующие переменные окружения:
TG_API_ID: ваш API_ID, используемый при создании файла сессии
TG_API_HASH: ваш API_HASH, используемый при создании файла сессии
TG_STATS_CHANNEL: канал, чью статистику вы будете собирать. Рекомендуется писать в формате @username.
TG_REPORT_CHAT: айди чата, в который будут присылаться отчеты
TG_TOP_POSTS: топ из скольких постов будет писаться в отчете
TG_SESSION_NAME: название файла сессии без расширения
.sessionTG_STATE_FILE: обязательно
/data/stats.jsonTG_PERIOD_DAYS: период сбора статистики из постов в днях (посты за сколько дней будут просмотрены)
Загрузим файлы. Переходим во вкладку "Репозиторий" и загружаем в Code файлы main.py, requirements.txt, файл сессии.
Остался последний шаг - настройка конфигурации приложения. Тут перейдите во вкладку "Конфигурация" и просто повторите настройки за мной:
После применения конфигурации нажимаем кнопку "Собрать" - и все, в течение пары минут приложение соберется и начнет работу!
Исходники проекта уже доступны на GitHub!
Итог
Давайте коротко подведем итог, что мы сделали.
Создали простого юзербота, которые от имени вашего аккаунта собирает статистики по постам и отправляет отчет в определенный чат
Упаковали это все в проект на Amvera.
Сэкономили на VPS.
Далее можно развивать идею как угодно: добавлять выгрузку в GoogleSheets, строить свои метрики или отправлять разные отчеты в разные чаты - тут работает только ваша фантазия!