certflow.main 源代码
# src/certflow/main.py
"""CertFlow 主入口
桌面应用启动入口：初始化日志、数据库管理器、状态推断与主窗口，
并进入 Qt 事件循环。
"""
from __future__ import annotations
import os
import subprocess
import sys
from pathlib import Path
from dotenv import load_dotenv
from loguru import logger
from PySide6 import QtCore
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication
from certflow.config.settings import cfg
from certflow.controllers.base_controller import BaseController
from certflow.utils.database import DatabaseManager, NutstoreWebDAVConfig
from certflow.utils.logger import setup_logger
from certflow.views.main_window import MainWindow
[文档]
def setup_webdav_config() -> NutstoreWebDAVConfig | None:
"""从环境变量加载WebDAV配置
Returns:
NutstoreWebDAVConfig | None: 当环境变量 ``NUTSTORE_WEBDAV_URL``、
``NUTSTORE_WEBDAV_USER``、``NUTSTORE_WEBDAV_PASSWORD`` 均存在时返回
配置对象，否则返回 None。
"""
load_dotenv() # 确保加载.env文件
url = os.getenv("NUTSTORE_WEBDAV_URL")
username = os.getenv("NUTSTORE_WEBDAV_USER")
password = os.getenv("NUTSTORE_WEBDAV_PASSWORD")
if all([url, username, password]):
from certflow.utils.database import NutstoreWebDAVConfig
logger.info("检测到WebDAV配置，将启用远程同步功能")
return NutstoreWebDAVConfig(
url=url,
username=username,
password=password,
remote_path=os.getenv("NUTSTORE_REMOTE_PATH", "CertFlow/backups"),
keep_backups=int(os.getenv("NUTSTORE_KEEP_BACKUPS", "10")),
)
return None
def _maybe_seed_base_data(db_manager: DatabaseManager) -> None:
"""若基础字典底座（材质/型号压力/口径）为空，则自动播种。
仅在新空白数据库场景下触发（业务数据缺失不影响，步骤均 upsert）。
播种失败不阻断应用启动；详细录入效果可通过 VALVE_SEED_DEBUG / VALVE_DEBUG
环境变量开启（subprocess 输出与最终计数会写入日志）。
设计要点：型号参数字典只播种「型号→压力」，标准号留空——标准号以合格证
打印指令回填最准确，底座不应写入可能不准的通用标准号（避免导入期标黄误报）。
"""
from certflow.models.caliber_mapping import CaliberMapping
from certflow.models.material_grade import MaterialGrade
from certflow.models.model_param_mapping import ModelParamMapping
session = db_manager.get_session()
try:
base_empty = (
session.query(MaterialGrade).count() == 0
or session.query(ModelParamMapping).count() == 0
or session.query(CaliberMapping).count() == 0
)
finally:
session.close()
if not base_empty:
logger.info("基础字典底座已存在，跳过自动播种")
return
debug = bool(os.getenv("VALVE_SEED_DEBUG") or os.getenv("VALVE_DEBUG"))
logger.info("检测到空白数据库，开始自动填充基础字典底座（型号→压力 / 口径 / 材质）...")
root = Path(__file__).resolve().parents[2]
cmd = [sys.executable, str(root / "scripts" / "data" / "seed_base_data.py"), "--skip-standard"]
if debug:
cmd.append("--debug")
# 子进程输出为 UTF-8；显式注入 PYTHONIOENCODING 保证跨平台（尤其 Windows GBK locale）
# 输出编码一致，避免 subprocess 内部读取线程按 GBK 解码 UTF-8 字节而抛 UnicodeDecodeError。
seed_env = {**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1"}
proc = subprocess.run(
cmd,
cwd=str(root),
capture_output=True,
env=seed_env,
)
stdout = proc.stdout.decode("utf-8", errors="replace") if proc.stdout else ""
stderr = proc.stderr.decode("utf-8", errors="replace") if proc.stderr else ""
if stdout:
for line in stdout.rstrip().splitlines():
(logger.info if debug else logger.debug)(f"[seed-base] {line}")
if proc.returncode != 0:
logger.warning(
f"基础字典底座自动播种失败(exit={proc.returncode})，应用将继续启动：\n{stderr[-1500:]}"
)
else:
logger.info("基础字典底座自动播种完成")
[文档]
def init_database_manager() -> DatabaseManager:
"""初始化数据库管理器（支持WebDAV同步）
Returns:
DatabaseManager: 已初始化（必要时从远程同步）的数据库管理器实例。
"""
webdav_config = setup_webdav_config()
# 创建数据库管理器
db_manager = DatabaseManager(webdav_config=webdav_config, auto_backup_before_sync=True)
# 初始化数据库（如果配置了WebDAV，自动从远程同步）
# 注意：这里只调用一次 init_db
db_manager.init_db(
create_tables=True,
sync_from_webdav=bool(webdav_config), # 有配置就启用同步
force_sync=False, # 不强制覆盖现有数据库
)
# 空白数据库自动填充基础字典底座（非阻断）
_maybe_seed_base_data(db_manager)
# 安全网：若自动播种后仍为空（如 seed 子进程失败），给出 seed 提示（只读，不自动 seed）
db_manager.warn_if_base_data_empty()
return db_manager
[文档]
def check_data_sources() -> None:
"""启动前检查关键数据源是否存在，缺失仅记 warning（不弹窗、不阻断启动）。
- 销售计划文件（SALES_PLAN_PATH）：缺失则提示用户将其放到该路径。
- 基线库（database/baseline.db 与 database/samples/baseline.sample.db）：
均缺失则提示放入本地基线 C 或接受新建空白库（详细提示由 db-reset 给出）。
"""
from certflow.config.settings import SALES_PLAN_PATH
from certflow.utils.path_utils import get_project_root
# 销售计划文件
if SALES_PLAN_PATH:
p = Path(SALES_PLAN_PATH)
if not p.is_absolute():
p = get_project_root() / p
if not p.is_file():
logger.warning(
f"销售计划文件缺失: {p}\n"
f" 请将销售计划 Excel 放到上述路径（或在 .env 中修改 "
f"VALVE_SALES_PLAN_PATH 指向正确文件），否则销售计划相关功能不可用。"
)
else:
logger.warning(
"未配置销售计划路径（VALVE_SALES_PLAN_PATH / paths.sales_plan.primary），"
"销售计划相关功能将不可用。"
)
# 基线库（仅提示，详细处置见 certflow db-reset）
# 路径统一从 config/paths.yaml 读取（paths.database.baseline_local / baseline_sample），
# 消除硬编码，与 db-reset / start.sh 口径一致。
root = get_project_root()
baseline_c = root / cfg("paths.database.baseline_local", "database/baseline.db")
baseline_d = root / cfg("paths.database.baseline_sample", "database/samples/baseline.sample.db")
if not baseline_c.is_file() and not baseline_d.is_file():
logger.warning(
f"未找到任何基线库：本地基线 C ({baseline_c}) 与样本基线 D ({baseline_d}) 均缺失。\n"
f" 如需历史参考数据，请将基线库放到 {baseline_c}（或配置 Git LFS 拉取样本库）；\n"
f" 否则首次启动将新建空白库并自动播种基础字典底座。"
)
[文档]
def main() -> None:
"""主函数
解析配置、初始化日志与数据库、构建主窗口并进入 Qt 事件循环。
Examples:
>>> from certflow.main import main
>>> main() # 启动桌面应用
"""
# 从配置读取 DEBUG 模块黑名单
debug_modules: list[str] = cfg("logging.debug_modules", [])
# 初始化日志
setup_logger(debug_modules=debug_modules)
# 启动前数据源检查（缺失仅 warning，不弹窗、不阻断）
check_data_sources()
# 过滤 Shiboken _pythonToCppCopy 噪声（Qt 内部通过 C++ fprintf(stderr) 输出，
# 绕过 Python sys.stderr wrapper，故用环境变量 QT_LOGGING_RULES 抑制）
# 原先由 start.sh 的 `| grep -v` 管道过滤，但管道缓冲会导致控制台日志延迟/丢失
os.environ.setdefault("QT_LOGGING_RULES", "*.debug=false")
# 仅抑制 Shiboken 的 _pythonToCppCopy 警告，保留其他 Qt 日志
_existing_rules = os.environ.get("QT_LOGGING_RULES", "")
if "shiboken" not in _existing_rules.lower():
os.environ["QT_LOGGING_RULES"] = (
_existing_rules + ";shiboken.warning=false"
if _existing_rules
else "shiboken.warning=false"
)
# 初始化状态推断模块（从 config.yaml 加载规则）
from certflow.utils.status_inference import StatusInference
StatusInference.init_from_config()
# ---- 高 DPI 适配 ----
# 启用 Qt 内置高 DPI 缩放，使控件/字体/行高在不同屏幕缩放（125%/150%）下
# 都按 devicePixelRatio 一致缩放，根除「同份代码在不同显示器上行高变高/错位」
# 的问题（之前未开启时 resizeRowsToContents 与 setFixedHeight 的像素度量不一致）。
# 尊重环境变量：设 QT_AUTO_SCALE_FACTOR=1 走自动感知，或显式 QT_SCALE_FACTOR=1.25
# 可固定缩放比；二者皆未设时回退到框架默认缩放。
if not os.environ.get("QT_SCALE_FACTOR") and not os.environ.get("QT_AUTO_SCALE_FACTOR"):
os.environ.setdefault("QT_AUTO_SCALE_FACTOR", "1")
# AA_EnableHighDpiScaling / AA_UseHighDpiPixmaps 在 Qt6 已废弃（高 DPI 默认开启），
# 设置会触发 DeprecationWarning 且无效果；仅在 Qt5 下保留以保持旧行为。
if QtCore.qVersion().startswith("5"):
if not QApplication.testAttribute(Qt.ApplicationAttribute.AA_EnableHighDpiScaling):
QApplication.setAttribute(Qt.ApplicationAttribute.AA_EnableHighDpiScaling, True)
QApplication.setAttribute(Qt.ApplicationAttribute.AA_UseHighDpiPixmaps, True)
else:
# Qt6：高 DPI 默认启用；统一缩放比取整策略，避免同份代码在不同显示器行高错位。
QApplication.setHighDpiScaleFactorRoundingPolicy(
Qt.HighDpiScaleFactorRoundingPolicy.PassThrough
)
# 创建应用
app = QApplication(sys.argv)
# 初始化数据库管理器（会自动创建表和同步）
db_manager = init_database_manager()
# 获取数据库会话
session = db_manager.get_session()
# 设置全局共享数据库管理器（让所有控制器自动使用）
BaseController.set_shared_db_manager(db_manager)
# 构建配置字典（使用 _cfg 读取）
config = {
"print": cfg("print", {}),
"printer": cfg("printer", {}),
"templates": cfg("templates", {}),
"print_templates": cfg("print_templates", {}),
"background": cfg("background", {}),
}
# 初始化设置控制器（主题/语言/自动保存）
from certflow.controllers.settings_controller import SettingsController
settings_controller = SettingsController()
# 创建主窗口（注入设置控制器）
window = MainWindow(
session, config, db_manager=db_manager, settings_controller=settings_controller
)
window.show()
# 运行应用
exit_code = app.exec()
# 清理资源
session.close()
db_manager.close()
BaseController.clear_shared_db_manager()
sys.exit(exit_code)
if __name__ == "__main__":
main()