Ithy Logo

将接口返回用水量单位转换为“万方”的详细指南

Metric system

概述

在现有的 api_device_water_usage_statistics 接口中,用水量数据以立方米(方)为单位返回。为了提高数据的可读性和符合业务需求,需要将用水量单位改为“万方”(即10,000立方米)。本文将提供一个全面的指南,详细说明如何在现有代码中进行必要的修改,以实现这一转换。

修改步骤

1. 理解现有代码结构

当前的 API 接口主要执行以下步骤:

  • 参数提取:从请求中提取参数,如 search_timestatistics_type 以及位置 ID。
  • 数据过滤:根据提供的位置 ID 过滤设备。
  • 数据检索:获取指定设备的日用水量或月用水量数据。
  • 数据处理:计算总用水量并准备响应数据。
  • 响应返回:以 JSON 格式返回处理后的数据。

2. 确定转换位置

为了将用水量转换为“万方”,需要在以下几个关键位置进行修改:

  • 日用水量统计:在计算每日用水量时进行单位转换。
  • 月用水量统计:在计算每月用水量时进行单位转换。
  • 总用水量统计:在汇总总用水量时进行单位转换。

3. 修改数据处理逻辑

3.1 转换因子的应用

1 万方等于 10,000 立方米。因此,需要在计算用水量时,将立方米数除以 10,000。

3.2 日用水量统计的修改

在处理日统计数据的循环中,进行单位转换如下:


for current_day in (first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1)):
    water_use = daily_usage.get(current_day, decimal.Decimal('0.000'))  # 无数据时填充 0
    daily_water_usage[current_day] = water_use / decimal.Decimal('10000')  # 转换为万方
    total_water_use += water_use / decimal.Decimal('10000')  # 转换总用水量为万方
    

同时,更新响应数据:


device_statistics_data = {
    "water_usage": [water_use for water_use in daily_water_usage.values()],
    "total": total_water_use,
}
    

3.3 月用水量统计的修改

在处理月统计数据的循环中,进行单位转换如下:


for current_month in (datetime.date(search_date.year, month, 1) for month in range(1, 13)):
    water_use = monthly_usage.get(current_month, decimal.Decimal('0.000'))  # 无数据时填充 0
    monthly_water_usage[current_month] = water_use / decimal.Decimal('10000')  # 转换为万方
    total_water_use += water_use / decimal.Decimal('10000')  # 转换总用水量为万方
    

同时,更新响应数据:


device_statistics_data = {
    "water_usage": [water_use for water_use in monthly_water_usage.values()],
    "total": total_water_use,
}
    

4. 更新返回数据结构

确保在响应中,所有涉及用水量的字段都已进行单位转换。具体来说:

  • water_usage 列表中的所有值均已转换为万方。
  • total 字段的值已转换为万方。

例如:


device_statistics_data = {
    "water_usage": [water_use for water_use in daily_water_usage.values()],  # 已经是万方
    "total": total_water_use,  # 已经是万方
}
    

5. 完整的修改代码示例

以下是完整修改后的代码示例,包括日统计和月统计部分的用水量单位转换:

5.1 日统计部分


