""" 日报生成核心逻辑 """ 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 from lib.field_mapping import get_field_display_name # 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'), } 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() for order in orders: val = order.get(field) if val is not None and val != '' and val != 0: id_values.add(str(val)) 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)) id_name_maps[field] = {str(row['id_val']): row['name_val'] for row in results} print(f"[ID转换] {field}: 查询到 {len(results)} 个名称映射") except Exception as e: print(f"[ID转换] {field} 查询失败: {e}") id_name_maps[field] = {} return id_name_maps 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') print(f"[日报生成] 开始生成日报") print(f" 配置ID: {config_id}") print(f" 配置名称: {config['config_name']}") print(f" 拆分方式: {config['split_type']} = {config['split_value']}") print(f" 时间范围: {start_time_str} 至 {end_time_str}") print(f" 选择字段: {len(selected_fields)} 个") # 构建查询SQL fields_str = ', '.join(selected_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)) print(f"[日报生成] 查询到 {len(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, 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）' } # 计算求和 sum_results = {} for field in sum_fields: if field in orders[0]: total = sum(float(order.get(field, 0) or 0) for order in orders) sum_results[field] = total # 构建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 ) print(f"[日报生成] Excel文件已生成: {file_path}") # 计算总金额 total_amount = sum(float(order.get('total_money', 0) or 0) for order in orders) # 保存历史记录 print(f"[日报生成] 准备保存历史记录，file_path={file_path}") 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, total_orders=len(orders), total_amount=total_amount, sum_results=sum_results, file_path=file_path, status=1 ) print(f"[日报生成] 历史记录已保存，ID={history_id}") print(f"[日报生成] ✓ 生成成功") print(f" 订单数量: {len(orders)}") print(f" 总金额: {total_amount}") print(f" 文件路径: {file_path}") print(f" 历史记录ID: {history_id}") 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: print(f"[日报生成] ✗ 生成失败: {str(e)}") import traceback traceback.print_exc() # 保存失败记录 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, total_orders=0, total_amount=0, sum_results={}, file_path='', status=0 ) print(f"[日报生成] 已保存失败记录到历史表") except Exception as save_error: print(f"[日报生成] 保存失败记录时出错: {str(save_error)}") 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): """ 生成Excel文件 Args: id_name_maps: ID到名称的映射字典，格式为 {field_name: {id_value: name_value}} 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) # 生成文件名 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'daily_report_{config_name}_{split_name}_{timestamp}.xlsx' file_path = os.path.join(reports_dir, filename) # 创建工作簿 wb = Workbook() ws = wb.active ws.title = '日报' # 写入标题 ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=len(selected_fields)) title_cell = ws.cell(row=1, column=1) title_cell.value = f'{config_name} - {split_name} 日报' title_cell.font = Font(size=16, bold=True) title_cell.alignment = Alignment(horizontal='center') # 写入时间范围 ws.merge_cells(start_row=2, start_column=1, end_row=2, end_column=len(selected_fields)) time_cell = ws.cell(row=2, column=1) time_cell.value = f'统计时间: {start_time} 至 {end_time}' time_cell.alignment = Alignment(horizontal='center') # 写入表头（中文） header_row = 4 for col_idx, field in enumerate(selected_fields, 1): cell = ws.cell(row=header_row, column=col_idx) # 如果是ID字段且有名称映射，表头显示为名称而不是ID if field in id_name_maps and id_name_maps[field]: # 将"用户ID"改为"用户"，"企业ID"改为"企业"等 display_name = get_field_display_name(field).replace('ID', '') else: display_name = get_field_display_name(field) cell.value = display_name cell.font = Font(bold=True) cell.fill = PatternFill(start_color='D3D3D3', end_color='D3D3D3', fill_type='solid') cell.alignment = Alignment(horizontal='center') # 写入数据 for row_idx, order in enumerate(orders, header_row + 1): for col_idx, field in enumerate(selected_fields, 1): cell = ws.cell(row=row_idx, column=col_idx) 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 # 写入求和行 if sum_fields: sum_row = header_row + len(orders) + 1 ws.cell(row=sum_row, column=1).value = '合计' ws.cell(row=sum_row, column=1).font = Font(bold=True) for col_idx, field in enumerate(selected_fields, 1): if field in sum_fields and field in sum_results: cell = ws.cell(row=sum_row, column=col_idx) cell.value = sum_results[field] cell.font = Font(bold=True) cell.number_format = '0.00' # 调整列宽 for col_idx, field in enumerate(selected_fields, 1): max_length = len(get_field_display_name(field)) for row in ws.iter_rows(min_row=header_row + 1, max_row=header_row + len(orders), min_col=col_idx, max_col=col_idx): for cell in row: if cell.value: max_length = max(max_length, len(str(cell.value))) # 使用 openpyxl 的列字母转换函数 from openpyxl.utils import get_column_letter col_letter = get_column_letter(col_idx) ws.column_dimensions[col_letter].width = min(max_length + 2, 50) # 保存文件 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, total_orders, total_amount, sum_results, file_path, status): """ 保存日报生成历史记录 Returns: int: 历史记录ID """ # 生成ID history_id = int(datetime.now().timestamp() * 1000) print(f"[保存历史] 准备保存历史记录") print(f" ID: {history_id}") print(f" file_path: '{file_path}'") print(f" status: {status}") sql = """ INSERT INTO t_daily_report_history (id, report_date, start_time, end_time, split_type, split_value, split_name, config_id, 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, NOW()) """ params = ( history_id, report_date, start_time, end_time, split_type, split_value, split_name, config_id, total_orders, total_amount, json.dumps(sum_results, ensure_ascii=False), file_path, status ) print(f"[保存历史] 执行SQL插入...") result = execute_insert(sql, params) print(f"[保存历史] 插入完成，返回ID: {result}") return history_id