CertFlow 系统架构设计文档¶

版本：v2.0 | 最后更新：2026-06-17 | 状态：✅ 与项目结构同步

📋 目录¶

一、概述
二、整体架构概览
三、完整目录结构
四、核心数据流
五、CLI 命令行接口
六、配置管理设计
七、数据库设计
八、日志设计
九、测试设计
十、扩展点
十一、技术栈总览
十二、附录

一、概述¶

1.1 文档目的¶

本文档描述 CertFlow 系统的整体架构设计，包括技术栈、分层架构、核心数据流、配置管理、数据库设计及扩展机制，为开发、运维和后续迭代提供指导。

1.2 系统定位¶

CertFlow 是一款桌面应用程序，用于销售计划管理、合格证打印、报告生成和数据查询，替代原有 VBA/Access 方案，提供更稳定、高效的工业品质量文档管理解决方案。

1.3 设计目标¶

目标	说明
可靠性	事务管理、数据一致性、打印稳定性
可维护性	分层架构、清晰职责、低耦合
可扩展性	插件化设计、配置驱动、扩展点
性能	批量操作、异步处理、缓存机制
兼容性	兼容 Access 数据库、Excel 样式保留

二、整体架构概览¶

2.1 架构分层图¶

flowchart TB
    subgraph 入口层["入口层"]
        GUI["GUI入口<br>main.py"]
        CLI["CLI入口<br>cli/__main__.py"]
    end

    subgraph 表现层["表现层 (Views)"]
        VW["views/<br>主窗口/对话框"]
        WG["widgets/<br>共享UI组件"]
    end

    subgraph 控制层["控制层 (Controllers)"]
        CT["controllers/<br>业务协调/信号响应"]
    end

    subgraph 业务层["业务层"]
        SV["services/<br>业务逻辑/事务管理"]
        HD["handlers/<br>基础功能处理"]
    end

    subgraph 数据层["数据层"]
        MD["models/<br>ORM模型"]
        UT["utils/<br>通用工具"]
    end

    subgraph 外部系统["外部系统"]
        DB[("SQLite<br>数据库")]
        EXCEL["Excel文件"]
        PRINTER["打印机"]
        ACCESS[("Access<br>数据库")]
    end

    GUI --> VW
    CLI --> CT
    VW --> CT
    WG --> VW
    CT --> SV
    SV --> HD
    HD --> MD
    MD --> DB
    UT -.-> 所有层
    SV --> EXCEL
    SV --> PRINTER
    SV --> ACCESS

2.2 分层职责¶

层级	模块路径	职责
入口层	`src/certflow/main.py` `src/certflow/cli/`	GUI 应用启动 / CLI 命令分发
视图层	`src/certflow/views/` `src/certflow/widgets/`	UI 界面组件，用户交互，信号发送
控制层	`src/certflow/controllers/`	业务协调，信号响应，调用服务层
服务层	`src/certflow/services/`	业务逻辑实现，事务管理
处理层	`src/certflow/handlers/`	基础功能处理，Excel 读写、PDF 生成
数据层	`src/certflow/models/`	ORM 模型定义，数据库表映射
工具层	`src/certflow/utils/`	通用工具类，数据库管理、日志配置

2.3 层间通信规则¶

flowchart LR
    subgraph 通信方向
        V[View] -->|信号/槽| C[Controller]
        C -->|方法调用| S[Service]
        S -->|方法调用| H[Handler]
        H -->|CRUD| M[Model]
        M -->|SQL| DB[(Database)]
    end

    subgraph 禁止规则
        V -.->|❌ 直接调用| S
        V -.->|❌ 直接操作| M
        C -.->|❌ 直接操作| M
        C -.->|❌ 直接操作| DB
        S -.->|❌ 直接操作| DB
        S -.->|❌ 直接操作| V
        H -.->|❌ 直接操作| V
        H -.->|❌ 直接操作| C
    end

规则	说明
单向依赖	上层依赖下层，下层不依赖上层
信号通信	View ↔ Controller 使用 Qt 信号/槽
方法调用	Controller → Service → Handler → Model
禁止跨层	View 不能直接调用 Service 或 Model
禁止反向	Service 不能直接操作 View

三、完整目录结构¶

