certflow.handlers package¶

Submodules¶

Module contents¶

基础处理器模块

提供数据处理、Excel操作、编号生成、报告生成和排序等基础功能。该模块是CertFlow系统的核心工具集，封装了各种通用的处理器组件。

主要功能: - 数据清洗: 销售计划数据的清洗和校验 - Excel处理: Excel文件的读取、样式读取和格式处理 - 编号生成: 合格证编号和唯一键生成 - 报告生成: PDF格式的试压报告、材质报告和质保书生成 - 排序分组: 多字段组合排序和数据分组

分层约束:: 本包为纯数据处理层，不得依赖 PySide6，须保证在无 Qt （headless / 无 X11）环境下可完整导入。需要向界面推送变更时，使用普通回调注册（如 TemplateManager.on_template_changed），由上层 View 桥接为 Qt 信号。UI 基础设施请放置于 certflow.views.common。
Exports:: DataCleaner: 数据清洗器，提供销售计划数据的清洗和校验功能 ExcelHandler: Excel文件处理器，提供文件读取、列校验和重命名功能 ExcelStyler: Excel样式处理器，提供单元格颜色、字体、批注等信息读取 IDGenerator: 编号生成器，提供合格证编号和唯一键生成功能 ReportGenerator: 报告生成器，提供PDF格式报告生成功能 Sorter: 排序器，提供多字段组合排序和分组功能

示例

>>> from certflow.handlers import DataCleaner, ExcelHandler, IDGenerator
>>>
>>> # 读取Excel文件
>>> df = ExcelHandler.read_excel("data.xlsx", header_row=0)
>>>
>>> # 清洗数据
>>> cleaned_df = DataCleaner.clean_sale_plan(df)
>>>
>>> # 生成唯一键
>>> unique_key = IDGenerator.generate_unique_key(data, ["contract_no", "product_model"])

class certflow.handlers.DataCleaner[源代码]¶

基类：object

数据清洗器

提供销售计划数据的清洗和校验功能. 负责处理原始导入数据中的格式问题、空值处理和字段规范化.

DATE_FIELDS¶: 日期字段列表，定义需要规范化的日期字段名称

DATE_FIELDS = ['plan_date', 'planned_delivery_date', 'contract_delivery_date']¶

static clean_sale_plan(df)[源代码]¶

清洗销售计划数据

对原始导入的 DataFrame 执行标准化清洗：去除字符串首尾空格、规范化日期字段、清洗订单号特殊字符、数值/文本字段规范化，并对销售计划日期填充当月默认值。

参数:: df (DataFrame) -- 原始销售计划 DataFrame，列应为字符串或数值类型
返回:: 清洗后的 DataFrame（字段已规范化，空值填充为【空白】标记）
返回类型:: pd.DataFrame

示例

>>> import pandas as pd
>>> df = pd.DataFrame({
...     "product_model": [" 阀门A ", ""],
...     "order_no": ["PO-001", "PO/002"],
...     "quantity": ["10", "5"],
... })
>>> cleaned = DataCleaner.clean_sale_plan(df)
>>> cleaned["product_model"][0]
"阀门A"

static validate_data(row)[源代码]¶

校验单条数据

检查数据行中产品型号是否为空、数量是否大于0.

参数:

row (dict[str, Any]) -- 单条数据字典，包含product_model和quantity等字段

返回:

校验结果元组

第一个元素: 是否通过校验，True表示通过，False表示不通过
第二个元素: 校验结果描述信息，通过时返回"校验通过"

返回类型:

Tuple[bool, str]

示例

>>> row = {"product_model": "阀门A", "quantity": 10}
>>> is_valid, message = DataCleaner.validate_data(row)
>>> print(is_valid)
True
>>>
>>> row = {"product_model": "", "quantity": 5}
>>> is_valid, message = DataCleaner.validate_data(row)
>>> print(message)
"产品型号不能为空"

class certflow.handlers.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"))

class certflow.handlers.ExcelStyler[源代码]¶

基类：object

Excel 样式处理器

负责读取 Excel 单元格的格式信息： - 背景色 - 字体色 - 批注内容 - 其他样式属性

classmethod get_status_from_color(color, color_type='bg')[源代码]¶

根据颜色获取对应的业务状态（配置驱动）

参数:

color (str) -- 原始颜色值（如 "FF0000"、"00B050" 或 "THEME_1"），可为空
color_type (str) -- 颜色类型，"font" 表示字体色，"bg" 表示背景色

返回:

对应的业务状态文本（如 "已开票"、"已发货"），无匹配时返回 None

返回类型:

str | None

static open_workbook(file_path, data_only=True)[源代码]¶

打开工作簿的上下文管理器

使用上下文管理器自动管理Excel工作簿的打开和关闭，确保资源正确释放。

参数:

file_path (Path) -- Excel文件路径
data_only (bool) -- 是否只读取数据值(不读取公式)，默认为True

生成器:

Workbook -- openpyxl工作簿对象

返回类型:

Iterator[Workbook]

示例

>>> with ExcelStyler.open_workbook(Path("data.xlsx")) as wb:
...     ws = wb.active
...     cell = ws["A1"]
...     print(cell.value)

static get_cell_color(cell, color_type='bg')[源代码]¶

获取单元格颜色

支持获取背景色和字体色，返回标准化的颜色值。

参数:

cell (Cell)
color_type (str)

返回类型:

str | None

static get_cell_comment(cell)[源代码]¶

获取单元格批注内容

参数:: cell (Cell) -- openpyxl单元格对象
返回:: 批注文本内容，没有批注时返回None
返回类型:: Optional[str]

示例

