Add Excel analysis and conversion scripts for order data processing

analyze_excel.py

@@ -0,0 +1,63 @@
+"""
+分析 Excel 文件的数据结构
+"""
+import pandas as pd
+import openpyxl
+import sys
+
+# 设置输出编码
+if sys.platform == 'win32':
+    import io
+    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
+
+# 读取 Excel 文件
+file_path = "data/导出文件.xlsx"
+
+print("=" * 80)
+print("1. 读取所有工作表名称")
+print("=" * 80)
+
+wb = openpyxl.load_workbook(file_path)
+sheet_names = wb.sheetnames
+print(f"工作表数量: {len(sheet_names)}")
+print(f"工作表名称: {sheet_names}")
+
+print("\n" + "=" * 80)
+print("2. 逐行读取原始数据（使用 openpyxl）")
+print("=" * 80)
+
+for sheet_name in sheet_names:
+    print(f"\n--- 工作表: {sheet_name} ---")
+    ws = wb[sheet_name]
+
+    for idx, row in enumerate(ws.iter_rows(values_only=True), 1):
+        # 检查是否为空行
+        is_empty = all(cell is None or str(cell).strip() == "" for cell in row)
+        if is_empty:
+            print(f"行 {idx}: [空行]")
+        else:
+            # 过滤掉 None 和空字符串，只显示有效数据
+            valid_data = [str(cell) if cell is not None else "" for cell in row]
+            print(f"行 {idx}: {valid_data}")
+
+print("\n" + "=" * 80)
+print("3. 使用 pandas 读取数据（观察列结构）")
+print("=" * 80)
+
+for sheet_name in sheet_names:
+    print(f"\n--- 工作表: {sheet_name} ---")
+    df = pd.read_excel(file_path, sheet_name=sheet_name)
+    print(f"DataFrame 形状: {df.shape}")
+    print(f"\n列名:")
+    for i, col in enumerate(df.columns):
+        print(f"  列 {i}: {col}")
+    print(f"\n数据预览:")
+    pd.set_option('display.max_rows', 25)
+    pd.set_option('display.max_columns', 20)
+    pd.set_option('display.width', 200)
+    pd.set_option('display.max_colwidth', 30)
+    print(df)
+    pd.reset_option('display.max_rows')
+    pd.reset_option('display.max_columns')
+    pd.reset_option('display.width')
+    pd.reset_option('display.max_colwidth')

convert_excel.py

