Ithy Logo

修改接口以返回万方单位的用水量统计

Water Cycle and Fresh Water Supply | Sustainability: A Comprehensive ...

概述

为了满足用户需求,需要将接口返回的用水量单位从立方米转换为“万方”。这一转换涉及对用水量数据进行适当的除法操作,并确保在整个数据处理和返回过程中保持数据的精确性和一致性。以下内容将详细介绍如何在现有的接口代码中进行相应的修改。

修改步骤

1. 修改用水量的计算逻辑

在处理用水量数据时,需要将原始的water_use值除以10,000,以便将单位转换为万方。这一操作应在所有统计计算中进行,包括按月和按日的统计。

2. 确保数据类型支持精确计算

在进行除法操作后,结果可能包含小数部分。确保DecimalField能够处理这些小数值,以避免数据截断或精度损失。如果需要,可以应用适当的四舍五入方法。

3. 更新返回的数据结构

所有返回的total_water_use字段应反映更新后的单位(万方)。确保前端或调用方能够正确理解和使用这些数据。

详细代码修改示例

完整的修改代码


@custom_require_http_methods(['GET'])
@json_request_handler
@token_validation_handler
def api_device_daily_usage_statistics(request):
    """用量统计接口"""
    # 按时间查找
    start_time = request.json_data.get('start_time')  # 开始时间
    end_time = request.json_data.get('end_time')  # 结束时间
    device_daily_usage_query = DeviceDailyUsageModel.objects.exclude(water_use__gt=10000000)
    device_query = DeviceInfoModel.objects
    statistics_type = request.json_data.get('statistics_type') or '日'  # 统计类型 年/月/日 默认为日
    
    if not start_time or not end_time:
        return error_response(message='请选择开始时间和结束时间')
    
    # 根据位置查询
    location_l1_id = request.json_data.get('location_l1_id')  # 一级位置信息
    location_l2_id = request.json_data.get('location_l2_id')  # 二级位置信息
    location_l3_id = request.json_data.get('location_l3_id')  # 三级位置信息
    location = None
    if location_l3_id:
        location = Location.objects.filter(id=location_l3_id).first()
    elif location_l2_id:
        location = Location.objects.filter(id=location_l2_id).first()
    elif location_l1_id:
        location = Location.objects.filter(id=location_l1_id).first()
    
    if location_l3_id or location_l2_id or location_l1_id:
        if not location:
            return error_response(message='指定位置不存在')
    
    if location:
        if location_l3_id:
            device_query = device_query.filter(location_id=location_l3_id)
            device_daily_usage_query = device_daily_usage_query.filter(
                license_no__in=device_query.values_list('license_no', flat=True))
        elif location_l2_id or location_l1_id:
            # 非叶子节点获取该节点的所有叶子节点
            leaf_location_query = location.get_leafnodes()
            device_query = device_query.filter(location__in=leaf_location_query)
            device_daily_usage_query = device_daily_usage_query.filter(
                license_no__in=device_query.values_list('license_no', flat=True))
    
    # 返回结果 每日充值次数、充值金额、总充值次数、总充值金额
    result = {
        'start_time': start_time,  # 开始时间
        'end_time': end_time,  # 结束时间
        'location_name': location.name if location else None,  # 区域名称
    }
    
    if statistics_type == '年':
        pass  # 年统计部分根据需求进行进一步实现
    elif statistics_type == '月':
        # 校验日期 月
        if start_time:
            if not str_is_datetime(start_time, date_format='%Y-%m'):
                return error_response('开始日期格式不正确')
        if end_time:
            if not str_is_datetime(end_time, date_format='%Y-%m'):
                return error_response('结束日期格式不正确')
        
        # 将字符串转换为 datetime 对象
        start_date = datetime.datetime.strptime(start_time, "%Y-%m")
        end_date = datetime.datetime.strptime(end_time, "%Y-%m")
        
        # 获取该日期所在月份的最后一天
        last_day_of_month = calendar.monthrange(end_date.year, end_date.month)[1]
        end_date = datetime.datetime(end_date.year, end_date.month, last_day_of_month)
        
        # 生成日期范围
        date_range = list(['{}-{}'.format(i.year, i.month) for i in get_month_range(start_date, end_date)])
        
        # 查询并按月份汇总用水量和用电量
        monthly_usage = device_daily_usage_query.filter(
            statistics_date__range=[start_date.date(), end_date.date()]) \
            .values('statistics_date__year', 'statistics_date__month') \
            .annotate(
                total_water_use=Sum('water_use'),
                total_electricity_use=Sum('electricity_use')
            )
        
        # 为确保每月都有统计,使用日期范围进行合并
        monthly_statistics_dict = {}
        for date in date_range:
            monthly_statistics_dict[date] = {
                'statistics_date': date,
                'total_device_count': 0,
                'total_water_use': 0,
                'total_electricity_use': 0
            }
        
        # 统计总用水量、用电量,并转换单位为万方
        total_water_use = 0
        total_electricity_use = 0
        for usage in monthly_usage:
            key_str = '{}-{}'.format(usage['statistics_date__year'], usage['statistics_date__month'])
            monthly_statistics_dict[key_str]['total_water_use'] = usage['total_water_use'] / 10000  # 转换为万方
            monthly_statistics_dict[key_str]['total_electricity_use'] = usage['total_electricity_use']
            total_water_use += usage['total_water_use']
            total_electricity_use += usage['total_electricity_use']
        
        result['daily_statistics'] = [v for v in monthly_statistics_dict.values()]
        result['total_statistics'] = {
            'total_water_use': total_water_use / 10000,  # 转换为万方
            'total_electricity_use': total_electricity_use
        }
    
    else:
        # 校验日期 日
        if start_time:
            if not str_is_datetime(start_time, date_format='%Y-%m-%d'):
                return error_response('开始日期格式不正确')
        if end_time:
            if not str_is_datetime(end_time, date_format='%Y-%m-%d'):
                return error_response('结束日期格式不正确')
        
        # 过滤时间
        device_daily_usage_query = device_daily_usage_query.filter(
            statistics_date__gte=start_time).filter(
            statistics_date__lte=end_time)
        
        # 将字符串转换为 datetime 对象
        start_date = datetime.datetime.strptime(start_time, "%Y-%m-%d")
        end_date = datetime.datetime.strptime(end_time, "%Y-%m-%d")
        
        # 生成日期范围
        date_range = [(start_date + timedelta(days=x)).date() 
                      for x in range((end_date - start_date).days + 1)]
        
        # 统计用水量、用电量
        device_daily_usage_query = device_daily_usage_query.all()
        daily_statistics = device_daily_usage_query.values('statistics_date').annotate(
            total_device_count=Coalesce(Count('id'), 0, output_field=IntegerField()),
            total_water_use=Coalesce(Sum('water_use'), 0, output_field=DecimalField()),
            total_electricity_use=Coalesce(Sum('electricity_use'), 0, output_field=DecimalField())
        )
        
        # 为确保每日都有统计,使用日期范围进行合并
        daily_statistics_dict = {}
        for date in date_range:
            daily_statistics_dict[date.strftime('%Y-%m-%d')] = {
                'statistics_date': date.strftime('%Y-%m-%d'),
                'total_device_count': 0,
                'total_water_use': 0,
                'total_electricity_use': 0
            }
        
        for daily_statistic in daily_statistics:
            daily_statistics_dict[daily_statistic['statistics_date'].strftime('%Y-%m-%d')] = daily_statistic
        
        # 统计总用水量、用电量,并转换单位为万方
        total_statistics = device_daily_usage_query.aggregate(
            total_water_use=Coalesce(Sum('water_use'), 0, output_field=DecimalField()),
            total_electricity_use=Coalesce(Sum('electricity_use'), 0, output_field=DecimalField())
        )
        
        result['daily_statistics'] = [
            {
                'statistics_date': v.get('statistics_date'), 
                'total_device_count': v.get('total_device_count'),
                'total_water_use': v.get('total_water_use') / 10000,  # 转换为万方
                'total_electricity_use': v.get('total_electricity_use')
            }
            for v in daily_statistics_dict.values()
        ]
        result['total_statistics'] = {
            'total_water_use': total_statistics['total_water_use'] / 10000,  # 转换为万方
            'total_electricity_use': total_statistics['total_electricity_use']
        }
    
    return success_response(data=result)
  

