Бот для сегментации аудитории по активности

Держите чатик или даже сетку чатов и хотите проанализировать, кто как и сколько пишет? Что ж, можно просто кикнуть мертвые души, и в чате останется исключительно актив. Но что если хочется больше инфы, чем просто «эти пишут», а «эти — нет»? С этим может помочь бот, создание которого мы опишем ниже.

Какие задачи решает бот для сегментации аудитории по активности

Но сначала уточним, для чего именно он может быть использован:

• Для создания сегментационных групп по количеству сообщений за указанное время.
• Для анализа активности юзеров в том или ином чате.
• Для анализа активности конкретных юзеров во всех отслеживаемых чатах.
• Ну и косвенно для разработки разного рода вовлекалок, основанных на стате актива.

А еще с помощью собираемой ботом статистики можно выявлять «суетологов» под последующую раскрутку других чатов — но это уже совсем другая история.

Принцип работы бота для сегментации аудитории по активности

Работает же бот следующим образом:

• Прослушивает эфир и реагирует на команды в «Избранном».

• Реагирует на команду /add ID добавлением чата с указанным ID в список отслеживаемых.

• Реагирует на команду /del ID удалением чата с указанным ID из списка отслеживаемых.

• Реагирует на команду /list выводом списка отслеживаемых.

• Реагирует на команду /set ID заданием чата с указанным ID в качестве выбранного.

• Реагирует на команду /stats выводом статистики по всем существующим сегментам аудитории для выбранного командой /set ID чата.

• Реагирует на команду /stats USER_ID по юзеру с указанным ID.

• Реагирует на команду /add_group NAME MIN-MAX h/d/w/m созданием сегмента с указанными параметрами для выбранного командой /set ID чата.

• Реагирует на команду /del_group удалением сегмента для выбранного командой /set ID чата.

• Реагирует на команду /show_group выводом информации о юзерах, относящихся к указанному сегменту для выбранного командой /set ID чата.

• Ведет учет активности в отслеживаемых чатах.

Звучит немного запутанно, но на деле все просто — бот запоминает, кто и сколько пишет в чатах, которые вы указали как чаты для отслеживания.

Пошаговая инструкция, как сделать бот для сегментации аудитории по активностиы

Начинается все, как впрочем и всегда, с развертывания сервера. Для этого подойдет любой Python-сервер — при желании можно даже сделать все на ноуте. Но лучше все же арендовать полноценный VPS. Затем настройка. Пишем в консоль:

pip install telethon

Затем получаем токены:

1. Открываем Telegram Web.
2. Находим «API development tools».
3. Жмем создать приложение и сохраняем api_id и api_hash.

Теперь сам бот:

1. Создаем файл bot.py и вставляем туда код:

import json

import os

import asyncio

from datetime import datetime

from telethon import TelegramClient, events

API_ID = ваш API_ID

API_HASH = "ваш API_HASH"

DB_FILE = "activity.json"

client = TelegramClient("activity_session", API_ID, API_HASH)

DB = {

"chats": {},

"selected_chat": None

}

def normalize_id(cid):

cid = str(cid)

if not cid.startswith("-"):

cid = "-" + cid

return cid

def load_db():

global DB

if os.path.exists(DB_FILE):

with open(DB_FILE, "r", encoding="utf-8") as f:

DB = json.load(f)

def save_db():

with open(DB_FILE, "w", encoding="utf-8") as f:

json.dump(DB, f, ensure_ascii=False, indent=2)

def period_key(p):

now = datetime.now()

if p == "d":

return now.strftime("%Y-%m-%d")

if p == "w":

return f"{now.year}-{now.isocalendar().week}"

if p == "m":

return now.strftime("%Y-%m")

if p == "y":

return str(now.year)

def get_count(user, period):

data = user.get(period, {})

if data.get("key") != period_key(period):

return 0

return data.get("count", 0)

def add_message(user):

for p in ("d", "w", "m", "y"):

key = period_key(p)

if user[p]["key"] != key:

user[p] = {

"key": key,

"count": 0

}

user[p]["count"] += 1

async def get_chat_name(cid):

try:

entity = await client.get_entity(int(cid))

return getattr(entity, "title", str(cid))

except:

return str(cid)

async def is_favorite(event):

me = await client.get_me()

return event.is_private and event.sender_id == me.id

def selected_chat():

cid = DB.get("selected_chat")

if not cid:

return None

return DB["chats"].get(cid)

def user_title(user, uid):

name = user.get("name", "").strip()

username = user.get("username")

if username:

name += f" @{username}"

if not name:

name = "Unknown"

return f"{name} ({uid})"

@client.on(events.NewMessage)

async def tracker(event):

cid = str(event.chat_id)

if cid not in DB["chats"]:

return

if not event.sender_id:

return

uid = str(event.sender_id)

users = DB["chats"][cid]["users"]

if uid not in users:

sender = await event.get_sender()