CertFlow-PySide6/                          # 项目根目录
├── README.md                               # 项目说明
├── pyproject.toml                          # uv 包管理配置
├── uv.lock                                 # 依赖锁定文件
├── mypy.ini                                # 类型检查配置
├── pyrightconfig.json                      # Pyright 配置
├── certflow.spec                           # PyInstaller 打包配置
├── run.py                                  # 启动脚本（Python）
├── run.bat                                 # 启动脚本（Windows）
├── start.sh                                # 启动脚本（Linux/Mac）
│
├── src/                                    # 源代码目录
│   └── certflow/                           # 主包
│       ├── __init__.py                     # 包初始化，版本信息
│       ├── main.py                         # GUI 程序入口
│       │
│       ├── cli/                            # CLI 命令行接口
│       │   ├── __init__.py
│       │   ├── __main__.py                 # python -m certflow.cli
│       │   ├── base_command.py             # 命令基类
│       │   └── commands/                   # 各子命令（18+ 个）
│       │       ├── auto_number.py          # 自动编号
│       │       ├── backfill_supply_type.py # 回填供货类型
│       │       ├── batch_update.py         # 批量更新
│       │       ├── check_consistency.py    # 一致性检查
│       │       ├── clean_duplicates.py     # 清理重复数据
│       │       ├── copy_cert_info.py       # 复制证书信息
│       │       ├── db_backup.py            # 数据库备份
│       │       ├── db_sync.py              # 数据库同步
│       │       ├── export_access.py        # 导出 Access
│       │       ├── export_excel.py         # 导出 Excel
│       │       ├── import_certs.py         # 导入证书
│       │       ├── import_sales_plan.py    # 导入销售计划
│       │       ├── mark_shipped.py         # 标记已发货
│       │       ├── print_cert.py           # 打印证书
│       │       ├── quick_query.py          # 快速查询
│       │       ├── serve.py                # 启动服务
│       │       ├── set_year_month.py       # 设置年月
│       │       ├── stats.py                # 统计信息
│       │       └── watch.py                # 文件监控
│       │
│       ├── config/                         # 配置模块
│       │   ├── __init__.py
│       │   ├── loader.py                   # YAML 加载器（支持 !include/环境变量）
│       │   ├── models.py                   # Pydantic 配置模型
│       │   ├── settings.py                 # 配置常量读取
│       │   ├── print_config.py             # 打印配置加载
│       │   └── ui_config.py                # UI 配置加载和样式构建
│       │
│       ├── controllers/                    # 控制层
│       │   ├── __init__.py
│       │   ├── base_controller.py          # 控制器基类（会话管理）
│       │   ├── certificate_controller.py   # 合格证控制器
│       │   ├── history_controller.py       # 打印历史控制器
│       │   ├── plaque_query_controller.py  # 标牌查询控制器
│       │   ├── printer_controller.py       # 打印控制器（QThread）
│       │   ├── query_controller.py         # 查询控制器
│       │   ├── report_controller.py        # 报告控制器
│       │   ├── scan_controller.py          # 扫描件控制器（QThread）
│       │   └── settings_controller.py      # 设置控制器
│       │
│       ├── services/                       # 服务层
│       │   ├── __init__.py
│       │   ├── auto_number_service.py      # 自动编号服务
│       │   ├── cert_export_service.py      # 合格证导出（Access/CSV）
│       │   ├── cert_log_service.py         # 打印日志服务
│       │   ├── cert_numbering.py           # 编号与 Certificate 写入
│       │   ├── cert_print_config.py        # 打印配置加载
│       │   ├── cert_print_engine.py        # GDI/HTML 打印引擎
│       │   ├── certificate_number_service.py # 证书编号服务
│       │   ├── certificate_print_service.py  # 合格证打印服务（QThread）
│       │   ├── certificate_service.py      # 合格证 CRUD 服务
│       │   ├── dn_service.py               # 口径（DN）解析服务
│       │   ├── pn_service.py               # 压力（PN）解析服务
│       │   ├── printer_manager.py          # 打印机统一管理
│       │   ├── query_builder.py            # 查询构建器
│       │   ├── query_service.py            # 高级查询服务（级联筛选）
│       │   ├── report_service.py           # 报告服务
│       │   ├── sale_plan_service.py        # 销售计划服务（Excel 导入）
│       │   ├── scan_service.py             # 扫描件服务（QThread）
│       │   ├── workflow_service.py         # 工作流服务
│       │   └── printer/                    # 打印机子模块
│       │       ├── __init__.py
│       │       ├── coordinate_converter.py # 坐标转换
│       │       ├── escp_commands.py        # ESC/P 命令生成
│       │       ├── lq635kii_printer.py     # LQ-635KII 打印机驱动
│       │       ├── print_strategy.py       # 打印策略（策略模式）
│       │       ├── printer_service.py      # 打印机服务
│       │       └── template_manager.py     # 模板管理
│       │
│       ├── handlers/                       # 处理层
│       │   ├── __init__.py
│       │   ├── data_cleaner.py             # 数据清洗器
│       │   ├── excel_handler.py            # Excel 文件处理器
│       │   ├── excel_styler.py             # Excel 样式读取（颜色/字体/批注）
│       │   ├── i18n_manager.py             # 国际化/语言切换
│       │   ├── id_generator.py             # 编号生成器（合格证编号/唯一键）
│       │   ├── report_generator.py         # PDF 报告生成器（ReportLab）
│       │   ├── save_handler.py             # 数据转换/变更检测（ORM 委托）
│       │   ├── scan_handler.py             # 扫描件处理器
│       │   ├── sorter.py                   # 多键排序器
│       │   ├── styled_excel_importer.py    # 带样式 Excel 导入处理器
│       │   ├── template_handler.py         # 模板配置管理
│       │   └── theme_manager.py            # 主题切换（亮色/暗色）
│       │
│       ├── models/                         # 数据层
│       │   ├── __init__.py
│       │   ├── base.py                     # SQLAlchemy 基类
│       │   ├── auto_number.py              # 自动编号规则和计数器
│       │   ├── caliber_mapping.py          # 口径映射表
│       │   ├── certificate.py              # 合格证模型
│       │   ├── material_grade.py           # 材质牌号表
│       │   ├── model_param_mapping.py      # 型号参数映射
│       │   ├── param_auto_learn_log.py     # 参数自动学习日志
│       │   ├── print_log.py                # 打印日志模型（51 字段）
│       │   ├── report_template.py          # 报告模板模型
│       │   ├── sale_plan.py                # 销售计划模型
│       │   ├── sale_plan_change.py         # 销售计划变更历史
│       │   ├── unmatched_certificate.py    # 未匹配合格证
│       │   └── vba_mapping.py              # VBA 字段映射
│       │
│       ├── utils/                          # 工具模块
│       │   ├── __init__.py
│       │   ├── access_db.py                # Access 数据库操作（pyodbc）
│       │   ├── database.py                 # 数据库管理器（引擎/会话/迁移）
│       │   ├── date_utils.py               # 日期工具
│       │   ├── image_utils.py              # 图像工具（PIL）
│       │   ├── logger.py                   # 日志管理器（单例/轮转/压缩）
│       │   ├── path_utils.py               # 路径工具
│       │   └── status_inference.py         # 状态推断
│       │
│       ├── views/                          # 视图层
│       │   ├── __init__.py
│       │   ├── main_window.py              # 主窗口
│       │   ├── config_view.py              # 配置总览视图
│       │   ├── dual_db_query_view.py       # 双数据库查询视图
│       │   ├── history_view.py             # 打印历史视图
│       │   ├── import_dialog.py            # 导入配置对话框
│       │   ├── param_supplement_dialog.py  # 参数补充对话框
│       │   ├── print_dialog.py             # 打印预览对话框
│       │   ├── print_view.py               # 合格证打印视图
│       │   ├── query_view.py               # 销售计划查询视图
│       │   ├── report_view.py              # 报告生成视图
│       │   ├── scan_view.py                # 扫描件生成视图
│       │   ├── settings_view.py            # 系统设置视图
│       │   └── query/                      # 查询子模块
│       │       ├── __init__.py
│       │       ├── context_menu.py         # 右键菜单
│       │       ├── filter_bar.py           # 筛选栏
│       │       ├── filter_row.py           # 筛选行
│       │       └── toolbar.py              # 工具栏
│       │
│       └── widgets/                        # 共享 UI 组件
│           ├── __init__.py
│           └── enhanced_table.py           # 增强表格控件（排序/筛选/编辑）
│
├── config/                                 # 配置文件目录
│   ├── config.yaml                         # 主配置（应用/路径/数据库/日志）
│   ├── userconfig.yaml                     # 用户配置（最近文件/导入历史/偏好）
│   ├── ui.yaml                             # UI 配置（窗口/样式/页面布局）
│   ├── views.py                            # 视图模板配置
│   ├── columns.yaml                        # 列定义
│   ├── dual_db_display.yaml                # 双数据库显示配置
│   ├── imperial_rules.yaml                 # 英制规则
│   ├── query_fields.yaml                   # 查询字段定义
│   ├── ui.yaml.example                     # UI 配置示例
│   ├── userconfig.yaml.example             # 用户配置示例
│   ├── schemas/                            # JSON Schema 验证
│   │   ├── columns_schema.json
│   │   ├── query_fields_schema.json
│   │   ├── root_schema.json
│   │   ├── ui_schema.json
│   │   └── userconfig_schema.json
│   └── templates/                          # 打印模板配置
│       └── coordinates.yaml                # 打印坐标模板
│
├── data/                                   # 数据文件目录
│   ├── db/                                 # 数据库文件
│   │   ├── certflow.db                     # 主数据库
│   │   └── SignDB.mdb                      # Access 遗留数据库
│   ├── images/                             # 图片资源
│   ├── templates/                          # 模板文件
│   ├── temp/                               # 临时文件
│   └── xlsx/                               # Excel 示例文件
│
├── database/                               # 数据库备份目录
│   ├── backups/                            # 自动备份
│   └── certflow*.db                        # 各版本数据库
│
├── docs/                                   # 文档目录
│   ├── README.md
│   ├── API_REFERENCE.md
│   ├── CHANGELOG.md
│   ├── CLI.md                              # CLI 使用文档
│   ├── INSTALLATION.md
│   ├── USER_GUIDE.md
│   ├── PRINT_USER_GUIDE.md
│   ├── TEMPLATE_PRINT.md
│   ├── IMPORT_CLI.md
│   ├── QUERY_CLI.md
│   ├── development/                        # 开发文档
│   │   ├── ARCHITECTURE.md
│   │   ├── CODING_STANDARDS.md
│   │   ├── DATA_FLOW.md
│   │   ├── DATA_MODEL_DESIGN.md
│   │   ├── DATABASE_SCHEMA.md
│   │   ├── VBA_FIELD_MAPPING.md
│   │   └── TODO.md
│   ├── deployment/                         # 部署文档
│   │   ├── BUILD.md
│   │   └── DOCKER.md
│   └── source/                             # Sphinx 文档源
│
├── scripts/                                # 脚本工具目录
│   ├── builder/                            # 构建脚本
│   │   ├── build.bat
│   │   ├── build.sh
│   │   ├── build_exe.py
│   │   ├── clean.sh
│   │   └── clean_for_build.sh
│   ├── data/                               # 数据迁移脚本
│   │   ├── add_progress_tracking_fields.py
│   │   ├── auto_number.py
│   │   ├── backfill_*.py
│   │   ├── import_*.py
│   │   └── update_*.py
│   ├── docs/                               # 文档生成脚本
│   │   ├── generate_db_schema.py
│   │   └── tree_structure.py
│   ├── monitor/                            # 打印机监控
│   │   ├── printer_monitor.py
│   │   └── analyze_printer_log.py
│   ├── setup/                              # 环境初始化
│   │   ├── init.sh
│   │   └── setup_uv.sh
│   ├── sync_databases_cli.py               # 数据库同步 CLI
│   └── utils/                              # 工具脚本
│
├── tests/                                   # 测试目录
│   ├── conftest.py                          # pytest 配置
│   ├── test_access.py
│   ├── test_config.py
│   ├── test_controller.py
│   ├── test_excel_handler.py
│   ├── test_excel_styler.py
│   ├── test_id_generator.py
│   ├── test_import.py
│   ├── test_logger.py
│   ├── test_main.py
│   ├── test_print_log_51fields.py
│   ├── test_printer_controller.py
│   ├── test_query_builder.py
│   ├── test_query_service.py
│   ├── test_query_view.py
│   ├── test_sale_plan_service.py
│   ├── test_settings_advanced.py
│   ├── test_ui_config.py
│   └── export_unique_keys.py
│
├── logs/                                   # 日志目录
│   ├── certflow.log                        # 当前日志
│   └── certflow.*.log.zip                  # 压缩的历史日志
│
├── output/                                 # 输出目录
│   ├── backups/                            # 手动备份
│   ├── certificates/                       # 导出的合格证
│   ├── images/                             # 生成的图片
│   ├── logs/                               # 导出日志
│   ├── nameplates/                         # 标牌
│   └── reports/                            # 生成的报告
│
├── resources/                              # 资源文件
│   ├── backgrounds/                        # 背景图
│   │   ├── cn_en_bg.png
│   │   ├── full_chinese_bg.png
│   │   └── ru_en_bg.png
│   ├── fonts/                              # 字体文件
│   ├── icons/                              # 图标
│   └── templates/                          # 报告模板
│
├── temp/                                   # 临时文件目录
├── htmlcov/                                # 测试覆盖率报告
├── coverage_improvement_plan.md            # 覆盖率改进计划
└── migration_preview.xlsx                  # 迁移预览

