为了满足用户需求,需要将接口返回的用水量单位从立方米转换为“万方”。这一转换涉及对用水量数据进行适当的除法操作,并确保在整个数据处理和返回过程中保持数据的精确性和一致性。以下内容将详细介绍如何在现有的接口代码中进行相应的修改。
在处理用水量数据时,需要将原始的water_use
值除以10,000,以便将单位转换为万方。这一操作应在所有统计计算中进行,包括按月和按日的统计。
在进行除法操作后,结果可能包含小数部分。确保DecimalField
能够处理这些小数值,以避免数据截断或精度损失。如果需要,可以应用适当的四舍五入方法。
所有返回的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_datetime
、get_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)
编写单元测试以验证单位转换的准确性。例如,输入特定的water_use
值,检查返回结果是否正确地反映了万方单位。
在整个应用程序中测试接口,确保在不同统计类型(年/月/日)和不同位置过滤条件下,接口均能正确返回转换后的用水量数据。
测试边界条件,如water_use
为零、大量数据的处理、日期范围的极端情况(如闰年、跨月等)。确保接口在这些情况下依然稳定且准确。
在处理大规模数据时,确保数据库查询效率。可以考虑以下优化措施:
statistics_date
、location_id
等字段上有适当的索引,以加快查询速度。select_related
或prefetch_related
减少数据库访问次数。通过上述步骤和代码修改,可以有效地将接口返回的用水量单位从立方米转换为万方。确保在整个修改过程中保持数据的准确性和一致性,并通过充分的测试验证修改结果,将有助于提升系统的可靠性和用户体验。