@@ -0,0 +1,213 @@
+"""
+将 Excel 报表数据转换为数据库记录形式
+"""
+import pandas as pd
+import openpyxl
+from typing import List, Dict
+import os
+
+def parse_sheet(ws) -> List[Dict]:
+    """
+    解析一个工作表，返回所有订单的数据
+
+    每个订单包含:
+    - order_info: 订单头信息
+    - materials: 物料数据列表
+    """
+    orders = []
+    all_rows = list(ws.iter_rows(values_only=True))
+
+    # 查找所有空行，用于分割订单
+    empty_rows = [i for i, row in enumerate(all_rows) if all(cell is None or str(cell).strip() == "" for cell in row)]
+
+    print(f"检测到空行索引: {empty_rows}")
+    print(f"总行数: {len(all_rows)}")
+
+    # 逐行扫描，按订单结构解析
+    i = 0
+    while i < len(all_rows):
+        row = all_rows[i]
+
+        # 检查是否是订单标题行
+        if row and '离散备料计划' in str(row[0]):
+            print(f"\n在行 {i + 1} 发现订单标题")
+
+            # 解析订单头信息（接下来的4行）
+            order_info = {}
+            for j in range(1, 5):
+                if i + j < len(all_rows) and all_rows[i + j]:
+                    parse_header_row(all_rows[i + j], order_info)
+
+            print(f"订单头信息: {order_info}")
+
+            # 跳过空行，找到表格标题行
+            table_row = i + 5
+            while table_row < len(all_rows) and (not all_rows[table_row] or not all_rows[table_row][0]):
+                table_row += 1
+
+            # 检查是否是表格标题行
+            if table_row < len(all_rows) and all_rows[table_row] and all_rows[table_row][0] == '序号':
+                print(f"在行 {table_row + 1} 发现表格标题")
+
+                # 解析物料数据
+                materials = []
+                footer_info = {}  # 页脚信息
+                data_row = table_row + 1
+                while data_row < len(all_rows) and all_rows[data_row]:
+                    # 检查是否是页脚信息（制单人、打印人）
+                    if all_rows[data_row][0] and ('制单人' in str(all_rows[data_row][0]) or '打印人' in str(all_rows[data_row][0])):
+                        print(f"在行 {data_row + 1} 发现页脚信息")
+                        # 解析页脚信息
+                        parse_header_row(all_rows[data_row], footer_info)
+                        # 检查下一行是否也是页脚信息
+                        if data_row + 1 < len(all_rows) and all_rows[data_row + 1]:
+                            parse_header_row(all_rows[data_row + 1], footer_info)
+                        print(f"页脚信息: {footer_info}")
+                        break
+
+                    # 检查是否是物料行（第一列是数字）
+                    if all_rows[data_row][0] and str(all_rows[data_row][0]).strip().isdigit():
+                        material_row = all_rows[data_row]
+                        material = {
+                            '序号': material_row[0],
+                            '材料编码': material_row[1],
+                            '材料名称': material_row[2],
+                            '规格': material_row[3],
+                            '型号': material_row[4],
+                            '图号': material_row[5],
+                            '物料材质': material_row[6],
+                            '计划数量': material_row[7],
+                            '单位': material_row[8],
+                            '需用日期': material_row[9],
+                            '发料仓库': material_row[10],
+                            '单位用量': material_row[11],
+                            '累计出库数量': material_row[12],
+                        }
+                        materials.append(material)
+                        print(f"  添加物料: {material['材料编码']} - {material['材料名称']}")
+
+                    data_row += 1
+
+                print(f"共解析到 {len(materials)} 条物料数据")
+
+                orders.append({
+                    'order_info': {**order_info, **footer_info},
+                    'materials': materials
+                })
+
+        i += 1
+
+    return orders
+
+def parse_header_row(row: tuple, info: Dict):
+    """
+    解析订单头信息的一行（字段名和值交错排列）
+    """
+    i = 0
+    while i < len(row):
+        cell = row[i]
+        if cell and str(cell).strip() and '：' in str(cell):
+            # 找到字段名
+            field_name = str(cell).replace('：', '').strip()
+
+            # 重命名冲突字段
+            field_name_mapping = {
+                '计划数量': '产品计划数量',
+                '单位': '产品单位'
+            }
+            if field_name in field_name_mapping:
+                field_name = field_name_mapping[field_name]
+
+            # 跳过空单元格，找到第一个非字段名的值
+            j = i + 1
+            while j < len(row) and (not row[j] or not str(row[j]).strip() or '：' in str(row[j])):
+                j += 1
+            if j < len(row) and row[j] and not '：' in str(row[j]):
+                info[field_name] = str(row[j]).strip()
+            # 跳过已处理的值，继续找下一个字段名
+            i = j + 1
+        else:
+            i += 1
+
+def convert_to_dataframe(orders: List[Dict]) -> pd.DataFrame:
+    """
+    将订单数据转换为扁平化的 DataFrame
+    """
+    all_records = []
+
+    for order in orders:
+        order_info = order['order_info']
+        materials = order['materials']
+
+        for material in materials:
+            record = {
+                **order_info,
+                **material
+            }
+            all_records.append(record)
+
+    return pd.DataFrame(all_records)
+
+def main():
+    input_file = "data/导出文件.xlsx"
+    output_file = "data/导出文件_转换.xlsx"
+
+    # 如果输出文件存在，先删除
+    if os.path.exists(output_file):
+        try:
+            os.remove(output_file)
+        except PermissionError:
+            print(f"警告: 无法删除 {output_file}，可能文件被其他程序打开")
+            output_file = "data/导出文件_转换_new.xlsx"
+
+    print("=" * 80)
+    print("开始转换 Excel 数据")
+    print("=" * 80)
+
+    # 读取工作表
+    wb = openpyxl.load_workbook(input_file)
+    ws = wb.active
+
+    # 解析订单数据
+    orders = parse_sheet(ws)
+
+    print(f"\n\n共解析到 {len(orders)} 个订单")
+
+    # 打印每个订单的摘要
+    for i, order in enumerate(orders, 1):
+        order_info = order['order_info']
+        materials = order['materials']
+        print(f"\n订单 {i}:")
+        print(f"  备料计划单号: {order_info.get('备料计划单号', 'N/A')}")
+        print(f"  来源单号: {order_info.get('来源单号', 'N/A')}")
+        print(f"  产品编码: {order_info.get('产品编码', 'N/A')}")
+        print(f"  产品名称: {order_info.get('产品名称', 'N/A')}")
+        print(f"  计划数量: {order_info.get('计划数量', 'N/A')}")
+        print(f"  物料数量: {len(materials)}")
+
+    # 转换为 DataFrame
+    df = convert_to_dataframe(orders)
+
+    print(f"\n转换后的数据形状: {df.shape}")
+
+    if not df.empty:
+        print(f"列名: {list(df.columns)}")
+
+        # 保存为 Excel
+        df.to_excel(output_file, index=False)
+        print(f"\n数据已保存到: {output_file}")
+
+        # 显示前几行数据
+        print("\n数据预览:")
+        pd.set_option('display.max_columns', None)
+        pd.set_option('display.width', 200)
+        pd.set_option('display.max_colwidth', 30)
+        print(df.head(20))
+        pd.reset_option('display.max_columns')
+        pd.reset_option('display.width')
+        pd.reset_option('display.max_colwidth')
+    else:
+        print("警告: 没有数据可保存")
+
+if __name__ == "__main__":
+    main()