CertFlow API 参考

配置模块

ConfigLoader 类

配置加载器，支持YAML文件加载、环境变量替换和文件包含。

from certflow.config import ConfigLoader, get_config

# 方式1: 使用加载器实例
loader = ConfigLoader()
config = loader.load("config.yaml")

# 方式2: 使用全局单例(推荐)
config = get_config()

# 访问配置
print(config.app.name)           # "CertFlow"
print(config.paths.database.dir) # "database"
print(config.database.type)      # "sqlite"

settings 模块 - 配置常量

直接导入配置常量使用(零硬编码)。

from certflow.config.settings import (
    APP_NAME, APP_VERSION, APP_DEBUG,
    DATABASE_PATH, LOGS_DIR,
    SALE_PLAN_COLUMNS, COLUMN_ALIASES,
    SORT_KEYS, REQUIRED_FIELDS,
    get_column_by_alias
)

print(APP_NAME)          # "CertFlow"
print(DATABASE_PATH)     # Path("/path/to/certflow.db")

# 根据别名获取标准列名
field = get_column_by_alias("产品型号")  # 返回 "product_model"

get_config_value 函数

获取配置值，优先级: 环境变量 > YAML > 默认值。

from certflow.config.settings import get_config_value

# 优先从环境变量VALVE_SALES_PLAN_PATH读取，否则从YAML读取
path = get_config_value("paths.sales_plan.primary", "VALVE_SALES_PLAN_PATH")

服务模块

SalePlanService

销售计划服务，提供Excel导入、查询、统计等功能。

from certflow.services import SalePlanService
from certflow.utils.database import DatabaseManager

db = DatabaseManager()
session = db.get_session()
service = SalePlanService(session)

# 导入Excel(带格式)
result = service.import_from_excel_with_config({
    "file_path": "sales.xlsx",
    "sheet_name": "Sheet1",
    "header_row": 1,           # 第2行作为表头
    "data_start_row": 2,       # 第3行开始为数据
    "skip_rows": 0,
    "column_mapping": {"产品型号": "product_model"},
    "preserve_formatting": True  # 保留格式(背景色/字体色/批注)
})

print(f"新增: {result['new_count']}, 重复: {result['duplicate_count']}")

# 查询
plans = service.get_all_sale_plans()
plans = service.get_by_product_model("阀门A")
plans = service.get_by_sort_group("产品A_订单001")

# 统计
stats = service.get_statistics()
print(stats['total_records'])    # 总记录数
print(stats['group_count'])      # 分组数
print(stats['status_stats'])     # 状态统计

# 清空数据
count = service.clear_all()

QueryService

高级查询服务，支持多条件级联查询和分页。

from certflow.services import QueryService

service = QueryService(session)

# 多条件查询
filters = {
    "product_model": "阀门",
    "customer": "某某公司",
    "date_range": {"start": "2024-01-01", "end": "2024-12-31"},
    "supply_type": "自制"
}
result = service.search(
    filters=filters,
    page=1,
    page_size=50,
    sort_by="plan_date",
    sort_order="desc"
)

print(f"共{result['total']}条，当前页{len(result['results'])}条")

# 获取级联选项
models = service.get_cascading_options({"customer": "某某公司"}, "product_model")

# 统计信息
stats = service.get_statistics(filters)

AutoNumberService

自动编号服务，支持分组计数和周期重置。

from certflow.services import AutoNumberService

service = AutoNumberService(session)

# 初始化默认规则
service.init_default_rules()

# 生成单个编号
number = service.generate_certificate_number(sale_plan_id=100, rule_code="certificate")

# 批量生成
results = service.batch_generate_numbers([100, 101, 102], "certificate")
for sid, num in results.items():
    print(f"ID {sid} -> {num}")

# 预览下一个编号
preview = service.get_next_serial_preview("certificate", "product_A")

# 获取可用规则
rules = service.get_available_rules()

ReportService

报告生成服务。

from certflow.services import ReportService

service = ReportService(session)

# 生成试压报告
result = service.generate_pressure_test_report(
    sale_plan_ids=[1, 2, 3],
    language="zh",
    records_per_page=6
)

if result["success"]:
    print(f"报告已生成: {result['filepath']}")

# 批量生成报告
configs = [
    {"type": "pressure_test", "data_ids": [1,2,3], "language": "zh"},
    {"type": "material", "data_ids": [4,5], "language": "en"}
]
result = service.generate_batch_reports(configs)

# 获取报告历史
history = service.get_report_history()

控制器模块

BaseController

