certflow.utils package

Submodules

• certflow.utils.access_db module
  • is_access_enabled()
  • AccessDatabase
    • AccessDatabase.query()
    • AccessDatabase.get_unsigned_records()
    • AccessDatabase.get_signed_records()
    • AccessDatabase.get_next_unsigned()
    • AccessDatabase.get_by_id()
    • AccessDatabase.search()
    • AccessDatabase.get_fields()
    • AccessDatabase.get_customer_stats()
    • AccessDatabase.get_model_stats()
    • AccessDatabase.get_statistics()
    • AccessDatabase.append_signed_records_to_form()
    • AccessDatabase.export_to_csv()
    • AccessDatabase.close()
    • AccessDatabase.__enter__()
    • AccessDatabase.__exit__()
  • AccessDatabaseManager
    • AccessDatabaseManager.get_instance()
    • AccessDatabaseManager.get_connection()
    • AccessDatabaseManager.close()
    • AccessDatabaseManager.__enter__()
    • AccessDatabaseManager.__exit__()
  • get_access_db()
  • get_access_db_new()
• certflow.utils.database module
  • NutstoreWebDAVConfig
    • NutstoreWebDAVConfig.get_auth()
    • NutstoreWebDAVConfig.get_remote_url()
  • DatabaseManager
    • DatabaseManager.db_path
    • DatabaseManager.engine
    • DatabaseManager.SessionLocal
    • DatabaseManager.webdav_config
    • DatabaseManager.init_db()
    • DatabaseManager.ensure_initialized_if_needed()
    • DatabaseManager.warn_if_base_data_empty()
    • DatabaseManager.commit_with_retry()
    • DatabaseManager.flush_with_retry()
    • DatabaseManager.pull_from_webdav()
    • DatabaseManager.push_to_webdav()
    • DatabaseManager.sync_with_webdav()
    • DatabaseManager.check_remote_status()
    • DatabaseManager.get_session()
    • DatabaseManager.close()
• certflow.utils.date_utils module
  • safe_qdate()
  • parse_qdate()
  • normalize_plan_date()
  • extract_yymm_from_date()
  • extract_yymm_int_from_date()
• certflow.utils.id_generator module
  • IDGenerator
    • IDGenerator.generate_certificate_no()
    • IDGenerator.generate_unique_key()
    • IDGenerator.generate_unique_key_readable()
    • IDGenerator.generate_identity_key()
    • IDGenerator.generate_change_signature()
    • IDGenerator.normalize_spec()
    • IDGenerator.normalize_pn()
    • IDGenerator.generate_batch_unique_keys()
    • IDGenerator.normalize_empty_values()
    • IDGenerator.normalize_dataframe_empty_values()
    • IDGenerator.get_all_unique_key_fields()
• certflow.utils.logger module
  • setup_logger()
  • get_logger()
• certflow.utils.path_utils module
  • IS_FROZEN
  • EXE_DIR
  • PROJECT_ROOT
  • CONFIG_DIR
  • SCRIPTS_DIR
  • BUILDER_DIR
  • CONFIG_PATH
  • BUILD_CONFIG_PATH
  • is_frozen()
  • get_exe_dir()
  • get_project_root()
  • get_config_dir()
  • get_config_path()
  • get_build_config_path()
  • get_scripts_dir()
  • get_builder_dir()
  • setup_path()
• certflow.utils.retry module
  • retry_on_locked()
  • with_db_retry()
• certflow.utils.status_inference module
  • StatusInference
    • StatusInference.STATUS_RULES
    • StatusInference.QUERY_KEYWORDS
    • StatusInference.init_from_config()
    • StatusInference.infer()
    • StatusInference.get_query_keyword()
    • StatusInference.load_rules_from_config()
    • StatusInference.extract_date()

Module contents

工具模块

提供数据库管理和日志配置等核心工具类与函数。 是CertFlow系统基础架构的重要组成部分。

class certflow.utils.DatabaseManager(db_path=None, webdav_config=None, auto_backup_before_sync=True)

基类：object

数据库管理器

管理数据库引擎、会话工厂和表创建，提供统一的数据库操作入口。 支持数据库初始化、表迁移(自动补齐缺失列)和会话管理。 支持从WebDAV服务器同步数据库文件。

参数:

• db_path (str | None)

