基于Peewee实现用户活动统计：事件计数与高频活动提取

我来帮你一步步搞定这个需求，先从模型定义开始，再分别解决活动次数统计和最频繁活动的问题，最后整合到汇总表中。

首先，先把对应数据库表的Peewee模型写好（替换成你实际用的数据库连接，比如PostgreSQL、MySQL都行）：

from peewee import *

# 初始化数据库连接，这里用SQLite举例，你可以换成其他数据库类型
db = SqliteDatabase('your_database.db')

class Users(Model):
    userid = IntegerField(primary_key=True)
    name = CharField()
    department = CharField()

    class Meta:
        database = db
        table_name = 'Users'  # 确保和你的数据库表名一致

class Activity(Model):
    userid = ForeignKeyField(Users, backref='activities', field='userid')  # 关联Users表的userid字段
    activity = CharField()

    class Meta:
        database = db
        table_name = 'Activity'

1. 统计每个用户的活动次数（activityCount） #

这一步类似pandas的group_by(userid).count()，用Peewee的聚合函数fn.Count就能实现，再关联Users表拿到用户的基本信息：

# 先按userid分组，统计每个用户的活动总数
activity_counts = (Activity
                   .select(Activity.userid, fn.Count(Activity.id).alias('activityCount'))
                   .group_by(Activity.userid))

# 把统计结果和Users表关联，得到用户信息+活动次数的组合
user_activity_data = (Users
                      .select(Users.userid, Users.name, Users.department, activity_counts.c.activityCount)
                      .join(activity_counts, on=(Users.userid == activity_counts.c.userid)))

如果要把结果存入新的Summary表，先定义这个表的模型：

class Summary(Model):
    userid = IntegerField(primary_key=True)
    name = CharField()
    department = CharField()
    activityCount = IntegerField()
    topActivity = CharField(null=True)

    class Meta:
        database = db
        table_name = 'Summary'

# 先创建表（如果还不存在的话）
db.create_tables([Summary], safe=True)

2. 获取每个用户的最频繁活动（topActivity） #

这部分要找每个用户活动的"众数"，推荐用窗口函数（只要你的数据库支持，比如SQLite 3.25+、PostgreSQL、MySQL 8.0+，效率很高）：

# 第一步：统计每个用户每个活动的出现次数
activity_frequency = (Activity
                      .select(Activity.userid, Activity.activity,
                              fn.Count(Activity.id).alias('freq'))
                      .group_by(Activity.userid, Activity.activity))

# 第二步：用窗口函数给每个用户的活动按次数排序（次数相同的话按活动名称排序，避免随机）
ranked_activities = (activity_frequency
                     .select(activity_frequency.c.userid, activity_frequency.c.activity,
                             fn.RowNumber().over(
                                 partition_by=activity_frequency.c.userid,  # 按用户分组
                                 order_by=[fn.Desc(activity_frequency.c.freq), activity_frequency.c.activity]  # 先按次数降序，再按活动名升序
                             ).alias('rank')))

# 第三步：筛选出每个用户排名第1的活动，就是次数最多的那个
top_activities = (ranked_activities
                  .select(ranked_activities.c.userid, ranked_activities.c.activity.alias('topActivity'))
                  .where(ranked_activities.c.rank == 1))

如果你的数据库不支持窗口函数，也可以用子查询的兼容方案（虽然效率稍低，但能跑）：

# 先找出每个用户的最大活动次数
max_activity_freq = (Activity
                     .select(Activity.userid, fn.Max(fn.Count(Activity.id)).alias('max_freq'))
                     .group_by(Activity.userid, Activity.activity)
                     .group_by(Activity.userid))

# 再匹配到对应次数的活动
top_activities = (Activity
                  .select(Activity.userid, Activity.activity.alias('topActivity'))
                  .join(max_activity_freq, on=(Activity.userid == max_activity_freq.c.userid))
                  .group_by(Activity.userid, Activity.activity)
                  .having(fn.Count(Activity.id) == max_activity_freq.c.max_freq))

3. 整合所有数据到Summary表 #

现在把用户基本信息、活动次数、最频繁活动三者关联起来，批量插入到Summary表中：

# 先清空Summary表（如果需要全量更新的话，不需要可以跳过这行）
Summary.delete().execute()

# 关联三个数据源，得到完整的汇总数据
full_summary_data = (Users
                     .select(Users.userid, Users.name, Users.department,
                             activity_counts.c.activityCount, top_activities.c.topActivity)
                     .join(activity_counts, on=(Users.userid == activity_counts.c.userid))
                     .join(top_activities, on=(Users.userid == top_activities.c.userid)))

# 批量插入到Summary表，用atomic保证事务
with db.atomic():
    for row in full_summary_data:
        # 可选：把活动名称首字母大写，和你示例中的格式一致
        formatted_activity = row.topActivity.capitalize() if row.topActivity else None
        Summary.create(
            userid=row.userid,
            name=row.name,
            department=row.department,
            activityCount=row.activityCount,
            topActivity=formatted_activity
        )

验证结果 #

你可以查询Summary表看看是不是符合预期：

for entry in Summary.select():
    print(f"{entry.userid}\t{entry.name}\t{entry.department}\t{entry.activityCount}\t{entry.topActivity}")

输出应该和你想要的一模一样：

123	Sam	Management	5	Browse
124	Joe	Employee	1	Signup

内容的提问来源于stack exchange，提问作者Llewellyn Hattingh