e7bbbbc194
Add centralized logging utility and optional logger parameters to all core functions for better observability and debugging capabilities. New modules: - utils/logging.py: Centralized logger configuration with console and optional file handlers Enhanced features: - Added optional logger parameter to all extractor_core functions - Added logger support to extractor, excel_converter, and auth modules - Functions remain silent when logger=None (backward compatible) - Improved environment variable validation in test files Documentation: - Added discrete_material_plan_extractor_core.md with complete API reference and usage patterns Benefits: - Consistent logging format across all components - Optional debug output for troubleshooting - No breaking changes - fully backward compatible - Better error messages and validation Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""
|
||
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)
|