feat: P0-6 Excel导入导出全套(客户/合同/库存/统计, 自动表头探测)

src/huamenchuang/services/excel_service.py

@@ -1,8 +1,8 @@
-"""Excel 生成服务（openpyxl）：BOM 单、订单导出、业务数据导入导出共用工具。"""
+"""Excel 生成/解析服务（openpyxl）：BOM 单、订单导出、业务数据导入导出共用工具。"""
 import io
 from typing import Any
 
-from openpyxl import Workbook
+from openpyxl import Workbook, load_workbook
 from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
 from openpyxl.utils import get_column_letter
 
@@ -96,3 +96,66 @@ def order_workbook(order: dict, bom: dict) -> bytes:
     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)