>>> with ExcelStyler.open_workbook(Path("data.xlsx")) as wb:
...     ws = wb.active
...     comment = ExcelStyler.get_cell_comment(ws["A1"])
...     if comment:
...         print(f"批注内容: {comment}")

static get_cell_style_info(cell)[源代码]¶

获取单元格完整样式信息

一次性获取单元格的值、背景色、字体色、状态、批注和字体属性。

参数:

cell (Cell) -- openpyxl单元格对象

返回:

包含以下键的字典:

value: 单元格值
bg_color: 背景色
font_color: 字体色
status: 根据颜色推断的业务状态
comment: 批注内容
font_name: 字体名称
font_size: 字体大小
bold: 是否粗体
italic: 是否斜体

返回类型:

Dict[str, Any]

示例

>>> with ExcelStyler.open_workbook(Path("data.xlsx")) as wb:
...     ws = wb.active
...     info = ExcelStyler.get_cell_style_info(ws["A1"])
...     print(f"值: {info['value']}, 状态: {info['status']}")

static read_with_styles(file_path, sheet_name=0, include_styles=True, include_comments=True)[源代码]¶

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

同时读取单元格的值和样式信息，返回多个DataFrame。

参数:

file_path (Path) -- Excel文件路径
sheet_name (str | int) -- 工作表名称或索引，默认为0(第一个工作表)
include_styles (bool) -- 是否包含样式信息(背景色、字体色)，默认为True
include_comments (bool) -- 是否包含批注信息，默认为True

返回:

包含5个元素的元组:

df_data: 数据值DataFrame
bg_colors: 背景色DataFrame(可选)
font_colors: 字体色DataFrame(可选)
comments: 批注DataFrame(可选)
statuses: 状态DataFrame(可选)

返回类型:

Tuple

示例

>>> data, bg, font, comments, status = ExcelStyler.read_with_styles(
...     Path("data.xlsx"),
...     sheet_name="Sheet1"
... )
>>> print(f"数据形状: {data.shape}")
>>> print(f"状态统计: {status.iloc[0, 0]}")

static get_comments_summary(file_path)[源代码]¶

获取批注汇总表

扫描整个工作簿，返回所有批注的结构化摘要。

参数:

file_path (Path) -- Excel文件路径

返回:

包含以下列的DataFrame:

工作表: 批注所在工作表名称
单元格: 批注所在单元格坐标(如"A1")
批注内容: 批注的文本内容

返回类型:

pd.DataFrame

示例

>>> summary = ExcelStyler.get_comments_summary(Path("data.xlsx"))
>>> print(summary)
>>> # 输出示例:
>>> #   工作表   单元格        批注内容
>>> # 0  Sheet1   A1   请注意此单元格
>>> # 1  Sheet1   B5   需要审核

static get_all_comments(file_path)[源代码]¶

获取整个工作簿的所有批注

遍历所有工作表和单元格，收集所有批注信息。

参数:

file_path (Path) -- Excel文件路径

返回:

嵌套字典结构: 外层键为工作表名称，内层键为单元格坐标，值为批注内容

返回类型:

Dict[str, Dict[str, str]]

示例

>>> all_comments = ExcelStyler.get_all_comments(Path("data.xlsx"))
>>> for sheet, cells in all_comments.items():
...     print(f"工作表: {sheet}")
...     for cell, comment in cells.items():
...         print(f"  {cell}: {comment}")

static get_comment_at(file_path, sheet_name, row, col)[源代码]¶

获取指定单元格的批注内容

参数:

file_path (Path) -- Excel 文件路径
sheet_name (str | int) -- 工作表名称或索引
row (int) -- 行号（1 基）
col (int) -- 列号（1 基）

返回:

批注文本，无批注时返回 None

返回类型:

Optional[str]

static get_comments_in_column(file_path, sheet_name, column)[源代码]¶

获取指定列的所有批注

参数:

file_path (Path) -- Excel 文件路径
sheet_name (str | int) -- 工作表名称或索引
column (int) -- 列号（1 基）

返回:

键为行号（1 基），值为批注文本；工作表不存在时返回空字典。

返回类型:

Dict[int, str]

static find_comments_by_keyword(file_path, keyword)[源代码]¶

按关键词搜索批注

参数:

file_path (Path) -- Excel 文件路径
keyword (str) -- 搜索关键词

返回:

包含工作表/单元格/批注内容三列，仅保留批注内容中包含关键词的行。

返回类型:

pd.DataFrame

static read_with_comments_merged(file_path, sheet_name=0, comment_column=None)[源代码]¶

读取数据并将指定列的批注合并为新列

参数:

file_path (Path) -- Excel 文件路径
sheet_name (str | int) -- 工作表名称或索引，默认为 0
comment_column (int | None) -- 批注所在列号（1 基）；为 None 时汇总所有批注

返回:

数据 DataFrame，额外包含 comment 列

返回类型:

pd.DataFrame

static get_cell_border(cell)[源代码]¶

获取单元格四边边框信息

参数:: cell (Cell) -- openpyxl 单元格对象
返回:: 包含 top/bottom/left/right 的字典，每条边为 {"style": ..., "color": ...}；无任何边框时返回 None
返回类型:: Optional[Dict]

static set_cell_border(cell, style='thin', color='000000')[源代码]¶

为单元格四边设置统一边框

参数:

cell (Cell) -- openpyxl 单元格对象
style (str) -- 边框样式（如 thin/medium/thick）
color (str) -- 边框颜色（ARGB 或 RGB 十六进制）

返回类型:

None

static add_border_to_range(ws, min_row, min_col, max_row, max_col, style='thin', color='000000')[源代码]¶

为矩形区域的所有单元格添加统一边框

