""" Excel Report Data Conversion Utility Converts Excel report data to database record format """ import pandas as pd import openpyxl from typing import List, Dict, Optional import os import logging from utils.logging import get_logger class ExcelConverter: """Excel Report Data Converter""" # Field name mapping (resolves field name conflicts) FIELD_NAME_MAPPING = {"计划数量": "Product planned quantity", "单位": "Product unit"} def __init__(self, verbose: bool = True, logger: Optional[logging.Logger] = None): """ Initialize the converter Args: verbose: Whether to print detailed logs (default: True). Deprecated, use logger instead. logger: Optional logging.Logger instance. If None and verbose=True, creates default logger. """ self.verbose = verbose # Create default logger if needed if logger is None and verbose: self.logger = get_logger('bipauto.converter') elif logger is None: # Silent mode - create a logger but disable output silent_logger = logging.getLogger('bipauto.converter.silent') silent_logger.setLevel(logging.CRITICAL + 1) # Higher than critical, never logs self.logger = silent_logger else: self.logger = logger def convert(self, input_file: str, output_file: Optional[str] = None) -> pd.DataFrame: """ Convert Excel file Args: input_file: Input file path output_file: Output file path (optional, not saved if not specified) Returns: Converted DataFrame """ # Handle output file if output_file: output_file = self._handle_output_file(output_file) # Read worksheet wb = openpyxl.load_workbook(input_file) ws = wb.active # Parse order data orders = self._parse_sheet(ws) # Convert to DataFrame df = self._convert_to_dataframe(orders) if not df.empty: # Save file if output_file: df.to_excel(output_file, index=False) # Print summary report (using logger) self.logger.info("=" * 60) self.logger.info("Conversion complete") self.logger.info("=" * 60) self.logger.info(f"Order count: {len(orders)}") self.logger.info(f"Data row count: {len(df)}") self.logger.info(f"Output file: {output_file if output_file else 'N/A'}") self.logger.info("=" * 60) return df def _handle_output_file(self, output_file: str) -> str: """ Handle output file, attempt to delete if it exists Args: output_file: Output file path Returns: Actual output file path used """ if os.path.exists(output_file): try: os.remove(output_file) except PermissionError: self.logger.warning( f"Warning: Could not delete {output_file}, file may be open by another program" ) # Modify filename base, ext = os.path.splitext(output_file) output_file = f"{base}_new{ext}" return output_file def _parse_sheet(self, ws) -> List[Dict]: """ Parse a worksheet and return data for all orders Each order contains: - order_info: Order header information (including footer) - materials: List of material data Args: ws: openpyxl worksheet object Returns: Order list """ orders = [] all_rows = list(ws.iter_rows(values_only=True)) # Scan row by row, parse by order structure i = 0 while i < len(all_rows): row = all_rows[i] # Check if this is the order title row if row and "离散备料计划" in str(row[0]): # Parse order header information (next 4 rows) order_info = {} for j in range(1, 5): if i + j < len(all_rows) and all_rows[i + j]: self._parse_header_row(all_rows[i + j], order_info) # Skip empty rows, find table header row 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 # Check if this is the table header row if ( table_row < len(all_rows) and all_rows[table_row] and all_rows[table_row][0] == "序号" ): # Check if the row below the header is empty to determine if data exists next_row = table_row + 1 is_empty_row = ( next_row < len(all_rows) and all_rows[next_row] and all( cell is None or str(cell).strip() == "" for cell in all_rows[next_row] ) ) if is_empty_row: # No data, find footer information materials = [] footer_info = {} data_row = next_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]) ): self._parse_header_row(all_rows[data_row], footer_info) if ( data_row + 1 < len(all_rows) and all_rows[data_row + 1] ): self._parse_header_row( all_rows[data_row + 1], footer_info ) break data_row += 1 orders.append( { "order_info": {**order_info, **footer_info}, "materials": materials, } ) else: # Has data, start extracting materials materials = [] footer_info = {} # Footer information data_row = table_row + 1 while data_row < len(all_rows) and all_rows[data_row]: # Check if this is footer information (creator, printer) if all_rows[data_row + 1][0] and "制单人" in str( all_rows[data_row + 1][0] ): # Parse footer information self._parse_header_row(all_rows[data_row], footer_info) # Check if the next row is also footer information if ( data_row + 1 < len(all_rows) and all_rows[data_row + 1] ): self._parse_header_row( all_rows[data_row + 1], footer_info ) break # Extract material data 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) data_row += 1 orders.append( { "order_info": {**order_info, **footer_info}, "materials": materials, } ) i += 1 return orders def _parse_header_row(self, row: tuple, info: Dict): """ Parse a row of order header information (field names and values interleaved) Args: row: Row data info: Dictionary to store parsing results """ i = 0 while i < len(row): cell = row[i] if cell and str(cell).strip() and "：" in str(cell): # Find field name field_name = str(cell).replace(":", "").strip() # Apply field name mapping if field_name in self.FIELD_NAME_MAPPING: field_name = self.FIELD_NAME_MAPPING[field_name] # Skip empty cells, find the first non-field-name value 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() # Skip processed value, continue to find next field name i = j + 1 else: i += 1 def _convert_to_dataframe(self, orders: List[Dict]) -> pd.DataFrame: """ Convert order data to a flattened DataFrame Args: orders: Order list Returns: Flattened 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)