certflow.utils.access_db module

Access 数据库操作模块

用于连接和查询 Access 数据库（.mdb/.accdb）， 独立于主 SQLite 数据库，专门处理铭牌数据。

控制方式（优先级从高到低）： 1. 环境变量 CERTFLOW_ENABLE_ACCESS_EXPORT=true 强制启用 2. 配置文件 config.yaml → export.access.enabled: true 3. 以上均未设置时默认禁用

certflow.utils.access_db.is_access_enabled()

检查 Access 功能是否启用

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

返回:

True 如果 Access 功能已启用

返回类型:

bool

class certflow.utils.access_db.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

class certflow.utils.access_db.AccessDatabaseManager

基类：object

Access 数据库连接管理器（单例模式）

管理全局唯一的数据库连接实例，避免频繁开关连接。

示例

>>> manager = AccessDatabaseManager.get_instance()
>>> db = manager.get_connection()
>>> records = db.search(customer="宁波")

返回类型:

AccessDatabaseManager

classmethod get_instance()

获取单例实例

返回:

单例实例。

返回类型:

AccessDatabaseManager

get_connection(db_path=None)

获取数据库连接（复用已存在的连接）

参数:

db_path (str | Path | None) -- 数据库路径，仅在首次连接时使用

返回:

AccessDatabase 实例

返回类型:

AccessDatabase

close()

关闭全局连接

关闭并释放全局数据库连接实例。

返回:

None

返回类型:

None

__enter__()

上下文管理器入口，自动获取连接

返回:

连接实例。

返回类型:

AccessDatabase

__exit__(exc_type, exc_val, exc_tb)

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

参数:

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

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

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

返回类型:

None

certflow.utils.access_db.get_access_db()

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

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

返回:

实例（全局单例）。

返回类型:

AccessDatabase

抛出:

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

certflow.utils.access_db.get_access_db_new()

获取新的 Access 数据库连接实例（不复用）

适用于短期使用的场景，使用后需要手动关闭或使用上下文管理器。

返回:

新的连接实例。

返回类型:

AccessDatabase

抛出:

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