Python Excel工具案例如何封装表格操作

wen python案例 2026-06-28 253

本文目录导读：

基础Excel操作封装类
高级功能封装
使用示例
批量处理工具
关键封装要点

我来为你展示几个实用的Python Excel操作封装案例，包括常见的读写、格式处理等功能。

基础Excel操作封装类

import openpyxl
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
from openpyxl.utils import get_column_letter
from pathlib import Path
from typing import List, Dict, Any, Optional
class ExcelOperator:
    """Excel操作工具类"""
    def __init__(self, file_path: str):
        self.file_path = Path(file_path)
        self.workbook = None
        self.sheet = None
    def create_workbook(self, sheet_name: str = "Sheet1"):
        """创建新工作簿"""
        self.workbook = openpyxl.Workbook()
        self.sheet = self.workbook.active
        self.sheet.title = sheet_name
        return self
    def load_workbook(self, sheet_name: Optional[str] = None):
        """加载现有工作簿"""
        self.workbook = openpyxl.load_workbook(self.file_path)
        if sheet_name:
            self.sheet = self.workbook[sheet_name]
        else:
            self.sheet = self.workbook.active
        return self
    def save(self, file_path: Optional[str] = None):
        """保存文件"""
        save_path = file_path or self.file_path
        self.workbook.save(save_path)
        return self
    def write_cell(self, row: int, col: int, value: Any):
        """写入单元格"""
        self.sheet.cell(row=row, column=col, value=value)
        return self
    def read_cell(self, row: int, col: int):
        """读取单元格"""
        return self.sheet.cell(row=row, column=col).value
    def write_row(self, row: int, data: List[Any], start_col: int = 1):
        """写入整行数据"""
        for col, value in enumerate(data, start=start_col):
            self.sheet.cell(row=row, column=col, value=value)
        return self
    def write_column(self, col: int, data: List[Any], start_row: int = 1):
        """写入整列数据"""
        for row, value in enumerate(data, start=start_row):
            self.sheet.cell(row=row, column=col, value=value)
        return self
    def get_all_data(self, include_header: bool = True) -> List[List[Any]]:
        """获取所有数据"""
        data = []
        for row in self.sheet.iter_rows(values_only=True):
            data.append(list(row))
        if not include_header and data:
            data = data[1:]
        return data
    def set_header_style(self, row: int = 1):
        """设置表头样式"""
        header_font = Font(bold=True, color="FFFFFF", size=11)
        header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
        header_alignment = Alignment(horizontal="center", vertical="center")
        for cell in self.sheet[row]:
            cell.font = header_font
            cell.fill = header_fill
            cell.alignment = header_alignment
        return self

高级功能封装

from datetime import datetime
import pandas as pd
class AdvancedExcel(ExcelOperator):
    """高级Excel操作类"""
    def auto_fit_columns(self):
        """自动调整列宽"""
        for column in self.sheet.columns:
            max_length = 0
            column_letter = get_column_letter(column[0].column)
            for cell in column:
                try:
                    if len(str(cell.value)) > max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            adjusted_width = min(max_length + 2, 50)
            self.sheet.column_dimensions[column_letter].width = adjusted_width
        return self
    def add_border(self, start_row: int = 1, end_row: Optional[int] = None):
        """添加边框"""
        if not end_row:
            end_row = self.sheet.max_row
        thin_border = Border(
            left=Side(style='thin'),
            right=Side(style='thin'),
            top=Side(style='thin'),
            bottom=Side(style='thin')
        )
        for row in self.sheet.iter_rows(min_row=start_row, max_row=end_row):
            for cell in row:
                cell.border = thin_border
        return self
    def insert_dataframe(self, df: pd.DataFrame, start_row: int = 1, 
                        start_col: int = 1, include_header: bool = True):
        """插入DataFrame数据"""
        # 写入表头
        if include_header:
            for col_idx, col_name in enumerate(df.columns, start=start_col):
                self.sheet.cell(row=start_row, column=col_idx, value=col_name)
            start_row += 1
        # 写入数据
        for row_idx, row_data in enumerate(df.values, start=start_row):
            for col_idx, value in enumerate(row_data, start=start_col):
                if pd.isna(value):
                    value = None
                self.sheet.cell(row=row_idx, column=col_idx, value=value)
        return self
    def export_to_dataframe(self, header_row: int = 1) -> pd.DataFrame:
        """导出为DataFrame"""
        data = self.get_all_data()
        if not data:
            return pd.DataFrame()
        headers = data[header_row - 1]
        data_rows = data[header_row:]
        return pd.DataFrame(data_rows, columns=headers)
