certflow.handlers.excel_handler module

Excel文件基础处理器模块

提供Excel文件的读取、DataFrame清洗和列校验重命名等基础操作. 支持自定义表头行、多表头行识别等功能.

v2.1.0: 新增 openpyxl 读写封装（load_workbook / write_excel）， 业务层应通过本 handler 间接操作 openpyxl，避免直连导致分散维护。

class certflow.handlers.excel_handler.ExcelHandler

基类：object

Excel文件处理器

提供Excel文件读取、数据清洗和列名校验的静态方法集合. 支持自定义表头行、多表头行识别等高级功能.

该处理器专注于Excel文件的基础操作，不涉及业务逻辑处理.

static read_excel(file_path, sheet_name=0, header_row=None, skiprows=None, usecols=None)

读取Excel文件

支持自定义表头行和数据起始行，提供灵活的数据读取选项.

参数:

• file_path (Path) -- Excel文件路径

• sheet_name (str | int) -- 工作表名称或索引，默认为0(第一个工作表)

• header_row (int | None) -- 表头所在行号(0-indexed)，如果指定则作为列名行

• skiprows (int | None) -- 跳过的行数，从文件开头算起

• usecols (str | list[int] | None) -- 指定读取的列，可以是列号列表或Excel列范围字符串如"A:D"

返回:

读取的DataFrame对象

返回类型:

pd.DataFrame

抛出:

Exception -- 当文件不存在、格式错误或读取失败时抛出异常

示例

>>> handler = ExcelHandler()
>>> # 读取第一个工作表，使用第一行作为表头
>>> df = handler.read_excel(Path("data.xlsx"))
>>>
>>> # 读取指定工作表，使用第三行作为表头
>>> df = handler.read_excel(
...     Path("data.xlsx"),
...     sheet_name="Sheet1",
...     header_row=2
... )

static read_excel_with_multi_header(file_path, sheet_name=0, header_rows=None, separator='_')

读取有多行表头的Excel文件

将多行表头合并为单行列名，适用于复杂表头结构的Excel文件.

参数:

• file_path (Path) -- Excel文件路径

• sheet_name (str | int) -- 工作表名称或索引，默认为0

• header_rows (list[int] | None) -- 作为表头的行号列表(0-indexed)，默认为[0, 1]

• separator (str) -- 多级表头连接符，默认使用下划线"_"

返回:

处理后的DataFrame，列名为合并后的字符串

返回类型:

pd.DataFrame

抛出:

Exception -- 当文件读取或表头合并失败时抛出异常

示例

>>> handler = ExcelHandler()
>>> # 合并前两行作为表头
>>> df = handler.read_excel_with_multi_header(
...     Path("data.xlsx"),
...     header_rows=[0, 1],
...     separator="_"
... )
>>>
>>> # 如果Excel表头结构为:
>>> # 第1行: 基本信息 | 基本信息 | 联系方式
>>> # 第2行: 姓名    | 年龄    | 电话
>>> # 合并后列名: 基本信息_姓名, 基本信息_年龄, 联系方式_电话

static detect_header_row(df_raw, keywords, max_rows=10)

自动检测表头行

根据关键词在数据中搜索，找到包含最多关键词的行作为表头.

参数:

• df_raw (DataFrame) -- 原始DataFrame(使用header=None读取)

• keywords (list[str]) -- 表头关键词列表，如["产品型号", "订单号", "数量"]

• max_rows (int) -- 搜索的最大行数，默认为10行

返回:

表头行索引(0-indexed)，如果未找到则返回None

返回类型:

Optional[int]

示例

>>> 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} 行")

static clean_dataframe(df)

清洗DataFrame

执行字符串去空格、删除全空行和重置索引等基础清洗操作.

参数:

df (DataFrame) -- 原始DataFrame对象

返回:

清洗后的DataFrame

返回类型:

pd.DataFrame

示例

>>> handler = ExcelHandler()
>>> df = pd.DataFrame({
...     "name": [" 张三 ", "李四"],
...     "age": [25, None]
... })
>>> cleaned = handler.clean_dataframe(df)
>>> print(cleaned["name"][0])
"张三"

