gb/bot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add 24-hour stats

Browse source
This commit is contained in:
草师傅 2025-08-08 10:46:20 +08:00
parent 6ce38d4602
commit 349299ee1b
2 changed files with 69 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

55
core/middleware/stats.py
View file

@ -2,6 +2,7 @@ from aiogram import BaseMiddleware
from aiogram.types import Message
from typing import Callable, Dict, Any, Awaitable
import json
from datetime import datetime, timedelta
class MessageStatsMiddleware(BaseMiddleware):
def __init__(self, stats_file: str = 'message_stats.json'):
@ -18,13 +19,19 @@ class MessageStatsMiddleware(BaseMiddleware):
if event.chat.type in ['group', 'supergroup']:
chat_id = str(event.chat.id)
user_id = str(event.from_user.id if event.from_user else 0)
current_time = datetime.now().isoformat()
# 初始化统计数据
if chat_id not in self.stats:
self.stats[chat_id] = {
'total_messages': 0,
'users': {},
'chat_title': event.chat.title
'chat_title': event.chat.title,
'messages_24h': {
'message_count': 0,
'active_users': {},
'messages': []
}
}
if user_id not in self.stats[chat_id]['users']:
@ -48,29 +55,75 @@ class MessageStatsMiddleware(BaseMiddleware):
# 更新统计
self.stats[chat_id]['total_messages'] += 1
self.stats[chat_id]['users'][user_id]['message_count'] += 1
self.stats[chat_id]['messages_24h']['message_count'] += 1
# 更新活跃用户统计
if user_id not in self.stats[chat_id]['messages_24h']['active_users']:
self.stats[chat_id]['messages_24h']['active_users'][user_id] = 0
self.stats[chat_id]['messages_24h']['active_users'][user_id] += 1
# 添加24小时消息记录
message_record = {
'user_id': user_id,
'timestamp': current_time,
'type': 'message'
}
# 羡慕、我菜统计
if event.text and any(keyword in event.text.lower() for keyword in ['xm','xmsl','羡慕','羡慕死了']):
if not self.stats[chat_id]['users'][user_id]['xm_count']:
self.stats[chat_id]['users'][user_id]['xm_count'] = 0
self.stats[chat_id]['users'][user_id]['xm_count'] += 1
message_record['special_type'] = 'xm'
if event.sticker and event.sticker.file_unique_id in ['AQADhhcAAs1rgFVy']:
if not self.stats[chat_id]['users'][user_id]['xm_count']:
self.stats[chat_id]['users'][user_id]['xm_count'] = 0
self.stats[chat_id]['users'][user_id]['xm_count'] += 1
message_record['special_type'] = 'xm'
if event.text and '我菜' in event.text:
if not self.stats[chat_id]['users'][user_id]['wocai_count']:
self.stats[chat_id]['users'][user_id]['xm_count'] = 0
self.stats[chat_id]['users'][user_id]['wocai_count'] += 1
message_record['special_type'] = 'wocai'
if event.sticker and event.sticker.file_unique_id in ['AQAD6AUAAgGeUVZy']:
if not self.stats[chat_id]['users'][user_id]['wocai_count']:
self.stats[chat_id]['users'][user_id]['wocai_count'] = 0
self.stats[chat_id]['users'][user_id]['wocai_count'] += 1
message_record['special_type'] = 'wocai'
# 添加消息记录到24小时列表
self.stats[chat_id]['messages_24h']['messages'].append(message_record)
# 清理超过24小时的记录
self.cleanup_old_messages(chat_id)
# 保存统计数据
self.save_stats()
return await handler(event, data)
def cleanup_old_messages(self, chat_id: str):
"""清理超过24小时的消息记录"""
if 'messages_24h' not in self.stats[chat_id]:
return
cutoff_time = datetime.now() - timedelta(hours=24)
self.stats[chat_id]['messages_24h']['messages'] = [
msg for msg in self.stats[chat_id]['messages_24h']['messages']
if datetime.fromisoformat(msg['timestamp']) > cutoff_time
]
# 更新消息计数和活跃用户列表
messages_24h = self.stats[chat_id]['messages_24h']['messages']
self.stats[chat_id]['messages_24h']['message_count'] = len(messages_24h)
# 重新计算活跃用户字典，统计每个用户的消息数量
active_users_dict = {}
for msg in messages_24h:
user_id = msg['user_id']
active_users_dict[user_id] = active_users_dict.get(user_id, 0) + 1
self.stats[chat_id]['messages_24h']['active_users'] = active_users_dict
def load_stats(self) -> dict:
try:
with open(self.stats_file, 'r', encoding='utf-8') as f:

16
core/stats.py
View file

@ -28,6 +28,11 @@ async def handle_stats_command(message: Message):
key=lambda x: x[1]['message_count'],
reverse=True
)
sorted_24h_users = sorted(
[(user_id, stats['users'][user_id]) for user_id in stats.get('messages_24h', {}).get('active_users', [])],
key=lambda x: x[1]['message_count'],
reverse=True
)
sorted_most_xm_users = sorted(
stats['users'].items(),
key=lambda x: x[1].get('xm_count',0),
@ -39,16 +44,25 @@ async def handle_stats_command(message: Message):
reverse=True
)
# 构建统计消息
text = f"📊 群组统计\n\n"
text += f"总消息数: {stats['total_messages']}\n"
text += f"活跃用户数: {len(stats['users'])}\n\n"
text += f"24小时内消息数: {stats['messages_24h']['message_count']}\n"
text += f"活跃用户数: {len(stats['users'])}\n"
text += f"24小时内活跃用户数:{len(stats['messages_24h']['active_users'])}\n\n"
text += "🏆 发言排行榜:\n"
text += "<blockquote expandable>"
for i, (user_id, user_data) in enumerate(sorted_users[:10], 1):
name = user_data['name'] or user_data['username'] or str(user_id)
text += f"{i}. {name}: {user_data['message_count']} 条\n"
text += "</blockquote>\n"
text += "📈 24小时内发言排行榜:\n"
text += "<blockquote expandable>"
for i, (user_id, user_data) in enumerate(sorted_24h_users[:10], 1):
name = user_data['name'] or user_data['username'] or str(user_id)
text += f"{i}. {name}: {stats['messages_24h']['active_users'][user_id]} 条\n"
text += "</blockquote>\n"
if sorted_most_xm_users and any(user_data['xm_count'] > 0 for _, user_data in sorted_most_xm_users):
text += "\n🍋 羡慕统计:\n"
text += "<blockquote expandable>"