参数:

ws (Any) -- openpyxl 工作表对象
min_row (int) -- 起始行（1 基）
min_col (int) -- 起始列（1 基）
max_row (int) -- 结束行（1 基）
max_col (int) -- 结束列（1 基）
style (str) -- 边框样式
color (str) -- 边框颜色

返回类型:

None

static get_range_borders(ws, min_row, min_col, max_row, max_col)[源代码]¶

获取矩形区域内所有单元格的边框信息

参数:

ws (Any) -- openpyxl 工作表对象
min_row (int) -- 起始行（1 基）
min_col (int) -- 起始列（1 基）
max_row (int) -- 结束行（1 基）
max_col (int) -- 结束列（1 基）

返回:

键为单元格坐标（如 "A1"），值为 get_cell_border 返回的四边边框信息（含 None 边）。

返回类型:

Dict[str, Dict]

static get_cell_hyperlink(cell)[源代码]¶

获取单元格超链接目标地址

参数:: cell (Cell) -- openpyxl 单元格对象
返回:: 超链接地址，无超链接时返回 None
返回类型:: Optional[str]

static get_style_dataframe(file_path, sheet_name=0)[源代码]¶

读取并返回单元格样式 DataFrame

逐行读取背景色、字体色与批注，组成与数据等形的 DataFrame。

参数:

file_path (Path) -- Excel 文件路径
sheet_name (str | int) -- 工作表名称或索引，默认为 0

返回:

包含 bg_color / font_color / comment 等列的样式表

返回类型:

pd.DataFrame

class certflow.handlers.IDGenerator[源代码]¶

基类：object

分层唯一键生成器（配置驱动）

提供合格证编号生成、基于数据内容的唯一键生成及批量生成功能。唯一键采用分层策略，通过配置文件 id_generator 节点驱动字段选择。

分层策略:

新数据模式：当配置的 new_data_fields 字段全部有值时使用（默认: sales_order_no + production_order_no）
老数据模式：当 new_data_fields 任一为空时降级使用（默认: 9个内容字段）

配置节点:

id_generator.new_data_fields: 新数据稳定字段列表 id_generator.old_data_fields: 老数据内容匹配字段列表 id_generator.empty_marker_prefix: 空值标记前缀（默认 "【空白"）

向后兼容:

generate_unique_key(data, keys) 指定 keys 参数时走旧逻辑

主要功能:

generate_certificate_no(): 生成合格证编号
generate_unique_key(): 生成MD5唯一键（支持分层策略）
generate_unique_key_readable(): 生成可读唯一键（调试用）
generate_batch_unique_keys(): 批量生成唯一键
get_all_unique_key_fields(): 获取所有唯一键字段（供去重配置使用）
normalize_empty_values(): 空值规范化
normalize_dataframe_empty_values(): DataFrame空值规范化

static generate_certificate_no(prefix='CERT', sequence=None)[源代码]¶

生成合格证编号

生成格式为 CERT-YYYYMMDD-NNNN 的合格证编号，序号部分自动补零至4位。

参数:

prefix (str) -- 编号前缀，默认为"CERT"
sequence (int | None) -- 序号，如果为None则默认为1

返回:

生成的合格证编号字符串

返回类型:

str

示例

>>> # 生成默认格式的编号
>>> cert_no = IDGenerator.generate_certificate_no()
>>> print(cert_no)  # 输出: CERT-20231201-0001
>>>
>>> # 生成自定义前缀和序号
>>> cert_no = IDGenerator.generate_certificate_no(prefix="QT", sequence=5)
>>> print(cert_no)  # 输出: QT-20231201-0005

classmethod generate_unique_key(data, keys=None)[源代码]¶

生成唯一键（分层策略；身份键 = MD5 哈希）

参数:

data (dict[str, Any]) -- 数据字典，包含需要生成唯一键的字段
keys (list[str] | None) -- 指定字段列表。有值时走旧逻辑（向后兼容，测试用）

返回:

16位MD5哈希值

返回类型:

str

示例

>>> # 分层策略（有生产令号时）
>>> data = {"sales_order_no": "SO001", "production_order_no": "PO001",
...         "product_model": "阀门A", "product_spec": "DN50"}
>>> key = IDGenerator.generate_unique_key(data)
>>> len(key)
16
>>>
>>> # 向后兼容：指定 keys 走旧逻辑
>>> key2 = IDGenerator.generate_unique_key(data, keys=["contract_no", "product_model"])
>>> len(key2)
16

classmethod generate_unique_key_readable(data, keys=None)[源代码]¶

生成可读唯一键（分层策略，调试用；字段间 :: 分隔）

参数:

data (dict[str, Any])
keys (list[str] | None)

返回类型:

str

classmethod generate_identity_key(data)[源代码]¶

生成可读身份键串（§9 落地待办②：身份键显式化，用于追溯/人工核对）

与 generate_unique_key 选用同一分层与字段，仅返回人工可读的拼接串（不哈希），作为 unique_key 的人类可读形式。

参数:: data (dict[str, Any])
返回类型:: str

classmethod generate_change_signature(data)[源代码]¶

生成变更签名（§9 落地待办②：内容摘要，用于更新 vs 跳过快速判定）

对受监控字段（import_deduplication.monitored_fields）做拼接+MD5，作为该行"内容指纹"。身份键命中已存在行时，若签名一致则判定无变更、跳过逐字段 diff（save_handler._handle_existing 使用），否则跑全量 diff。

参数:: data (dict[str, Any])
返回类型:: str

classmethod normalize_spec(record, session=None)[源代码]¶

导入前口径规范化（R1 总前置）。

