64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
# @Time : 2025/3/15 15:22
|
||
# @Author : AngesZhu
|
||
# @File : data_file_utils.py
|
||
# @Desc : excel表格相关操作
|
||
import pandas as pd
|
||
from utils.logger_utils import logger
|
||
|
||
|
||
class ExcelHandler:
|
||
def __init__(self, file_path: str):
|
||
"""
|
||
初始化 ExcelHandler 类。
|
||
:param file_path: Excel 文件路径
|
||
"""
|
||
self.file_path = file_path
|
||
|
||
def read_excel(self, sheet_name=0, header=0, usecols=None):
|
||
"""
|
||
读取 Excel 文件。
|
||
:param sheet_name: 需要读取的 sheet 名称或索引(默认第一个 sheet)
|
||
:param header: 作为列名的行号(默认第一行)
|
||
:param usecols: 指定需要读取的列(默认读取所有列)
|
||
:return: pandas DataFrame
|
||
"""
|
||
try:
|
||
df = pd.read_excel(self.file_path, sheet_name=sheet_name, header=header, usecols=usecols)
|
||
# print(df)
|
||
return df.fillna("")
|
||
except Exception as e:
|
||
logger.error(f"读取 Excel 文件失败: {e}")
|
||
return None
|
||
|
||
def write_excel(self, data: pd.DataFrame, sheet_name='Sheet1', mode='w', index=False):
|
||
"""
|
||
写入 Excel 文件。
|
||
:param data: 需要写入的 DataFrame
|
||
:param sheet_name: 目标 sheet 名称(默认 'Sheet1')
|
||
:param mode: 写入模式('w' 覆盖写入,'a' 追加写入)
|
||
:param index: 是否写入索引(默认 False)
|
||
"""
|
||
try:
|
||
with pd.ExcelWriter(self.file_path, engine='openpyxl', mode=mode) as writer:
|
||
data.to_excel(writer, sheet_name=sheet_name, index=index)
|
||
logger.info("Excel 文件写入成功。")
|
||
except Exception as e:
|
||
logger.error(f"写入 Excel 文件失败: {e}")
|
||
|
||
def append_to_excel(self, data: pd.DataFrame, sheet_name='Sheet1', index=False):
|
||
"""
|
||
追加数据到 Excel 文件的指定 sheet。
|
||
:param data: 需要追加的 DataFrame
|
||
:param sheet_name: 目标 sheet 名称(默认 'Sheet1')
|
||
:param index: 是否写入索引(默认 False)
|
||
"""
|
||
try:
|
||
with pd.ExcelWriter(self.file_path, engine='openpyxl', mode='a', if_sheet_exists='overlay') as writer:
|
||
startrow = writer.sheets[sheet_name].max_row if sheet_name in writer.sheets else 0
|
||
data.to_excel(writer, sheet_name=sheet_name, startrow=startrow, index=index, header=(startrow == 0))
|
||
logger.info("数据成功追加到 Excel 文件。")
|
||
except Exception as e:
|
||
logger.error(f"追加 Excel 文件失败: {e}")
|