#!/usr/bin/env python3
|
"""
|
群组统计功能测试脚本
|
"""
|
|
import sys
|
import os
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
from app.services.group_stats_service import group_stats_service
|
from config import settings
|
|
|
def test_group_stats():
|
"""测试群组统计功能"""
|
print("=== 群组统计功能测试 ===\n")
|
|
# 测试群组ID
|
test_group_id = "test_group_123"
|
test_users = ["user1", "user2", "user3"]
|
|
print(f"1. 测试群组: {test_group_id}")
|
print(f"2. 测试用户: {test_users}")
|
print(f"3. 默认客服名称: {settings.customer_service_default_name}")
|
print()
|
|
# 清空测试数据
|
print("4. 清空测试数据...")
|
group_stats_service.clear_group_stats(test_group_id)
|
print(" 清空完成")
|
print()
|
|
# 测试初始状态
|
print("5. 测试初始状态...")
|
initial_stats = group_stats_service.get_group_message_stats(test_group_id)
|
initial_nickname = group_stats_service.get_most_active_user_nickname(test_group_id)
|
print(f" 初始统计: {initial_stats}")
|
print(f" 初始最活跃用户昵称: {initial_nickname}")
|
print()
|
|
# 模拟用户发言
|
print("6. 模拟用户发言...")
|
# user1 发言 5 次
|
for i in range(5):
|
group_stats_service.increment_user_message_count(test_group_id, test_users[0])
|
print(f" {test_users[0]} 发言第 {i+1} 次")
|
|
# user2 发言 3 次
|
for i in range(3):
|
group_stats_service.increment_user_message_count(test_group_id, test_users[1])
|
print(f" {test_users[1]} 发言第 {i+1} 次")
|
|
# user3 发言 7 次
|
for i in range(7):
|
group_stats_service.increment_user_message_count(test_group_id, test_users[2])
|
print(f" {test_users[2]} 发言第 {i+1} 次")
|
print()
|
|
# 查看统计结果
|
print("7. 查看统计结果...")
|
final_stats = group_stats_service.get_group_message_stats(test_group_id)
|
final_nickname = group_stats_service.get_most_active_user_nickname(test_group_id)
|
print(f" 最终统计: {final_stats}")
|
print(f" 最活跃用户昵称: {final_nickname}")
|
print()
|
|
# 获取统计摘要
|
print("8. 获取统计摘要...")
|
summary = group_stats_service.get_group_stats_summary(test_group_id)
|
print(f" 统计摘要: {summary}")
|
print()
|
|
# 测试相同发言次数的情况
|
print("9. 测试相同发言次数的情况...")
|
test_group_id_2 = "test_group_equal"
|
group_stats_service.clear_group_stats(test_group_id_2)
|
|
# 两个用户都发言 3 次
|
for i in range(3):
|
group_stats_service.increment_user_message_count(test_group_id_2, "equal_user1")
|
group_stats_service.increment_user_message_count(test_group_id_2, "equal_user2")
|
|
equal_stats = group_stats_service.get_group_message_stats(test_group_id_2)
|
equal_nickname = group_stats_service.get_most_active_user_nickname(test_group_id_2)
|
print(f" 相同发言次数统计: {equal_stats}")
|
print(f" 相同发言次数最活跃用户昵称: {equal_nickname}")
|
print()
|
|
# 测试无发言的情况
|
print("10. 测试无发言的情况...")
|
test_group_id_3 = "test_group_empty"
|
group_stats_service.clear_group_stats(test_group_id_3)
|
|
empty_stats = group_stats_service.get_group_message_stats(test_group_id_3)
|
empty_nickname = group_stats_service.get_most_active_user_nickname(test_group_id_3)
|
print(f" 无发言统计: {empty_stats}")
|
print(f" 无发言最活跃用户昵称: {empty_nickname}")
|
print()
|
|
print("=== 测试完成 ===")
|
|
|
def test_config():
|
"""测试配置功能"""
|
print("=== 配置测试 ===\n")
|
|
print(f"客服名称列表: {settings.customer_service_names}")
|
print(f"默认客服名称: {settings.customer_service_default_name}")
|
print()
|
|
|
if __name__ == "__main__":
|
try:
|
test_config()
|
test_group_stats()
|
except Exception as e:
|
print(f"测试失败: {str(e)}")
|
import traceback
|
traceback.print_exc()
|