四、核心数据流¶

4.1 系统整体数据流¶

flowchart LR
    subgraph 输入
        EXCEL[Excel 文件]
        ACCESS[Access 数据库]
        USER[用户操作]
    end

    subgraph 处理
        IMP[导入处理]
        GEN[生成处理]
        PRINT[打印处理]
        QUERY[查询处理]
    end

    subgraph 存储
        DB[(SQLite 数据库)]
        CONFIG[配置文件]
    end

    subgraph 输出
        CERT[合格证]
        REPORT[报告 PDF]
        SCAN[扫描件 PNG]
        LOG[日志文件]
    end

    EXCEL --> IMP
    ACCESS --> IMP
    USER --> GEN
    USER --> PRINT
    USER --> QUERY

    IMP --> DB
    IMP --> CONFIG
    GEN --> DB
    PRINT --> DB
    QUERY --> DB

    DB --> CERT
    DB --> REPORT
    DB --> SCAN
    DB --> LOG

4.2 销售计划导入流程¶

flowchart TD
    A[用户选择 Excel 文件] --> B[ImportConfigDialog<br>配置参数]
    B --> C[SalePlanService<br>.import_from_excel_with_config]

    C --> D[StyledExcelImporter<br>.read_styled_excel]
    D --> E[DataCleaner<br>.clean_data]
    E --> F[Sorter<br>.sort_records]
    F --> G[SaveHandler<br>.save_records_with_stats]

    G --> H{检查唯一键}
    H -->|不存在| I[新增记录]
    H -->|已存在| J[更新记录]

    I --> K[保存到 SalePlan 表]
    J --> L[记录变更历史]
    K --> M[返回导入统计]
    L --> M