static validate_columns(df, required_columns, use_aliases=True, alias_mapping=None)

校验并重命名列

检查DataFrame中是否包含所有必需列，并将列名按映射关系重命名. 支持列别名匹配，提高对不同Excel文件结构的兼容性.

参数:

• df (DataFrame) -- 原始DataFrame对象

• required_columns (dict[str, str]) -- 列名映射字典，键为原始列名，值为目标列名

• use_aliases (bool) -- 是否使用别名匹配，默认为True

• alias_mapping (dict[str, list[str]] | None) -- 别名映射字典，格式为{目标列名: [别名列表]}

返回:

重命名后的DataFrame

返回类型:

pd.DataFrame

抛出:

ValueError -- 当缺少必需列时抛出异常，并列出可用列

示例

>>> 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"]

static get_column_mapping_from_config(direct_mapping=None, alias_mapping=None, excel_columns=None)

根据配置和Excel实际列名生成列映射

结合直接映射和别名映射，自动匹配Excel列名到目标字段.

参数:

• direct_mapping (dict[str, str] | None) -- 直接映射字典，格式{Excel列名: 目标字段}

• alias_mapping (dict[str, list[str]] | None) -- 别名映射字典，格式{目标字段: [别名列表]}

• excel_columns (list[str] | None) -- Excel文件的实际列名列表

返回:

列映射字典，格式{原始列名: 目标列名}

返回类型:

Dict[str, str]

示例

>>> 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"}

static get_data_start_row(file_path, sheet_name=0, header_row=0, min_data_rows=1)

获取数据起始行号

自动检测数据起始行，跳过空行和说明行.

参数:

• file_path (Path) -- Excel文件路径

• sheet_name (str | int) -- 工作表名称或索引，默认为0

• header_row (int) -- 表头行号(0-indexed)

• min_data_rows (int) -- 最小数据行数，用于判断是否为有效数据行

返回:

数据起始行号(0-indexed)，如果检测失败则返回header_row + 1

返回类型:

int

示例

>>> 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}")

read_excel_with_styles(file_path, sheet_name=0)

读取Excel并返回数据和样式信息

使用ExcelStyler读取Excel文件，同时获取单元格数据和样式信息.

参数:

• file_path (Path) -- Excel文件路径

• sheet_name (str | int) -- 工作表名称或索引，默认为0

返回:

Tuple[pd.DataFrame, pd.DataFrame | None, pd.DataFrame | None,

pd.DataFrame | None, pd.DataFrame | None]: (df_data, 背景色, 字体色, 批注, 状态) 五元组，后四项可能为 None

返回类型:

tuple[DataFrame, DataFrame | None, DataFrame | None, DataFrame | None, DataFrame | None]

static load_workbook(file_path, data_only=True, read_only=False)

加载 Excel 工作簿（openpyxl 统一入口）

业务层应通过此方法间接操作 openpyxl，避免各模块直连 openpyxl.load_workbook 导致参数/错误处理分散。

参数:

• file_path (Path) -- Excel 文件路径

• data_only (bool) -- 仅读取公式计算结果（默认 True）

• read_only (bool) -- 只读模式，适合超大文件扫描（默认 False）

返回:

openpyxl.Workbook 对象

抛出:

• FileNotFoundError -- 文件不存在

• Exception -- openpyxl 原生异常透传

返回类型:

Workbook

示例

>>> handler = ExcelHandler()
>>> wb = handler.load_workbook(Path("template.xlsx"), data_only=False)
>>> ws = wb.active
>>> ws["A1"] = "新值"
>>> handler.write_excel(wb, Path("template.xlsx"))

static write_excel(wb, file_path, auto_close=True)

保存 openpyxl Workbook 到文件

参数:

• wb (Workbook) -- 已修改的 openpyxl Workbook 对象

• file_path (Path) -- 目标文件路径

• auto_close (bool) -- 保存后自动关闭工作簿（默认 True）

返回类型:

None

示例

>>> handler = ExcelHandler()
>>> wb = handler.load_workbook(Path("data.xlsx"))
>>> ...  # 修改 wb
>>> handler.write_excel(wb, Path("data.xlsx"))