data_factory/utils/data_file_utils.py

64 lines
2.6 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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}")