Traffic Cardinal
написал 18.05.2026
Traffic Cardinal
написал 18.05.2026
Мониторинг активности под постами — основа эффективности при работе в Телеге. В особенности когда речь идет о мониторинге активности под постами в чужих каналах. Но делать это ручками — удовольствие то еще. Впрочем, оно и не обязательно, ведь можно сделать бот, который будет делать это за вас.
О том, как его сделать, рассказываем ниже.
Какие задачи решает бот для отслеживания активности под постами
Впрочем, сначала чуть подробнее остановимся на том, что именно можно делать с помощью данного бота. А делать с его помощью можно следующее:
Анализ активности — в самых широких смыслах.
Анализ просмотров — чтобы увидеть, какие посты привлекают больше всего внимания.
Анализ реакций — чтобы видеть, какие посты провоцируют на взаимодействие.
Анализ комментов — чтобы видеть, какие посты вызывают резонанс.
Разумеется, как и в случае с любым другим ботом, любые изначально не задуманные, но обнаруженные в процессе способы применения — также актуальны.
Принцип работы бота для отслеживания активности под постами
Алгоритмически же данный бот делает следующее:
Слушает эфир в чате управления — реагируя на вводимые команды.
Слушает эфир в каналах из json-файла — проверяя соответствие постов заданным критериям.
Реагирует на /add ссылка, частота (минуты), период (дни), просмотры, реакции, комменты (например, /add ссылка 60 1 1000 100 5) — записывая канал и его атрибуты в json-файл.
Реагирует на /del ссылка — удаляет канал и его атрибуты из json-файла.
Реагирует на /show ссылка — выводит атрибуты канала из json-файла.
Реагирует на /force ссылка — форсирует проверку канала.
Реагирует на /force — форсирует проверку всех каналов.
Реагирует на /list — выводит список всех каналов.
Мониторит активность — проверяет посты в каналы из json-файла в соответствии с заданными каналу атрибутами.
Выводит оповещения — в случае если пост набирает указанное количество просмотров, реакций или комментариев, бот пересылает его в чат управления.
Кроме того, при вводе /add «поверх» уже прописанной ранее инфы, бот ее перезаписывает.
Также в случае неправильного ввода команд оповещает о правильном формате.
Пошаговая инструкция, как сделать бот для отслеживания активности под постами
Как и в случае с любыми другими телеграм-ботами, сначала нужно развернуть Python-сервер. Сделать это можно как описанным тут способом, так и просто арендовав его. Затем настройка. Пишем в консоль:
pip install telethon
pip install asyncio
Далее делаем вот что:
- Заходим в веб-версию Телеги.
- Ищем «API development tools».
- Заполняем данные.
- Получаем api_id и api_hash.
- Создаем чат управления с самим собой (или другими админами бота) и копируем его ID.
Затем:
1. Создаем файл bot.py.
2. Вставляем в него:
import asyncio
import json
from datetime import datetime, timedelta, timezone
from telethon import TelegramClient, events
API_ID = ваш API_ID
API_HASH = "ваш API_HASH"
CONTROL_CHAT_ID = ID чата для управления
DATA_FILE = "channels.json"
client = TelegramClient("session", API_ID, API_HASH)
def load_data():
try:
with open(DATA_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except:
return {"channels": {}}
def save_data(data):
with open(DATA_FILE, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def normalize_channel(ch: str):
return ch.replace("https://t.me/", "").replace("/", "")
async def get_metrics(msg):
views = msg.views or 0
comments = 0
try:
if msg.replies:
comments = msg.replies.replies or 0
except:
comments = 0
reactions = 0
try:
if msg.reactions and msg.reactions.results:
reactions = sum(
r.count for r in msg.reactions.results if hasattr(r, "count")
)
except:
reactions = 0
return views, comments, reactions
async def check_channel(channel, config):
data = load_data()
period = config["period_days"]
thresholds = config["thresholds"]
sent = set(config.get("sent_posts", []))
cutoff = datetime.now(timezone.utc) - timedelta(days=period)
processed_albums = set()
async for msg in client.iter_messages(channel, limit=100):
msg_date = msg.date
if msg_date.tzinfo is None:
msg_date = msg_date.replace(tzinfo=timezone.utc)
if msg_date < cutoff:
continue
if msg.grouped_id:
album_key = f"{channel}:{msg.grouped_id}"
if album_key in processed_albums:
continue
processed_albums.add(album_key)
views, comments, reactions = await get_metrics(msg)
checks = []
if thresholds["views"] > 0:
checks.append(views >= thresholds["views"])
if thresholds["reactions"] > 0:
checks.append(reactions >= thresholds["reactions"])
if thresholds["comments"] > 0:
checks.append(comments >= thresholds["comments"])
ok = True if checks and any(checks) else False
key = f"{channel}:{msg.id}"
if ok and key not in sent:
try:
await client.forward_messages(CONTROL_CHAT_ID, msg)
except:
link = f"https://t.me/{normalize_channel(channel)}/{msg.id}"
await client.send_message(
CONTROL_CHAT_ID,
f"🔥 MATCH\n{link}\nviews={views} comments={comments} reactions={reactions}"
)
sent.add(key)
config["sent_posts"] = list(sent)
async def force_all():
data = load_data()
await client.send_message(CONTROL_CHAT_ID, "⚡ FORCE ALL")
for ch, cfg in data["channels"].items():
try:
await check_channel(ch, cfg)
except Exception as e:
await client.send_message(CONTROL_CHAT_ID, f"❌ ERROR {ch}: {e}")
save_data(data)
async def force_one(channel):
data = load_data()
cfg = data["channels"].get(channel)
if not cfg:
await client.send_message(CONTROL_CHAT_ID, "❌ Not found")
return
await client.send_message(CONTROL_CHAT_ID, f"⚡ FORCE {channel}")
try:
await check_channel(channel, cfg)
save_data(data)
except Exception as e:
await client.send_message(CONTROL_CHAT_ID, f"❌ ERROR {e}")
async def scheduler():
while True:
data = load_data()
now = datetime.now(timezone.utc)
for channel, cfg in data["channels"].items():
last = cfg.get("last_check")
interval = cfg["interval_minutes"]
should_run = True
if last:
last_dt = datetime.fromisoformat(last)
if last_dt.tzinfo is None:
last_dt = last_dt.replace(tzinfo=timezone.utc)
should_run = (now - last_dt).total_seconds() >= interval * 60
if should_run:
try:
await check_channel(channel, cfg)
cfg["last_check"] = now.isoformat()
except Exception as e:
print("[ERROR]", channel, e)
save_data(data)
await asyncio.sleep(10)
@client.on(events.NewMessage(chats=CONTROL_CHAT_ID))
async def handler(event):
text = event.raw_text.strip()
parts = text.split()
data = load_data()
if not parts:
return
cmd = parts[0]
if cmd == "/force":
if len(parts) == 1:
await force_all()
else:
await force_one(parts[1])
return
if cmd == "/add":
if len(parts) < 7:
await event.reply("❌ /add <channel> <interval> <days> <views> <reactions> <comments>")
return
ch = parts[1]
data["channels"][ch] = {
"interval_minutes": int(parts[2]),
"period_days": int(parts[3]),
"thresholds": {
"views": int(parts[4]),
"reactions": int(parts[5]),
"comments": int(parts[6])
},
"sent_posts": [],
"last_check": None
}
save_data(data)
await event.reply(f"✅ Added {ch}")
return
if cmd == "/del":
if len(parts) < 2:
await event.reply("❌ /del <channel>")
return
data["channels"].pop(parts[1], None)
save_data(data)
await event.reply("🗑 Deleted")
return
if cmd == "/list":
await event.reply("\n".join(data["channels"].keys()) or "📭 Empty")
return
if cmd == "/show":
if len(parts) < 2:
await event.reply("❌ /show <channel>")
return
cfg = data["channels"].get(parts[1])
if not cfg:
await event.reply("❌ Not found")
return
await event.reply(json.dumps(cfg, indent=2, ensure_ascii=False))
return
await event.reply("❌ Unknown command")
async def main():
await client.start()
await client.send_message(CONTROL_CHAT_ID, "🤖 Бот запущен")
asyncio.create_task(scheduler())
await client.run_until_disconnected()
with client:
client.loop.run_until_complete(main())
3. Заменяем API_ID = ваш API_ID, API_HASH = "ваш API_HASH" и CONTROL_CHAT_ID = ID чата для управления на свои значения.
4. Сохраняем файл.
5. Проверяем работу.
Демонстрация работы
Подводя итоги
Как видите, отслеживать активность под постами не так уж и сложно. В особенности если за вас это делает бот. Следите за нашей Телегой, чтобы первыми получать исходный код других ботов для арбитража трафика.