调 DNService.resolve_caliber 解析 product_spec，将合法数字口径写回 record["product_spec_norm"]；解析失败（含 φ50 伪归一）保留原值并标记 record["spec_needs_manual"]=True，供 R4 标黄。

参数:

record (dict[str, Any]) -- 记录字典（含 product_spec）
session (Session | None) -- SQLAlchemy 会话（DB 映射查询依赖；None 时退化为纯文本解析）

返回:

同一 record（就地补充 product_spec_norm / spec_needs_manual）

返回类型:

dict

classmethod normalize_pn(record, session=None)[源代码]¶

导入前公称压力/标准号规范化（#30 P0 DN/PN 字典 DB 化）。

调 PNService.resolve_pn 解析 product_model，将公称压力写回 record["pressure_value"]；从型号压力转换表(db) 补全省录 record["test_standard"]``（标准号导入期默认空、由合格证打印时回填，不计入黄标）；黄标仅由**公称压力无法解析**驱动——``pn_needs_manual 在 pressure_value 为空时置 True，对标 VBA 的 Stop 硬断点，改为**非中断黄标**（更优：批量导入不被单点卡死）。

参数:

record (dict[str, Any]) -- 记录字典（含 product_model / 可选 pressure / test_standard）
session (Session | None) -- SQLAlchemy 会话（DB 映射查询依赖；None 时退化为纯文本解析）

返回:

同一 record（就地补充 pressure_value / test_standard / pn_needs_manual）

返回类型:

dict

static generate_batch_unique_keys(data_list, keys=None)[源代码]¶

批量生成唯一键

为数据列表中的每条记录生成唯一键，并将键值写入"unique_key"字段。

参数:

data_list (list[dict[str, Any]]) -- 数据字典列表
keys (list[str] | None) -- 用于生成唯一键的字段列表，如果为None则使用所有字段

返回:

添加了unique_key字段的数据列表

返回类型:

List[Dict[str, Any]]

示例

>>> data_list = [
...     {"contract_no": "PO-001", "product_model": "阀门A"},
...     {"contract_no": "PO-002", "product_model": "阀门B"}
... ]
>>> result = IDGenerator.generate_batch_unique_keys(data_list, ["contract_no"])
>>> for item in result:
...     print(item["unique_key"])

classmethod normalize_empty_values(data, keys=None)[源代码]¶

规范化字典中的空值

将 None 或空字符串转换为【空白中文字段名】格式。用于统一空值表示，便于后续处理和识别。

参数:

data (dict[str, Any]) -- 原始数据字典
keys (list[str] | None) -- 需要处理的字段列表，如果为None则处理所有字段

返回:

空值规范化后的字典

返回类型:

Dict[str, Any]

示例

>>> data = {"contract_no": "", "product_model": "阀门A", "quantity": None}
>>> normalized = IDGenerator.normalize_empty_values(data, ["contract_no", "quantity"])
>>> print(normalized["contract_no"])  # 输出: "【空白合同号】"
>>> print(normalized["quantity"])    # 输出: "【空白数量】"

classmethod normalize_dataframe_empty_values(df, fields)[源代码]¶

规范化DataFrame中指定字段的空值

将DataFrame中指定字段的空值(NaN、None、空字符串)统一替换为格式化的空值标记。适用于批量数据处理场景。

参数:

df (DataFrame) -- 需要处理的DataFrame
fields (list[str]) -- 需要处理的字段列表

返回:

空值规范化后的DataFrame副本

返回类型:

pd.DataFrame

示例

>>> import pandas as pd
>>> df = pd.DataFrame({
...     "contract_no": ["PO-001", "", None],
...     "product_model": ["阀门A", "阀门B", "阀门C"]
... })
>>> cleaned = IDGenerator.normalize_dataframe_empty_values(df, ["contract_no"])
>>> print(cleaned["contract_no"][1])  # 输出: "【空白合同号】"

classmethod get_all_unique_key_fields()[源代码]¶

获取所有唯一键字段（去重合并，供 import_deduplication 使用）

返回:: 去重后的唯一键字段列表
返回类型:: list[str]

class certflow.handlers.ReportGenerator(output_dir='reports')[源代码]¶

基类：object

报告生成处理器 - 支持复杂模板和合并单元格

基于ReportLab生成PDF报告，支持试压报告、材质报告和质保书三种类型，提供中文字体注册、自定义样式、分页和签名栏等功能。

参数:: output_dir (str)

output_dir¶: 输出目录路径，用于存放生成的PDF文件

styles¶: 报告样式集合，包含自定义的中文段落样式

generate_pressure_test_report(data, template_name='pressure_test', language='zh', records_per_page=5)[源代码]¶

生成试压报告

根据测试数据生成PDF格式的试压报告，支持分页和签名栏。

参数:

data (list[dict[str, Any]]) -- 测试数据字典列表，每条记录包含试验压力、保压时间等字段
template_name (str) -- 模板名称，默认为"pressure_test"
language (str) -- 语言，可选"zh"(中文)或"en"(英文)，默认为"zh"
records_per_page (int) -- 每页记录数，可选5/6/9，默认为5

返回:

生成的PDF文件完整路径

返回类型:

str

示例

>>> generator = ReportGenerator("output/reports")
>>> test_data = [
...     {"test_pressure": "1.5", "hold_time": "60", "test_result": "合格"},
...     {"test_pressure": "2.0", "hold_time": "90", "test_result": "合格"}
... ]
>>> filepath = generator.generate_pressure_test_report(test_data)
>>> print(f"报告已生成: {filepath}")

generate_material_certificate(data, material_type='steel', language='zh', records_per_page=6)[源代码]¶

生成材质报告/质保书

