"""Excel文件基础处理器模块
提供Excel文件的读取、DataFrame清洗和列校验重命名等基础操作.
支持自定义表头行、多表头行识别等功能.
v2.1.0: 新增 openpyxl 读写封装（``load_workbook`` / ``write_excel``），
业务层应通过本 handler 间接操作 openpyxl，避免直连导致分散维护。
"""
from __future__ import annotations
import contextlib
from pathlib import Path
import pandas as pd
from loguru import logger
from openpyxl import Workbook
from openpyxl import load_workbook as _openpyxl_load_workbook
from certflow.handlers.excel_styler import ExcelStyler
[文档]
class ExcelHandler:
"""Excel文件处理器
提供Excel文件读取、数据清洗和列名校验的静态方法集合.
支持自定义表头行、多表头行识别等高级功能.
该处理器专注于Excel文件的基础操作，不涉及业务逻辑处理.
"""
[文档]
@staticmethod
def read_excel(
file_path: Path,
sheet_name: str | int = 0,
header_row: int | None = None,
skiprows: int | None = None,
usecols: str | list[int] | None = None,
) -> pd.DataFrame:
"""读取Excel文件
支持自定义表头行和数据起始行，提供灵活的数据读取选项.
Args:
file_path: Excel文件路径
sheet_name: 工作表名称或索引，默认为0(第一个工作表)
header_row: 表头所在行号(0-indexed)，如果指定则作为列名行
skiprows: 跳过的行数，从文件开头算起
usecols: 指定读取的列，可以是列号列表或Excel列范围字符串如"A:D"
Returns:
pd.DataFrame: 读取的DataFrame对象
Raises:
Exception: 当文件不存在、格式错误或读取失败时抛出异常
Examples:
>>> handler = ExcelHandler()
>>> # 读取第一个工作表，使用第一行作为表头
>>> df = handler.read_excel(Path("data.xlsx"))
>>>
>>> # 读取指定工作表，使用第三行作为表头
>>> df = handler.read_excel(
... Path("data.xlsx"),
... sheet_name="Sheet1",
... header_row=2
... )
"""
try:
# 构建读取参数
read_kwargs = {
"sheet_name": sheet_name,
"dtype": str,
}
if header_row is not None:
read_kwargs["header"] = header_row
else:
read_kwargs["header"] = 0 # 默认第一行为表头
if skiprows is not None:
read_kwargs["skiprows"] = skiprows
if usecols is not None:
read_kwargs["usecols"] = usecols
df = pd.read_excel(file_path, **read_kwargs)
logger.info(f"成功读取Excel: {file_path}, 行数: {len(df)}, 列数: {len(df.columns)}")
if header_row is not None:
logger.debug(f"使用第 {header_row + 1} 行作为表头")
return df
except Exception as e:
logger.error(f"读取Excel失败: {e}")
raise
[文档]
@staticmethod
def read_excel_with_multi_header(
file_path: Path,
sheet_name: str | int = 0,
header_rows: list[int] | None = None,
separator: str = "_",
) -> pd.DataFrame:
"""读取有多行表头的Excel文件
将多行表头合并为单行列名，适用于复杂表头结构的Excel文件.
Args:
file_path: Excel文件路径
sheet_name: 工作表名称或索引，默认为0
header_rows: 作为表头的行号列表(0-indexed)，默认为[0, 1]
separator: 多级表头连接符，默认使用下划线"_"
Returns:
pd.DataFrame: 处理后的DataFrame，列名为合并后的字符串
Raises:
Exception: 当文件读取或表头合并失败时抛出异常
Examples:
>>> handler = ExcelHandler()
>>> # 合并前两行作为表头
>>> df = handler.read_excel_with_multi_header(
... Path("data.xlsx"),
... header_rows=[0, 1],
... separator="_"
... )
>>>
>>> # 如果Excel表头结构为:
>>> # 第1行: 基本信息 | 基本信息 | 联系方式
>>> # 第2行: 姓名 | 年龄 | 电话
>>> # 合并后列名: 基本信息_姓名, 基本信息_年龄, 联系方式_电话
"""
if header_rows is None:
header_rows = [0, 1]
try:
# 读取指定行作为表头
headers = []
for row_idx in header_rows:
df_header = pd.read_excel(
file_path, sheet_name=sheet_name, header=None, nrows=row_idx + 1
)
headers.append(df_header.iloc[row_idx].fillna("").astype(str))
# 合并多级表头
merged_headers = []
for i in range(len(headers[0])):
parts = [h[i] for h in headers if h[i] and h[i] != "nan"]
if parts:
merged_headers.append(separator.join(parts))
else:
merged_headers.append(f"col_{i}")
# 重新读取数据，跳过表头行
skiprows = max(header_rows) + 1
df = pd.read_excel(
file_path,
sheet_name=sheet_name,
header=None,
skiprows=skiprows,
dtype=str,
)
# 设置合并后的列名
df.columns = merged_headers
logger.info(f"成功读取多级表头Excel: {file_path}, 行数: {len(df)}")
return df
except Exception as e:
logger.error(f"读取多级表头Excel失败: {e}")
raise
[文档]
@staticmethod
def detect_header_row(
df_raw: pd.DataFrame, keywords: list[str], max_rows: int = 10
) -> int | None:
"""自动检测表头行
根据关键词在数据中搜索，找到包含最多关键词的行作为表头.
Args:
df_raw: 原始DataFrame(使用header=None读取)
keywords: 表头关键词列表，如["产品型号", "订单号", "数量"]
max_rows: 搜索的最大行数，默认为10行
Returns:
Optional[int]: 表头行索引(0-indexed)，如果未找到则返回None
Examples:
>>> handler = ExcelHandler()
>>> # 读取原始数据(不指定表头)
>>> df_raw = pd.read_excel("data.xlsx", header=None)
>>> # 检测表头行
>>> header_row = handler.detect_header_row(
... df_raw,
... keywords=["产品型号", "订单号", "数量"],
... max_rows=10
... )
>>> if header_row is not None:
... print(f"表头在第 {header_row + 1} 行")
"""
for i in range(min(max_rows, len(df_raw))):
try:
# 安全地转换每一行为字符串
row_values = []
for val in df_raw.iloc[i]:
if pd.isna(val):
row_values.append("")
else:
row_values.append(str(val))
row_str = " ".join(row_values)
except Exception as e:
logger.debug(f"第 {i + 1} 行转换失败: {e}")
continue
# 检查是否包含关键词
matches = sum(1 for kw in keywords if kw in row_str)
if matches >= len(keywords) // 2:
logger.info(f"检测到表头行: 第 {i + 1} 行, 匹配关键词数: {matches}")
return i
logger.warning(f"未检测到表头行，关键词: {keywords}")
return None
[文档]
@staticmethod
def clean_dataframe(df: pd.DataFrame) -> pd.DataFrame:
"""清洗DataFrame
执行字符串去空格、删除全空行和重置索引等基础清洗操作.
Args:
df: 原始DataFrame对象
Returns:
pd.DataFrame: 清洗后的DataFrame
Examples:
>>> handler = ExcelHandler()
>>> df = pd.DataFrame({
... "name": [" 张三 ", "李四"],
... "age": [25, None]
... })
>>> cleaned = handler.clean_dataframe(df)
>>> print(cleaned["name"][0])
"张三"
"""
# 去除字符串空格
df = df.map(lambda x: x.strip() if isinstance(x, str) else x)
# 删除全空行
before_rows = len(df)
df = df.dropna(how="all")
after_rows = len(df)
if before_rows != after_rows:
logger.debug(f"删除了 {before_rows - after_rows} 个全空行")
# 重置索引
return df.reset_index(drop=True)
[文档]
@staticmethod
def validate_columns(
df: pd.DataFrame,
required_columns: dict[str, str],
use_aliases: bool = True,
alias_mapping: dict[str, list[str]] | None = None,
) -> pd.DataFrame:
"""校验并重命名列
检查DataFrame中是否包含所有必需列，并将列名按映射关系重命名.
支持列别名匹配，提高对不同Excel文件结构的兼容性.
Args:
df: 原始DataFrame对象
required_columns: 列名映射字典，键为原始列名，值为目标列名
use_aliases: 是否使用别名匹配，默认为True
alias_mapping: 别名映射字典，格式为{目标列名: [别名列表]}
Returns:
pd.DataFrame: 重命名后的DataFrame
Raises:
ValueError: 当缺少必需列时抛出异常，并列出可用列
Examples:
>>> handler = ExcelHandler()
>>> df = pd.DataFrame({
... "产品名称": ["阀门A", "阀门B"],
... "数量": [5, 10]
... })
>>> required = {"产品名称": "product_name", "数量": "quantity"}
>>> renamed = handler.validate_columns(df, required)
>>> print(renamed.columns.tolist())
["product_name", "quantity"]
"""
# 记录原始列名
original_columns = list(df.columns)
if use_aliases and alias_mapping:
# 使用别名匹配
column_mapping = {}
missing_columns = []
for target_col, aliases in alias_mapping.items():
# 查找匹配的列
matched = None
for alias in aliases:
if alias in original_columns:
matched = alias
break
if matched:
column_mapping[matched] = target_col
else:
# 检查是否在 required_columns 中
if target_col in required_columns.values():
missing_columns.append(aliases[0] if aliases else target_col)
# 检查是否有缺失的必需列
if missing_columns:
available_cols = ", ".join(original_columns)
logger.error(f"可用的列: {available_cols}")
raise ValueError(f"缺少必需列: {', '.join(missing_columns)}")
# 执行重命名
df = df.rename(columns=column_mapping)
logger.info(
f"列映射成功: {list(column_mapping.keys())} → {list(column_mapping.values())}"
)
else:
# 使用直接映射
for col in required_columns:
if col not in df.columns:
# 提供友好的错误信息
available_cols = ", ".join(df.columns)
logger.error(f"可用的列: {available_cols}")
raise ValueError(f"缺少必需列: {col}")
# 重命名列
df = df.rename(columns=required_columns)
return df
[文档]
@staticmethod
def get_column_mapping_from_config(
direct_mapping: dict[str, str] | None = None,
alias_mapping: dict[str, list[str]] | None = None,
excel_columns: list[str] | None = None,
) -> dict[str, str]:
"""根据配置和Excel实际列名生成列映射
结合直接映射和别名映射，自动匹配Excel列名到目标字段.
Args:
direct_mapping: 直接映射字典，格式{Excel列名: 目标字段}
alias_mapping: 别名映射字典，格式{目标字段: [别名列表]}
excel_columns: Excel文件的实际列名列表
Returns:
Dict[str, str]: 列映射字典，格式{原始列名: 目标列名}
Examples:
>>> handler = ExcelHandler()
>>> direct = {"产品型号": "product_model"}
>>> aliases = {"quantity": ["数量", "数量(件)", "数量(台)"]}
>>> excel_cols = ["产品型号", "数量(件)"]
>>> mapping = handler.get_column_mapping_from_config(direct, aliases, excel_cols)
>>> print(mapping)
{"产品型号": "product_model", "数量(件)": "quantity"}
"""
column_mapping = {}
direct_mapping = direct_mapping or {}
alias_mapping = alias_mapping or {}
excel_columns = excel_columns or []
for excel_col in excel_columns:
# 先检查直接映射
if excel_col in direct_mapping:
column_mapping[excel_col] = direct_mapping[excel_col]
continue
# 使用别名查找
for target_col, aliases in alias_mapping.items():
if excel_col in aliases:
column_mapping[excel_col] = target_col
break
logger.debug(f"生成列映射: {column_mapping}")
return column_mapping
[文档]
@staticmethod
def get_data_start_row(
file_path: Path,
sheet_name: str | int = 0,
header_row: int = 0,
min_data_rows: int = 1,
) -> int:
"""获取数据起始行号
自动检测数据起始行，跳过空行和说明行.
Args:
file_path: Excel文件路径
sheet_name: 工作表名称或索引，默认为0
header_row: 表头行号(0-indexed)
min_data_rows: 最小数据行数，用于判断是否为有效数据行
Returns:
int: 数据起始行号(0-indexed)，如果检测失败则返回header_row + 1
Examples:
>>> handler = ExcelHandler()
>>> # 假设表头在第2行(索引1)，数据从第4行开始
>>> data_start = handler.get_data_start_row(
... Path("data.xlsx"),
... header_row=1,
... min_data_rows=2
... )
>>> print(f"数据起始行: {data_start}")
"""
try:
# 读取表头后的数据
df = pd.read_excel(file_path, sheet_name=sheet_name, header=header_row, dtype=str)
# 找到第一个非空行
for i in range(min(len(df), 100)): # 最多检查100行
row = df.iloc[i]
# 检查行是否有有效数据
non_empty = row.dropna().tolist()
if len(non_empty) >= min_data_rows:
logger.debug(f"数据起始行: {header_row + i + 1}")
return header_row + i
return header_row + 1
except Exception as e:
logger.warning(f"检测数据起始行失败: {e}")
return header_row + 1
[文档]
def read_excel_with_styles(
self, file_path: Path, sheet_name: str | int = 0
) -> tuple[
pd.DataFrame,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
pd.DataFrame | None,
]:
"""读取Excel并返回数据和样式信息
使用ExcelStyler读取Excel文件，同时获取单元格数据和样式信息.
Args:
file_path: Excel文件路径
sheet_name: 工作表名称或索引，默认为0
Returns:
Tuple[pd.DataFrame, pd.DataFrame | None, pd.DataFrame | None,
pd.DataFrame | None, pd.DataFrame | None]:
(df_data, 背景色, 字体色, 批注, 状态) 五元组，后四项可能为 None
"""
return ExcelStyler.read_with_styles(file_path, sheet_name)
# ============================================================
# openpyxl 读写封装（v2.1.0 统一入口）
# ============================================================
[文档]
@staticmethod
def load_workbook(
file_path: Path,
data_only: bool = True,
read_only: bool = False,
) -> Workbook:
"""加载 Excel 工作簿（openpyxl 统一入口）
业务层应通过此方法间接操作 openpyxl，避免各模块直连
``openpyxl.load_workbook`` 导致参数/错误处理分散。
Args:
file_path: Excel 文件路径
data_only: 仅读取公式计算结果（默认 True）
read_only: 只读模式，适合超大文件扫描（默认 False）
Returns:
openpyxl.Workbook 对象
Raises:
FileNotFoundError: 文件不存在
Exception: openpyxl 原生异常透传
Examples:
>>> handler = ExcelHandler()
>>> wb = handler.load_workbook(Path("template.xlsx"), data_only=False)
>>> ws = wb.active
>>> ws["A1"] = "新值"
>>> handler.write_excel(wb, Path("template.xlsx"))
"""
if not file_path.exists():
raise FileNotFoundError(f"Excel 文件不存在: {file_path}")
try:
wb = _openpyxl_load_workbook(str(file_path), data_only=data_only, read_only=read_only)
logger.debug(f"加载工作簿成功: {file_path} (data_only={data_only})")
return wb
except Exception as e:
logger.error(f"加载工作簿失败 [{file_path}]: {e}")
raise
[文档]
@staticmethod
def write_excel(
wb: Workbook,
file_path: Path,
auto_close: bool = True,
) -> None:
"""保存 openpyxl Workbook 到文件
Args:
wb: 已修改的 openpyxl Workbook 对象
file_path: 目标文件路径
auto_close: 保存后自动关闭工作簿（默认 True）
Examples:
>>> handler = ExcelHandler()
>>> wb = handler.load_workbook(Path("data.xlsx"))
>>> ... # 修改 wb
>>> handler.write_excel(wb, Path("data.xlsx"))
"""
try:
wb.save(str(file_path))
logger.info(f"工作簿已保存: {file_path}")
except Exception as e:
logger.error(f"保存工作簿失败 [{file_path}]: {e}")
raise
finally:
if auto_close:
with contextlib.suppress(Exception):
wb.close()