Peewee+MySQL：如何用自定义SQL子查询补全无数据周统计

核心思路 #

要实现每个用户的所有目标周都有记录（无日志时bin=0），关键是先生成包含所有目标周数的序列，再和用户列表做笛卡尔积，最后左连接日志统计结果。因为数据库只读，不能用临时表，所以用CTE或UNION ALL生成周数序列是最优解，完全通过单条SQL完成。

方案实现（基于Peewee 3.x+） #

假设你的日志模型定义如下：

from peewee import *

db = MySQLDatabase('your_db', user='user', password='pass', host='host')

class UserLog(Model):
    userid = IntegerField()
    log_time = DateTimeField()  # 日志时间字段

    class Meta:
        database = db
        table_name = 'user_log'

1. 生成自定义周数序列（CTE方式，推荐MySQL 8+）

用CTE生成0到n_weeks-1的周数，这里以8周为例（可自定义）：

from peewee import CTE, Select, Value, fn, SQL

n_weeks = 8  # 自定义周数，比如8周

# 生成周数CTE：周0到周7
weeks_cte = CTE(
    Select([Value(i).alias('week')]) for i in range(n_weeks)
).alias('weeks')

2. 获取所有唯一用户ID

# 提取日志中所有不重复的userid
users_subq = UserLog.select(UserLog.userid).distinct().alias('users')

3. 生成用户-周数笛卡尔积

确保每个用户都对应所有目标周：

# 笛卡尔积：每个userid匹配所有周数
user_weeks = (
    users_subq
    .cross_join(weeks_cte)
    .select(users_subq.c.userid, weeks_cte.c.week)
    .alias('user_weeks')
)

4. 统计用户每周日志数量

计算每条日志对应的周数（这里用TIMESTAMPDIFF计算日志时间与当前日期的周差，周0为最近一周，可根据需求调整）：

# 统计每个用户每周的日志数
log_counts = (
    UserLog
    .select(
        UserLog.userid,
        # 计算日志所属的周编号（周0=最近一周，周1=上一周...）
        fn.TIMESTAMPDIFF(WEEK, UserLog.log_time, fn.CURDATE()).alias('week'),
        fn.COUNT(UserLog.id).alias('bin')
    )
    .group_by(UserLog.userid, SQL('week'))
    .alias('log_counts')
)

5. 左连接并补全空周的0值

用COALESCE将无日志周的NULL转为0：

# 最终查询：左连接笛卡尔积和统计结果，补0
final_query = (
    user_weeks
    .left_join(
        log_counts,
        on=(
            (user_weeks.c.userid == log_counts.c.userid) &
            (user_weeks.c.week == log_counts.c.week)
        )
    )
    .select(
        user_weeks.c.userid,
        user_weeks.c.week,
        fn.COALESCE(log_counts.c.bin, 0).alias('bin')
    )
)

# 执行查询并获取结果
results = final_query.dicts().execute()
for row in results:
    print(row)  # 输出格式：{'userid': 123, 'week': 0, 'bin': 5}, 等

兼容旧版MySQL（无CTE支持） #

如果你的MySQL版本低于8.0，不支持CTE，可以用UNION ALL生成周数序列：

def generate_week_subquery(n_weeks):
    # 动态生成UNION ALL的周数子查询
    base_query = Select([Value(0).alias('week')])
    for i in range(1, n_weeks):
        base_query = base_query.union_all(Select([Value(i)]))
    return base_query.alias('weeks')

# 替换之前的weeks_cte
weeks_subq = generate_week_subquery(n_weeks=8)

后续步骤和CTE方案完全一致，只需把weeks_cte换成weeks_subq即可。

关键说明 #

• 周数计算逻辑：示例中用TIMESTAMPDIFF(WEEK, log_time, CURDATE())，若你需要以自然周（周一/周日为起始）划分，可调整为fn.WEEKOFYEAR(log_time)结合当前周数计算差值。
• 单条SQL执行：整个查询通过Peewee的链式调用生成，最终只会执行一条包含多子查询的SQL，符合只读权限要求。
• 替代RawQuery：用Peewee原生的Select/CTE生成的子查询具备.c属性，可直接用于关联，解决了RawQuery无法关联的问题。

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