Django：如何计算店铺近1时/日/周基于营业时间与状态的运行、停机时长？

模型定义 #

ShopHour（营业时间模型） #

class ShopHour(models.Model):
    DAY_CHOICES = [
        (0, 'Monday'),
        (1, 'Tuesday'),
        (2, 'Wednesday'),
        (3, 'Thursday'),
        (4, 'Friday'),
        (5, 'Saturday'),
        (6, 'Sunday'),
    ]

    store = models.ForeignKey(
        Shop,
        on_delete=models.CASCADE,
        related_name='business_hours'
    )
    day_of_week = models.IntegerField(choices=DAY_CHOICES)
    start_time_local = models.TimeField()
    end_time_local = models.TimeField()

Status（状态记录模型） #

class Status(models.Model):
    STATUS_CHOICES = [
        ('active', 'Active'),
        ('inactive', 'Inactive'),
    ]

    shop = models.ForeignKey(
        Shop,
        on_delete=models.CASCADE,
        related_name='updates'
    )
    timestamp_utc = models.DateTimeField()
    status = models.CharField(max_length=10, choices=STATUS_CHOICES)

核心需求 #

• 统计店铺近1小时、1天、1周内，处于active（运行）和inactive（停机）状态的时长
• 计算需基于店铺营业时间：无配置时默认24*7营业；报表需转换为店铺所在时区展示

现有代码的关键问题 #

你提供的create_report函数存在以下逻辑错误：

1. 状态时长计算重复：直接累加每条状态记录到当前时间的时长，会重复计算重叠区间（比如连续两条active记录，会把中间时间段算两次）
2. 营业时间处理失效：仅在无配置时用默认值，但有配置时完全没用到ShopHour的星期几和对应时间，逻辑混乱
3. 时区处理错误：timestamp_utc是UTC时间，直接和店铺时区的last_hour比较，会导致时间区间偏移
4. 未区分营业/非营业时间：需求是基于营业时间统计，现有代码未排除非营业时间的状态
5. 代码冗余：三个时间段的计算逻辑完全重复，可维护性差

优化后的实现方案 #

1. 工具函数：获取时间区间内的有效营业时间段 #

from django.utils import timezone
from datetime import datetime, timedelta

def get_business_intervals(shop, start_tz, end_tz):
    """
    获取店铺在[start_tz, end_tz]区间内的有效营业时间段（转换为UTC）
    start_tz/end_tz：店铺时区的datetime对象
    """
    store_tz = timezone(shop.timezone_str)
    business_hours = shop.business_hours.all()
    intervals = []

    if not business_hours:
        # 默认24*7营业，直接返回整个区间的UTC转换
        return [(start_tz.astimezone(timezone.utc), end_tz.astimezone(timezone.utc))]

    # 遍历时间区间内的每一天
    current_date = start_tz.date()
    end_date = end_tz.date()

    while current_date <= end_date:
        day_of_week = current_date.weekday()  # 0=周一，对应ShopHour的DAY_CHOICES
        shop_hour = business_hours.filter(day_of_week=day_of_week).first()

        if shop_hour:
            # 构造当天的营业起止时间（店铺时区）
            start_local = datetime.combine(current_date, shop_hour.start_time_local)
            end_local = datetime.combine(current_date, shop_hour.end_time_local)
            start_local = store_tz.localize(start_local)
            end_local = store_tz.localize(end_local)

            # 与统计区间取交集
            interval_start = max(start_local, start_tz)
            interval_end = min(end_local, end_tz)

            if interval_start < interval_end:
                # 转换为UTC时间
                intervals.append((interval_start.astimezone(timezone.utc), interval_end.astimezone(timezone.utc)))

        current_date += timedelta(days=1)

    return intervals

2. 工具函数：计算状态在营业区间内的时长 #

def calculate_status_durations(shop, start_utc, end_utc):
    """
    计算店铺在[start_utc, end_utc]区间内的active/inactive时长（秒）
    """
    # 获取排序后的状态记录（UTC时间）
    statuses = shop.updates.filter(
        timestamp_utc__gte=start_utc,
        timestamp_utc__lte=end_utc
    ).order_by('timestamp_utc')

    # 初始化状态区间：默认初始状态为inactive（可根据业务调整）
    status_intervals = []
    prev_time = start_utc
    prev_status = 'inactive'

    for status in statuses:
        if prev_time < status.timestamp_utc:
            status_intervals.append((prev_time, status.timestamp_utc, prev_status))
        prev_time = status.timestamp_utc
        prev_status = status.status

    # 添加最后一个状态到结束时间的区间
    if prev_time < end_utc:
        status_intervals.append((prev_time, end_utc, prev_status))

    # 统计各状态在营业区间内的时长
    active_total = 0
    inactive_total = 0

    business_intervals = get_business_intervals(shop, start_utc.astimezone(timezone(shop.timezone_str)), end_utc.astimezone(timezone(shop.timezone_str)))

    for biz_start, biz_end in business_intervals:
        for stat_start, stat_end, stat in status_intervals:
            # 计算状态区间与营业区间的交集
            overlap_start = max(stat_start, biz_start)
            overlap_end = min(stat_end, biz_end)

            if overlap_start < overlap_end:
                duration = (overlap_end - overlap_start).total_seconds()
                if stat == 'active':
                    active_total += duration
                else:
                    inactive_total += duration

    return active_total, inactive_total

3. 主函数实现 #

def create_report(shop_id):
    shop = Shop.objects.get(id=shop_id)
    store_tz = timezone(shop.timezone_str)
    now_tz = datetime.now(tz=store_tz)

    # 定义三个统计区间（店铺时区）
    intervals = [
        ('hour', now_tz - timedelta(hours=1), now_tz, 60),  # 近1小时，转换为分钟
        ('day', now_tz - timedelta(days=1), now_tz, 3600),  # 近1天，转换为小时
        ('week', now_tz - timedelta(days=7), now_tz, 3600), # 近1周，转换为小时
    ]

    report_data = {}

    for name, start_tz, end_tz, divisor in intervals:
        # 转换为UTC时间用于数据库查询
        start_utc = start_tz.astimezone(timezone.utc)
        end_utc = end_tz.astimezone(timezone.utc)

        active_sec, inactive_sec = calculate_status_durations(shop, start_utc, end_utc)

        # 转换为对应单位（分钟/小时）
        report_data[f'uptime_last_{name}'] = round(active_sec / divisor, 2)
        report_data[f'downtime_last_{name}'] = round(inactive_sec / divisor, 2)

    # 创建报表并返回
    report = Report.objects.create(shop=shop, **report_data)
    return report

关键优化点说明 #

• 时区统一处理：所有时间计算先转换为店铺时区处理营业时间，再转UTC匹配状态记录，避免时区偏移
• 状态区间化：将离散的状态记录转换为连续的时间区间，避免重复计算
• 营业区间交集计算：仅统计营业时间内的状态时长，符合需求逻辑
• 代码复用：通过工具函数封装核心逻辑，减少冗余，提升可维护性
• 精度控制：保留两位小数，避免整数除法导致的精度丢失

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