代码修改说明

  • 单位转换:所有涉及total_water_use的计算处,均通过除以10,000将单位从立方米转换为万方。例如:
    total_water_use = usage['total_water_use'] / 10000  # 转换为万方
  • 数据类型处理:确保DecimalField能够处理转换后的数据,避免精度损失。如果必要,应用四舍五入:
    usage['total_water_use'] = round(usage['total_water_use'] / 10000, 2)
  • 错误处理:在进行除法操作时,虽然本例中不太可能出现除以零的情况,但根据业务逻辑,可能需要处理water_use为零或缺失的情况。
  • 代码一致性:确保所有相关部分(如月统计和日统计)都进行了相同的单位转换,以保持数据的一致性。

外部依赖和库

请确保所有必要的库已导入,并且辅助函数(如str_is_datetimeget_month_range)已正确定义和实现。例如:


from decimal import Decimal
import datetime
import calendar
from django.db.models import Sum, Count, DecimalField, IntegerField
from django.db.models.functions import Coalesce
from datetime import timedelta

# 确保辅助函数已定义
def str_is_datetime(string, date_format):
    try:
        datetime.datetime.strptime(string, date_format)
        return True
    except ValueError:
        return False

def get_month_range(start_date, end_date):
    # 生成月份范围的生成器
    current = start_date
    while current <= end_date:
        yield current
        # 增加一个月
        if current.month == 12:
            current = current.replace(year=current.year + 1, month=1)
        else:
            current = current.replace(month=current.month + 1)
  

测试和验证

1. 单元测试

编写单元测试以验证单位转换的准确性。例如,输入特定的water_use值,检查返回结果是否正确地反映了万方单位。

2. 集成测试

在整个应用程序中测试接口,确保在不同统计类型(年/月/日)和不同位置过滤条件下,接口均能正确返回转换后的用水量数据。

3. 边界条件测试

测试边界条件,如water_use为零、大量数据的处理、日期范围的极端情况(如闰年、跨月等)。确保接口在这些情况下依然稳定且准确。

性能优化建议

在处理大规模数据时,确保数据库查询效率。可以考虑以下优化措施:

  • 索引优化:确保statistics_datelocation_id等字段上有适当的索引,以加快查询速度。
  • 查询优化:避免不必要的查询,使用select_relatedprefetch_related减少数据库访问次数。
  • 缓存策略:对于频繁请求的数据,可以考虑使用缓存机制,减少数据库负载。

总结

通过上述步骤和代码修改,可以有效地将接口返回的用水量单位从立方米转换为万方。确保在整个修改过程中保持数据的准确性和一致性,并通过充分的测试验证修改结果,将有助于提升系统的可靠性和用户体验。


Last updated January 6, 2025
Ask me more