根据材质数据生成PDF格式的材质证明书。

参数:

data (list[dict[str, Any]]) -- 材质数据字典列表，包含材质牌号、规格、炉号等字段
material_type (str) -- 材质类型，可选"steel"(钢材)/"valve"(阀门)/"pipe"(管材)，默认为"steel"
language (str) -- 语言，可选"zh"(中文)或"en"(英文)，默认为"zh"
records_per_page (int) -- 每页记录数，默认为6

返回:

生成的PDF文件完整路径

返回类型:

str

示例

>>> generator = ReportGenerator()
>>> material_data = [
...     {"material_grade": "Q235B", "specification": "10mm", "heat_no": "H12345"},
...     {"material_grade": "304", "specification": "8mm", "heat_no": "H67890"}
... ]
>>> filepath = generator.generate_material_certificate(material_data)
>>> print(f"材质报告已生成: {filepath}")

generate_quality_certificate(data, language='zh')[源代码]¶

生成质保书(单页)

根据质保数据生成单页PDF格式的产品质量保证书。

参数:

data (dict[str, Any]) -- 质保书数据字典，包含产品名称、规格型号、数量等字段
language (str) -- 语言，可选"zh"(中文)或"en"(英文)，默认为"zh"

返回:

生成的PDF文件完整路径

返回类型:

str

示例

>>> generator = ReportGenerator()
>>> quality_data = {
...     "product_name": "阀门",
...     "specification": "DN100",
...     "quantity": "100",
...     "batch_no": "B2024001"
... }
>>> filepath = generator.generate_quality_certificate(quality_data)
>>> print(f"质保书已生成: {filepath}")

class certflow.handlers.ShippedExportHandler(session)[源代码]¶

基类：object

已发货导出处理器

负责: - 将 SalePlan 记录拷贝到 SalePlanShipped 归档表 - 导出完整字段 Excel 文件 - 按列定义导出选中行到 Excel

示例

>>> handler = ShippedExportHandler(session)
>>> count = handler.archive_to_shipped(plans)
>>> result = handler.export_full_excel(plans, "/output")

参数:: session (Session)

copy_to_shipped(plan, shipped_by, shipped_at=None)[源代码]¶

将单条 SalePlan 记录拷贝为 SalePlanShipped 归档记录

不执行 session.add，由调用方控制事务。

参数:

plan (SalePlan) -- 源 SalePlan 记录
shipped_by (str) -- 归档来源标识 ("archive" 或 "export_delete")
shipped_at (datetime | None) -- 归档时间，默认当前时间

返回:

归档记录对象

返回类型:

SalePlanShipped

batch_archive(plans, shipped_by)[源代码]¶

批量归档 SalePlan → SalePlanShipped 并从主表删除

参数:

plans (list[SalePlan]) -- 要归档的 SalePlan 记录列表
shipped_by (str) -- 归档来源标识

返回:

归档记录数

返回类型:

int

export_full_excel(plans, output_dir)[源代码]¶

导出 SalePlan 完整字段到 Excel 文件

参数:

plans (list[SalePlan]) -- 要导出的 SalePlan 记录列表
output_dir (str) -- 输出目录

返回:

N, "filepath": "..."}

返回类型:

