diff --git a/analyze_excel.py b/analyze_excel.py new file mode 100644 index 0000000..aa99310 --- /dev/null +++ b/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') diff --git a/convert_excel.py b/convert_excel.py new file mode 100644 index 0000000..d61c79c --- /dev/null +++ b/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()