"""Access 数据库操作模块
用于连接和查询 Access 数据库（.mdb/.accdb），
独立于主 SQLite 数据库，专门处理铭牌数据。
控制方式（优先级从高到低）：
1. 环境变量 CERTFLOW_ENABLE_ACCESS_EXPORT=true 强制启用
2. 配置文件 config.yaml → export.access.enabled: true
3. 以上均未设置时默认禁用
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
import pandas as pd
from loguru import logger
[文档]
def is_access_enabled() -> bool:
"""检查 Access 功能是否启用
优先级：环境变量 > 配置文件 > 默认禁用
Returns:
True 如果 Access 功能已启用
"""
env_val = os.getenv("CERTFLOW_ENABLE_ACCESS_EXPORT")
if env_val is not None:
return env_val.lower() in ("true", "1", "yes")
try:
from certflow.config.settings import cfg
return bool(cfg("export.access.enabled", False))
except Exception:
return False
def _import_pyodbc() -> Any:
"""延迟导入 pyodbc，仅在 Access 功能启用时才加载"""
if not is_access_enabled():
raise ImportError(
"Access 功能已禁用"
"（设置 export.access.enabled=true 或 CERTFLOW_ENABLE_ACCESS_EXPORT=true 启用）"
)
import pyodbc as _pyodbc
return _pyodbc
[文档]
class AccessDatabase:
"""Access 数据库管理器
专门用于连接和查询 Access 数据库，独立于主 SQLite 数据库。
支持连接复用，避免频繁开关连接。
Examples:
>>> # 推荐：使用上下文管理器（自动管理连接）
>>> with get_access_db() as db:
... records = db.search(customer="宁波大隆")
>>> # 或：获取全局单例（保持连接）
>>> db = get_access_db()
>>> records = db.search(customer="宁波大隆")
"""
def __init__(self, db_path: str | Path, max_retries: int = 3, keep_alive: bool = True):
"""初始化 Access 数据库连接
Args:
db_path: Access 数据库文件路径 (.mdb 或 .accdb)
max_retries: 连接失败时的最大重试次数
keep_alive: 是否保持连接（False 时每次查询后自动关闭）
Raises:
ImportError: Access 功能被配置禁用时抛出。
"""
self.db_path = Path(db_path)
self.conn = None
self.cursor = None
self.max_retries = max_retries
self.keep_alive = keep_alive
self._pyodbc = _import_pyodbc() # 延迟导入 + 配置检查
self._connect()
def _connect(self) -> None:
"""建立数据库连接（支持重试）
:raises FileNotFoundError: 数据库文件不存在时抛出
:raises ConnectionError: 所有重试均失败后抛出
"""
if not self.db_path.exists():
raise FileNotFoundError(f"数据库文件不存在: {self.db_path}")
conn_str = f"DRIVER={{Microsoft Access Driver (*.mdb, *.accdb)}};DBQ={self.db_path};"
for attempt in range(1, self.max_retries + 1):
try:
self.conn = self._pyodbc.connect(conn_str, timeout=10)
self.cursor = self.conn.cursor()
logger.info(
"Access 数据库连接成功",
extra={
"db_path": str(self.db_path),
"attempt": attempt,
"keep_alive": self.keep_alive,
},
)
return
except self._pyodbc.Error as e:
logger.warning(
f"连接失败 (尝试 {attempt}/{self.max_retries})",
extra={"db_path": str(self.db_path), "error": str(e)},
)
if attempt == self.max_retries:
logger.error("数据库连接最终失败", extra={"db_path": str(self.db_path)})
raise ConnectionError(f"无法连接到数据库: {self.db_path}") from e
def _ensure_connection(self) -> None:
"""确保连接有效，如果断开则重连
:raises pyodbc.Error: 重连失败时抛出
"""
try:
# 测试连接是否有效
self.cursor.execute("SELECT 1")
except (self._pyodbc.Error, AttributeError):
logger.warning("数据库连接已断开，尝试重新连接", extra={"db_path": str(self.db_path)})
self._connect()
def _execute_query(self, sql: str, params: list | None = None) -> tuple[list, list]:
"""执行查询并返回列名和数据（内部方法，避免 pandas 警告）
:param sql: SQL 查询语句
:param params: 查询参数列表，默认为 None
:return: 包含列名列表和数据列表的元组 (columns, data)
:raises pyodbc.Error: SQL 执行失败时抛出
"""
self._ensure_connection() # 确保连接有效
try:
if params:
self.cursor.execute(sql, params)
else:
self.cursor.execute(sql)
columns = (
[column[0] for column in self.cursor.description] if self.cursor.description else []
)
data = self.cursor.fetchall()
return columns, data
except self._pyodbc.Error as e:
logger.error(
"SQL 执行失败", extra={"sql": sql[:200], "params": params, "error": str(e)}
)
raise
[文档]
def query(self, sql: str, params: list | None = None) -> pd.DataFrame:
"""执行查询并返回 DataFrame
Args:
sql: SQL 查询语句。
params: 查询参数列表，默认为 None。
Returns:
pd.DataFrame: 查询结果 DataFrame，无结果时返回空 DataFrame（保留列名）。
Raises:
pyodbc.Error: SQL 执行失败时抛出。
"""
columns, data = self._execute_query(sql, params)
if not data:
return pd.DataFrame(columns=columns)
return pd.DataFrame.from_records(data, columns=columns)
def _safe_top_n(self, top_n: int) -> str:
"""安全处理 TOP N 值
:param top_n: 期望返回的记录数
:return: 经过校验和限制后的 TOP N 值（字符串格式）
"""
if not isinstance(top_n, int) or top_n <= 0:
logger.warning(f"无效的 top_n 值: {top_n}，使用默认值 10")
top_n = 10
if top_n > 10000:
logger.warning(f"top_n 值过大: {top_n}，限制为 10000")
top_n = 10000
return str(top_n)
[文档]
def get_unsigned_records(self) -> pd.DataFrame:
"""查询未刻印的记录
Returns:
pd.DataFrame: 未刻印记录的 DataFrame，按 ID 升序排列。
"""
sql = "SELECT * FROM FormResumeDataTb WHERE 刻印完毕标记 = 0 ORDER BY ID"
logger.debug("执行查询: 未刻印记录")
return self.query(sql)
[文档]
def get_signed_records(self, top_n: int = 10) -> pd.DataFrame:
"""查询已刻印的记录
Args:
top_n: 返回的记录数，默认为 10。
Returns:
pd.DataFrame: 已刻印记录的 DataFrame，按 ID 降序排列。
"""
safe_top = self._safe_top_n(top_n)
sql = f"SELECT TOP {safe_top} * FROM FormResumeDataTb WHERE 刻印完毕标记 = -1 ORDER BY ID DESC"
logger.debug(f"执行查询: 已刻印记录 (TOP {safe_top})")
return self.query(sql)
[文档]
def get_next_unsigned(self) -> pd.DataFrame:
"""获取下一条未刻印的记录
Returns:
pd.DataFrame: 按 ID 升序排列的第一条未刻印记录（TOP 1）。
"""
sql = "SELECT TOP 1 * FROM FormResumeDataTb WHERE 刻印完毕标记 = 0 ORDER BY ID"
logger.debug("执行查询: 下一条未刻印记录")
return self.query(sql)
[文档]
def get_by_id(self, record_id: int) -> pd.DataFrame:
"""根据ID查询记录
Args:
record_id: 记录的主键 ID。
Returns:
pd.DataFrame: 匹配记录的 DataFrame。
"""
sql = "SELECT * FROM FormResumeDataTb WHERE ID = ?"
logger.debug(f"执行查询: 根据ID查询 (ID={record_id})")
return self.query(sql, [record_id])
[文档]
def search(
self,
customer: str | None = None,
project: str | None = None,
model: str | None = None,
sn: str | None = None,
signed: bool | None = None,
) -> pd.DataFrame:
"""多条件搜索
Args:
customer: 订货单位名称（模糊匹配），为 None 时不筛选。
project: 项目名称（模糊匹配），为 None 时不筛选。
model: 产品型号（模糊匹配），为 None 时不筛选。
sn: 出厂编号（模糊匹配），为 None 时不筛选。
signed: 刻印状态筛选，True=已刻印，False=未刻印，None=不筛选。
Returns:
pd.DataFrame: 符合所有条件的记录 DataFrame，按 ID 降序排列。
"""
sql = "SELECT * FROM FormResumeDataTb WHERE 1=1"
params = []
if customer:
sql += " AND 订货单位 LIKE ?"
params.append(f"%{customer}%")
if project:
sql += " AND 项目名称 LIKE ?"
params.append(f"%{project}%")
if model:
sql += " AND 产品型号 LIKE ?"
params.append(f"%{model}%")
if sn:
sql += " AND 出厂编号 LIKE ?"
params.append(f"%{sn}%")
if signed is True:
sql += " AND 刻印完毕标记 = -1"
elif signed is False:
sql += " AND 刻印完毕标记 = 0"
sql += " ORDER BY ID DESC"
logger.debug(f"执行搜索: 条件数={len(params)}")
return self.query(sql, params)
[文档]
def get_fields(self, table_name: str) -> list[str]:
"""获取表的字段列表
Args:
table_name: 表名
Returns:
字段名列表
"""
self._ensure_connection()
self.cursor.execute(f"SELECT TOP 1 * FROM [{table_name}]")
return [column[0] for column in self.cursor.description]
[文档]
def get_customer_stats(self) -> pd.DataFrame:
"""按订货单位统计
Returns:
pd.DataFrame: 包含订货单位和数量的 DataFrame，按数量降序排列。
"""
sql = """
SELECT 订货单位, COUNT(*) as 数量
FROM FormResumeDataTb
GROUP BY 订货单位
ORDER BY COUNT(*) DESC
"""
logger.debug("执行统计: 订货单位统计")
return self.query(sql)
[文档]
def get_model_stats(self) -> pd.DataFrame:
"""按产品型号统计
Returns:
pd.DataFrame: 包含产品型号和数量的 DataFrame，按数量降序排列。
"""
sql = """
SELECT 产品型号, COUNT(*) as 数量
FROM FormResumeDataTb
GROUP BY 产品型号
ORDER BY COUNT(*) DESC
"""
logger.debug("执行统计: 产品型号统计")
return self.query(sql)
[文档]
def get_statistics(self) -> dict[str, Any]:
"""获取统计信息
Returns:
dict[str, Any]: 包含以下键的统计字典：
- 总记录数: int
- 已刻印记录数: int
- 未刻印记录数: int
- 最新刻印时间: str | None
- ID范围: str
Raises:
pyodbc.Error: 查询失败时抛出。
"""
stats = {}
logger.debug("开始获取统计信息")
self._ensure_connection() # 确保连接有效
try:
self.cursor.execute("SELECT COUNT(*) FROM FormResumeDataTb")
stats["总记录数"] = self.cursor.fetchone()[0]
self.cursor.execute("SELECT COUNT(*) FROM FormResumeDataTb WHERE 刻印完毕标记 = -1")
stats["已刻印记录数"] = self.cursor.fetchone()[0]
self.cursor.execute("SELECT COUNT(*) FROM FormResumeDataTb WHERE 刻印完毕标记 = 0")
stats["未刻印记录数"] = self.cursor.fetchone()[0]
self.cursor.execute(
"SELECT MAX(铭牌打印时间) FROM FormResumeDataTb WHERE 刻印完毕标记 = -1"
)
stats["最新刻印时间"] = self.cursor.fetchone()[0]
self.cursor.execute("SELECT MIN(ID), MAX(ID) FROM FormResumeDataTb")
min_id, max_id = self.cursor.fetchone()
stats["ID范围"] = f"{min_id} - {max_id}"
logger.info(
"统计信息获取成功",
extra={
"总记录数": stats["总记录数"],
"已刻印": stats["已刻印记录数"],
"未刻印": stats["未刻印记录数"],
},
)
except self._pyodbc.Error as e:
logger.error("获取统计信息失败", extra={"error": str(e)})
raise
return stats
[文档]
def export_to_csv(self, output_path: str, table_name: str = "FormResumeDataTb") -> pd.DataFrame:
"""导出表到CSV
Args:
output_path: CSV 输出文件路径。
table_name: 要导出的表名，默认为 "FormResumeDataTb"。
Returns:
pd.DataFrame: 导出的数据 DataFrame。
"""
logger.info(f"开始导出表: {table_name}")
df = self.query(f"SELECT * FROM [{table_name}]")
df.to_csv(output_path, index=False, encoding="utf-8-sig")
logger.info(f"导出完成: {len(df)} 条记录 -> {output_path}")
return df
[文档]
def close(self) -> None:
"""关闭数据库连接
关闭游标和连接，释放资源。
Returns:
None
"""
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()
logger.info("Access 数据库连接已关闭", extra={"db_path": str(self.db_path)})
[文档]
def __enter__(self) -> AccessDatabase:
"""上下文管理器入口
Returns:
AccessDatabase: 当前 AccessDatabase 实例。
"""
return self
[文档]
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""上下文管理器出口，自动关闭连接
Args:
exc_type: 异常类型（无异常时为 None）。
exc_val: 异常值（无异常时为 None）。
exc_tb: 异常回溯（无异常时为 None）。
"""
self.close()
# ==================== 单例管理器 ====================
[文档]
class AccessDatabaseManager:
"""Access 数据库连接管理器（单例模式）
管理全局唯一的数据库连接实例，避免频繁开关连接。
Examples:
>>> manager = AccessDatabaseManager.get_instance()
>>> db = manager.get_connection()
>>> records = db.search(customer="宁波")
"""
_instance = None
_connection = None
def __new__(cls) -> AccessDatabaseManager:
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
[文档]
@classmethod
def get_instance(cls) -> AccessDatabaseManager:
"""获取单例实例
Returns:
AccessDatabaseManager: 单例实例。
"""
if cls._instance is None:
cls._instance = cls()
return cls._instance
[文档]
def get_connection(self, db_path: str | Path | None = None) -> AccessDatabase:
"""获取数据库连接（复用已存在的连接）
Args:
db_path: 数据库路径，仅在首次连接时使用
Returns:
AccessDatabase 实例
"""
if self._connection is None:
if db_path is None:
from certflow.utils.path_utils import get_project_root
db_path = get_project_root() / "data" / "SignDB.mdb"
self._connection = AccessDatabase(db_path, keep_alive=True)
return self._connection
[文档]
def close(self) -> None:
"""关闭全局连接
关闭并释放全局数据库连接实例。
Returns:
None
"""
if self._connection:
self._connection.close()
self._connection = None
logger.info("全局数据库连接已关闭")
[文档]
def __enter__(self) -> AccessDatabase:
"""上下文管理器入口，自动获取连接
Returns:
AccessDatabase: 连接实例。
"""
return self.get_connection()
[文档]
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""上下文管理器出口，自动关闭全局连接
Args:
exc_type: 异常类型（无异常时为 None）。
exc_val: 异常值（无异常时为 None）。
exc_tb: 异常回溯（无异常时为 None）。
"""
self.close()
# ==================== 便捷函数 ====================
[文档]
def get_access_db() -> AccessDatabase:
"""获取 Access 数据库连接实例（复用全局连接）
推荐使用此函数获取连接，内部会自动管理连接生命周期。
适用于需要频繁查询的场景。
Returns:
AccessDatabase: 实例（全局单例）。
Raises:
ImportError: Access 功能被配置禁用时抛出。
"""
if not is_access_enabled():
raise ImportError(
"Access 功能已禁用"
"（设置 export.access.enabled=true 或 CERTFLOW_ENABLE_ACCESS_EXPORT=true 启用）"
)
return AccessDatabaseManager.get_instance().get_connection()
[文档]
def get_access_db_new() -> AccessDatabase:
"""获取新的 Access 数据库连接实例（不复用）
适用于短期使用的场景，使用后需要手动关闭或使用上下文管理器。
Returns:
AccessDatabase: 新的连接实例。
Raises:
ImportError: Access 功能被配置禁用时抛出。
"""
if not is_access_enabled():
raise ImportError(
"Access 功能已禁用"
"（设置 export.access.enabled=true 或 CERTFLOW_ENABLE_ACCESS_EXPORT=true 启用）"
)
from certflow.utils.path_utils import get_project_root
db_path = get_project_root() / "data" / "SignDB.mdb"
return AccessDatabase(str(db_path))