调用链路示例：

# ========== 调用链路示例 ==========

# View层: views/import_dialog.py
class ImportConfigDialog(QDialog):
    def on_import_clicked(self):
        self.import_requested.emit(config)

# Controller层: controllers/query_controller.py
class QueryController:
    def on_import_requested(self, config):
        result = self.sale_plan_service.import_from_excel_with_config(
            file_path=config.file_path,
            sheet_name=config.sheet_name,
            header_row=config.header_row,
            column_mapping=config.column_mapping
        )
        self.view.show_import_result(result)

# Service层: services/sale_plan_service.py
class SalePlanService:
    def import_from_excel_with_config(self, **kwargs):
        styled_data = StyledExcelImporter.read_styled_excel(...)
        cleaned_data = DataCleaner.clean_data(styled_data)
        sorted_data = Sorter.sort_records(cleaned_data)
        stats = SaveHandler.save_records_with_stats(
            records=sorted_data,
            model_class=SalePlan,
            unique_key='unique_key'
        )
        return stats

4.3 合格证生成与打印流程¶

flowchart TD
    A[用户选择销售计划] --> B[选择编号规则]
    B --> C[CertificatePrintService<br>启动打印线程]

    C --> D[CertNumberingService<br>.auto_number]
    D --> E[按 supply_type 分组]
    E --> F[生成编号: 前缀+日期+流水号+后缀]
    F --> G[CertNumberingService<br>.create_certificates]
    G --> H[写入 Certificate 表]

    H --> I[check_and_supplement_params]
    I --> J{参数完整?}
    J -->|否| K[弹窗补充: 口径/压力/温度/介质/材质/工号]
    K --> L[保存补充参数]
    J -->|是| L

    L --> M[批量打印循环]
    M --> N[CertPrintEngine<br>.print_single - GDI 精确打印]
    N --> O{打印成功?}
    O -->|是| P[记录打印日志]
    O -->|否| Q[CertPrintEngine<br>.print_html_fallback - 降级]
    Q --> P

    P --> R[CertLogService<br>.auto_save - 51字段日志]
    R --> S[更新打印状态]
    S --> T[完成]

调用链路示例：

# ========== 调用链路示例 ==========

# Service层: services/certificate_print_service.py
class CertificatePrintService(QThread):
    def run(self):
        # 1. 编号服务
        cert_numbering = CertNumberingService()
        cert_data = cert_numbering.auto_number(
            sale_plan_ids=self.selected_ids,
            rule_id=self.rule_id
        )

        # 2. 写入Certificate表
        for data in cert_data:
            certificate = Certificate(
                certificate_number=data['cert_number'],
                sale_plan_id=data['sale_plan_id'],
                supply_type=data['supply_type']
            )
            self.db_session.add(certificate)

        # 3. 补充参数（弹窗）
        self.check_and_supplement_params(cert_data)

        # 4. 批量打印
        print_engine = CertPrintEngine()
        for cert in certificates:
            print_engine.print_single(cert, self.template, self.printer_name)
            CertLogService.auto_save(cert, print_result)

        self.print_completed.emit(results)

4.4 查询与删除流程¶

flowchart TD
    A[用户输入筛选条件] --> B[QueryView 收集筛选条件]
    B --> C[QueryController<br>.on_search_requested]
    C --> D[QueryService.query]

    D --> E[构建 SQLAlchemy 查询]
    E --> F[应用过滤器]
    F --> G[排序]
    G --> H[分页]
    H --> I[返回 QueryResult]

    I --> J[更新表格显示]

    K[用户选择记录] --> L[点击删除]
    L --> M[QueryController<br>.on_delete_requested]
    M --> N{检查关联证书}
    N -->|有| O[级联删除]
    N -->|无| P[直接删除]

    O --> Q[删除 Certificate]
    Q --> R[删除 SalePlan]
    P --> R

    R --> S[事务提交]
    S --> T[刷新查询结果]

调用链路示例：

# ========== 调用链路示例 ==========

# Service层: services/query_service.py
class QueryService:
    def query(self, filters, page, page_size):
        query = self.db_session.query(SalePlan)

        for field, value in filters.items():
            if value:
                if field == 'customer_name':
                    query = query.filter(SalePlan.customer_name.like(f'%{value}%'))
                elif field == 'create_date_range':
                    query = query.filter(SalePlan.create_date.between(value[0], value[1]))

        query = query.order_by(SalePlan.create_date.desc())
        total = query.count()
        data = query.offset((page - 1) * page_size).limit(page_size).all()

        return QueryResult(total=total, data=data, page=page)

    def delete_with_certificates(self, ids):
        # 级联删除
        self.db_session.query(Certificate).filter(
            Certificate.sale_plan_id.in_(ids)
        ).delete(synchronize_session=False)

        self.db_session.query(SalePlan).filter(
            SalePlan.id.in_(ids)
        ).delete(synchronize_session=False)

4.5 扫描件生成流程¶

