Files
ylt_diy/lib/sum_data_collector.py
2026-07-15 16:01:06 +08:00

772 lines
29 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
日报求和数据采集与管理
"""
import json
import os
from datetime import datetime
from lib.db import execute_query, execute_update, execute_insert
from lib.field_mapping import get_field_display_name
from lib.logger import log_info, log_error, log_warning
def collect_sum_data_from_history(history_id=None):
"""
从历史记录中采集求和数据并写入求和数据表
充电电量charge_degree是必须采集的其他求和字段有就采集没有就不采集
Args:
history_id: 历史记录ID可选不填则采集所有成功的历史记录
Returns:
dict: 采集结果
"""
try:
# 查询历史记录
if history_id:
history_list = execute_query(
'SELECT * FROM t_daily_report_history WHERE id = %s AND status = 1',
(history_id,)
)
else:
history_list = execute_query(
'SELECT * FROM t_daily_report_history WHERE status = 1 ORDER BY create_time DESC'
)
if not history_list:
return {'success': False, 'message': '没有找到符合条件的历史记录'}
collected_count = 0
skipped_count = 0
config_cache = {} # 缓存配置的排序号和自定义服务费单价
for history in history_list:
sum_results = history.get('sum_results')
# 解析 sum_results JSON
if isinstance(sum_results, str):
sum_results = json.loads(sum_results) if sum_results else {}
elif sum_results is None:
sum_results = {}
# 获取配置信息(排序号、自定义服务费单价)(缓存)
config_id = history.get('config_id')
sort_order = None
custom_service_fee_price = None
if config_id:
if config_id in config_cache:
sort_order = config_cache[config_id].get('sort_order')
custom_service_fee_price = config_cache[config_id].get('custom_service_fee_price')
custom_service_fee_name = config_cache[config_id].get('custom_service_fee_name')
else:
try:
cfg_result = execute_query(
'SELECT sort_order, custom_service_fee_price, custom_service_fee_name FROM t_daily_report_config WHERE id = %s LIMIT 1',
(config_id,)
)
if cfg_result:
sort_order = cfg_result[0].get('sort_order')
custom_service_fee_price = cfg_result[0].get('custom_service_fee_price')
custom_service_fee_name = cfg_result[0].get('custom_service_fee_name')
config_cache[config_id] = {
'sort_order': sort_order,
'custom_service_fee_price': custom_service_fee_price,
'custom_service_fee_name': custom_service_fee_name
}
except Exception:
config_cache[config_id] = {
'sort_order': None,
'custom_service_fee_price': None,
'custom_service_fee_name': None
}
# 强制确保有充电电量charge_degree
if 'charge_degree' not in sum_results or sum_results['charge_degree'] is None:
# 从订单表重新计算充电电量
try:
charge_degree = _calculate_charge_degree(history)
if charge_degree is not None:
sum_results['charge_degree'] = charge_degree
log_info(f"[采集] 历史记录 {history['id']} 补充计算充电电量: {charge_degree} kWh", 'sum_data')
except Exception as calc_error:
log_error(f"[采集] 历史记录 {history['id']} 计算充电电量失败: {calc_error}", 'sum_data')
# 如果还是没有任何求和数据,跳过
if not sum_results:
skipped_count += 1
continue
# 先删除该历史记录对应的旧数据(避免重复)
execute_update(
'''DELETE FROM t_daily_report_sum_data
WHERE report_date = %s AND config_id = %s AND split_value = %s AND data_source = 'auto'
''',
(history['report_date'], history['config_id'], history['split_value'])
)
# 逐条插入求和字段数据
for field_key, field_value in sum_results.items():
sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count
sql = '''
INSERT INTO t_daily_report_sum_data
(id, report_date, config_id, config_name, split_type, split_value, split_name,
sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark,
create_time, update_time)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
'''
params = (
sum_data_id,
history['report_date'],
history['config_id'],
history.get('config_name', ''),
history.get('split_type', ''),
history.get('split_value', ''),
history.get('split_name', ''),
field_key,
get_field_display_name(field_key),
float(field_value) if field_value is not None else 0,
history.get('total_orders', 0),
sort_order,
'auto',
f'从历史记录自动采集历史ID: {history["id"]}'
)
execute_insert(sql, params)
collected_count += 1
# 如果配置了自定义服务费单价,自动计算并采集自定义服务费
if custom_service_fee_price is not None and custom_service_fee_price != '' and sum_results.get('charge_degree') is not None:
try:
charge_degree = float(sum_results['charge_degree'])
custom_service_fee = round(charge_degree * float(custom_service_fee_price), 2)
if custom_service_fee_name and str(custom_service_fee_name).strip():
fee_display_name = str(custom_service_fee_name).strip()
else:
fee_display_name = f'自定义服务费({custom_service_fee_price}元/kWh)'
sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count
sql = '''
INSERT INTO t_daily_report_sum_data
(id, report_date, config_id, config_name, split_type, split_value, split_name,
sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark,
create_time, update_time)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
'''
params = (
sum_data_id,
history['report_date'],
history['config_id'],
history.get('config_name', ''),
history.get('split_type', ''),
history.get('split_value', ''),
history.get('split_name', ''),
'custom_service_fee',
fee_display_name,
custom_service_fee,
history.get('total_orders', 0),
sort_order,
'auto',
f'自动计算:充电电量{charge_degree}kWh × 单价{custom_service_fee_price}元/kWh'
)
execute_insert(sql, params)
collected_count += 1
except Exception as fee_error:
log_error(f"[采集] 计算自定义服务费失败: {fee_error}", 'sum_data')
log_info(f'[采集] 采集完成,共采集 {collected_count} 条求和数据,跳过 {skipped_count} 条无求和数据的记录', 'sum_data')
return {
'success': True,
'message': f'采集完成,共采集 {collected_count} 条求和数据,跳过 {skipped_count} 条无求和数据的记录'
}
except Exception as e:
log_error(f'[采集] 采集失败: {e}', 'sum_data')
return {'success': False, 'message': f'采集失败: {str(e)}'}
def _calculate_charge_degree(history):
"""
从订单表中计算指定历史记录对应的充电电量总和
Args:
history: 历史记录字典
Returns:
float: 充电电量总和(保留三位小数),失败返回 None
"""
try:
split_type = history.get('split_type', '')
split_value = history.get('split_value', '')
start_time = history.get('start_time', '')
end_time = history.get('end_time', '')
if not split_type or not split_value or not start_time or not end_time:
return None
# 检查是否为多选值(逗号分隔)
split_values = [v.strip() for v in str(split_value).split(',') if v.strip()]
if len(split_values) > 1:
placeholders = ', '.join(['%s'] * len(split_values))
sql = f"""
SELECT SUM(charge_degree) as total_degree
FROM t_equipment_charge_order
WHERE state = 3
AND report_time >= %s
AND report_time < %s
AND {split_type} IN ({placeholders})
"""
params = [start_time, end_time] + split_values
else:
sql = f"""
SELECT SUM(charge_degree) as total_degree
FROM t_equipment_charge_order
WHERE state = 3
AND report_time >= %s
AND report_time < %s
AND {split_type} = %s
"""
params = [start_time, end_time, split_value]
result = execute_query(sql, tuple(params))
if result and result[0].get('total_degree') is not None:
return round(float(result[0]['total_degree']), 3)
return 0.0
except Exception as e:
log_error(f"[计算充电电量] 失败: {e}", 'sum_data')
return None
def get_sum_data_list(page=1, page_size=20, start_date=None, end_date=None, config_id=None,
config_name=None, sum_field_key=None, data_source=None):
"""
查询求和数据列表
Args:
page: 页码
page_size: 每页大小
start_date: 开始日期(可选)
end_date: 结束日期(可选)
config_id: 配置ID可选
config_name: 配置名称(可选,模糊搜索)
sum_field_key: 求和字段键名(可选)
data_source: 数据来源(可选)
Returns:
dict: 查询结果
"""
try:
offset = (page - 1) * page_size
where_clauses = ['1=1']
params = []
if start_date:
where_clauses.append('report_date >= %s')
params.append(start_date)
if end_date:
where_clauses.append('report_date <= %s')
params.append(end_date)
if config_id:
where_clauses.append('config_id = %s')
params.append(config_id)
if config_name:
where_clauses.append('config_name LIKE %s')
params.append(f'%{config_name}%')
if sum_field_key:
where_clauses.append('sum_field_key = %s')
params.append(sum_field_key)
if data_source:
where_clauses.append('data_source = %s')
params.append(data_source)
where_sql = ' AND '.join(where_clauses)
# 查询总数
total_result = execute_query(
f'SELECT COUNT(*) as total FROM t_daily_report_sum_data WHERE {where_sql}',
tuple(params)
)
total = total_result[0]['total']
# 查询数据
list_result = execute_query(
f'''SELECT * FROM t_daily_report_sum_data
WHERE {where_sql}
ORDER BY report_date DESC, sort_order ASC, config_id ASC, id DESC
LIMIT %s OFFSET %s''',
tuple(params) + (page_size, offset)
)
return {
'success': True,
'data': {
'list': list_result,
'total': total,
'page': page,
'page_size': page_size
}
}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}
def add_sum_data(data):
"""
手动添加求和数据
Args:
data: 求和数据字典
Returns:
dict: 操作结果
"""
try:
sum_data_id = int(datetime.now().timestamp() * 1000)
# 获取配置的排序号
sort_order = None
config_id = data.get('config_id')
if config_id:
try:
cfg_result = execute_query(
'SELECT sort_order FROM t_daily_report_config WHERE id = %s LIMIT 1',
(config_id,)
)
if cfg_result and cfg_result[0]['sort_order'] is not None:
sort_order = cfg_result[0]['sort_order']
except Exception:
pass
sql = '''
INSERT INTO t_daily_report_sum_data
(id, report_date, config_id, config_name, split_type, split_value, split_name,
sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark,
create_time, update_time)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
'''
params = (
sum_data_id,
data.get('report_date', ''),
data.get('config_id'),
data.get('config_name', ''),
data.get('split_type', ''),
data.get('split_value', ''),
data.get('split_name', ''),
data.get('sum_field_key', ''),
data.get('sum_field_name', ''),
float(data.get('sum_value', 0)),
data.get('total_orders', 0),
sort_order,
'manual',
data.get('remark', '')
)
execute_insert(sql, params)
return {'success': True, 'message': '添加成功', 'id': sum_data_id}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}
def update_sum_data(sum_data_id, data):
"""
更新求和数据
Args:
sum_data_id: 求和数据ID
data: 更新的数据字典
Returns:
dict: 操作结果
"""
try:
update_fields = []
params = []
if 'report_date' in data:
update_fields.append('report_date = %s')
params.append(data['report_date'])
if 'config_id' in data:
update_fields.append('config_id = %s')
params.append(data['config_id'])
if 'config_name' in data:
update_fields.append('config_name = %s')
params.append(data['config_name'])
if 'split_type' in data:
update_fields.append('split_type = %s')
params.append(data['split_type'])
if 'split_value' in data:
update_fields.append('split_value = %s')
params.append(data['split_value'])
if 'split_name' in data:
update_fields.append('split_name = %s')
params.append(data['split_name'])
if 'sum_field_key' in data:
update_fields.append('sum_field_key = %s')
params.append(data['sum_field_key'])
if 'sum_field_name' in data:
update_fields.append('sum_field_name = %s')
params.append(data['sum_field_name'])
if 'sum_value' in data:
update_fields.append('sum_value = %s')
params.append(float(data['sum_value']))
if 'total_orders' in data:
update_fields.append('total_orders = %s')
params.append(data['total_orders'])
if 'remark' in data:
update_fields.append('remark = %s')
params.append(data['remark'])
if not update_fields:
return {'success': False, 'message': '没有需要更新的字段'}
update_fields.append('update_time = NOW()')
params.append(sum_data_id)
sql = f'''
UPDATE t_daily_report_sum_data
SET {', '.join(update_fields)}
WHERE id = %s
'''
rowcount = execute_update(sql, tuple(params))
if rowcount > 0:
return {'success': True, 'message': '更新成功'}
else:
return {'success': False, 'message': '记录不存在'}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}
def delete_sum_data(sum_data_id):
"""
删除求和数据
Args:
sum_data_id: 求和数据ID
Returns:
dict: 操作结果
"""
try:
rowcount = execute_update(
'DELETE FROM t_daily_report_sum_data WHERE id = %s',
(sum_data_id,)
)
if rowcount > 0:
return {'success': True, 'message': '删除成功'}
else:
return {'success': False, 'message': '记录不存在'}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}
def batch_delete_sum_data(ids):
"""
批量删除求和数据
Args:
ids: ID列表
Returns:
dict: 操作结果
"""
try:
placeholders = ', '.join(['%s'] * len(ids))
rowcount = execute_update(
f'DELETE FROM t_daily_report_sum_data WHERE id IN ({placeholders})',
tuple(ids)
)
return {'success': True, 'message': f'成功删除 {rowcount} 条记录'}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}
def export_sum_data_to_excel(start_date=None, end_date=None, config_id=None, data_source=None, keyword=None):
"""
导出求和数据为Excel
Args:
start_date: 开始日期(可选)
end_date: 结束日期(可选)
config_id: 配置ID可选
data_source: 数据来源(可选)
keyword: 配置名称关键词(可选)
Returns:
dict: {success, message, file_path}
"""
try:
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
# 构建查询条件
where_clauses = []
params = []
if start_date:
where_clauses.append('report_date >= %s')
params.append(start_date)
if end_date:
where_clauses.append('report_date <= %s')
params.append(end_date)
if config_id:
where_clauses.append('config_id = %s')
params.append(config_id)
if data_source:
where_clauses.append('data_source = %s')
params.append(data_source)
if keyword:
where_clauses.append('config_name LIKE %s')
params.append(f'%{keyword}%')
where_sql = ''
if where_clauses:
where_sql = 'WHERE ' + ' AND '.join(where_clauses)
# 查询数据(不分页,全量导出)
sql = f'''
SELECT id, report_date, config_id, config_name, split_type, split_value, split_name,
sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark,
create_time, update_time
FROM t_daily_report_sum_data
{where_sql}
ORDER BY report_date DESC, sort_order ASC, config_id ASC, sum_field_key ASC
'''
rows = execute_query(sql, tuple(params) if params else ())
# 创建Excel
wb = Workbook()
ws = wb.active
ws.title = '求和数据'
# 表头
headers = ['序号', '报表日期', '配置名称', '拆分方式', '拆分值', '求和字段', '求和值', '订单数', '数据来源', '修改原因/备注', '创建时间', '更新时间']
header_fill = PatternFill(start_color='4472C4', end_color='4472C4', fill_type='solid')
header_font = Font(bold=True, color='FFFFFF', size=11)
header_alignment = Alignment(horizontal='center', vertical='center')
thin_border = Border(
left=Side(style='thin', color='000000'),
right=Side(style='thin', color='000000'),
top=Side(style='thin', color='000000'),
bottom=Side(style='thin', color='000000')
)
even_fill = PatternFill(start_color='F2F2F2', end_color='F2F2F2', fill_type='solid')
# 写入表头
for col_idx, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col_idx)
cell.value = header
cell.font = header_font
cell.fill = header_fill
cell.alignment = header_alignment
cell.border = thin_border
ws.row_dimensions[1].height = 25
# 拆分方式映射
split_type_map = {
'company_id': '按企业ID',
'user_id': '按用户ID',
'station_id': '按场站ID'
}
# 数据来源映射
source_map = {
'auto': '自动采集',
'manual': '手动编辑'
}
# 写入数据
for row_idx, row in enumerate(rows, 2):
is_even = (row_idx - 2) % 2 == 1
values = [
row.get('sort_order') or '',
row.get('report_date', ''),
row.get('config_name', ''),
split_type_map.get(row.get('split_type', ''), row.get('split_type', '')),
row.get('split_name', '') or row.get('split_value', ''),
row.get('sum_field_name', ''),
float(row.get('sum_value', 0) or 0),
int(row.get('total_orders', 0) or 0),
source_map.get(row.get('data_source', ''), row.get('data_source', '')),
row.get('remark', ''),
str(row.get('create_time', '')) if row.get('create_time') else '',
str(row.get('update_time', '')) if row.get('update_time') else ''
]
for col_idx, val in enumerate(values, 1):
cell = ws.cell(row=row_idx, column=col_idx)
cell.value = val
cell.border = thin_border
# 对齐方式:数字靠右,其他居中或靠左
if col_idx in [7, 8]: # 求和值、订单数
cell.alignment = Alignment(horizontal='right', vertical='center')
if col_idx == 7:
cell.number_format = '0.000'
elif col_idx in [1, 2, 4, 9]: # 序号、日期、拆分方式、数据来源
cell.alignment = Alignment(horizontal='center', vertical='center')
else:
cell.alignment = Alignment(horizontal='left', vertical='center')
if is_even:
cell.fill = even_fill
ws.row_dimensions[row_idx].height = 22
# 调整列宽
col_widths = [8, 14, 20, 12, 20, 18, 14, 10, 12, 30, 20, 20]
for col_idx, width in enumerate(col_widths, 1):
from openpyxl.utils import get_column_letter
col_letter = get_column_letter(col_idx)
ws.column_dimensions[col_letter].width = width
# 冻结首行
ws.freeze_panes = 'A2'
# 保存文件
reports_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'public', 'reports')
os.makedirs(reports_dir, exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f'求和数据导出_{timestamp}.xlsx'
file_path = os.path.join(reports_dir, filename)
wb.save(file_path)
return {'success': True, 'message': '导出成功', 'file_path': f'/public/reports/{filename}', 'filename': filename, 'total': len(rows)}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}
def get_monthly_charge_degree_total(config_id=None, month=None, split_type=None, split_value=None):
"""
获取当月充电电量累计(从求和数据表查询,对账后的准确数据)
- 如果当天已经生成了日报并采集,自动包含在内,不会重复计算
- 以配置当前拆分为准,只统计匹配的拆分值数据
Args:
config_id: 配置ID可选不传则统计所有配置
month: 月份,格式 YYYY-MM可选默认当月
split_type: 拆分类型可选company_id/user_id/station_id
split_value: 拆分值,逗号分隔的字符串(可选,用于过滤)
Returns:
dict: {
success: bool,
data: {
total_degree: float, // 当月累计充电电量(kWh)
total_days: int, // 有数据的天数
total_orders: int, // 累计订单数
today_degree: float, // 当天充电电量(kWh)(如果有的话)
month: str, // 统计月份
detail: list // 按天明细
},
message: str
}
"""
try:
# 默认当月
if not month:
month = datetime.now().strftime('%Y-%m')
# 拼接日期模糊匹配前缀
date_prefix = month + '-'
# 构建查询SQL
sql = """
SELECT
report_date,
SUM(sum_value) as total_degree,
SUM(total_orders) as total_orders
FROM t_daily_report_sum_data
WHERE sum_field_key = 'charge_degree'
AND report_date LIKE %s
"""
params = [date_prefix + '%']
if config_id:
sql += ' AND config_id = %s'
params.append(config_id)
# 按拆分值过滤(以配置当前拆分为准)
if split_type and split_value:
values = [v.strip() for v in split_value.split(',') if v.strip()]
if values:
placeholders = ','.join(['%s'] * len(values))
sql += f' AND split_type = %s AND split_value IN ({placeholders})'
params.append(split_type)
params.extend(values)
sql += ' GROUP BY report_date ORDER BY report_date ASC'
rows = execute_query(sql, tuple(params))
total_degree = 0.0
total_orders = 0
total_days = len(rows)
today_degree = 0.0
today_str = datetime.now().strftime('%Y-%m-%d')
detail = []
for row in rows:
degree = float(row['total_degree'] or 0)
orders = int(row['total_orders'] or 0)
total_degree += degree
total_orders += orders
detail.append({
'report_date': row['report_date'],
'degree': round(degree, 3),
'orders': orders
})
if row['report_date'] == today_str:
today_degree = degree
return {
'success': True,
'data': {
'total_degree': round(total_degree, 3),
'total_days': total_days,
'total_orders': total_orders,
'today_degree': round(today_degree, 3),
'month': month,
'detail': detail
},
'message': '查询成功'
}
except Exception as e:
import traceback
traceback.print_exc()
return {'success': False, 'message': str(e)}