在Telegram社群运营中,了解成员的活跃度是制定管理策略的核心基础。无论是清理潜水成员、激励优质用户,还是调整内容方向,都需要一份清晰的数据支撑。遗憾的是,Telegram原生功能并未提供活跃度统计工具,但通过自建机器人,我们可以精准记录每位成员的发言轨迹,并计算出活跃度评分。本文将手把手教你从零搭建一个统计机器人,实现从消息采集到活跃排行的完整流程。
统计成员活跃度的常见思路
在动手开发前,我们需要明确统计的逻辑。Telegram Bot API不允许直接读取历史消息,因此最常见的做法是:从机器人加入群聊的那一刻起,监听所有新消息,并将发送者、时间、群组等关键信息存入数据库。积累一段时间后,即可通过查询数据库计算每个成员在指定周期内的发言次数、活跃天数等指标。
如果某些群组已经拥有海量历史记录,也可以先用Telegram桌面版导出聊天记录,再通过脚本导入数据库,实现历史语料的回溯统计。但本文聚焦于实时采集方案,更适合长期运营。
准备工作:机器人与数据库
我们使用Python编写机器人,依赖以下组件:
- python-telegram-bot(官方推荐的Bot库)
- SQLite3(轻量级数据库,无需额外配置)
1. 创建机器人
在Telegram中搜索@BotFather,发送/newbot,按提示设置昵称和用户名,最终会得到一串HTTP API Token,请妥善保存。
2. 设置环境
pip install python-telegram-bot
创建一个Python文件,如stat_bot.py,导入所需模块并初始化数据库。
编写机器人统计代码
数据库设计
我们建立一张messages表,存储每个发送者的ID、用户名、群组ID、消息时间等字段。为了统计活跃天数,还需要将时间戳格式化为日期。
import sqlite3
conn = sqlite3.connect('activity.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS messages
(id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
username TEXT,
chat_id INTEGER,
date TEXT)''')
conn.commit()
消息监听与存储
利用MessageHandler捕获群组中的文本消息、图片、视频等所有有效内容,只要消息不是来自机器人本身即可记录。
from telegram.ext import Application, MessageHandler, filters
from datetime import datetime
async def record(update, context):
msg = update.message
if not msg or msg.from_user.is_bot:
return
if msg.chat.type in ['group', 'supergroup']:
ts = datetime.fromtimestamp(msg.date).strftime('%Y-%m-%d')
c.execute("INSERT INTO messages (user_id, username, chat_id, date) VALUES (?,?,?,?)",
(msg.from_user.id, msg.from_user.username, msg.chat.id, ts))
conn.commit()
app = Application.builder().token('YOUR_TOKEN').build()
app.add_handler(MessageHandler(filters.ALL, record))
活跃度统计命令
我们添加一个/active命令,接收天数参数(默认7天),返回活跃排名。定义活跃度为:在指定天数内的发言次数,同时可附加“活跃天数”作为辅助指标。
from telegram.ext import CommandHandler
def calculate_active(days):
since = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
since = since - timedelta(days=days)
since_str = since.strftime('%Y-%m-%d')
c.execute('''SELECT username, user_id, COUNT(*) as cnt, COUNT(DISTINCT date) as act_days
FROM messages
WHERE chat_id = ? AND date >= ?
GROUP BY user_id
ORDER BY cnt DESC
LIMIT 20''', (update.effective_chat.id, since_str))
rows = c.fetchall()
reply = f"近天活跃成员 TOP 20:\n"
for i, row in enumerate(rows, 1):
reply += f". @{row[0]} - 发言{row[2]}次, 活跃{row[3]}天\n"
return reply
async def active_cmd(update, context):
args = context.args
days = int(args[0]) if args and args[0].isdigit() else 7
reply = calculate_active(days)
await update.message.reply_text(reply)
完善启动逻辑
在主函数中注册active命令处理器,并启动应用。
app.add_handler(CommandHandler('active', active_cmd))
if __name__ == '__main__':
app.run_polling()
部署与运行
- 将机器人加入目标群组,并提升为管理员(保证能接收所有消息)。
- 运行Python脚本:
python stat_bot.py。 - 在群组中发送
/active 7即可查看最近7天的活跃成员排名。
注意:如果群组开启“限制保存内容”,机器人将无法获取媒体消息,但文本消息仍可记录。
进阶技巧:活跃度打分模型
简单的发言数排名容易刷屏刷量,你可以引入更科学的评分模型,例如:
- 将不同消息类型赋予权重:文字1分、图片2分、视频3分。
- 结合消息发送时段:工作日晚间活跃加权。
- 加入“潜水惩罚”:连续N天未发言扣分。
修改数据库结构和统计查询即可实现,灵活扩展。
总结
通过本文的教程,你已经掌握了用Telegram机器人统计群成员活跃度的核心方法。从消息监听、数据存储到活跃度计算,一套完整的闭环能够让你对群组状态一目了然。未来你还可以将统计结果接入外部可视化面板,或者与定时清理机器人联动,真正实现群组管理自动化。