控制器基类，提供会话管理和事务处理。

from certflow.controllers import BaseController

# 使用上下文管理器(推荐)
with BaseController() as controller:
    # 执行数据库操作
    result = controller.session.query(SalePlan).all()
    controller.commit()

# 手动管理
controller = BaseController()
try:
    result = controller.session.query(SalePlan).all()
    controller.commit()
finally:
    controller.close()

CertificateController

合格证控制器。

from certflow.controllers import CertificateController

controller = CertificateController(session)

# 批量生成合格证
result = controller.generate_certificates_for_sale_plans([1, 2, 3])
print(f"生成成功: {result['generated']}/{result['total']}")

# 批量打印
result = controller.batch_print_certificates(
    certificate_ids=[1, 2, 3],
    printer_name="HP LaserJet",
    copies=2
)

# 标记刻印
result = controller.mark_as_engraved([1, 2, 3])

# 查询
certs = controller.get_certificates_by_status("待打印")
certs = controller.get_certificates_by_product("阀门A")

# 统计
stats = controller.get_print_statistics(days=30)

处理器模块

ExcelHandler

Excel文件处理器。

from certflow.handlers import ExcelHandler

# 读取Excel
df = ExcelHandler.read_excel(
    file_path="sales.xlsx",
    sheet_name=0,
    header_row=1,      # 第2行作为表头
    skiprows=0
)

# 读取多级表头
df = ExcelHandler.read_excel_with_multi_header(
    file_path="sales.xlsx",
    header_rows=[0, 1],
    separator="_"
)

# 自动检测表头行
df_raw = pd.read_excel("sales.xlsx", header=None)
header_row = ExcelHandler.detect_header_row(
    df_raw,
    keywords=["产品型号", "订货单位"],
    max_rows=10
)

# 清洗DataFrame
df = ExcelHandler.clean_dataframe(df)

# 验证并重命名列
df = ExcelHandler.validate_columns(
    df,
    required_columns={},
    use_aliases=True,
    alias_mapping=COLUMN_ALIASES
)

DataCleaner

数据清洗器。

from certflow.handlers import DataCleaner

# 清洗销售计划数据
df_clean = DataCleaner.clean_sale_plan(df)

# 校验单条数据
is_valid, message = DataCleaner.validate_data(row)

IDGenerator

编号生成器。

from certflow.handlers import IDGenerator

# 生成合格证编号
cert_no = IDGenerator.generate_certificate_no(prefix="CERT", sequence=5)

# 生成唯一键(基于MD5)
unique_key = IDGenerator.generate_unique_key(data, keys=["product_model", "order_no"])

# 生成可读唯一键(用于调试)
readable = IDGenerator.generate_unique_key_readable(data, keys=["product_model"])

# 批量生成唯一键
data_list = IDGenerator.generate_batch_unique_keys(data_list, keys=["product_model"])

# 规范化空值
normalized = IDGenerator.normalize_empty_values(data, keys=["product_model"])

ReportGenerator

PDF报告生成器。

from certflow.handlers import ReportGenerator

generator = ReportGenerator(output_dir="output/reports")

# 生成试压报告
filepath = generator.generate_pressure_test_report(
    data=test_data,
    language="zh",
    records_per_page=6
)

# 生成材质报告
filepath = generator.generate_material_certificate(
    data=material_data,
    material_type="steel",
    language="zh"
)

# 生成质保书
filepath = generator.generate_quality_certificate(
    data=quality_data,
    language="zh"
)

Sorter

多键排序器。

from certflow.handlers import Sorter

# 多键排序
sorted_data = Sorter.multi_key_sort(
    data,
    sort_keys=["product_model", "specification"],
    keep_original_order=True
)

# 按排序键分组
groups = Sorter.group_by_sort_key(sorted_data)

# 恢复原始顺序
original_data = Sorter.restore_original_order(sorted_data)

ExcelStyler

Excel样式读取器。

from certflow.handlers import ExcelStyler

# 读取带样式的Excel
data, bg_colors, font_colors, comments, statuses = ExcelStyler.read_with_styles(
    file_path="sales.xlsx",
    sheet_name=0,
    include_styles=True,
    include_comments=True
)

# 获取所有批注
comments = ExcelStyler.get_all_comments("sales.xlsx")

# 获取批注汇总表
df = ExcelStyler.get_comments_summary("sales.xlsx")

视图模块

ImportConfigDialog

导入配置对话框。

from certflow.views import ImportConfigDialog

dialog = ImportConfigDialog(file_path, parent)
dialog.config_confirmed.connect(on_config_confirmed)