class ExcelReport(AdvancedExcel):
    """报表生成类"""
    def create_report_from_template(self, template_path: str, data: Dict):
        """从模板创建报表"""
        # 加载模板
        self.load_workbook(template_path)
        # 替换数据
        for cell_ref, value in data.items():
            if ':' in cell_ref:  # 范围
                cell_range = self.sheet[cell_ref]
                for row in cell_range:
                    for cell in row:
                        if cell.value and '{' in str(cell.value):
                            try:
                                cell.value = cell.value.format(**value if isinstance(value, dict) else {'value': value})
                            except:
                                pass
            else:  # 单个单元格
                cell = self.sheet[cell_ref]
                if isinstance(value, dict):
                    if cell.value and '{' in str(cell.value):
                        cell.value = cell.value.format(**value)
                else:
                    cell.value = value
        return self
    def add_summary_row(self, row: int, data: List, func: str = 'sum'):
        """添加汇总行"""
        for col, value in enumerate(data, start=1):
            if value == func and col > 1:  # 跳过第一列
                # 计算列范围
                col_letter = get_column_letter(col)
                formula = f"=SUM({col_letter}2:{col_letter}{row-1})"
                self.sheet.cell(row=row, column=col, value=formula)
            else:
                self.sheet.cell(row=row, column=col, value=value)
        return self

使用示例

def demo_excel_operations():
    """演示Excel操作"""
    # 1. 创建新Excel文件
    excel = ExcelOperator("output.xlsx")
    excel.create_workbook("销售数据")
    # 2. 写入表头和数据
    headers = ["产品", "销量", "单价", "总金额"]
    excel.write_row(1, headers)
    excel.set_header_style()
    # 写入数据
    data = [
        ["产品A", 100, 50, 5000],
        ["产品B", 200, 30, 6000],
        ["产品C", 150, 45, 6750]
    ]
    for row_idx, row_data in enumerate(data, start=2):
        excel.write_row(row_idx, row_data)
    # 3. 添加合计行
    excel.write_cell(5, 1, "合计")
    excel.write_cell(5, 2, "=SUM(B2:B4)")
    excel.write_cell(5, 4, "=SUM(D2:D4)")
    # 4. 保存文件
    excel.save()
def demo_advanced_operations():
    """演示高级操作"""
    # 使用高级功能
    excel = AdvancedExcel("report.xlsx")
    excel.create_workbook("月度报表")
    # 创建DataFrame数据
    import pandas as pd
    df = pd.DataFrame({
        '日期': pd.date_range('2024-01-01', periods=5, freq='D'),
        '销售额': [1000, 1500, 1200, 1800, 2000],
        '成本': [600, 800, 700, 1000, 1100]
    })
    df['利润'] = df['销售额'] - df['成本']
    # 插入DataFrame
    excel.insert_dataframe(df)
    # 自动调整列宽
    excel.auto_fit_columns()
    # 添加边框
    excel.add_border()
    # 保存
    excel.save()
def demo_report_creation():
    """演示报表创建"""
    # 创建报表
    report = ExcelReport("template_report.xlsx")
    # 准备模板数据
    template_data = {
        'A1': {'company': 'ABC科技有限公司', 'date': datetime.now().strftime('%Y-%m-%d')},
        'A3': '报表标题: 月度销售分析',
        'B3': '2024年1月',
        'A5': '部门名称: 销售部',
        'B5': '负责人: 张三'
    }
    # 创建报表
    report.create_report_from_template("template.xlsx", template_data)
    # 添加实际数据
    sales_data = [
        ["产品A", 100, 50, 5000],
        ["产品B", 200, 30, 6000],
        ["产品C", 150, 45, 6750]
    ]
    for row_idx, row_data in enumerate(sales_data, start=7):
        report.write_row(row_idx, row_data)
    # 添加汇总
    report.add_summary_row(11, ["总计", "sum", "", "sum"])
    # 保存为新文件
    report.save("completed_report.xlsx")
if __name__ == "__main__":
    demo_excel_operations()
    demo_advanced_operations()
    demo_report_creation()

批量处理工具

import os
from concurrent.futures import ThreadPoolExecutor
class BatchExcelProcessor:
    """批量Excel处理类"""
    def __init__(self, operations: List[Dict]):
        self.operations = operations
    def process_file(self, file_path: str, output_dir: str = None):
        """处理单个文件"""
        excel = ExcelOperator(file_path)
        excel.load_workbook()
        for operation in self.operations:
            method = getattr(excel, operation['method'])
            result = method(**operation.get('params', {}))
        if output_dir:
            output_path = os.path.join(output_dir, os.path.basename(file_path))
        else:
            output_path = file_path
        excel.save(f"processed_{output_path}")
        return output_path
    def batch_process(self, files: List[str], output_dir: str = None, 
                     max_workers: int = 4):
        """批量处理文件"""
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [
                executor.submit(self.process_file, file, output_dir)
                for file in files
            ]
            results = []
            for future in futures:
                try:
                    result = future.result()
                    results.append(result)
                    print(f"处理完成: {result}")
                except Exception as e:
                    print(f"处理失败: {e}")
        return results
# 批量处理使用示例
if __name__ == "__main__":
    # 定义操作列表
    operations = [
        {'method': 'set_header_style', 'params': {'row': 1}},
        {'method': 'auto_fit_columns'},
        {'method': 'add_border'}
    ]
    # 获取所有Excel文件
    files = [f for f in os.listdir('.') if f.endswith('.xlsx')]
    # 批量处理
    processor = BatchExcelProcessor(operations)
    results = processor.batch_process(files, 'processed', max_workers=2)