flowchart TD
    A[用户选择合格证] --> B[选择模板]
    B --> C[ScanController 启动 QThread]
    C --> D[ScanService.run]

    D --> E[加载证书数据]
    E --> F[循环处理每个证书]

    F --> G[ScanImageHandler.generate]
    G --> H[ReportLab 生成 PDF]
    H --> I[PyMuPDF 转换为图像]

    I --> J[ScanLayoutHandler.layout]
    J --> K[添加水印]
    K --> L[添加边框]

    L --> M[保存为 PNG 文件]
    M --> N[更新进度]
    N --> O{还有证书?}
    O -->|是| F
    O -->|否| P[打开输出目录]

调用链路示例：

# ========== 调用链路示例 ==========

# Service层: services/scan_service.py
class ScanService(QThread):
    def run(self):
        certificates = self.load_certificates(self.cert_ids)
        image_handler = ScanImageHandler()
        layout_handler = ScanLayoutHandler()

        for idx, cert in enumerate(certificates):
            image = image_handler.generate(
                certificate=cert,
                template=self.template,
                dpi=300
            )
            final_image = layout_handler.layout(
                image=image,
                watermark=self.watermark,
                border=self.border
            )
            output_path = Path(self.output_dir) / f"{cert.certificate_number}.png"
            final_image.save(output_path)
            self.progress_updated.emit(idx + 1, len(certificates))

        QDesktopServices.openUrl(QUrl.fromLocalFile(str(self.output_dir)))

4.6 主题切换流程¶

flowchart TD
    A[用户点击 🌙/☀️] --> B[MainWindow.on_theme_toggle]
    B --> C[SettingsController.switch_theme]

    C --> D[ThemeManager.apply_theme]
    D --> E[加载主题 QSS]
    E --> F[QApplication.setStyleSheet]
    F --> G[更新配置]

    C --> H[更新按钮图标]
    H --> I[保存用户配置]

    C --> J[_sync_timer_from_prefs]
    J --> K[同步自动保存定时器]

    K --> L[界面实时更新]

五、CLI 命令行接口¶

5.1 CLI 架构¶

flowchart TB
    subgraph CLI入口
        CMD["python -m certflow.cli"]
    end

    subgraph 命令注册
        REG["命令注册表"]
    end

    subgraph 命令实现
        C1["auto_number"]
        C2["import_sales_plan"]
        C3["print_cert"]
        C4["db_sync"]
        C5["stats"]
        C6["... 18+ 命令"]
    end

    subgraph 共享服务
        SV["Services 层"]
        DB[("数据库")]
    end

    CMD --> REG
    REG --> C1 & C2 & C3 & C4 & C5 & C6
    C1 --> SV
    C2 --> SV
    C3 --> SV
    C4 --> SV
    C5 --> SV
    SV --> DB

5.2 核心命令列表¶

分类	命令	用途	示例
数据导入	`import_sales_plan`	导入销售计划	`import_sales_plan -f plan.xlsx`
	`import_certs`	导入合格证	`import_certs -f certs.xlsx`
	`backfill_supply_type`	回填供货类型	`backfill_supply_type --file data.xlsx`
打印管理	`print_cert`	打印合格证	`print_cert --ids 1,2,3`
	`auto_number`	自动编号	`auto_number --rule-id 1`
数据管理	`db_backup`	备份数据库	`db_backup`
	`db_sync`	同步数据库	`db_sync --source source.db`
	`clean_duplicates`	清理重复数据	`clean_duplicates --dry-run`
	`check_consistency`	一致性检查	`check_consistency`
数据导出	`export_excel`	导出 Excel	`export_excel -o output.xlsx`
	`export_access`	导出 Access	`export_access -o output.mdb`
查询统计	`quick_query`	快速查询	`quick_query --customer "XX"`
	`stats`	统计信息	`stats --type daily`
数据维护	`batch_update`	批量更新	`batch_update -f updates.csv`
	`mark_shipped`	标记已发货	`mark_shipped --ids 1,2,3`
	`copy_cert_info`	复制证书信息	`copy_cert_info --from 1 --to 2`
服务	`serve`	启动服务	`serve --host 0.0.0.0 --port 8000`
	`watch`	文件监控	`watch --dir ./data`

5.3 命令基类设计¶

# cli/base_command.py
from abc import ABC, abstractmethod
import argparse

class BaseCommand(ABC):
    """CLI 命令基类"""

    name: str = ""
    help: str = ""

    @abstractmethod
    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        """添加命令行参数"""
        pass

    @abstractmethod
    def execute(self, args: argparse.Namespace) -> int:
        """执行命令，返回退出码"""
        pass

    def setup(self) -> None:
        """命令执行前的初始化"""
        pass

    def teardown(self) -> None:
        """命令执行后的清理"""
        pass

5.4 命令实现示例¶

# cli/commands/import_sales_plan.py
from ..base_command import BaseCommand

class ImportSalesPlanCommand(BaseCommand):
    name = "import_sales_plan"
    help = "从 Excel 文件导入销售计划"

    def add_arguments(self, parser):
        parser.add_argument("-f", "--file", required=True, help="Excel 文件路径")
        parser.add_argument("--sheet", default="Sheet1", help="工作表名称")
        parser.add_argument("--header-row", type=int, default=1, help="表头行号")
        parser.add_argument("--dry-run", action="store_true", help="预览模式")

    def execute(self, args):
        service = SalePlanService()
        result = service.import_from_excel_with_config(
            file_path=args.file,
            sheet_name=args.sheet,
            header_row=args.header_row,
            dry_run=args.dry_run
        )
        print(f"导入完成: {result}")
        return 0

5.5 全局选项¶

选项	说明	默认值
`--config`	指定配置文件路径	`config/config.yaml`
`--db`	指定数据库路径	`data/db/certflow.db`
`--verbose`	显示详细日志	`False`
`--quiet`	静默模式	`False`
`--version`	显示版本信息	-

5.6 环境变量¶

变量	说明	默认值
`CERTFLOW_CONFIG`	配置文件路径	`config/config.yaml`
`CERTFLOW_DB`	数据库路径	`data/db/certflow.db`
`CERTFLOW_LOG_LEVEL`	日志级别	`INFO`

5.7 使用示例¶

# 查看帮助
python -m certflow.cli --help

