""" 日报生成核心逻辑 """ import os import json from datetime import datetime, timedelta from openpyxl import Workbook from openpyxl.styles import Font, Alignment, PatternFill import pytz from lib.db import execute_query, execute_insert, execute_update from lib.field_mapping import get_field_display_name from lib.logger import log_info, log_error, log_warning # ID字段到名称的映射配置 # key: 订单表中的ID字段名 # value: (关联表名, 关联表ID字段, 关联表名称字段) ID_TO_NAME_MAPPING = { 'user_id': ('t_user', 'id', 'user_name'), 'company_id': ('t_company', 'id', 'company_name'), 'station_id': ('t_station', 'id', 'station_name'), 'equipment_id': ('t_equipment', 'id', 'equipment_name'), 'connector_id': ('t_connector', 'id', 'connector_name'), } # 特来电表字段映射到报表字段 # t_hlht_tld 字段 → 报表字段 TELECOM_FIELD_MAPPING = { 'order_id': 'order_no', # 订单号 'station_id': 'station_id', # 场站ID 'operator_name': 'station_name', # 运营商名称 → 场站名称 'connector_name': 'equipment_name', # 连接器名称 → 设备名称 'charge_start_time': 'charge_begin_time', # 开始充电时间 'charge_end_time': 'charge_end_time', # 结束充电时间 'total_power': 'charge_degree', # 总电量 'start_soc': 'charge_begin_soc', # 开始SOC 'end_soc': 'charge_end_soc', # 结束SOC 'elec_money': 'charge_elecfee_amount', # 电费 'service_money': 'charge_service_amount', # 服务费 'total_money': 'total_money', # 总金额 'sharp_power': 'sharp_electricity', # 尖时段电量 'peak_power': 'peak_electricity', # 峰时段电量 'flat_power': 'flat_electricity', # 平时段电量 'valley_power': 'valley_electricity', # 谷时段电量 'vehicle_self_no': 'company_name', # 车量自编号 → 企业名称 'plate_no': 'charge_plate_no', # 车牌号 'vin_code': 'charge_vin', # VIN码 'stop_reason': 'finish_msg', # 结束原因 'charge_duration_sec': 'charge_duration', # 充电时长(秒) } # 特来电数据中名称字段到ID字段的映射（用于匹配selected_fields中的ID字段） TELECOM_NAME_TO_ID_FIELDS = { 'company_name': 'company_id', 'station_name': 'station_id', 'equipment_name': 'equipment_id', } def query_telecom_orders(vehicle_self_nos, start_time_str, end_time_str): """ 查询特来电平台的订单数据 Args: vehicle_self_nos: 车量自编号列表（对应企业名称） start_time_str: 开始时间字符串 end_time_str: 结束时间字符串 Returns: list: 转换后的订单数据，字段映射为报表字段 """ if not vehicle_self_nos: return [] # 构建查询条件 placeholders = ', '.join(['%s'] * len(vehicle_self_nos)) sql = f""" SELECT order_id, station_id, operator_name, connector_name, charge_start_time, charge_end_time, charge_duration_sec, total_power, start_soc, end_soc, elec_money, service_money, total_money, sharp_power, peak_power, flat_power, valley_power, vehicle_self_no, plate_no, vin_code, stop_reason FROM t_hlht_tld WHERE vehicle_self_no IN ({placeholders}) AND second_source != '驿来特' AND charge_end_time >= %s AND charge_end_time < %s ORDER BY charge_end_time DESC """ params = tuple(vehicle_self_nos) + (start_time_str, end_time_str) try: results = execute_query(sql, params) log_info(f"[特来电] 查询到 {len(results)} 条订单", 'report') except Exception as e: log_error(f"[特来电] 查询失败: {e}", 'report') return [] # 转换字段名 converted_orders = [] for row in results: converted = {} for telecom_field, report_field in TELECOM_FIELD_MAPPING.items(): if telecom_field in row: converted[report_field] = row[telecom_field] # 将名称字段复制到对应的ID字段，以便ID转名称逻辑能正确处理 # 例如：company_name → company_id（值相同），这样 build_id_to_name_map 会查询 t_company 表 # 但由于值是名称而不是ID，查询会失败，所以我们需要特殊处理 for name_field, id_field in TELECOM_NAME_TO_ID_FIELDS.items(): if name_field in converted: converted[id_field] = converted[name_field] # 添加数据来源标记 converted['data_source'] = '特来电' converted_orders.append(converted) return converted_orders def build_id_to_name_map(orders, selected_fields): """ 构建ID到名称的映射字典 Args: orders: 订单数据列表 selected_fields: 选中的字段列表 Returns: dict: {field_name: {id_value: name_value}} """ id_name_maps = {} for field in selected_fields: if field not in ID_TO_NAME_MAPPING: continue table_name, id_field, name_field = ID_TO_NAME_MAPPING[field] # 收集所有需要查询的ID值 id_values = set() telecom_name_values = {} # 特来电数据中的名称值（直接使用） for order in orders: val = order.get(field) if val is not None and val != '' and val != 0: # 检查是否是特来电订单 if order.get('data_source') == '特来电' and field in TELECOM_NAME_TO_ID_FIELDS.values(): # 特来电数据中的ID字段实际是名称，直接使用 telecom_name_values[str(val)] = str(val) else: id_values.add(str(val)) # 合并特来电的名称映射 if telecom_name_values: id_name_maps[field] = telecom_name_values log_info(f"[ID转换] {field}: 特来电数据直接使用名称 {len(telecom_name_values)} 个", 'report') # 查询驿来特数据的ID映射 if not id_values: continue # 批量查询名称 placeholders = ', '.join(['%s'] * len(id_values)) sql = f"SELECT {id_field} AS id_val, {name_field} AS name_val FROM {table_name} WHERE {id_field} IN ({placeholders})" try: results = execute_query(sql, tuple(id_values)) if field not in id_name_maps: id_name_maps[field] = {} id_name_maps[field].update({str(row['id_val']): row['name_val'] for row in results}) log_info(f"[ID转换] {field}: 查询到 {len(results)} 个名称映射", 'report') except Exception as e: log_error(f"[ID转换] {field} 查询失败: {e}", 'report') if field not in id_name_maps: id_name_maps[field] = {} return id_name_maps def calculate_time_period_electricity(orders, time_periods_config): """ 计算分时段电量 从 t_equipment_charge_order_detail 表查询分时电量数据， 根据时段配置计算各时段的电量总和。 Args: orders: 订单数据列表 time_periods_config: 时段配置 {sharp: [11,12], peak: [9,10], flat: [7,8], valley: [0,1,2,3]} Returns: dict: {order_no: {sharp_electricity: x, peak_electricity: y, flat_electricity: z, valley_electricity: w}} """ if not time_periods_config or not any(time_periods_config.values()): return {} # 收集所有订单号 order_nos = [order.get('order_no') for order in orders if order.get('order_no')] if not order_nos: return {} # 构建小时到时段的映射 hour_to_period = {} for period_key, hours in time_periods_config.items(): for hour in hours: hour_to_period[hour] = period_key if not hour_to_period: return {} # 批量查询订单的分时电量 placeholders = ', '.join(['%s'] * len(order_nos)) sql = f""" SELECT order_no, charge_degree, charge_startr_time, time_flag FROM t_equipment_charge_order_detail WHERE order_no IN ({placeholders}) ORDER BY order_no, charge_startr_time """ try: details = execute_query(sql, tuple(order_nos)) log_info(f"[时段电量] 查询到 {len(details)} 条分时记录", 'report') except Exception as e: log_error(f"[时段电量] 查询失败: {e}", 'report') return {} # 按订单号分组计算各时段电量 result = {} for order_no in order_nos: result[order_no] = { 'sharp_electricity': 0.0, 'peak_electricity': 0.0, 'flat_electricity': 0.0, 'valley_electricity': 0.0 } for detail in details: order_no = detail.get('order_no') if order_no not in result: continue charge_degree = float(detail.get('charge_degree', 0) or 0) charge_start_time = detail.get('charge_startr_time') if not charge_start_time: continue # 获取小时数 if isinstance(charge_start_time, datetime): hour = charge_start_time.hour elif isinstance(charge_start_time, str): try: hour = datetime.strptime(str(charge_start_time), '%Y-%m-%d %H:%M:%S').hour except: continue else: continue # 根据小时确定时段 period_key = hour_to_period.get(hour) if period_key and f'{period_key}_electricity' in result[order_no]: result[order_no][f'{period_key}_electricity'] += charge_degree return result def get_beijing_time(): """获取北京时间""" beijing_tz = pytz.timezone('Asia/Shanghai') return datetime.now(beijing_tz) def generate_daily_report(config_id, start_time=None, end_time=None): """ 生成日报 Args: config_id: 配置ID start_time: 开始时间（可选，默认为昨天8点） end_time: 结束时间（可选，默认为今天8点） Returns: dict: 生成结果，包含 success, message, file_path, total_orders, total_amount, sum_results """ try: # 获取配置信息 configs = execute_query( 'SELECT * FROM t_daily_report_config WHERE id = %s', (config_id,) ) if not configs: return {'success': False, 'message': '配置不存在'} config = configs[0] # 解析字段 selected_fields = json.loads(config['selected_fields']) sum_fields = json.loads(config['sum_fields']) if config['sum_fields'] else [] # 设置时间范围 if not start_time or not end_time: now = get_beijing_time() end_time = now.replace(hour=8, minute=0, second=0, microsecond=0) start_time = end_time - timedelta(days=1) # 格式化时间 start_time_str = start_time.strftime('%Y-%m-%d %H:%M:%S') end_time_str = end_time.strftime('%Y-%m-%d %H:%M:%S') log_info(f"[日报生成] 开始生成日报", 'report') log_info(f" 配置ID: {config_id}", 'report') log_info(f" 配置名称: {config['config_name']}", 'report') log_info(f" 拆分方式: {config['split_type']} = {config['split_value']}", 'report') log_info(f" 时间范围: {start_time_str} 至 {end_time_str}", 'report') log_info(f" 选择字段: {len(selected_fields)} 个", 'report') # 删除同一配置、同一日期、同一拆分的旧数据（覆盖） report_date = start_time.strftime('%Y-%m-%d') delete_old_history(config_id, report_date, config['split_type'], config['split_value']) # 构建查询SQL - 过滤掉虚拟字段（时段电量、自定义服务费、实收金额等） # 虚拟字段不是数据库表中的实际字段，需要动态计算 VIRTUAL_FIELDS = {'sharp_electricity', 'peak_electricity', 'flat_electricity', 'valley_electricity', 'custom_service_fee', 'total_amount'} db_fields = [f for f in selected_fields if f not in VIRTUAL_FIELDS] # 如果选择了实收金额字段，需要确保查询 charge_elecfee_amount（电费金额） if 'total_amount' in selected_fields and 'charge_elecfee_amount' not in db_fields: db_fields.append('charge_elecfee_amount') # 确保至少有 order_no 字段用于关联分时数据 if 'order_no' not in db_fields: db_fields.append('order_no') fields_str = ', '.join(db_fields) # 检查是否为多选值（逗号分隔） split_values = [v.strip() for v in config['split_value'].split(',') if v.strip()] if len(split_values) > 1: # 多选：使用 IN 查询 placeholders = ', '.join(['%s'] * len(split_values)) sql = f""" SELECT {fields_str} FROM t_equipment_charge_order WHERE state = 3 AND report_time >= %s AND report_time < %s AND {config['split_type']} IN ({placeholders}) ORDER BY report_time DESC """ params = [start_time_str, end_time_str] + split_values else: # 单选：使用 = 查询 sql = f""" SELECT {fields_str} FROM t_equipment_charge_order WHERE state = 3 AND report_time >= %s AND report_time < %s AND {config['split_type']} = %s ORDER BY report_time DESC """ params = [start_time_str, end_time_str, config['split_value']] # 执行查询 orders = execute_query(sql, tuple(params)) # 转换为列表，方便后续添加特来电数据 orders = list(orders) if orders else [] log_info(f"[日报生成] 驿来特查询到 {len(orders)} 条订单", 'report') # 添加数据来源标记 for order in orders: order['data_source'] = '驿来特' # 检查是否有补单记录（finish_type = 2 表示补单结束） has_supplement = any(order.get('finish_type') == 2 for order in orders) if has_supplement: log_info(f"[日报生成] 检测到补单记录，将在文件名中添加标记", 'report') # 检查是否需要合并特来电数据 merge_telecom = config.get('merge_telecom') or 0 telecom_vehicle_no = config.get('telecom_vehicle_no') or '' log_info(f"[日报生成] 特来电合并配置: merge_telecom={merge_telecom}, split_type={config['split_type']}, vehicle_no={telecom_vehicle_no}", 'report') if merge_telecom and config['split_type'] == 'company_id' and telecom_vehicle_no: # 解析车量自编号（支持逗号分隔的多个） vehicle_nos = [v.strip() for v in telecom_vehicle_no.split(',') if v.strip()] if vehicle_nos: log_info(f"[日报生成] 开始合并特来电数据，车量自编号: {vehicle_nos}", 'report') telecom_orders = query_telecom_orders(vehicle_nos, start_time_str, end_time_str) log_info(f"[日报生成] 特来电查询到 {len(telecom_orders)} 条订单", 'report') if telecom_orders: # 将特来电数据添加到订单列表 orders.extend(telecom_orders) log_info(f"[日报生成] 合并后总订单数: {len(orders)}", 'report') # 检查特来电数据中是否有补单记录 if not has_supplement: has_supplement = any(order.get('finish_type') == 2 for order in telecom_orders) if not orders: # 即使没有数据，也保存一条失败记录 save_report_history( report_date=start_time.strftime('%Y-%m-%d'), start_time=start_time_str, end_time=end_time_str, split_type=config['split_type'], split_value=config['split_value'], split_name=config.get('split_name', config['split_value']), config_id=config_id, config_name=config.get('config_name', ''), total_orders=0, total_amount=0, sum_results={}, file_path='', status=0 ) return { 'success': False, 'message': f'时间范围内没有符合条件的订单数据（查询条件：{start_time_str} 至 {end_time_str}，{config["split_type"]}={config["split_value"]}，state=3）' } # 解析时段配置并计算分时段电量 time_periods_config = json.loads(config['time_periods']) if config.get('time_periods') else {} if time_periods_config and any(time_periods_config.values()): log_info(f"[日报生成] 时段配置: {time_periods_config}", 'report') time_period_electricity = calculate_time_period_electricity(orders, time_periods_config) # 将分时段电量添加到订单数据中（仅对驿来特订单，特来电订单已有数据） for order in orders: # 特来电订单已有尖峰平谷电量数据，不需要重新计算 if order.get('data_source') == '特来电': continue order_no = order.get('order_no') if order_no and order_no in time_period_electricity: ep = time_period_electricity[order_no] order['sharp_electricity'] = ep['sharp_electricity'] order['peak_electricity'] = ep['peak_electricity'] order['flat_electricity'] = ep['flat_electricity'] order['valley_electricity'] = ep['valley_electricity'] log_info(f"[日报生成] 已计算 {len(time_period_electricity)} 条订单的分时段电量", 'report') # 计算求和（保留三位小数） sum_results = {} for field in sum_fields: # 检查是否至少有一条订单包含该字段 has_field = any(field in order for order in orders) if has_field: total = sum(float(order.get(field, 0) or 0) for order in orders) sum_results[field] = round(total, 3) # 强制采集充电电量（charge_degree），即使不在求和字段配置中也要采集 if 'charge_degree' not in sum_results and orders: has_charge_degree = any('charge_degree' in order for order in orders) if has_charge_degree: total_charge_degree = sum(float(order.get('charge_degree', 0) or 0) for order in orders) sum_results['charge_degree'] = round(total_charge_degree, 3) log_info(f"[日报生成] 强制采集充电电量: {sum_results['charge_degree']} kWh", 'report') # 构建ID到名称的映射 id_name_maps = build_id_to_name_map(orders, selected_fields) # 生成Excel文件 file_path = generate_excel( orders=orders, selected_fields=selected_fields, sum_fields=sum_fields, sum_results=sum_results, config_name=config['config_name'], split_name=config.get('split_name', config['split_value']), start_time=start_time_str, end_time=end_time_str, id_name_maps=id_name_maps, field_custom_names=config.get('field_custom_names'), has_supplement=has_supplement, show_monthly_total=config.get('show_monthly_total', 0) == 1, config_id=config['id'], report_date=start_time.strftime('%Y-%m-%d'), sort_order=config.get('sort_order'), split_type=config.get('split_type'), split_value=config.get('split_value'), custom_service_fee_price=config.get('custom_service_fee_price'), custom_service_fee_name=config.get('custom_service_fee_name'), show_custom_service_fee=config.get('show_custom_service_fee', 0) == 1, show_total_amount=config.get('show_total_amount', 0) == 1, total_amount_name=config.get('total_amount_name') ) log_info(f"[日报生成] Excel文件已生成: {file_path}", 'report') # 计算总金额 total_amount = sum(float(order.get('total_money', 0) or 0) for order in orders) # 保存历史记录 log_info(f"[日报生成] 准备保存历史记录，file_path={file_path}", 'report') history_id = save_report_history( report_date=start_time.strftime('%Y-%m-%d'), start_time=start_time_str, end_time=end_time_str, split_type=config['split_type'], split_value=config['split_value'], split_name=config.get('split_name', config['split_value']), config_id=config_id, config_name=config.get('config_name', ''), total_orders=len(orders), total_amount=total_amount, sum_results=sum_results, file_path=file_path, status=1 ) log_info(f"[日报生成] 历史记录已保存，ID={history_id}", 'report') # 自动采集求和数据到求和数据表 try: if sum_results: from lib.sum_data_collector import collect_sum_data_from_history collect_result = collect_sum_data_from_history(history_id) if collect_result['success']: log_info(f"[日报生成] 求和数据采集成功", 'report') else: log_warning(f"[日报生成] 求和数据采集失败: {collect_result['message']}", 'report') except Exception as collect_error: log_error(f"[日报生成] 求和数据采集异常: {str(collect_error)}", 'report') log_info(f"[日报生成] ✓ 生成成功", 'report') log_info(f" 订单数量: {len(orders)}", 'report') log_info(f" 总金额: {total_amount}", 'report') log_info(f" 文件路径: {file_path}", 'report') log_info(f" 历史记录ID: {history_id}", 'report') return { 'success': True, 'message': '日报生成成功', 'file_path': file_path, 'total_orders': len(orders), 'total_amount': total_amount, 'sum_results': sum_results, 'history_id': history_id } except Exception as e: log_error(f"[日报生成] ✗ 生成失败: {str(e)}", 'report') # 保存失败记录 try: save_report_history( report_date=start_time.strftime('%Y-%m-%d') if start_time else datetime.now().strftime('%Y-%m-%d'), start_time=start_time_str if 'start_time_str' in locals() else '', end_time=end_time_str if 'end_time_str' in locals() else '', split_type=config['split_type'] if 'config' in locals() and config else '', split_value=config['split_value'] if 'config' in locals() and config else '', split_name=config.get('split_name', '') if 'config' in locals() and config else '', config_id=config_id, config_name=config.get('config_name', '') if 'config' in locals() and config else '', total_orders=0, total_amount=0, sum_results={}, file_path='', status=0 ) log_info(f"[日报生成] 已保存失败记录到历史表", 'report') except Exception as save_error: log_error(f"[日报生成] 保存失败记录时出错: {str(save_error)}", 'report') return { 'success': False, 'message': f'生成日报失败: {str(e)}' } def generate_excel(orders, selected_fields, sum_fields, sum_results, config_name, split_name, start_time, end_time, id_name_maps=None, field_custom_names=None, has_supplement=False, show_monthly_total=False, config_id=None, report_date=None, sort_order=None, split_type=None, split_value=None, custom_service_fee_price=None, custom_service_fee_name=None, show_custom_service_fee=False, show_total_amount=False, total_amount_name=None): """ 生成Excel文件 Args: id_name_maps: ID到名称的映射字典，格式为 {field_name: {id_value: name_value}} field_custom_names: 字段自定义名头字典，格式为 {field_key: custom_name} has_supplement: 是否有补单记录 show_monthly_total: 是否显示当月充电量总计 config_id: 配置ID（用于查询月度累计） report_date: 报表日期，格式 YYYY-MM-DD Returns: str: 文件路径 """ if id_name_maps is None: id_name_maps = {} # 创建reports目录 reports_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'public', 'reports') os.makedirs(reports_dir, exist_ok=True) # 生成文件名 - 格式：配置名称_几月几日消费记录.xlsx # 从 start_time 提取日期（报表数据对应的是开始时间那天的数据） try: start_dt = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S') date_str = f'{start_dt.month}月{start_dt.day}日' except: date_str = datetime.now().strftime('%m月%d日') # 如果有补单，添加标记 supplement_suffix = '_有补单' if has_supplement else '' filename = f'{config_name}_{date_str}消费记录{supplement_suffix}.xlsx' file_path = os.path.join(reports_dir, filename) # 创建工作簿 wb = Workbook() ws = wb.active ws.title = '日报' # 表头从第1行开始 header_row = 1 if field_custom_names: if isinstance(field_custom_names, str): try: field_custom_names = json.loads(field_custom_names) except (json.JSONDecodeError, TypeError): field_custom_names = {} else: field_custom_names = {} # 导入边框样式 from openpyxl.styles import Border, Side 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') ) 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') even_row_fill = PatternFill(start_color='F2F2F2', end_color='F2F2F2', fill_type='solid') sum_fill = PatternFill(start_color='FFF2CC', end_color='FFF2CC', fill_type='solid') sum_font = Font(bold=True, size=11) monthly_fill = PatternFill(start_color='D6E4F0', end_color='D6E4F0', fill_type='solid') monthly_font = Font(bold=True, color='1F4E78', size=11) # 数字字段列表（靠右对齐，保留三位小数） numeric_fields = { # 电量类 'charge_degree', 'charge_begin_degree', 'charge_end_degree', 'charge_ah', 'sharp_electricity', 'peak_electricity', 'flat_electricity', 'valley_electricity', 'charge_times_degree', 'subsidy_degree', # 金额类（应收） 'receivable_electric_fee', 'receivable_service_fee', 'receivable_total_fee', # 金额类（实际） 'charge_elecfee_amount', 'charge_elecfee_cost_amount', 'charge_service_amount', 'charging_amt', # 金额类（支付） 'actual_pay_amount', 'pay_amount', # 金额类（单价） 'charge_unit_price', 'charge_unit_cost', 'charge_unit_service_fee', # 金额类（结算） 'settle_electric_fee', 'settle_service_fee', 'settle_fee', 'settle_coupon_amount', # 金额类（优惠券） 'coupon_amount', 'coupon_total_amount', 'elecfee_coupon_amount', 'servicefee_coupon_amount', # 金额类（发票） 'invoice_fee', # 金额类（活动） 'activity_electric_fee', 'activity_service_fee', 'activity_total_fee', 'company_activity_total_fee', 'company_activity_service_fee', 'company_activity_electric_fee', # 金额类（补贴/运营商/佣金） 'subsidy_fee', 'operator_income', 'plat_service_fee', 'com_service_fee', 'com_electric_fee', 'com_total_fee', 'com_charge_discount', # 兼容字段（旧版命名） 'total_money', 'total_amount', 'pay_money', 'discount_money', 'service_money', 'electricity_money', 'parking_money', 'other_money', 'peak_valley_diff_money', # 时长类 'total_duration', 'charging_duration', 'charge_duration', # SOC类 'start_soc', 'end_soc', 'total_soc', 'charge_begin_soc', 'charge_end_soc', 'charge_cur_soc', } # 保留三位小数的字段（金额、电量等） decimal3_fields = numeric_fields # 写入表头 # 第1列：序号 cell = ws.cell(row=header_row, column=1) cell.value = '序号' cell.font = header_font cell.fill = header_fill cell.alignment = header_alignment cell.border = thin_border has_custom_service_fee = custom_service_fee_price is not None and custom_service_fee_price != '' for col_idx, field in enumerate(selected_fields, 2): cell = ws.cell(row=header_row, column=col_idx) if field == 'custom_service_fee': if custom_service_fee_name and str(custom_service_fee_name).strip(): display_name = str(custom_service_fee_name).strip() elif has_custom_service_fee: display_name = f'自定义服务费({custom_service_fee_price}元/kWh)' else: display_name = '自定义服务费' elif field == 'total_amount': if total_amount_name and str(total_amount_name).strip(): display_name = str(total_amount_name).strip() else: display_name = '实收金额' elif field in field_custom_names and field_custom_names[field]: display_name = field_custom_names[field] elif field in id_name_maps and id_name_maps[field]: display_name = get_field_display_name(field).replace('ID', '') else: display_name = get_field_display_name(field) cell.value = display_name cell.font = header_font cell.fill = header_fill cell.alignment = header_alignment cell.border = thin_border ws.row_dimensions[header_row].height = 25 # 写入数据 data_start_row = header_row + 1 for row_idx, order in enumerate(orders, data_start_row): is_even = (row_idx - data_start_row) % 2 == 1 # 第1列：序号 seq_num = row_idx - data_start_row + 1 seq_cell = ws.cell(row=row_idx, column=1) seq_cell.value = seq_num seq_cell.border = thin_border seq_cell.alignment = Alignment(horizontal='center', vertical='center') if is_even: seq_cell.fill = even_row_fill for col_idx, field in enumerate(selected_fields, 2): cell = ws.cell(row=row_idx, column=col_idx) if field == 'custom_service_fee': if has_custom_service_fee: charge_degree = float(order.get('charge_degree', 0) or 0) value = round(charge_degree * float(custom_service_fee_price), 2) else: value = '' elif field == 'total_amount': actual_money = float(order.get('charge_elecfee_amount', 0) or 0) if has_custom_service_fee: charge_degree = float(order.get('charge_degree', 0) or 0) custom_fee = round(charge_degree * float(custom_service_fee_price), 2) value = round(actual_money + custom_fee, 2) else: value = round(actual_money, 2) else: value = order.get(field, '') # 格式化时间字段 if field in ['report_time', 'start_time', 'end_time', 'create_time', 'update_time', 'pay_time']: if isinstance(value, datetime): value = value.strftime('%Y-%m-%d %H:%M:%S') # 将ID转换为名称 if field in id_name_maps and value is not None and value != '': name_map = id_name_maps[field] name_value = name_map.get(str(value)) if name_value: value = name_value cell.value = value cell.border = thin_border # 数字字段靠右对齐，其他居中或靠左 if field in numeric_fields or field in ['custom_service_fee', 'total_amount']: cell.alignment = Alignment(horizontal='right', vertical='center') if isinstance(value, (int, float)): cell.number_format = '0.00' elif field in ['order_no', 'vehicle_no', 'gun_no', 'connector_no']: cell.alignment = Alignment(horizontal='center', vertical='center') else: cell.alignment = Alignment(horizontal='left', vertical='center') # 斑马纹（偶数行） if is_even: cell.fill = even_row_fill ws.row_dimensions[row_idx].height = 22 # 写入求和行 last_data_row = header_row + len(orders) sum_row = last_data_row + 1 if sum_fields: ws.cell(row=sum_row, column=1).value = '合计' ws.cell(row=sum_row, column=1).font = sum_font ws.cell(row=sum_row, column=1).fill = sum_fill ws.cell(row=sum_row, column=1).alignment = Alignment(horizontal='center', vertical='center') ws.cell(row=sum_row, column=1).border = thin_border for col_idx, field in enumerate(selected_fields, 2): cell = ws.cell(row=sum_row, column=col_idx) cell.fill = sum_fill cell.border = thin_border if field == 'custom_service_fee': if has_custom_service_fee: total_charge_degree = sum_results.get('charge_degree', 0) value = round(float(total_charge_degree) * float(custom_service_fee_price), 2) cell.value = value cell.font = sum_font cell.alignment = Alignment(horizontal='right', vertical='center') cell.number_format = '0.00' elif field == 'total_amount': total_actual_money = sum_results.get('charge_elecfee_amount', 0) if has_custom_service_fee: total_charge_degree = sum_results.get('charge_degree', 0) total_custom_fee = round(float(total_charge_degree) * float(custom_service_fee_price), 2) value = round(float(total_actual_money) + total_custom_fee, 2) else: value = round(float(total_actual_money), 2) cell.value = value cell.font = sum_font cell.alignment = Alignment(horizontal='right', vertical='center') cell.number_format = '0.00' elif field in sum_fields and field in sum_results: cell.value = sum_results[field] cell.font = sum_font cell.alignment = Alignment(horizontal='right', vertical='center') cell.number_format = '0.000' ws.row_dimensions[sum_row].height = 25 # 写入当月充电量累计总计行（从求和采集表查询，对账后数据） if show_monthly_total and config_id and report_date: try: from lib.sum_data_collector import get_monthly_charge_degree_total month = report_date[:7] monthly_result = get_monthly_charge_degree_total( config_id=config_id, month=month, split_type=split_type, split_value=split_value, end_date=report_date ) if monthly_result['success'] and monthly_result['data']: total_degree = monthly_result['data']['total_degree'] total_days = monthly_result['data']['total_days'] today_degree_from_table = monthly_result['data'].get('today_degree', 0) missing_dates = monthly_result['data'].get('missing_dates', []) expected_days = monthly_result['data'].get('expected_days', 0) # 如果当天的数据还没采集进 sum_data 表，手动加上当天的 today_charge_degree = sum_results.get('charge_degree', 0) if today_degree_from_table == 0 and today_charge_degree > 0: total_degree = round(total_degree + today_charge_degree, 3) total_days = total_days + 1 # 加上当天后，缺失日期也相应减少一天 if report_date in missing_dates: missing_dates = [d for d in missing_dates if d != report_date] # 月度总计行（在合计行下面空一行） monthly_row = sum_row + 2 if sum_fields else sum_row + 1 # 找 charge_degree 字段所在的列（+1因为序号占了第1列） charge_degree_col = None for col_idx, field in enumerate(selected_fields, 2): if field == 'charge_degree': charge_degree_col = col_idx break # 总列数 total_cols = len(selected_fields) + 2 # 标题文字 title_text = f'当月累计充电量（共{total_days}天，截止{report_date}）' if missing_dates and len(missing_dates) > 0: if len(missing_dates) <= 5: missing_str = '、'.join([d[5:] for d in missing_dates]) else: missing_str = '、'.join([d[5:] for d in missing_dates[:5]]) + f'...共{len(missing_dates)}天' title_text += f'\n（缺：{missing_str}）' # 标题跨列：从第1列到 charge_degree_col 的前一列 # 如果没有找到充电电量列，就跨到倒数第二列 title_end_col = charge_degree_col - 1 if charge_degree_col else total_cols - 1 if title_end_col < 1: title_end_col = 1 # 合并标题单元格 if title_end_col > 1: ws.merge_cells(start_row=monthly_row, start_column=1, end_row=monthly_row, end_column=title_end_col) # 设置标题单元格 title_cell = ws.cell(row=monthly_row, column=1) title_cell.value = title_text title_cell.font = monthly_font title_cell.fill = monthly_fill title_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) title_cell.border = thin_border # 其他列填充背景色和边框 for col_idx in range(2, total_cols): cell = ws.cell(row=monthly_row, column=col_idx) cell.fill = monthly_fill cell.border = thin_border # 数值放在 charge_degree 列 if charge_degree_col: cell = ws.cell(row=monthly_row, column=charge_degree_col) cell.value = total_degree cell.font = monthly_font cell.alignment = Alignment(horizontal='right', vertical='center') cell.number_format = '0.000' # 行高 if missing_dates and len(missing_dates) > 0: ws.row_dimensions[monthly_row].height = 45 else: ws.row_dimensions[monthly_row].height = 28 except Exception as e: log_error(f"[日报生成] 写入月度总计时出错: {e}", 'report') # 调整列宽（中文按2个字符宽度计算） from openpyxl.utils import get_column_letter # 序号列固定宽度 ws.column_dimensions['A'].width = 8 # 计算所有字段的宽度 col_widths = {} numeric_cols = [] for col_idx, field in enumerate(selected_fields, 2): header_name = get_field_display_name(field) if field in field_custom_names and field_custom_names[field]: header_name = field_custom_names[field] cn_count = sum(1 for c in str(header_name) if '\u4e00' <= c <= '\u9fff') max_length = len(str(header_name)) + cn_count for row in ws.iter_rows(min_row=data_start_row, max_row=last_data_row, min_col=col_idx, max_col=col_idx): for cell in row: if cell.value is not None and cell.value != '': cell_str = str(cell.value) cn_count = sum(1 for c in cell_str if '\u4e00' <= c <= '\u9fff') cell_len = len(cell_str) + cn_count max_length = max(max_length, cell_len) col_letter = get_column_letter(col_idx) col_widths[col_idx] = min(max_length + 2, 50) # 标记数值类型字段（求和类列） if field in numeric_fields or field in ['custom_service_fee', 'total_amount']: numeric_cols.append(col_idx) # 计算数值类型列的最大宽度，使所有求和类列宽度一致 if numeric_cols: max_numeric_width = max(col_widths[col] for col in numeric_cols) for col_idx in numeric_cols: col_letter = get_column_letter(col_idx) ws.column_dimensions[col_letter].width = max_numeric_width # 设置非数值类型列的宽度 for col_idx, width in col_widths.items(): if col_idx not in numeric_cols: col_letter = get_column_letter(col_idx) ws.column_dimensions[col_letter].width = width # 冻结首行 ws.freeze_panes = 'A2' # 保存文件 wb.save(file_path) # 返回相对路径 return f'/reports/{filename}' def save_report_history(report_date, start_time, end_time, split_type, split_value, split_name, config_id, config_name, total_orders, total_amount, sum_results, file_path, status): """ 保存日报生成历史记录 Returns: int: 历史记录ID """ # 生成ID history_id = int(datetime.now().timestamp() * 1000) log_info(f"[保存历史] 准备保存历史记录", 'report') log_info(f" ID: {history_id}", 'report') log_info(f" file_path: '{file_path}'", 'report') log_info(f" status: {status}", 'report') sql = """ INSERT INTO t_daily_report_history (id, report_date, start_time, end_time, split_type, split_value, split_name, config_id, config_name, total_orders, total_amount, sum_results, file_path, status, create_time) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) """ params = ( history_id, report_date, start_time, end_time, split_type, split_value, split_name, config_id, config_name, total_orders, total_amount, json.dumps(sum_results, ensure_ascii=False), file_path, status ) log_info(f"[保存历史] 执行SQL插入...", 'report') result = execute_insert(sql, params) log_info(f"[保存历史] 插入完成，返回ID: {result}", 'report') return history_id def delete_old_history(config_id, report_date, split_type, split_value): """ 删除同一配置、同一日期、同一拆分的旧历史记录，包括： 1. 旧的报表文件 2. 旧的求和数据（自动采集的） 3. 旧的历史记录 Args: config_id: 配置ID report_date: 报表日期（YYYY-MM-DD） split_type: 拆分方式 split_value: 拆分值 """ try: log_info(f"[删除旧数据] 检查是否有旧数据，配置ID={config_id}, 日期={report_date}, 拆分={split_type}={split_value}", 'report') # 查询旧的历史记录 old_histories = execute_query( '''SELECT id, file_path FROM t_daily_report_history WHERE config_id = %s AND report_date = %s AND split_type = %s AND split_value = %s''', (config_id, report_date, split_type, split_value) ) if not old_histories: log_info("[删除旧数据] 没有找到旧数据", 'report') return log_info(f"[删除旧数据] 找到 {len(old_histories)} 条旧记录", 'report') reports_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'public', 'reports') for history in old_histories: history_id = history['id'] file_path = history.get('file_path', '') # 删除报表文件 if file_path: try: if file_path.startswith('/reports/'): filename = file_path.replace('/reports/', '') elif file_path.startswith('/public/reports/'): filename = file_path.replace('/public/reports/', '') else: filename = os.path.basename(file_path) full_path = os.path.join(reports_dir, filename) if os.path.exists(full_path): os.remove(full_path) log_info(f"[删除旧数据] 已删除文件: {full_path}", 'report') except Exception as e: log_warning(f"[删除旧数据] 删除文件失败: {e}", 'report') # 删除求和数据（自动采集的） try: execute_update( '''DELETE FROM t_daily_report_sum_data WHERE report_date = %s AND config_id = %s AND split_type = %s AND split_value = %s AND data_source = 'auto' ''', (report_date, config_id, split_type, split_value) ) log_info(f"[删除旧数据] 已删除旧的求和数据", 'report') except Exception as e: log_warning(f"[删除旧数据] 删除求和数据失败: {e}", 'report') # 删除历史记录 try: execute_update( 'DELETE FROM t_daily_report_history WHERE id = %s', (history_id,) ) log_info(f"[删除旧数据] 已删除历史记录 ID={history_id}", 'report') except Exception as e: log_warning(f"[删除旧数据] 删除历史记录失败: {e}", 'report') log_info("[删除旧数据] 旧数据清理完成", 'report') except Exception as e: log_error(f"[删除旧数据] 异常: {e}", 'report')