excel_processor.py 3.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283
  1. import pandas as pd
  2. from typing import List
  3. import logging
  4. from mylib.logging_config import setup_logging
  5. # Setup custom logging
  6. setup_logging()
  7. logger = logging.getLogger('excel_tool')
  8. class ExcelProcessor:
  9. def __init__(self):
  10. """Initialize Excel processor"""
  11. pass
  12. def _column_letter_to_index(self, col: str) -> int:
  13. """将Excel列字母转换为索引(A=0, B=1,...)"""
  14. index = 0
  15. for char in col.upper():
  16. if not 'A' <= char <= 'Z':
  17. raise ValueError(f"无效的列字母: {col}")
  18. index = index * 26 + (ord(char) - ord('A') + 1)
  19. return index - 1
  20. def read_excel_file(self, file_path: str) -> pd.DataFrame:
  21. """读取Excel文件并返回DataFrame"""
  22. try:
  23. df = pd.read_excel(file_path)
  24. logger.info(f"成功读取文件: {file_path}")
  25. return df
  26. except Exception as e:
  27. logger.error(f"读取文件失败: {file_path}, 错误: {str(e)}")
  28. raise
  29. def save_excel_file(self, df: pd.DataFrame, output_path: str) -> None:
  30. """将DataFrame保存为Excel文件"""
  31. try:
  32. df.to_excel(output_path, index=False)
  33. logger.info(f"成功保存文件: {output_path}")
  34. except Exception as e:
  35. logger.error(f"保存文件失败: {output_path}, 错误: {str(e)}")
  36. raise
  37. def insert_column(self, df: pd.DataFrame, ref_column: str, new_column_name: str, position: str = 'right') -> pd.DataFrame:
  38. """
  39. 在指定列旁边插入空列
  40. Args:
  41. df (pd.DataFrame): 输入的DataFrame
  42. ref_column (str): 参考列名或Excel列字母(如'A', 'B')
  43. new_column_name (str): 新列名
  44. position (str): 插入位置,'left'或'right',默认为'right'
  45. Returns:
  46. pd.DataFrame: 包含新列的DataFrame
  47. """
  48. try:
  49. # 如果ref_column是字母,转换为列索引
  50. if ref_column.isalpha():
  51. col_index = self._column_letter_to_index(ref_column)
  52. ref_column = df.columns[col_index]
  53. # 获取参考列的位置
  54. ref_index = df.columns.get_loc(ref_column)
  55. # 计算插入位置
  56. insert_index = ref_index + 1 if position == 'right' else ref_index
  57. # 插入新列
  58. df.insert(insert_index, new_column_name, '')
  59. logger.info(f"成功在列 {ref_column} 的{position}插入新列 {new_column_name}")
  60. return df
  61. except Exception as e:
  62. logger.error(f"插入列失败: {str(e)}")
  63. raise
  64. if __name__ == '__main__':
  65. # 测试代码
  66. processor = ExcelProcessor()
  67. df = processor.read_excel_file('/home/mrh/code/excel_tool/temp/测试.xlsx')
  68. df = processor.insert_column(df, 'B', 'New Column')
  69. processor.save_excel_file(df, '/home/mrh/code/excel_tool/temp/测试_process.xlsx')