""" Excel 导出工具 用于将采购订单等数据导出为 Excel 文件 """ import io from datetime import datetime from typing import List, Dict, Any from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter class ExcelExporter: """Excel 文件导出器""" # 定义导出字段和显示名称 PROCUREMENT_COLUMNS = { '原始订单编号': 'A', '线上宝贝名称': 'B', '品牌': 'C', '货号': 'D', '线上销售属性': 'E', 'procurement_method': 'F', '颜色': 'G', '尺寸': 'H', '卖家备注': 'I', '忽略采购': 'J', '数量': 'K', '订单单价': 'L', '订单金额': 'M', '网店名称': 'N', '付款时间': 'O', 'procurement_status': 'P', '已推送': 'Q', 'is_out_of_stock': 'R', '交易状态': 'S', '退款状态': 'T', 'invoice_code': 'U', 'procurement_cost': 'V', 'price_alert': 'W', '票据核对': 'X' } # 待发货订单导出字段 PENDING_SHIPMENTS_COLUMNS = { '子订单编号': 'A', '主订单编号': 'B', '网店名称': 'C', '商品标题': 'D', '商品属性': 'E', '购买数量': 'F', '已发货数量': 'G', '买家实付金额': 'H', '订单状态': 'I', '退款状态': 'J', '订单付款时间': 'K', '系统发货状态': 'L', '发货状态详情': 'M', '系统发货时间': 'N', '系统快递公司': 'O', '系统快递单号': 'P', '所有快递单号': 'Q', '发货操作员': 'R', '商家备注': 'S', '主订单买家留言': 'T', '仓库备注': 'U', '仓库备注时间': 'V', '品牌': 'W', '颜色': 'X', '尺码': 'Y', '货号': 'Z', '采购方式': 'AA', '商品扫码记录': 'AB' } # 列头显示名称映射 COLUMN_DISPLAY_NAMES = { '原始订单编号': '订单编号', '线上宝贝名称': '商品名称', '品牌': '品牌', '货号': '货号', '线上销售属性': '销售属性', 'procurement_method': '采购方式', '颜色': '颜色', '尺寸': '尺寸', '卖家备注': '卖家备注', '忽略采购': '忽略采购', '数量': '数量', '订单单价': '单价', '订单金额': '金额', '网店名称': '店铺', '付款时间': '付款时间', 'procurement_status': '已采购', '已推送': '已推送', 'is_out_of_stock': '是否缺货', '交易状态': '交易状态', '退款状态': '退款状态', 'invoice_code': '票据编码', 'procurement_cost': '采购金额', 'price_alert': '价格预警', '票据核对': '票据核对' } @classmethod def _parse_scanned_codes(cls, scanned_info: Any) -> str: """ 解析商品扫码记录JSON，转换为逗号分隔的字符串 Args: scanned_info: JSON格式的扫码信息，可能是: - {"scanned_at": "...", "codes": ["code1", "code2"], "operator": "..."} - ["code1", "code2"] - None Returns: 逗号分隔的扫码字符串，如 "code1, code2" """ if not scanned_info: return '' try: # 如果是字符串，尝试解析JSON if isinstance(scanned_info, str): import json scanned_info = json.loads(scanned_info) # 如果是字典格式 {"codes": [...]} if isinstance(scanned_info, dict) and 'codes' in scanned_info: codes = scanned_info['codes'] if isinstance(codes, list): return ', '.join(str(c) for c in codes) # 如果直接是列表 if isinstance(scanned_info, list): return ', '.join(str(c) for c in scanned_info) except Exception: pass return '' @classmethod def export_procurement_orders(cls, orders: List[Dict[str, Any]]) -> bytes: """ 导出采购订单为 Excel 文件 Args: orders: 订单数据列表 Returns: Excel 文件的二进制内容 """ # 创建工作簿 wb = Workbook() ws = wb.active ws.title = '采购订单' # 定义样式 header_fill = PatternFill(start_color='1F4E78', end_color='1F4E78', fill_type='solid') header_font = Font(bold=True, color='FFFFFF', size=11) border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) center_alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) left_alignment = Alignment(horizontal='left', vertical='center', wrap_text=True) # 添加表头 col_num = 1 for field_key, col_letter in cls.PROCUREMENT_COLUMNS.items(): display_name = cls.COLUMN_DISPLAY_NAMES.get(field_key, field_key) cell = ws[f'{col_letter}1'] cell.value = display_name cell.fill = header_fill cell.font = header_font cell.alignment = center_alignment cell.border = border col_num += 1 # 添加数据行 for row_idx, order in enumerate(orders, start=2): for field_key, col_letter in cls.PROCUREMENT_COLUMNS.items(): cell = ws[f'{col_letter}{row_idx}'] value = order.get(field_key) # 格式化特殊字段 if field_key == '付款时间' and value: # 格式化时间字段 if isinstance(value, str): try: dt = datetime.fromisoformat(value.replace('Z', '+00:00')) cell.value = dt.strftime('%Y-%m-%d %H:%M') except: cell.value = value elif isinstance(value, datetime): cell.value = value.strftime('%Y-%m-%d %H:%M') else: cell.value = value elif field_key == '订单单价' or field_key == '订单金额': # 格式化金额字段 try: cell.value = float(value) if value else 0 cell.number_format = '0.00' except: cell.value = value elif field_key == '数量': # 数量字段为整数 try: cell.value = int(value) if value else 0 except: cell.value = value elif field_key == 'procurement_status': # 采购状态转换 status_map = { 'PENDING': '否', 'ORDERED': '是', 'RECEIVED': '是', 'CANCELLED': '否' } cell.value = status_map.get(value, str(value) if value else '否') elif field_key == 'procurement_method': # 采购方式显示 cell.value = str(value) if value else 'NY' elif field_key == '忽略采购': # 布尔值转换 cell.value = '是' if value else '否' elif field_key == 'is_out_of_stock': # 缺货状态转换 cell.value = '已缺货' if value else '未缺货' elif field_key == 'procurement_cost': # 采购金额字段 - 格式化为货币 try: cell.value = float(value) if value else None if cell.value is not None: cell.number_format = '0.00' except: cell.value = value elif field_key == 'price_alert': # 价格预警 - 布尔值转换 cell.value = '是' if value else '否' else: cell.value = value # 应用样式 cell.border = border if field_key in ['订单单价', '订单金额', '采购金额', '数量']: cell.alignment = Alignment(horizontal='right', vertical='center') else: cell.alignment = left_alignment # 设置列宽 column_widths = { 'A': 16, # 订单编号 'B': 35, # 商品名称 'C': 12, # 品牌 'D': 12, # 货号 'E': 20, # 销售属性 'F': 10, # 采购方式 'G': 10, # 颜色 'H': 10, # 尺寸 'I': 20, # 卖家备注 'J': 10, # 忽略采购 'K': 8, # 数量 'L': 10, # 单价 'M': 10, # 金额 'N': 12, # 店铺 'O': 16, # 付款时间 'P': 10, # 已采购 'Q': 10, # 已推送 'R': 10, # 是否缺货 'S': 12, # 交易状态 'T': 10, # 退款状态 'U': 14, # 票据编码 'V': 12, # 采购金额 'W': 10, # 价格预警 'X': 10 # 票据核对 } for col_letter, width in column_widths.items(): ws.column_dimensions[col_letter].width = width # 冻结表头 ws.freeze_panes = 'A2' # 保存到字节流 buffer = io.BytesIO() wb.save(buffer) buffer.seek(0) return buffer.getvalue() @classmethod def export_pending_shipments(cls, orders: List[Dict[str, Any]]) -> bytes: """ 导出待发货订单为 Excel 文件 Args: orders: 订单数据列表（包含procurement_info） Returns: Excel 文件的二进制内容 """ wb = Workbook() ws = wb.active ws.title = '待发货订单' # 设置列头样式 header_font = Font(bold=True, color='FFFFFF') header_fill = PatternFill(start_color='4472C4', end_color='4472C4', fill_type='solid') header_alignment = Alignment(horizontal='center', vertical='center') # 写入表头 headers = list(cls.PENDING_SHIPMENTS_COLUMNS.keys()) for idx, header in enumerate(headers, 1): cell = ws.cell(row=1, column=idx, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = header_alignment # 写入数据行 for row_idx, order in enumerate(orders, 2): # 基本订单信息 ws.cell(row=row_idx, column=1, value=order.get('子订单编号', '')) ws.cell(row=row_idx, column=2, value=order.get('主订单编号', '')) ws.cell(row=row_idx, column=3, value=order.get('网店名称', '')) ws.cell(row=row_idx, column=4, value=order.get('商品标题', '')) ws.cell(row=row_idx, column=5, value=order.get('商品属性', '')) ws.cell(row=row_idx, column=6, value=order.get('购买数量', 0)) ws.cell(row=row_idx, column=7, value=order.get('已发货数量', 0)) ws.cell(row=row_idx, column=8, value=order.get('买家实付金额', 0.0)) ws.cell(row=row_idx, column=9, value=order.get('订单状态', '')) ws.cell(row=row_idx, column=10, value=order.get('退款状态', '')) # 时间字段 payment_time = order.get('订单付款时间') if payment_time: try: if isinstance(payment_time, str): payment_time = datetime.fromisoformat(payment_time.replace('Z', '+00:00')) ws.cell(row=row_idx, column=11, value=payment_time.strftime('%Y-%m-%d %H:%M:%S')) except: ws.cell(row=row_idx, column=11, value=payment_time) # 发货信息 ws.cell(row=row_idx, column=12, value=order.get('系统发货状态', '')) ws.cell(row=row_idx, column=13, value=order.get('发货状态详情', '')) shipping_time = order.get('系统发货时间') if shipping_time: try: if isinstance(shipping_time, str): shipping_time = datetime.fromisoformat(shipping_time.replace('Z', '+00:00')) ws.cell(row=row_idx, column=14, value=shipping_time.strftime('%Y-%m-%d %H:%M:%S')) except: ws.cell(row=row_idx, column=14, value=shipping_time) ws.cell(row=row_idx, column=15, value=order.get('系统快递公司', '')) ws.cell(row=row_idx, column=16, value=order.get('系统快递单号', '')) # 所有快递单号（从shipment_records中提取） shipment_records = order.get('shipment_records', []) if shipment_records: # 合并所有快递单号，格式: "快递公司:单号(数量)" all_tracking = '; '.join([ f"{r.get('快递公司', '')}:{r.get('快递单号', '')}({r.get('发货数量', 1)}件)" for r in shipment_records ]) ws.cell(row=row_idx, column=17, value=all_tracking) else: ws.cell(row=row_idx, column=17, value='') ws.cell(row=row_idx, column=18, value=order.get('发货操作员', '')) # 备注信息 ws.cell(row=row_idx, column=19, value=order.get('商家备注', '')) ws.cell(row=row_idx, column=20, value=order.get('主订单买家留言', '')) # 仓库备注 ws.cell(row=row_idx, column=21, value=order.get('warehouse_notes', '')) # 仓库备注时间 ws.cell(row=row_idx, column=22, value=order.get('warehouse_notes_updated_at', '')) # 采购信息（从procurement_info中提取） procurement_info = order.get('procurement_info', {}) ws.cell(row=row_idx, column=23, value=procurement_info.get('品牌', '')) ws.cell(row=row_idx, column=24, value=procurement_info.get('颜色', '')) ws.cell(row=row_idx, column=25, value=procurement_info.get('尺码', '')) ws.cell(row=row_idx, column=26, value=procurement_info.get('货号', '')) ws.cell(row=row_idx, column=27, value=procurement_info.get('采购方式', '')) # 商品扫码记录（解析JSON格式） scanned_codes_str = cls._parse_scanned_codes(order.get('商品扫码记录')) ws.cell(row=row_idx, column=28, value=scanned_codes_str) # 自动调整列宽 for column_cells in ws.columns: length = max(len(str(cell.value or '')) for cell in column_cells) ws.column_dimensions[get_column_letter(column_cells[0].column)].width = min(length + 2, 50) # 冻结首行 ws.freeze_panes = 'A2' # 将工作簿保存到字节流 buffer = io.BytesIO() wb.save(buffer) buffer.seek(0) return buffer.getvalue() # 快递单号维度导出字段 TRACKING_NUMBER_COLUMNS = { '系统快递单号': 'A', '快递子单号': 'B', '快递公司': 'C', '发货时间': 'D', '子订单编号': 'E', '主订单编号': 'F', '网店名称': 'G', '商品标题': 'H', '商品属性': 'I', 'SKU实付单价': 'J', '本次发货数量': 'K', '订单数量': 'L', '合计总发货数量': 'M', '剩余未发货数量': 'N', '所有快递单号': 'O', '品牌': 'P', '货号': 'Q', '颜色': 'R', '尺码': 'S', '采购方式': 'T', '操作员': 'U', '商品扫码记录': 'V' } @classmethod def export_by_tracking_number(cls, orders: List[Dict[str, Any]]) -> bytes: """ 按快递单号维度导出为 Excel 文件 每个SKU一行，同一快递单号下的不同SKU生成子单号 Args: orders: 订单数据列表（包含shipment_records和procurement_info） Returns: Excel 文件的二进制内容 """ wb = Workbook() ws = wb.active ws.title = '快递单号明细' # 设置列头样式 header_font = Font(bold=True, color='FFFFFF') header_fill = PatternFill(start_color='2E7D32', end_color='2E7D32', fill_type='solid') header_alignment = Alignment(horizontal='center', vertical='center') # 写入表头 headers = list(cls.TRACKING_NUMBER_COLUMNS.keys()) for idx, header in enumerate(headers, 1): cell = ws.cell(row=1, column=idx, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = header_alignment # 按快递单号分组收集数据 # 结构: {快递单号: [(订单数据, 发货记录), ...]} tracking_groups = {} for order in orders: shipment_records = order.get('shipment_records', []) if not shipment_records: # 如果没有发货记录但有快递单号（已发货状态） tracking_number = order.get('系统快递单号') if tracking_number: if tracking_number not in tracking_groups: tracking_groups[tracking_number] = [] # 创建虚拟发货记录 virtual_record = { '快递单号': tracking_number, '快递公司': order.get('系统快递公司', ''), '发货数量': order.get('已发货数量') or order.get('购买数量') or 1, '发货时间': order.get('系统发货时间'), '操作员': order.get('发货操作员', '') } tracking_groups[tracking_number].append((order, virtual_record)) else: # 有发货记录，按记录分组 for record in shipment_records: tracking_number = record.get('快递单号') if tracking_number: if tracking_number not in tracking_groups: tracking_groups[tracking_number] = [] tracking_groups[tracking_number].append((order, record)) # 写入数据行 row_idx = 2 for tracking_number, items in tracking_groups.items(): # 同一快递单号下的多个SKU，生成子单号 for sub_idx, (order, record) in enumerate(items, 1): sub_tracking = f"{tracking_number}-{sub_idx}" if len(items) > 1 else tracking_number # 系统快递单号 ws.cell(row=row_idx, column=1, value=tracking_number) # 快递子单号 ws.cell(row=row_idx, column=2, value=sub_tracking) # 快递公司 ws.cell(row=row_idx, column=3, value=record.get('快递公司', '')) # 发货时间 shipping_time = record.get('发货时间') if shipping_time: try: if isinstance(shipping_time, str): shipping_time = datetime.fromisoformat(shipping_time.replace('Z', '+00:00')) ws.cell(row=row_idx, column=4, value=shipping_time.strftime('%Y-%m-%d %H:%M:%S')) except: ws.cell(row=row_idx, column=4, value=str(shipping_time)) else: ws.cell(row=row_idx, column=4, value='') # 订单信息 ws.cell(row=row_idx, column=5, value=order.get('子订单编号', '')) ws.cell(row=row_idx, column=6, value=order.get('主订单编号', '')) ws.cell(row=row_idx, column=7, value=order.get('网店名称', '')) ws.cell(row=row_idx, column=8, value=order.get('商品标题', '')) ws.cell(row=row_idx, column=9, value=order.get('商品属性', '')) # SKU实付单价 = 买家实付金额 / 购买数量 total_payment = order.get('买家实付金额') or 0 qty = order.get('购买数量') or 1 unit_price = round(total_payment / qty, 2) if qty > 0 else 0 ws.cell(row=row_idx, column=10, value=unit_price) # SKU维度的数量信息 current_ship_quantity = record.get('发货数量', 1) # 本次发货数量 order_quantity = order.get('购买数量') or 0 # 订单数量 shipped_quantity = order.get('已发货数量') or 0 # 合计总发货数量 remaining_quantity = order_quantity - shipped_quantity # 剩余未发货数量 ws.cell(row=row_idx, column=11, value=current_ship_quantity) ws.cell(row=row_idx, column=12, value=order_quantity) ws.cell(row=row_idx, column=13, value=shipped_quantity) ws.cell(row=row_idx, column=14, value=remaining_quantity) # 所有快递单号（从shipment_records中提取） shipment_records = order.get('shipment_records', []) if shipment_records: all_tracking = '; '.join([ f"{r.get('快递公司', '')}:{r.get('快递单号', '')}({r.get('发货数量', 1)}件)" for r in shipment_records ]) ws.cell(row=row_idx, column=15, value=all_tracking) else: ws.cell(row=row_idx, column=15, value='') # 采购信息 procurement_info = order.get('procurement_info', {}) ws.cell(row=row_idx, column=16, value=procurement_info.get('品牌', '')) ws.cell(row=row_idx, column=17, value=procurement_info.get('货号', '')) ws.cell(row=row_idx, column=18, value=procurement_info.get('颜色', '')) ws.cell(row=row_idx, column=19, value=procurement_info.get('尺码', '')) ws.cell(row=row_idx, column=20, value=procurement_info.get('采购方式', '')) # 操作员 ws.cell(row=row_idx, column=21, value=record.get('操作员', '')) # 商品扫码记录 scanned_codes_str = cls._parse_scanned_codes(order.get('商品扫码记录')) ws.cell(row=row_idx, column=22, value=scanned_codes_str) row_idx += 1 # 自动调整列宽 for column_cells in ws.columns: length = max(len(str(cell.value or '')) for cell in column_cells) ws.column_dimensions[get_column_letter(column_cells[0].column)].width = min(length + 2, 50) # 冻结首行 ws.freeze_panes = 'A2' # 将工作簿保存到字节流 buffer = io.BytesIO() wb.save(buffer) buffer.seek(0) return buffer.getvalue() @classmethod def get_filename(cls, prefix: str = '待采购订单') -> str: """ 生成导出文件名 Args: prefix: 文件前缀 Returns: 文件名字符串 """ timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') return f'{prefix}_{timestamp}.xlsx'