• webdav_config (NutstoreWebDAVConfig | None)

• auto_backup_before_sync (bool)

db_path

数据库文件路径

engine

SQLAlchemy引擎实例，负责数据库连接

SessionLocal

会话工厂实例，用于创建数据库会话

webdav_config

WebDAV配置（如果启用）

示例

>>> from certflow.utils.database import DatabaseManager
>>>
>>> # 创建数据库管理器
>>> db = DatabaseManager("path/to/database.db")
>>>
>>> # 初始化数据库
>>> db.init_db(create_tables=True)
>>>
>>> # 从WebDAV拉取数据库
>>> webdav_config = NutstoreWebDAVConfig(
...     url="https://dav.jianguoyun.com/dav/",
...     username="user@example.com",
...     password="app_password"
... )
>>> db.init_db(sync_from_webdav=True)
>>>
>>> # 获取会话并执行操作
>>> session = db.get_session()
>>> try:
...     result = session.query(SalePlan).all()
... finally:
...     session.close()

init_db(create_tables=True, sync_from_webdav=False, force_sync=False)

初始化数据库

创建数据库目录、SQLAlchemy引擎和会话工厂， 可选创建所有数据表。已有表会自动补齐缺失列。 可选从WebDAV同步数据库文件。

参数:

• create_tables (bool) -- 是否创建数据表，默认为True

• sync_from_webdav (bool) -- 是否从WebDAV同步数据库，默认为False

• force_sync (bool) -- 是否强制同步（即使本地数据库存在也尝试从远程拉取）

抛出:

Exception -- 当数据库初始化失败时抛出异常

返回类型:

None

示例

>>> db = DatabaseManager("test.db")
>>> db.init_db(create_tables=True)
>>> print("数据库初始化完成")

ensure_initialized_if_needed()

仅在引擎尚未创建时初始化数据库（幂等、低开销）

供装配层（BaseController / AppContext）在「不确定当前是否已完成 引导」的构造路径上调用：已初始化则直接跳过，未初始化则补建表/迁移。 区别于 init_db 的「总是执行完整初始化」，本方法避免在每次 Controller 构造时重复跑建表与迁移流程。

返回类型:

None

warn_if_base_data_empty()

检查基础字典底座是否为空，为空则给出 seed 提示（只读、不自动播种）。

对应 DATABASE_BASE_DATA.md §4.2 的「可选加固」：在 init_db 之后挂载 一个轻量检查——若 material_grades / model_param_mappings / caliber_mappings 任一为空，记录 WARNING 并提示运行 scripts/data/seed_base_data.py。

刻意**不**自动播种（§4.2 明确「不要每次启动自动 seed」，避免依赖 xlsx / 拖慢启动）；自动播种由 main._maybe_seed_base_data 在 GUI 启动路径单独负责。 CLI / 双库等不走自动播种的入口调用本方法即可获得「空库提示」安全网。

返回:

字典表是否为空（True=空，需 seed）。

返回类型:

bool

static commit_with_retry(session, max_attempts=12)

带退避重试的 session.commit，专治瞬时写锁。

直接在业务 session 上调用，遇 database is locked 时指数退避 重试（base 0.05s / 上限 0.8s / 累计约 8s）。配合 WAL+busy_timeout 使用，几乎可消除打印流程中的写保护锁现象。

参数:

• session (Any) -- SQLAlchemy Session（其引擎需已设 busy_timeout）。

• max_attempts (int) -- 最大尝试次数。

抛出:

非锁类异常或重试耗尽后仍失败时原样抛出。 --

返回类型:

None

static flush_with_retry(session, max_attempts=12)

带退避重试的 session.flush（写库前先落盘也可能需要抢锁）。

参数:

• session (Any)

• max_attempts (int)

返回类型:

None

pull_from_webdav(backup_before_pull=True)

从坚果云拉取最新数据库备份

参数:

backup_before_pull (bool) -- 拉取前是否备份本地数据库

返回:

是否成功拉取并替换

返回类型:

bool

push_to_webdav(backup_remote=True)

上传本地数据库到坚果云

参数:

backup_remote (bool) -- 是否备份远程数据库（保留旧版本）

返回:

是否成功上传

返回类型:

bool

sync_with_webdav(direction='pull')

与WebDAV服务器同步数据库

参数:

