"""Excel 生成/解析服务(openpyxl):BOM 单、订单导出、业务数据导入导出共用工具。""" import io from typing import Any from openpyxl import Workbook, load_workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter HEADER_FILL = PatternFill("solid", fgColor="2F5B8F") HEADER_FONT = Font(color="FFFFFF", bold=True, size=11) TITLE_FONT = Font(bold=True, size=14) THIN = Side(style="thin", color="C0C0C0") BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN) def new_workbook() -> Workbook: return Workbook() def save_bytes(wb: Workbook) -> bytes: """工作簿 → xlsx 字节(供响应返回)。""" buf = io.BytesIO() wb.save(buf) return buf.getvalue() def write_title(ws: Any, row: int, text: str, span: int) -> int: """写合并标题行,返回下一行行号。""" ws.cell(row=row, column=1, value=text).font = TITLE_FONT ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=span) ws.cell(row=row, column=1).alignment = Alignment(horizontal="center", vertical="center") ws.row_dimensions[row].height = 24 return row + 1 def write_table(ws: Any, start_row: int, headers: list[str], rows: list[list[Any]], widths: list[int] | None = None) -> int: """写带样式表格(表头填充+全边框),返回下一行行号。""" for c, head in enumerate(headers, 1): cell = ws.cell(row=start_row, column=c, value=head) cell.fill = HEADER_FILL cell.font = HEADER_FONT cell.border = BORDER cell.alignment = Alignment(horizontal="center", vertical="center") for r, row_data in enumerate(rows, start_row + 1): for c, v in enumerate(row_data, 1): cell = ws.cell(row=r, column=c, value=v) cell.border = BORDER if isinstance(v, (int, float)): cell.alignment = Alignment(horizontal="right") if widths: for c, w in enumerate(widths, 1): ws.column_dimensions[get_column_letter(c)].width = w return start_row + 1 + len(rows) def order_workbook(order: dict, bom: dict) -> bytes: """订单 Excel:基本信息 + BOM 型材表 + 玻璃/五金汇总。""" wb = new_workbook() ws = wb.active ws.title = "订单BOM" row = write_title(ws, 1, f"生产订单 — {order['order_no']}", 5) info_rows = [ ["客户", order.get("customer_name") or "-", "电话", order.get("phone") or "-", ""], ["地址", order.get("address") or "-", "状态", order.get("status_label") or "-", ""], ["面积", f"{order.get('area') or 0} ㎡", "樘数", order.get("count") or 0, ""], ["窗型尺寸", f"{bom.get('width_mm', 0)} × {bom.get('height_mm', 0)} mm", "玻璃面积", f"{bom.get('glass_area_m2', 0)} ㎡", ""], ["下单时间", str(order.get("created_at"))[:19], "金额", f"¥{order.get('money') or 0}", ""], ] for r in info_rows: for c, v in enumerate(r, 1): cell = ws.cell(row=row, column=c, value=v) if c in (1, 3): cell.font = Font(bold=True) row += 1 row += 1 row = write_title(ws, row, "型材下料清单", 5) bar_rows = [[b["note"] or b["kind"], b["length"], b["qty"], b["length"] * b["qty"] / 1000, ""] for b in bom.get("bars", [])] row = write_table(ws, row, ["用途", "单件长度(mm)", "数量(件)", "总长(m)", "备注"], bar_rows, widths=[18, 16, 12, 12, 20]) total_m = sum(b["length"] * b["qty"] for b in bom.get("bars", [])) / 1000 cell = ws.cell(row=row, column=1, value="型材合计") cell.font = Font(bold=True) ws.cell(row=row, column=4, value=round(total_m, 2)).font = Font(bold=True) row += 2 row = write_title(ws, row, "玻璃/五金", 5) glass_hw = [["玻璃", f"{bom.get('glass_area_m2', 0)} ㎡", "", "", ""], ["五金", f"{bom.get('sash_count', 0)} 套", "", "", ""]] row = write_table(ws, row, ["项目", "数量", "", "", ""], glass_hw) row += 1 ws.cell(row=row, column=1, value="制单:____________ 审核:____________ 日期:____________") return save_bytes(wb) # ---------------- 导入解析 ---------------- def read_rows(data: bytes, sheet_index: int = 0) -> list[dict]: """解析上传的 xlsx → 行字典列表(自动探测表头行,兼容首行大标题的导出格式)。 探测规则:前 3 行中非空单元格最多的一行视为表头;空行跳过。 """ wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True) ws = wb.worksheets[sheet_index] all_rows = [list(r) for r in ws.iter_rows(values_only=True)] if not all_rows: return [] # 探测表头行:取前 3 行中非空单元格数最多的行 scan = all_rows[:3] header_idx = max(range(len(scan)), key=lambda i: sum(1 for v in scan[i] if v is not None and str(v).strip())) headers = [str(v).strip() if v is not None else "" for v in all_rows[header_idx]] out: list[dict] = [] for values in all_rows[header_idx + 1:]: row = {headers[i]: values[i] for i in range(min(len(headers), len(values)))} if any(v is not None and str(v).strip() for v in row.values()): out.append(row) return out def pick(row: dict, *names: str, default=None): """按候选列名(含中文表头)取值,全部缺失返回 default。""" for n in names: if n in row and row[n] not in (None, ""): return row[n] return default # ---------------- 业务导出工作簿 ---------------- def simple_table_workbook(title: str, headers: list[str], rows: list[list], widths: list[int] | None = None) -> bytes: """单表导出(客户/合同/库存等通用)。""" wb = new_workbook() ws = wb.active ws.title = title[:28] or "导出" row = write_title(ws, 1, title, len(headers)) write_table(ws, row, headers, rows, widths) return save_bytes(wb) def statistics_workbook(orders_summary: dict, receipts: dict) -> bytes: """统计报表导出:订单汇总 + 收款统计双 Sheet。""" wb = new_workbook() ws = wb.active ws.title = "订单汇总" row = write_title(ws, 1, "订单/业务统计", 4) rows = [[k, v, "", ""] for k, v in orders_summary.items()] row = write_table(ws, row, ["指标", "数值", "", ""], rows, widths=[24, 18, 8, 8]) ws2 = wb.create_sheet("收款统计") row = write_title(ws2, 1, f"收款统计 {receipts['year']} 年", 14) row = write_table(ws2, row, ["年度收款", "合同应收", "未收"] + [f"{m}月" for m in range(1, 13)], [[receipts["total_received"], receipts["contract_total"], receipts["unreceived"], *receipts["monthly"]]], widths=[12] * 15) return save_bytes(wb)