{"count"

export_selected_to_excel(records, columns, file_path)[源代码]¶

按列定义导出选中记录到 Excel

参数:

records (list[Any]) -- ORM 记录列表（SalePlan 对象）
columns (list[dict[str, str]]) -- 列定义列表 [{"field": "...", "label": "..."}, ...]
file_path (str) -- 目标文件路径

返回:

导出记录数

返回类型:

int

export_query_result_to_excel(records, columns, file_path)[源代码]¶

按列定义导出查询结果到 Excel（带 plan_date 格式化）

参数:

records (list[Any]) -- ORM 记录列表
columns (list[dict[str, str]]) -- 列定义列表
file_path (str) -- 目标文件路径

返回:

导出记录数

返回类型:

int

class certflow.handlers.Sorter[源代码]¶

基类：object

分组排序处理器

提供两级分组排序功能： 1. 第一层：按业务分组键分组（如订单信息） 2. 第二层：每组内按产品排序键排序（如产品名称、型号、规格） 3. 生成全局唯一的序号（分组前缀 + 组内序号）

示例

>>> sorter = Sorter()
>>> data = [
...     {"plan_date": "2024-01-01", "customer": "客户A", "product_name": "阀门B", "product_model": "DN80"},
...     {"plan_date": "2024-01-01", "customer": "客户A", "product_name": "阀门A", "product_model": "DN100"},
...     {"plan_date": "2024-01-01", "customer": "客户A", "product_name": "阀门A", "product_model": "DN80"},
... ]
>>> result = sorter.group_and_sort(
...     data,
...     group_keys=["plan_date", "customer"],
...     sort_keys=["product_name", "product_model"]
... )
>>> for item in result:
...     print(f"{item['full_seq']}: {item['product_name']}")
G001-001: 阀门A
G001-002: 阀门A
G001-003: 阀门B

static group_and_sort(data, group_keys, sort_keys=None, keep_original_order=True, generate_seq=True, use_global_prefix=True, prefix_format='G{:03d}', seq_format='{:03d}', seq_field='group_seq', group_key_field='_group_key', group_index_field='_group_index', source_file='', source_sheet='')[源代码]¶

两级分组排序

先按业务分组键分组，再在每组内按指定排序键排序，并可生成组内序号。

参数:

data (list[dict[str, Any]]) -- 原始数据列表，每个元素为字典格式的记录
group_keys (list[str]) -- 分组键列表，如["plan_date", "customer", "project_name", "sales_order_no"]
sort_keys (list[str] | None) -- 组内排序键列表，如["product_name", "product_model", "product_spec"]， None表示保持原始顺序
keep_original_order (bool) -- 是否保留原始顺序标记，默认为True
generate_seq (bool) -- 是否生成组内序号，默认为True
use_global_prefix (bool) -- 是否使用全局分组前缀（用于区分不同分组），默认为True
prefix_format (str) -- 分组前缀格式，默认"G{:03d}"，生成G001, G002...
seq_format (str) -- 组内序号格式，默认"{:03d}"，生成001,002...
seq_field (str) -- 序号存储的字段名，默认"group_seq"
group_key_field (str) -- 分组键存储的字段名，默认"_group_key"
group_index_field (str) -- 分组索引存储的字段名，默认"_group_index"
source_file (str) -- 源文件路径，用于提取年份生成_original_order，如"2026年销售计划.xlsx"
source_sheet (str) -- 工作表名称，用于提取月份生成_original_order，如"1月"

返回:

处理后的数据列表，每条记录包含:

_group_key: 分组键值（由group_keys组合而成）
_group_index: 分组索引(1-based)
group_seq: 组内序号（如"001"）
full_seq: 完整序号（如"G001-001"），仅当use_global_prefix=True时
_original_order: 原始顺序键(YYMMxxxx)，仅当keep_original_order=True时

返回类型:

list[dict[str, Any]]

示例

>>> data = [
...     {"order_no": "SO001", "product": "阀门B", "spec": "DN80"},
...     {"order_no": "SO001", "product": "阀门A", "spec": "DN100"},
...     {"order_no": "SO002", "product": "阀门C", "spec": "DN50"}
... ]
>>> result = Sorter.group_and_sort(
...     data,
...     group_keys=["order_no"],
...     sort_keys=["product", "spec"]
... )
>>> for item in result:
...     print(f"{item['full_seq']}: {item['product']}")
G001-001: 阀门A
G001-002: 阀门B
G002-001: 阀门C

static get_group_info(data, group_index=None, group_key=None)[源代码]¶

获取指定分组的记录

参数:

data (list[dict[str, Any]]) -- 已分组的数据列表(由group_and_sort生成)
group_index (int | None) -- 分组索引(1-based)
group_key (str | None) -- 分组键值

返回:

指定分组的记录列表

返回类型:

list[dict[str, Any]]

示例

>>> result = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> # 按索引获取
>>> group1 = Sorter.get_group_info(result, group_index=1)
>>> # 按键值获取
>>> group = Sorter.get_group_info(result, group_key="SO001")

static get_all_groups(data, seq_field='group_seq', group_key_field='_group_key')[源代码]¶

获取所有分组的信息

参数:

data (list[dict[str, Any]]) -- 已分组的数据列表(由group_and_sort生成)
seq_field (str) -- 序号字段名，与group_and_sort中的seq_field一致，默认"group_seq"
group_key_field (str) -- 分组键字段名，与group_and_sort中的group_key_field一致，默认"_group_key"

返回:

分组信息字典，键为分组索引，值为分组信息

返回类型:

dict[int, dict[str, Any]]

示例

>>> result = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> groups = Sorter.get_all_groups(result)
>>> for idx, info in groups.items():
...     print(f"分组{idx}: {info['count']}条记录")

static restore_original_order(data)[源代码]¶

恢复原始顺序

根据"_original_order"字段恢复数据的原始顺序，并清理所有临时字段。

参数:: data (list[dict[str, Any]]) -- 需要恢复顺序的数据列表，应包含"_original_order"字段 (由group_and_sort方法生成)
返回:: 恢复原始顺序并清理临时字段后的数据列表
返回类型:: list[dict[str, Any]]

备注

如果数据中没有"_original_order"字段，会记录警告并返回原数据
恢复顺序后会删除所有临时字段，保持数据干净

示例

>>> sorted_data = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> # 执行某些处理后...
>>> original_data = Sorter.restore_original_order(sorted_data)

class certflow.handlers.TemplateManager(templates_xlsx_path='')[源代码]¶

基类：object

模板管理器

管理合格证打印模板的配置，支持内置模板和用户自定义模板。提供 templates.xlsx 历史模板数据的查询和回填功能。

通知机制:: 模板切换时回调 on_template_changed 注册的监听器（携带模板名称）。本类不依赖 Qt；如需 Qt 信号，由上层 View 注册回调后自行转发。

参数:: templates_xlsx_path (str)

on_template_changed(callback)[源代码]¶

注册模板切换监听回调

参数:: callback (Callable[[str], None]) -- 模板切换时被调用，入参为模板内部名称
返回:: None
返回类型:: None

add_template(template)[源代码]¶

添加模板

参数:: template (TemplateConfig) -- 要添加的模板配置对象
返回:: None
返回类型:: None

remove_template(template_name)[源代码]¶

删除模板

参数:: template_name (str) -- 要删除的模板内部名称
返回:: None
返回类型:: None

get_template(name)[源代码]¶

获取模板

参数:: name (str) -- 模板内部名称
返回:: 命中的模板配置，未找到时返回 None
返回类型:: TemplateConfig | None

set_current_template(name)[源代码]¶

设置当前模板

参数:: name (str) -- 模板内部名称
返回:: 设置成功后回调已注册的模板切换监听器
返回类型:: None

get_template_list()[源代码]¶

获取模板名称列表（用户可见）

参数:: 无
返回:: 各模板的 display_name 列表
返回类型:: list[str]

get_template_names()[源代码]¶

获取模板内部名称列表

参数:: 无
返回:: 各模板的内部 name 列表
返回类型:: list[str]

get_template_by_display_name(display_name)[源代码]¶

根据显示名称获取模板

参数:: display_name (str) -- 模板用户可见名称
返回:: 命中的模板配置，未找到时返回 None
返回类型:: TemplateConfig | None

export_template(template_name, file_path)[源代码]¶

导出模板配置为 JSON

参数:

template_name (str) -- 要导出的模板内部名称
file_path (str) -- 目标 JSON 文件路径

返回:

导出成功返回 True，模板不存在时返回 False

返回类型:

bool

import_template(file_path)[源代码]¶

从 JSON 导入模板配置

参数:: file_path (str) -- 源 JSON 文件路径
返回:: 导入成功的模板配置，异常时为 None
返回类型:: TemplateConfig | None

load_from_xlsx(xlsx_path=None)[源代码]¶

从 templates.xlsx 的「数据库」工作表加载历史模板数据

参数:: xlsx_path (str | None) -- xlsx 文件路径，默认使用初始化时指定的路径
返回:: DataFrame 或 None
返回类型:: DataFrame | None

query_by_model(product_model, xlsx_path=None)[源代码]¶

按产品型号从 templates.xlsx 查询历史模板

在「数据库」工作表的第4列（产品型号）进行部分匹配。

参数:

product_model (str) -- 产品型号（支持部分匹配）
xlsx_path (str | None) -- xlsx 文件路径

返回:

匹配的模板列表，每条包含完整打印参数

返回类型:

list[dict[str, Any]]

writeback_to_xlsx(certificate_data, xlsx_path=None)[源代码]¶

打印完成后回填到 templates.xlsx「数据库」表

如果同型号+同编号已有记录则更新，否则新增行。

参数:

certificate_data (dict[str, Any]) -- 合格证数据字典
xlsx_path (str | None) -- xlsx 文件路径

返回:

是否回填成功

返回类型:

bool

class certflow.handlers.TemplateConfig(name, display_name, template_type='全中文', page_width=60, page_height=100, background_image='', fields=<factory>)[源代码]¶

基类：object

合格证打印模板配置

参数:

name (str)
display_name (str)
template_type (str)
page_width (float)
page_height (float)
background_image (str)
fields (list[FieldMapping])

name: str¶

display_name: str¶

template_type: str = '全中文'¶

page_width: float = 60¶

page_height: float = 100¶

background_image: str = ''¶

fields: list[FieldMapping]¶

to_dict()[源代码]¶

将模板配置序列化为字典

参数:: 无
返回:: 模板配置字典，包含 name/display_name/fields 等键
返回类型:: dict[str, Any]

classmethod from_dict(data)[源代码]¶

从字典反序列化为 TemplateConfig 实例

参数:: data (dict[str, Any]) -- 模板配置字典（含 name/display_name/fields 等键）
返回:: 重建的模板配置对象
返回类型:: TemplateConfig

class certflow.handlers.FieldMapping(source_field, target_cell='', display_name='', show_label=False, x=0, y=0, align='left', font_size=8, is_required=False, default_value='', format_string='')[源代码]¶

基类：object

字段映射 — 数据源字段到打印目标位置的映射

参数:

source_field (str)
target_cell (str)
display_name (str)
show_label (bool)
x (float)
y (float)
align (str)
font_size (int)
is_required (bool)
default_value (str)
format_string (str)

source_field: str¶

target_cell: str = ''¶

display_name: str = ''¶

show_label: bool = False¶

x: float = 0¶

y: float = 0¶

align: str = 'left'¶

font_size: int = 8¶

is_required: bool = False¶

default_value: str = ''¶

format_string: str = ''¶

class certflow.handlers.ScanImageHandler[源代码]¶

基类：object

扫描件图片处理器（底层绘制）

RENDER_DPI = 300¶

PX_PER_MM = 11.811023622047244¶

FONT_SCALE = 3.125¶

MM_TO_PX = 11.811023622047244¶

mm_to_px(mm)[源代码]¶

毫米转像素

参数:: mm (float) -- 毫米值
返回:: 像素值
返回类型:: int

classmethod load_background_rgb(background_path)[源代码]¶

加载背景图并归一化到干净 sRGB（修复俄英文背景在 Linux/IDE 偏色）。

问题根因（实测 ru_en_bg.png）：该 PNG 像素为 RGB 模式，却内嵌了约 557KB 的 CMYK 打印机 profile**（``Japan Color 2001 Coated``, device class=prtr, space=CMYK）——profile 与像素模式**不匹配。支持色彩管理的看图器（Linux/IDE 预览）会试图按该 CMYK profile 解读 RGB 像素 → 顶栏/印章偏色；不做色彩管理的看图器则正常。全中文/中英文背景无 ICC，故不受影响。

解法： - profile 的色彩空间与图像 mode **匹配**（如 RGB profile 配 RGB 图）→ 用

ImageCms 正常转换到 sRGB；

profile 与 mode 不匹配**（本例 CMYK profile 配 RGB 图，无法转换）→ 判定为错误标签，**剥离 profile，直接使用 RGB 像素值；
无 ICC → 仅确保 mode == RGB。

最终一律**清除 ``icc_profile``**，使生成结果在任意看图器下渲染一致（消除用户所述「Linux 环境 / IDE 打开导致偏色」的不确定性）。

参数:: background_path (str) -- 背景图片路径。
返回:: 无 ICC profile 的 sRGB Image.Image。
返回类型:: Image

create_canvas(width_mm, height_mm, background_path=None)[源代码]¶

创建画布

参数:

width_mm (float) -- 宽度（毫米）
height_mm (float) -- 高度（毫米）
background_path (str | None) -- 背景图片路径（可选）

返回:

PIL Image 对象

返回类型:

Image

draw_text(img, text, x_mm, y_mm, font_size=9, color='black')[源代码]¶

在画布上绘制文字（单段，不自动换行；框内换行请用 draw_field）。

参数:

img (Image) -- PIL Image 对象
text (str) -- 文字内容
x_mm (float) -- X 坐标（毫米）
y_mm (float) -- Y 坐标（毫米）
font_size (int) -- 字号（像素）
color (str) -- 颜色

返回:

直接在 img 上绘制，无返回值

返回类型:

None

draw_field(img, value, x_mm, y_mm, font_size=9, width_mm=None, height_mm=None, align='left', v_align='top', color='black')[源代码]¶

绘制字段：按框宽自动换行、超框高自动缩小字号、按对齐绘制。

解决阶段5扫描件两类渲染缺陷（BUG-006）： - 长字符串（如产品型号 BESDZY-320）按框宽折行，不再溢出整行； - 字号按基准 font_size 给出，不再被放大 2 倍； - 内容超高时自动缩小字号以适配框高； - 支持水平(align)/垂直(v_align)对齐，使用坐标配置中的 width/height。

参数:

img (Image) -- PIL Image 对象
value (str) -- 字段文字内容
x_mm (float) -- 字段框左上角坐标（毫米）
y_mm (float) -- 字段框左上角坐标（毫米）
font_size (int) -- 基准字号（像素）
width_mm (float | None) -- 字段框宽/高（毫米）；提供后启用换行与缩放
height_mm (float | None) -- 字段框宽/高（毫米）；提供后启用换行与缩放
align (str) -- 水平对齐 left/center/right
v_align (str) -- 垂直对齐 top/center/bottom
color (str) -- 文字颜色

返回:

直接在 img 上绘制，无返回值

返回类型:

None

save_jpg(img, file_path, quality=95)[源代码]¶

保存为 JPG 文件

参数:

img (Image) -- PIL Image 对象
file_path (str) -- 文件路径
quality (int) -- 质量 (1-100)

返回:

文件路径

返回类型:

str

save_pdf(img, file_path)[源代码]¶

保存为 PDF 文件

参数:

img (Image) -- PIL Image 对象
file_path (str) -- 文件路径

返回:

文件路径

返回类型:

str

class certflow.handlers.ScanLayoutHandler(rows=1, cols=1)[源代码]¶

基类：object

扫描件布局处理器（负责位置计算）

参数:

rows (int)
cols (int)

calculate_canvas_size(cert_width_mm, cert_height_mm)[源代码]¶

计算画布尺寸

参数:

cert_width_mm (float) -- 单张合格证宽度（毫米）
cert_height_mm (float) -- 单张合格证高度（毫米）

返回:

(width_px, height_px) 像素尺寸

返回类型:

tuple[int, int]

calculate_canvas_size_mm(cert_width_mm, cert_height_mm)[源代码]¶

计算画布尺寸（毫米）

参数:

cert_width_mm (float) -- 单张合格证宽度（毫米）
cert_height_mm (float) -- 单张合格证高度（毫米）

返回:

(width_mm, height_mm) 毫米尺寸

返回类型:

tuple[float, float]

get_position(index, cert_width_mm, cert_height_mm)[源代码]¶

获取指定索引的位置

参数:

index (int) -- 合格证索引（0-based）
cert_width_mm (float) -- 单张宽度（毫米）
cert_height_mm (float) -- 单张高度（毫米）

返回:

(x_offset_mm, y_offset_mm, row, col)

返回类型:

tuple[float, float, int, int]

is_multi_page()[源代码]¶

是否为多张拼接模式

参数:: 无
返回:: 当 rows*cols > 1（多张拼接）时为 True
返回类型:: bool

get_total_pages(total_certs)[源代码]¶

计算总页数

参数:: total_certs (int) -- 总合格证数量
返回:: 总页数
返回类型:: int

effective_grid(n)[源代码]¶

返回实际占用网格 (used_rows, used_cols)，用于最后一页（非满）裁剪画布。

满页（n == per_page）时返回完整 (rows, cols)，行为不变；非满页时去除多余空白单元格：

仅当所有单元落在**单行**（used_rows == 1）时，按实际列数横向裁剪，彻底消除尾部空白；
否则（上方已有满行，仅末行不满）保留整列宽，仅裁剪纵向空行。

这样最后一页不会生成整块 180×200 的空白合格证，画布尺寸贴合实际内容。

参数:: n (int) -- 本页实际单元（单台编号）数量。
返回:: (used_rows, used_cols) 实际参与排版的行列数。
返回类型:: tuple[int, int]

class certflow.handlers.ScanExportHandler(output_dir, output_format='JPG')[源代码]¶

基类：object

扫描件导出处理器（负责文件命名和组织）

参数:

output_dir (str)
output_format (str)

ensure_output_dir()[源代码]¶

确保输出目录存在

参数:: 无
返回:: 递归创建 output_dir，已存在则跳过
返回类型:: None

generate_filename_single(certificate_no)[源代码]¶

生成单张模式文件名

参数:: certificate_no (str) -- 合格证编号
返回:: 完整文件路径
返回类型:: str

generate_filename_multi(page_num, total_pages)[源代码]¶

生成多张拼接模式文件名

参数:

page_num (int) -- 当前页码
total_pages (int) -- 总页数

返回:

完整文件路径

返回类型:

str

generate_filename(base_name, page_num=1, total_pages=1)[源代码]¶

生成文件名（兼容接口）

参数:

base_name (str) -- 基础名称
page_num (int) -- 页码
total_pages (int) -- 总页数

返回:

完整文件路径

返回类型:

str