direction (str) -- 同步方向，"pull"（从服务器拉取）或 "push"（推送到服务器）

返回:

是否同步成功

返回类型:

bool

check_remote_status()

检查远程备份状态

返回:

包含远程备份信息的字典

返回类型:

dict[str, Any]

get_session()

获取数据库会话

如果会话工厂未初始化则先初始化数据库，然后创建并返回新会话。

返回:

SQLAlchemy数据库会话实例，可用于执行数据库操作

返回类型:

Session

示例

>>> db = DatabaseManager()
>>> session = db.get_session()
>>> try:
...     # 执行数据库操作
...     result = session.query(SalePlan).filter_by(id=1).first()
... finally:
...     session.close()

close()

关闭数据库连接

释放数据库引擎资源，关闭所有活动连接。

示例

>>> db = DatabaseManager()
>>> db.init_db()
>>> # 使用数据库...
>>> db.close()  # 程序退出前释放资源

返回类型:

None

class certflow.utils.AccessDatabase(db_path, max_retries=3, keep_alive=True)

基类：object

Access 数据库管理器

专门用于连接和查询 Access 数据库，独立于主 SQLite 数据库。 支持连接复用，避免频繁开关连接。

示例

>>> # 推荐：使用上下文管理器（自动管理连接）
>>> with get_access_db() as db:
...     records = db.search(customer="宁波大隆")

>>> # 或：获取全局单例（保持连接）
>>> db = get_access_db()
>>> records = db.search(customer="宁波大隆")

参数:

• db_path (str | Path)

• max_retries (int)

• keep_alive (bool)

query(sql, params=None)

执行查询并返回 DataFrame

参数:

• sql (str) -- SQL 查询语句。

• params (list | None) -- 查询参数列表，默认为 None。

返回:

查询结果 DataFrame，无结果时返回空 DataFrame（保留列名）。

返回类型:

pd.DataFrame

抛出:

pyodbc.Error -- SQL 执行失败时抛出。

get_unsigned_records()

查询未刻印的记录

返回:

未刻印记录的 DataFrame，按 ID 升序排列。

返回类型:

pd.DataFrame

get_signed_records(top_n=10)

查询已刻印的记录

参数:

top_n (int) -- 返回的记录数，默认为 10。

返回:

已刻印记录的 DataFrame，按 ID 降序排列。

返回类型:

pd.DataFrame

get_next_unsigned()

获取下一条未刻印的记录

返回:

按 ID 升序排列的第一条未刻印记录（TOP 1）。

返回类型:

pd.DataFrame

get_by_id(record_id)

根据ID查询记录

参数:

record_id (int) -- 记录的主键 ID。

返回:

匹配记录的 DataFrame。

返回类型:

pd.DataFrame

search(customer=None, project=None, model=None, sn=None, signed=None)

多条件搜索

参数:

• customer (str | None) -- 订货单位名称（模糊匹配），为 None 时不筛选。

• project (str | None) -- 项目名称（模糊匹配），为 None 时不筛选。

• model (str | None) -- 产品型号（模糊匹配），为 None 时不筛选。

• sn (str | None) -- 出厂编号（模糊匹配），为 None 时不筛选。

• signed (bool | None) -- 刻印状态筛选，True=已刻印，False=未刻印，None=不筛选。

返回:

符合所有条件的记录 DataFrame，按 ID 降序排列。

返回类型:

pd.DataFrame

get_fields(table_name)

获取表的字段列表

参数:

table_name (str) -- 表名

返回:

字段名列表

返回类型:

list[str]

get_customer_stats()

按订货单位统计

返回:

包含订货单位和数量的 DataFrame，按数量降序排列。

返回类型:

pd.DataFrame

get_model_stats()

按产品型号统计

返回:

包含产品型号和数量的 DataFrame，按数量降序排列。

返回类型:

pd.DataFrame

get_statistics()

获取统计信息

返回:

包含以下键的统计字典：

• 总记录数: int

• 已刻印记录数: int

• 未刻印记录数: int

• 最新刻印时间: str | None

• ID范围: str

返回类型:

dict[str, Any]

抛出:

pyodbc.Error -- 查询失败时抛出。

append_signed_records_to_form(plan_date)

将已刻印记录追加到 FormResumeDataTb

参数:

plan_date (str) -- 计划日期