@custom_require_http_methods(['GET'])
@json_request_handler
@token_validation_handler
def api_device_water_usage_statistics(request):
    """用水量统计接口"""
    # 按时间查找
    search_time = request.json_data.get('search_time')  # 查询时间 2024 或 2024-3
    # 字段校验
    if not search_time:
        return error_response('请指定查询时间')

    device_query = DeviceInfoModel.objects.all()
    statistics_type = request.json_data.get('statistics_type') or '日'  # 统计类型 月/日 默认为日
    if not statistics_type:
        return error_response('请指定查询类型')

    # 根据位置查询
    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)
        elif location_l2_id or location_l1_id:
            leaf_location_query = location.get_leafnodes()
            device_query = device_query.filter(location__in=leaf_location_query)

    if device_query.count() == 0:
        return error_response('未查询到设备数据')

    # 返回结果
    if statistics_type == '日':
        try:
            # 将字符串转换为 datetime 对象
            search_date = datetime.datetime.strptime(search_time, "%Y-%m")
        except Exception as e:
            return error_response('日期格式错误:%Y-%m {}'.format(e))
        # 确定月份的第一天和最后一天
        first_day = datetime.date(search_date.year, search_date.month, 1)
        last_day = (datetime.date(search_date.year, search_date.month + 1, 1) - timedelta(days=1)
                    if search_date.month < 12
                    else datetime.date(search_date.year + 1, 1, 1) - timedelta(days=1))

        # 查询设备数据
        device_info_dict = {}
        sql_query = """
            SELECT 
                l1.`name` AS location_1_name, 
                l2.`name` AS location_2_name, 
                l3.`name` AS location_3_name, 
                owner_name, 
                well_number, 
                license_no, 
                water_permit_e_certificate_code, 
                device_id 
            FROM device_info_table 
            INNER JOIN location_table AS l3 ON device_info_table.location_id = l3.id 
            INNER JOIN location_table AS l2 ON l3.parent_id=l2.id 
            INNER JOIN location_table AS l1 ON l2.parent_id = l1.id 
            WHERE device_info_table.license_no IN ({})
        """.format(','.join(["'{}'".format(device.license_no) for device in device_query]))
        print(sql_query)
        sql_results = execute_raw_sql_dict(sql_query)
        for row in sql_results:
            device_info_dict[row.get('license_no')] = row

        # 查询所有相关数据
        all_daily_data = (
            DeviceDailyUsageModel.objects.filter(statistics_date__range=(first_day, last_day)).exclude(water_use__gt=100000000)
            .values("license_no", "statistics_date")
            .annotate(daily_water_use=Sum("water_use"))
        )

        # 按 license_no 分组
        data_by_license_no = defaultdict(dict)
        for entry in all_daily_data:
            license_no = entry["license_no"]
            statistics_date = entry["statistics_date"]
            daily_water_use = entry["daily_water_use"]
            data_by_license_no[license_no][statistics_date] = daily_water_use

        # 设备数据初始化
        specific = []

        # 遍历所有设备并生成统计数据
        for device in device_query:
            daily_usage = data_by_license_no.get(device.license_no, {})
            daily_water_usage = {}
            total_water_use = 0

            # 填补日期和计算总和
            for current_day in (first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1)):
                water_use = daily_usage.get(current_day, decimal.Decimal('0.000'))  # 无数据时填充 0
                daily_water_usage[current_day] = water_use / decimal.Decimal('10000')  # 转换为万方
                total_water_use += water_use / decimal.Decimal('10000')  # 转换总用水量为万方

            device_statistics_data = {
                "water_usage": [water_use for water_use in daily_water_usage.values()],
                "total": total_water_use,
            }
            try:
                device_statistics_data.update(device_info_dict.get(device.license_no))  # 合并设备其他字段
            except Exception as e:
                print(e)
                print(device_statistics_data)
            specific.append(device_statistics_data)

        return success_response(data=specific)

    else:
        # 校验日期 月
        # 假设传入的 search_time 为年份字符串,如 "2024"
        search_date = datetime.datetime.strptime(search_time, "%Y")

        # 确定年份的第一天和最后一天
        first_day = datetime.date(search_date.year, 1, 1)
        last_day = datetime.date(search_date.year, 12, 31)

        # 查询设备数据
        device_info_dict = {}
        sql_query = """
            SELECT 
                l1.`name` AS location_1_name, 
                l2.`name` AS location_2_name, 
                l3.`name` AS location_3_name, 
                owner_name, 
                well_number, 
                license_no, 
                water_permit_e_certificate_code, 
                device_id 
            FROM device_info_table 
            INNER JOIN location_table AS l3 ON device_info_table.location_id = l3.id 
            INNER JOIN location_table AS l2 ON l3.parent_id=l2.id 
            INNER JOIN location_table AS l1 ON l2.parent_id = l1.id 
            WHERE device_info_table.license_no IN ({})
        """.format(','.join(["'{}'".format(device.license_no) for device in device_query]))
        print(sql_query)
        sql_results = execute_raw_sql_dict(sql_query)
        for row in sql_results:
            device_info_dict[row.get('license_no')] = row

        # 查询所有相关数据
        all_monthly_data = (
            DeviceDailyUsageModel.objects.filter(statistics_date__range=(first_day, last_day)).exclude(water_use__gt=100000000)
            .annotate(month=TruncMonth("statistics_date"))  # 按月份截断日期
            .values("license_no", "month")
            .annotate(monthly_water_use=Sum("water_use"))
        )

        # 按 license_no 分组
        data_by_license_no = defaultdict(dict)
        for entry in all_monthly_data:
            license_no = entry["license_no"]
            month = entry["month"]
            monthly_water_use = entry["monthly_water_use"]
            data_by_license_no[license_no][month] = monthly_water_use

        # 初始化结果
        specific = []

        # 遍历所有设备并生成统计数据
        for device in device_query:
            monthly_usage = data_by_license_no.get(device.license_no, {})
            monthly_water_usage = {}
            total_water_use = 0

            # 按月填补数据
            for current_month in (datetime.date(search_date.year, month, 1) for month in range(1, 13)):
                water_use = monthly_usage.get(current_month, decimal.Decimal('0.000'))  # 无数据时填充 0
                monthly_water_usage[current_month] = water_use / decimal.Decimal('10000')  # 转换为万方
                total_water_use += water_use / decimal.Decimal('10000')  # 转换总用水量为万方

            device_statistics_data = {
                "water_usage": [water_use for water_use in monthly_water_usage.values()],
                "total": total_water_use,
            }
            try:
                device_statistics_data.update(device_info_dict.get(device.license_no))  # 合并设备其他字段
            except Exception as e:
                print(e)
                print(device_statistics_data)
            specific.append(device_statistics_data)

        return success_response(data=specific)
    