def on_config_confirmed(config_params):
    print(f"表头行: {config_params['header_row']}")
    print(f"列映射: {config_params['column_mapping']}")

dialog.exec()

QueryView

销售计划查询视图。

from certflow.views import QueryView

view = QueryView(session)
view.show()

# 连接信号(需自定义)
view.data_selected.connect(on_data_selected)

CertificateNumberView

合格证编号生成视图。

from certflow.views import CertificateNumberView

view = CertificateNumberView(session)
view.show()

工具模块

DatabaseManager

数据库管理器。

from certflow.utils import DatabaseManager

# 创建管理器
db = DatabaseManager(db_path="database/certflow.db")

# 初始化数据库(创建表)
db.init_db(create_tables=True)

# 获取会话
session = db.get_session()

# 执行查询
results = session.query(SalePlan).all()

# 关闭连接
db.close()

LoggerManager / setup_logger

日志系统。

from certflow.utils.logger import setup_logger, get_logger

# 初始化日志(应用启动时调用一次)
setup_logger(
    log_path="logs/certflow.log",
    console_level="INFO",
    file_level="DEBUG"
)

# 获取日志实例
logger = get_logger()
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告信息")
logger.error("错误信息")

# 动态调整级别
from certflow.utils.logger import _logger_manager
_logger_manager.set_console_level("DEBUG")

数据模型

SalePlan 模型

from certflow.models import SalePlan

# 创建实例
plan = SalePlan(
    product_model="阀门A",
    customer="某某公司",
    quantity=10,
    sales_order_no="SO-001"
)

# 访问属性
print(plan.product_model)   # "阀门A"
print(plan.customer)        # "某某公司"
print(plan.unique_key)      # 自动生成UUID

Certificate 模型

from certflow.models import Certificate

cert = Certificate(
    certificate_no="CERT-001",
    sale_plan_id=100,
    unique_key="abc123",
    product_model="阀门A",
    status="待打印"
)

异常处理

from sqlalchemy.exc import SQLAlchemyError
from pydantic import ValidationError

try:
    result = service.import_from_excel_with_config(config)
except FileNotFoundError as e:
    logger.error(f"文件不存在: {e}")
    QMessageBox.critical(self, "错误", f"文件不存在: {e}")
except ValidationError as e:
    logger.error(f"配置验证失败: {e}")
    QMessageBox.warning(self, "配置错误", str(e))
except SQLAlchemyError as e:
    logger.error(f"数据库错误: {e}")
    session.rollback()
    QMessageBox.critical(self, "数据库错误", str(e))
except Exception as e:
    logger.exception(f"未知错误: {e}")
    QMessageBox.critical(self, "错误", f"操作失败: {e}")

钩子函数

导入完成钩子

def on_import_completed(result: Dict[str, Any]) -> None:
    """导入完成后的回调"""
    print(f"新增 {result['new_count']} 条记录")

# 在SalePlanService中注册
service.on_import_completed = on_import_completed

配置变更钩子

def on_config_reloaded() -> None:
    """配置重载后的回调"""
    print("配置已更新")

# 在settings模块中
reload_config()

使用示例

完整导入流程

from certflow.utils.database import DatabaseManager
from certflow.services import SalePlanService
from certflow.utils.logger import setup_logger

# 1. 初始化日志
setup_logger()

# 2. 初始化数据库
db = DatabaseManager()
db.init_db(create_tables=True)
session = db.get_session()

# 3. 创建服务
service = SalePlanService(session)

# 4. 导入数据
config = {
    "file_path": "销售计划.xlsx",
    "sheet_name": "1月",
    "header_row": 1,
    "data_start_row": 2,
    "skip_rows": 0,
    "column_mapping": {
        "产品型号": "product_model",
        "订单号": "sales_order_no",
        "数量": "quantity"
    }
}

result = service.import_from_excel_with_config(config)

# 5. 查看结果
print(f"导入完成: 新增{result['new_count']}条, 重复{result['duplicate_count']}条")

查询与导出

from certflow.services import QueryService
import pandas as pd

# 创建查询服务
query_service = QueryService(session)

# 执行查询
filters = {"product_model": "阀门A", "supply_type": "自制"}
result = query_service.search(filters, page=1, page_size=100)

# 导出到Excel
data = [{
    "产品型号": r.product_model,
    "订单号": r.sales_order_no,
    "客户": r.customer,
    "数量": r.quantity
} for r in result["results"]]

df = pd.DataFrame(data)
df.to_excel("导出结果.xlsx", index=False)