返回:

插入的记录数

返回类型:

int

export_to_csv(output_path, table_name='FormResumeDataTb')

导出表到CSV

参数:

• output_path (str) -- CSV 输出文件路径。

• table_name (str) -- 要导出的表名，默认为 "FormResumeDataTb"。

返回:

导出的数据 DataFrame。

返回类型:

pd.DataFrame

close()

关闭数据库连接

关闭游标和连接，释放资源。

返回:

None

返回类型:

None

__enter__()

上下文管理器入口

返回:

当前 AccessDatabase 实例。

返回类型:

AccessDatabase

__exit__(exc_type, exc_val, exc_tb)

上下文管理器出口，自动关闭连接

参数:

• exc_type (Any) -- 异常类型（无异常时为 None）。

• exc_val (Any) -- 异常值（无异常时为 None）。

• exc_tb (Any) -- 异常回溯（无异常时为 None）。

返回类型:

None

certflow.utils.get_access_db()

获取 Access 数据库连接实例（复用全局连接）

推荐使用此函数获取连接，内部会自动管理连接生命周期。 适用于需要频繁查询的场景。

返回:

实例（全局单例）。

返回类型:

AccessDatabase

抛出:

ImportError -- Access 功能被配置禁用时抛出。

certflow.utils.is_access_enabled()

检查 Access 功能是否启用

优先级：环境变量 > 配置文件 > 默认禁用

返回:

True 如果 Access 功能已启用

返回类型:

bool

certflow.utils.setup_logger(log_path='logs/certflow.log', console_level='INFO', file_level='DEBUG', debug_modules=None)

配置日志系统(兼容旧接口)

提供简化的函数式接口，用于快速配置日志系统。

参数:

• log_path (str) -- 日志文件路径，默认为"logs/certflow.log"

• console_level (str) -- 控制台输出级别，默认为"INFO"

• file_level (str) -- 文件输出级别，默认为"DEBUG"

• debug_modules (list[str] | None) -- DEBUG 模块黑名单，["*"] 屏蔽全部，[] 全部输出

返回:

配置好的日志实例

返回类型:

loguru.Logger

示例

>>> from certflow.utils.logger import setup_logger
>>>
>>> # 使用默认配置
>>> logger = setup_logger()
>>>
>>> # 自定义配置
>>> logger = setup_logger(
...     log_path="logs/app.log",
...     console_level="DEBUG",
...     file_level="INFO"
... )
>>> logger.info("应用启动成功")

certflow.utils.get_logger()

获取全局日志实例

获取已配置的全局日志实例，如果尚未配置则使用默认配置。

返回:

全局日志实例

返回类型:

loguru.Logger

示例

>>> from certflow.utils.logger import get_logger
>>>
>>> logger = get_logger()
>>> logger.info("这是一条信息日志")
>>> logger.error("这是一条错误日志")

certflow.utils.extract_yymm_from_date(plan_date)

从计划日期提取 YYMM 格式（后两位年份+两位月份）

先通过 normalize_plan_date 统一格式化为 YYYY-MM-DD， 再从中提取年份后两位和月份。

参数:

plan_date (Any) -- 日期对象或日期字符串（支持任意 normalize_plan_date 支持的格式）

返回:

格式如 "2603" 的 YYMM 字符串，失败返回 None

返回类型:

Optional[str]

certflow.utils.extract_yymm_int_from_date(plan_date)

从计划日期提取 YYMM 整数格式

参数:

plan_date (Any) -- 日期对象或日期字符串

返回:

如 2603 的整数，失败返回 None

返回类型:

Optional[int]

certflow.utils.normalize_plan_date(date_value)

将日期值统一格式化为 YYYY-MM-DD 标准格式（不带时分秒）。

全局统一的日期格式化方法，所有模块导入 plan_date 时均应调用此方法。

支持格式：

• datetime / pandas.Timestamp 对象

• 字符串: 2024-01-15 14:30:00, 2024-01-15, 2024/01/15, 2024.01.15, 15-01-2024, 15/01/2024, 20240115, 240115

参数:

date_value (Any) -- 原始日期值，可以是 datetime、Timestamp、字符串或 None

返回:

格式化后的 YYYY-MM-DD 日期字符串，

如果值为空、无效或无法解析，返回 None

返回类型:

str | None