Django(PostgreSQL)如何通过ORM按日期拆分DurationField覆盖的时长

该需求完全可实现，依托PostgreSQL原生的时间序列生成、时区转换能力，配合Django ORM即可完成，无需引入第三方依赖，性能可满足最长1年时间跨度的查询要求。

1. 基础模型定义 #

首先确保活动模型按Django最佳实践存储UTC格式的时间，示例模型如下：

from django.db import models
from datetime import timedelta

class Event(models.Model):
    user = models.ForeignKey("auth.User", on_delete=models.CASCADE, related_name="events")
    # 统一存储UTC时区的活动开始时间
    start_datetime = models.DateTimeField()
    # 活动时长
    duration = models.DurationField()

    @property
    def end_datetime(self):
        return self.start_datetime + self.duration

2. 封装PostgreSQL原生函数 #

Django ORM支持直接封装数据库内置函数，我们需要用到时区转换、日期序列生成两个核心能力，封装如下：

from django.db.models import Func, DateTimeField, DateField, Value, F
from django.db.models.functions import Cast

class ATTimeZone(Func):
    """将UTC时间转换为指定时区的本地时间"""
    function = "AT TIME ZONE"
    arity = 2

    def as_sql(self, compiler, connection, **extra_context):
        timezone_sql, timezone_params = compiler.compile(self.source_expressions[1])
        datetime_sql, datetime_params = compiler.compile(self.source_expressions[0])
        return f"{datetime_sql} AT TIME ZONE {timezone_sql}", datetime_params + timezone_params

class GenerateDateSeries(Func):
    """生成指定范围内的连续日期序列"""
    function = "GENERATE_SERIES"
    output_field = DateField()
    arity = 3

3. 分日时长统计查询实现 #

核心逻辑：

• 先过滤指定用户、在查询时间范围内的活动
• 为每个活动生成其覆盖的所有自然日序列
• 逐天计算活动与当日自然日的时间重叠时长
• 自动处理指定时区的偏移、夏令时问题

from django.db.models import DurationField
from django.db.models.expressions import RawSQL
from datetime import datetime, timedelta

def get_user_event_daily_duration(user_id: int, timezone: str, start_date: datetime, end_date: datetime):
    """
    获取指定用户的活动分日时长统计
    :param user_id: 用户ID
    :param timezone: 统计使用的时区，格式如'UTC'、'Asia/Shanghai'
    :param start_date: 统计范围开始日期（统计时区下的自然日，含）
    :param end_date: 统计范围结束日期（统计时区下的自然日，含）
    """
    # 强制校验时间跨度不超过1年，避免性能问题
    if (end_date - start_date).days > 366:
        raise ValueError("统计时间跨度不能超过1年")

    # 过滤符合范围的活动
    events = Event.objects.filter(
        user_id=user_id,
        start_datetime__lt=end_date + timedelta(days=1),
        end_datetime__gte=start_date
    ).annotate(
        event_end=F("start_datetime") + F("duration")
    )

    # 关联生成连续日期，计算每日占用时长
    daily_stats = events.annotate(
        event_local_start=ATTimeZone(F("start_datetime"), Value(timezone)),
        event_local_end=ATTimeZone(F("event_end"), Value(timezone)),
        series_date=GenerateDateSeries(
            RawSQL("DATE(%s)", ["event_local_start"]),
            RawSQL("DATE(%s)", ["event_local_end"]),
            Value(timedelta(days=1))
        )
    ).annotate(
        # 计算当日0点、次日0点对应的UTC时间
        day_start_utc=ATTimeZone(Cast(F("series_date"), DateTimeField()), Value(timezone)),
        day_end_utc=ATTimeZone(Cast(F("series_date") + timedelta(days=1), DateTimeField()), Value(timezone))
    ).annotate(
        # 计算活动与当日的时间交集起止点
        overlap_start=RawSQL("GREATEST(%s, %s)", ["start_datetime", "day_start_utc"]),
        overlap_end=RawSQL("LEAST(%s, %s)", ["event_end", "day_end_utc"])
    ).annotate(
        # 计算当日占用时长
        daily_duration=F("overlap_end") - F("overlap_start")
    ).filter(
        # 过滤掉时长为0的无效记录
        daily_duration__gt=timedelta(0)
    ).values(
        "id", "series_date", "daily_duration"
    ).order_by("id", "series_date")

    return daily_stats

针对你给出的示例：活动开始时间为2022-12-24T12:00:00 UTC，时长timedelta(hours=48)，传入参数timezone='UTC'，查询返回结果为：

• 对应活动id、2022-12-24、timedelta(hours=12)
• 对应活动id、2022-12-25、timedelta(hours=24)
• 对应活动id、2022-12-26、timedelta(hours=12)
  和预期结果完全一致。

• 数据库中存储的start_datetime必须为UTC时区的时间，不要存储本地时间，否则会出现时区转换错误
• 夏令时、冬令时切换的日期会被PostgreSQL自动处理，无需手动计算偏移
• 1年跨度的硬限制不可随意放开，避免生成过多日期序列导致查询变慢
• 查询返回的daily_duration为Python原生timedelta类型，可直接换算为小时、分钟等单位使用

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