users[uid] = {

"name": (

f"{sender.first_name or ''} "

f"{sender.last_name or ''}"

).strip(),

"username": getattr(

sender,

"username",

None

),

"d": {

"key": period_key("d"),

"count": 0

},

"w": {

"key": period_key("w"),

"count": 0

},

"m": {

"key": period_key("m"),

"count": 0

},

"y": {

"key": period_key("y"),

"count": 0

}

}

add_message(users[uid])

save_db()

@client.on(events.NewMessage)

async def commands(event):

if not await is_favorite(event):

return

text = event.raw_text.strip()

if not text.startswith("/"):

return

args = text.split()

cmd = args[0].lower()

if cmd == "/add":

cid = normalize_id(args[1])

DB["chats"][cid] = {

"name": await get_chat_name(cid),

"users": {},

"groups": {}

}

save_db()

await event.reply(

"Чат добавлен"

)

elif cmd == "/del":

cid = normalize_id(args[1])

DB["chats"].pop(cid, None)

if DB["selected_chat"] == cid:

DB["selected_chat"] = None

save_db()

await event.reply(

"Удалено"

)

elif cmd == "/list":

if not DB["chats"]:

await event.reply(

"Нет чатов"

)

return

await event.reply(

"\n".join(

f"{cid} {chat['name']}"

for cid, chat in DB["chats"].items()

)

)

elif cmd == "/set":

value = args[1]

if value.lower() == "unset":

DB["selected_chat"] = None

save_db()

await event.reply(

"Выбор снят"

)

return

value = normalize_id(value)

if value not in DB["chats"]:

await event.reply(

"Чат не найден"

)

return

DB["selected_chat"] = value

save_db()

await event.reply(

"Выбран"

)

elif cmd == "/stats":

if len(args) > 1:

await stats_user(

event,

args[1]

)

return

chat = selected_chat()

if not chat:

await event.reply(

"Сначала /set"

)

return

out = [

chat["name"]

]

for uid, user in chat["users"].items():

out.append(

f"{user_title(user, uid)}: "

f"d:{get_count(user,'d')} "

f"w:{get_count(user,'w')} "

f"m:{get_count(user,'m')} "

f"y:{get_count(user,'y')}"

)

await event.reply(

"\n".join(out)

)

elif cmd == "/add_group":

chat = selected_chat()

if not chat:

await event.reply(

"Сначала /set"

)

return

name = args[1]

mn, mx = args[2].split("-")

chat["groups"][name] = {

"min": int(mn),

"max": int(mx),

"period": args[3]

}

save_db()

await event.reply(

"Создано"

)

elif cmd == "/edit_group":

chat = selected_chat()

name = args[1]

mn, mx = args[2].split("-")

chat["groups"][name] = {

"min": int(mn),

"max": int(mx),

"period": args[3]

}

save_db()

await event.reply(

"Изменено"

)

elif cmd == "/del_group":

chat = selected_chat()

chat["groups"].pop(

args[1],

None

)

save_db()

await event.reply(

"Удалено"

)

elif cmd == "/show_group":

chat = selected_chat()

await event.reply(

await group_text(

chat,

args[1]

)

)

elif cmd == "/show":

chat = selected_chat()

if not chat:

await event.reply(

"Сначала /set"

)

return

result = []

for group in chat["groups"]:

result.append(

await group_text(

chat,

group

)

)

await event.reply(

"\n\n".join(result)

)

async def group_text(chat, name):

if name not in chat["groups"]:

return "Нет группы"

g = chat["groups"][name]

out = [

f"{name} ({g['min']}-{g['max']} {g['period']})"

]

for uid, user in chat["users"].items():

count = get_count(

user,

g["period"]

)

if g["min"] <= count <= g["max"]:

out.append(

f"{user_title(user, uid)}: {count}"

)

if len(out) == 1:

out.append(

"никого"

)

return "\n".join(out)

async def stats_user(event, uid):

uid = str(uid)

out = [

f"User {uid}"

]

for chat in DB["chats"].values():

if uid in chat["users"]:

user = chat["users"][uid]

out.append(

f"{chat['name']}: "

f"d:{get_count(user,'d')} "

f"w:{get_count(user,'w')} "

f"m:{get_count(user,'m')} "

f"y:{get_count(user,'y')}"

)

await event.reply(

"\n".join(out)

)

async def main():

load_db()

await client.start()

print("Started")

await client.run_until_disconnected()

asyncio.run(main())

2. Заменяем API_ID = ваш API_ID и API_HASH = "ваш API_HASH" на свои.

3. Сохраняем файл.

4. Проверяем работоспособность бота.

Демонстрация работы

Подводя итоги

Как видите, бот может превратить обычную статистику сообщений в вполне рабочий инструмент для сегментации аудитории. Не придется вручную вылавливать самых активных юзеров или часами считать сообщения по чатам — достаточно задать нужные сегменты и дать ему немного поработать. А дальше уже можно использовать собранную статистику для анализа аудитории, вовлекалок и других арбитражных задач. Подписывайтесь на нашу Телегу, чтобы первыми получать исходный код полезных Telegram-ботов для арбитража.