# 导入销售计划（预览模式）
python -m certflow.cli import_sales_plan -f data.xlsx --dry-run

# 批量打印
python -m certflow.cli print_cert --ids 1,2,3,4,5

# 数据库备份 + 同步
python -m certflow.cli db_backup
python -m certflow.cli db_sync --source backup.db

# 统计查询
python -m certflow.cli stats --type daily --month 2026-06

5.8 错误码¶

代码	说明
0	成功
1	通用错误
2	配置错误
3	数据库错误
4	文件错误
5	打印错误
6	参数错误
7	权限错误

六、配置管理设计¶

6.1 配置文件结构¶

flowchart LR
    subgraph 配置源
        ENV[环境变量]
        YAML[YAML 文件]
        DEFAULT[代码默认值]
    end

    subgraph 配置加载
        LOADER[ConfigLoader]
        SCHEMA[Schema 验证]
        CACHE[内存缓存]
    end

    subgraph 配置文件
        C1[config.yaml<br>主配置]
        C2[userconfig.yaml<br>用户配置]
        C3[ui.yaml<br>UI 配置]
        C4[columns.yaml<br>列定义]
        C5[query_fields.yaml<br>查询字段]
        C6[templates/coordinates.yaml<br>打印模板]
    end

    ENV --> LOADER
    YAML --> LOADER
    DEFAULT --> LOADER
    LOADER --> SCHEMA
    SCHEMA --> CACHE
    C1 & C2 & C3 & C4 & C5 & C6 --> LOADER

6.2 配置文件详情¶

文件	用途	关键配置项
`config.yaml`	主配置	应用名称、版本、数据库路径、日志级别
`userconfig.yaml`	用户配置	最近文件列表、导入历史、窗口位置
`ui.yaml`	UI 配置	主题、字体、布局、颜色方案
`columns.yaml`	列定义	表格列名、宽度、可见性、排序
`query_fields.yaml`	查询字段	筛选条件字段、类型、可选值
`dual_db_display.yaml`	双数据库显示	对比显示配置
`imperial_rules.yaml`	英制规则	英制单位转换规则
`templates/coordinates.yaml`	打印模板	打印坐标、字体、尺寸

6.3 配置加载特性¶

特性	说明	实现方式
惰性加载	配置只在首次访问时加载	`@property` + `_loaded` 标志
缓存机制	加载后缓存在内存中	`_config_cache` 字典
环境变量替换	支持 `${ENV_VAR:default}`	正则表达式替换
文件包含	支持 `!include` 标签	YAML 自定义构造器
热重载	打印配置支持热重载	`reload_print_config()`
Schema 验证	配置结构校验	JSON Schema

6.4 配置优先级¶

flowchart LR
    ENV[环境变量<br>最高优先级] --> APP[应用配置]
    YAML[YAML 配置<br>中等优先级] --> APP
    DEFAULT[代码默认值<br>最低优先级] --> APP

6.5 配置加载器实现¶

# config/loader.py
class ConfigLoader:
    """配置加载器，支持惰性加载、缓存、环境变量替换"""

    _instance = None
    _config_cache = {}

    @classmethod
    def get_instance(cls) -> "ConfigLoader":
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    def load(self, name: str) -> dict:
        """加载指定配置"""
        if name in self._config_cache:
            return self._config_cache[name]

        path = self._get_config_path(name)
        content = self._read_file(path)
        content = self._resolve_env_vars(content)
        content = self._resolve_includes(content)
        config = yaml.safe_load(content)

        self._validate_schema(config, name)
        self._config_cache[name] = config
        return config

    def _resolve_env_vars(self, content: str) -> str:
        """解析环境变量 ${VAR:default}"""
        pattern = r'\${([^}:]+)(?::([^}]*))?}'
        def replacer(match):
            var_name = match.group(1)
            default = match.group(2) or ''
            return os.environ.get(var_name, default)
        return re.sub(pattern, replacer, content)

七、数据库设计¶

7.1 ER 图¶