6. 额外考虑事项

  • 错误处理:确保在转换过程中,任何潜在的错误都能被适当处理,以避免 API 中断。例如,处理除以零错误或数据类型不匹配的问题。
  • 文档更新:更新 API 文档,明确说明返回的 water_usagetotal 字段现在以“万方”为单位。
  • 测试:全面测试修改后的 API,确保数据转换准确且响应格式正确。包括边界情况测试,如无数据或异常日期格式。
  • 代码优化
    • 考虑将单位转换逻辑封装为独立的函数,以提高代码复用性和可维护性。
    • 将转换因子(如 10,000)配置化,便于未来调整。

7. 测试与验证

在完成代码修改后,进行以下测试以确保功能的正确性:

  • 功能测试
    • 输入不同的 statistics_type(如“日”或“月”)。
    • 输入不同的 search_time(如“2024-03”或“2024”)。
    • 包含多种设备和位置的数据输入。
  • 验证点
    • water_usagetotal 字段是否已正确转换为万方。
    • 设备信息和位置信息是否正确无误。
    • API 对无数据或错误输入的处理是否符合预期。

8. 优化与扩展

  • 代码复用:将单位转换逻辑封装到一个函数中,例如:
  • 
    def convert_to_wanfang(water_use):
        return water_use / decimal.Decimal('10000')
            
  • 配置管理:将转换因子 10,000 放入配置文件,以便未来根据需要进行调整。
  • 接口文档更新:确保前端开发人员和API用户了解单位的变更,避免数据误解。

总结

通过上述步骤,我们成功地将 api_device_water_usage_statistics 接口中返回的用水量单位从“方”转换为“万方”。这种转换不仅提升了数据的可读性,也更符合实际业务需求。在实际开发过程中,类似的单位转换需求应遵循以下原则:

  • 明确需求:清晰理解单位转换的具体要求和范围。
  • 全面修改:确保所有相关逻辑和数据字段都进行了相应的修改。
  • 充分测试:验证修改的正确性和完整性,确保系统稳定运行。
  • 更新文档:同步更新相关文档,确保团队成员和API用户了解最新的接口规范。

通过遵循这些最佳实践,可以确保 API 的可靠性和可维护性,为用户提供准确和有用的数据。


Last updated January 6, 2025
Ask me more