erDiagram
    sale_plans ||--o{ certificates : "生成"
    sale_plans ||--o{ sale_plan_changes : "变更历史"
    certificates ||--o{ print_logs : "打印记录"
    certificates }o--|| material_grades : "材质"
    certificates }o--|| caliber_mappings : "口径"

    sale_plans {
        integer id PK
        string unique_key UK
        string customer_name
        string product_model
        string supply_type
        date create_date
        string execution_status
    }

    certificates {
        integer id PK
        string certificate_number UK
        integer sale_plan_id FK
        string supply_type
        string caliber
        string pressure
        string material_grade
        string temperature
        string medium
        date print_date
    }

    print_logs {
        integer id PK
        integer certificate_id FK
        string certificate_number
        string printer_name
        datetime print_time
        string status
        text error_message
    }

    auto_number_rules ||--o{ auto_number_counters : "计数"

    auto_number_rules {
        integer id PK
        string name
        string prefix
        string date_format
        integer padding
        string suffix_pattern
        string reset_period
    }

    auto_number_counters {
        integer id PK
        integer rule_id FK
        string group_key
        integer current_value
        date last_reset_date
    }

7.2 核心表说明¶

表名	用途	关键字段	说明
`sale_plans`	销售计划主表	`unique_key`, `customer_name`, `supply_type`, `execution_status`	核心业务数据表
`sale_plan_changes`	变更历史	`sale_plan_id`, `changed_fields`, `change_time`	自动记录变更
`certificates`	合格证表	`certificate_number`, `sale_plan_id`, `supply_type`, `caliber`	合格证数据
`print_logs`	打印日志（51字段）	`certificate_id`, `printer_name`, `print_time`, `status`	兼容 Access SignTb
`material_grades`	材质牌号	`grade_name`, `standard`, `description`	材质字典表
`caliber_mappings`	口径映射	`display_name`, `standard_value`, `sort_order`	口径字典表
`model_param_mappings`	型号参数映射	`model`, `caliber`, `pressure`, `temperature`	自动补全参数
`param_auto_learn_logs`	参数学习日志	`model`, `field`, `learned_value`, `confidence`	自动学习记录
`auto_number_rules`	编号规则	`prefix`, `date_format`, `padding`, `reset_period`	编号规则配置
`auto_number_counters`	编号计数器	`rule_id`, `group_key`, `current_value`	流水号计数

7.3 自动编号特性¶

特性	说明
分组计数	按产品型号/订单号/供货类型分组
周期重置	按日/月/年重置流水号
自定义格式	前缀 + 日期 + 流水号 + 后缀
后缀判定	按 `supply_type` 自动判定（L/B/G/Y/XL）

编号格式示例：

FL202606170001L
││  ││      ││││ └─ 后缀（L = 法兰）
││  ││      │││└─── 流水号（0001）
││  ││      ││└──── 日（17）
││  ││      │└───── 月（06）
││  ││      └────── 年（2026）
││  │└─────────── 前缀（FL = 法兰）
││  └──────────── 分隔符

7.4 数据库管理器¶

# utils/database.py
class DatabaseManager:
    """数据库管理器（单例模式）"""

    _instance = None
    _engine = None
    _session_factory = None

    @classmethod
    def get_instance(cls) -> "DatabaseManager":
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    def initialize(self, connection_string: str) -> None:
        """初始化数据库连接"""
        self._engine = create_engine(connection_string, echo=False)
        self._session_factory = sessionmaker(bind=self._engine)
        Base.metadata.create_all(self._engine)

    def get_session(self) -> Session:
        """获取数据库会话"""
        return self._session_factory()

    def backup(self, backup_path: str) -> None:
        """备份数据库"""
        shutil.copy2(self._engine.url.database, backup_path)

    def migrate(self, migration_script: str) -> None:
        """执行数据库迁移"""
        with self.get_session() as session:
            session.execute(text(migration_script))
            session.commit()

八、日志设计¶

8.1 日志架构¶

flowchart LR
    APP[应用程序] --> LOG[LoggerManager 单例]
    LOG --> CONSOLE[控制台输出<br>INFO 级别 彩色]
    LOG --> FILE[文件输出<br>DEBUG 级别]
    FILE --> ROTATE[每天轮转]
    ROTATE --> COMPRESS[压缩为 ZIP]
    COMPRESS --> RETENTION[保留 30 天]

8.2 日志配置¶

特性	说明
模式	单例模式，全局唯一日志管理器
控制台输出	INFO 级别，彩色格式（适用于开发调试）
文件输出	DEBUG 级别，每天轮转
保留策略	保留 30 天，自动压缩为 ZIP
日志文件	`logs/certflow.log`（当前），`logs/certflow.*.log.zip`（历史）

8.3 日志管理器实现¶

# utils/logger.py
from loguru import logger
import sys

class LoggerManager:
    """日志管理器（单例模式）"""

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._setup()
        return cls._instance

    def _setup(self):
        """配置日志"""
        # 移除默认处理器
        logger.remove()

        # 控制台输出（彩色）
        logger.add(
            sys.stdout,
            format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
            level="INFO",
            colorize=True
        )

        # 文件输出（每天轮转）
        logger.add(
            "logs/certflow.log",
            format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
            level="DEBUG",
            rotation="1 day",
            retention="30 days",
            compression="zip"
        )

    def get_logger(self):
        return logger

8.4 日志使用示例¶

from certflow.utils.logger import LoggerManager

# 获取日志实例
log = LoggerManager().get_logger()

# 不同级别日志
log.debug("调试信息")
log.info("系统启动")
log.warning("配置缺失，使用默认值")
log.error("打印失败", exc_info=True)
log.success("导入完成")

# 带上下文的日志
log.bind(user="admin").info("用户登录")

九、测试设计¶

9.1 测试架构¶

flowchart TB
    subgraph 测试层
        UNIT[单元测试<br>test_*.py]
        INT[集成测试<br>test_*_integration.py]
    end

    subgraph 测试框架
        PYT[pytest]
        COV[coverage.py]
        FIX[fixtures]
    end

    subgraph 被测试模块
        SRC[src/certflow/]
    end

    UNIT --> PYT
    INT --> PYT
    PYT --> COV
    PYT --> FIX
    FIX --> SRC
    COV --> REPORT[覆盖率报告<br>htmlcov/]

9.2 测试文件结构¶

测试文件	测试目标	状态
`test_config.py`	配置加载器、配置模型	✅
`test_controller.py`	控制器逻辑	✅
`test_excel_handler.py`	Excel 文件处理	✅
`test_excel_styler.py`	Excel 样式读取	✅
`test_query_service.py`	查询服务	✅
`test_query_builder.py`	查询构建器	✅
`test_sale_plan_service.py`	销售计划服务	✅
`test_printer_controller.py`	打印控制器	✅
`test_print_log_51fields.py`	打印日志 51 字段	✅
`test_id_generator.py`	ID 生成器	✅
`test_logger.py`	日志系统	✅
`test_import.py`	导入功能	✅
`test_main.py`	主入口	✅
`test_access.py`	Access 数据库操作	✅
`test_ui_config.py`	UI 配置	✅
`test_settings_advanced.py`	高级设置	✅
`test_query_view.py`	查询视图	✅

9.3 测试 Fixtures¶

# tests/conftest.py
import pytest
from src.certflow.utils.database import DatabaseManager

@pytest.fixture
def db_session():
    """测试用数据库会话"""
    db = DatabaseManager.get_instance()
    db.initialize("sqlite:///:memory:")  # 使用内存数据库
    session = db.get_session()
    yield session
    session.rollback()
    session.close()

@pytest.fixture
def sample_sale_plan():
    """示例销售计划数据"""
    return {
        "customer_name": "测试客户",
        "product_model": "DN100",
        "supply_type": "法兰",
        "quantity": 10
    }

@pytest.fixture
def config_loader():
    """配置加载器"""
    from src.certflow.config.loader import ConfigLoader
    return ConfigLoader.get_instance()

9.4 运行测试¶

# 运行所有测试
uv run pytest tests/

# 运行指定测试文件
uv run pytest tests/test_query_service.py

# 运行指定测试函数
uv run pytest tests/test_query_service.py::test_query_with_filters

# 运行测试并显示详细输出
uv run pytest -v tests/

# 运行测试并显示打印输出
uv run pytest -s tests/

# 运行测试并生成覆盖率报告
uv run pytest --cov=src/certflow tests/

# 生成 HTML 覆盖率报告
uv run pytest --cov=src/certflow --cov-report=html tests/

9.5 覆盖率目标¶

模块	目标	说明
`services/`	≥ 80%	核心业务逻辑
`handlers/`	≥ 75%	基础功能处理器
`controllers/`	≥ 70%	控制器（含 UI 交互）
`utils/`	≥ 80%	工具类
`models/`	≥ 90%	数据模型（简单）
`config/`	≥ 70%	配置管理
整体	≥ 70%	全模块平均

十、扩展点¶

10.1 扩展点总览¶

flowchart LR
    subgraph 扩展方式
        E1[继承基类]
        E2[配置驱动]
        E3[注册机制]
        E4[新增文件]
    end

    subgraph 扩展点
        P1[报告类型]
        P2[编号规则]
        P3[数据模型]
        P4[处理器]
        P5[UI 主题]
        P6[视图]
        P7[控制器]
        P8[服务]
        P9[CLI 命令]
    end

    E1 --> P1 & P3 & P7 & P8
    E2 --> P2 & P5
    E3 --> P9
    E4 --> P4 & P6

10.2 扩展点详情¶

扩展点	方式	示例代码
新增报告类型	继承 `ReportGenerator`	`class CustomReport(ReportGenerator): ...`
新增编号规则	数据库配置	`INSERT INTO auto_number_rules ...`
新增数据模型	继承 `Base`	`class NewModel(Base): ...`
新增处理器	添加到 `handlers/`	`class NewHandler: ...`
UI 主题定制	修改 `ui.yaml`	`themes: { custom: {...} }`
新增视图	添加到 `views/`，注册到 `main_window.py`	`class NewView(QWidget): ...`
新增控制器	添加到 `controllers/`	`class NewController: ...`
新增服务	添加到 `services/`	`class NewService: ...`
新增 CLI 命令	添加到 `cli/commands/`，继承 `BaseCommand`	`class NewCommand(BaseCommand): ...`

10.3 扩展示例¶

新增报告类型¶

# handlers/report_generator.py
class CustomReportGenerator(ReportGenerator):
    """自定义报告生成器"""

    def generate(self, data: dict, output_path: str) -> str:
        """生成自定义报告"""
        pdf = canvas.Canvas(output_path)
        # 自定义绘制逻辑
        pdf.drawString(100, 800, "自定义报告")
        pdf.save()
        return output_path

新增 CLI 命令¶

# cli/commands/export_custom.py
from ..base_command import BaseCommand

class ExportCustomCommand(BaseCommand):
    name = "export_custom"
    help = "导出自定义格式"

    def add_arguments(self, parser):
        parser.add_argument("-o", "--output", required=True, help="输出路径")

    def execute(self, args):
        # 实现导出逻辑
        print(f"导出到: {args.output}")
        return 0

十一、技术栈总览¶

分类	组件	技术	版本	用途
GUI	框架	PySide6	≥6.5	跨平台桌面界面
ORM	数据库	SQLAlchemy	≥2.0	数据库 ORM
PDF	生成	ReportLab	≥4.0	报告 PDF 生成
Excel	读取	pandas	≥2.0	数据处理
	样式	openpyxl	≥3.1	Excel 样式保留
日志	管理	loguru	≥0.7	统一日志管理
配置	解析	PyYAML	≥6.0	YAML 配置解析
包管理	依赖	uv	≥0.5	Python 包管理
测试	框架	pytest	≥7.0	单元测试
	覆盖率	pytest-cov	≥4.0	覆盖率报告
类型检查	静态	mypy	≥1.0	类型检查
	静态	pyright	≥1.0	类型检查
打包	可执行	PyInstaller	≥5.0	打包为 exe
文档	生成	Sphinx	≥5.0	文档生成
数据库	Access	pyodbc	≥4.0	Access 数据库操作
图像	处理	Pillow	≥10.0	图像处理

十二、附录¶

12.1 缩略语¶

缩略语	全称	说明
ORM	Object-Relational Mapping	对象关系映射
QSS	Qt Style Sheets	Qt 样式表
ESC/P	Epson Standard Code for Printers	爱普生打印控制命令
GDI	Graphics Device Interface	Windows 图形设备接口
DN	Diameter Nominal	公称通径
PN	Pressure Nominal	公称压力
CLI	Command Line Interface	命令行接口
GUI	Graphical User Interface	图形用户界面
CRUD	Create, Read, Update, Delete	增删改查
MVC	Model-View-Controller	模型-视图-控制器

12.2 设计模式应用¶

设计模式	应用场景	示例
单例模式	日志、配置、数据库	`LoggerManager`, `ConfigLoader`, `DatabaseManager`
工厂模式	处理器创建	`create_handler()`
策略模式	打印策略	`GDIPrintStrategy`, `HTMLPrintStrategy`
观察者模式	信号/槽通信	Qt 信号机制
委托模式	数据保存	`SaveHandler` 委托给 Service
模板方法模式	报告生成	`ReportGenerator.generate()`
命令模式	CLI 命令	`BaseCommand` 子类

12.3 性能优化设计¶

优化点	方式	说明
批量操作	`bulk_insert_mappings()`	批量插入/更新
分页查询	`offset() + limit()`	大数据量分页
配置缓存	`_config_cache`	避免重复加载
QSS 缓存	`_qss_cache`	避免重复渲染
异步处理	QThread	耗时操作不阻塞 UI
延迟加载	`@property`	按需加载

12.4 安全设计¶

安全项	方式	说明
SQL 注入防护	SQLAlchemy 参数化查询	防止 SQL 注入
事务管理	`session.begin()` / `rollback()`	数据一致性
文件路径验证	`path_utils.validate_path()`	防止路径遍历
日志脱敏	自定义过滤器	敏感信息脱敏
输入验证	Pydantic 模型	配置数据验证

12.5 版本历史¶

版本	日期	变更说明
v1.0	2026-06-01	初始版本
v2.0	2026-06-17	同步实际项目结构：新增 CLI、widgets、测试模块，完善流程图

📌 文档维护：本文档应与 src/certflow/ 代码保持同步。如有架构变更，请及时更新此文档。