# certflow/config/settings.py
"""CertFlow 配置管理 - 所有值从 config.yaml / .env 读取，零硬编码
提供模块级配置常量(直接通过 cfg() 读取 YAML)和工具函数.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from typing import Any
import yaml
from loguru import logger
# 统一配置读取入口（覆盖层 > base，按值类型智能解析）。
# 放在顶部：paths_override 现已改为惰性导入 settings，不会形成循环导入。
from certflow.config.paths_override import cfg # noqa: E402
# ── .env 文件加载 ────────────────────────────────────────────
try:
from dotenv import load_dotenv as _dotenv_load
except ImportError:
_dotenv_load = None
def _load_env_file() -> None:
"""查找并加载 .env 文件，使环境变量在 YAML 解析前生效.
搜索优先级：
1. 环境变量 CERTFLOW_ENV_FILE 指定的路径
2. 当前工作目录 .env
3. 打包环境：EXE 同级 .env、EXE 同级 config/.env
4. 开发环境：项目根目录 .env、项目根目录 config/.env
仅在 python-dotenv 可用时加载，否则静默跳过.
"""
if _dotenv_load is None:
return
search_paths: list[Path] = []
# 1. 环境变量指定
if env_file := os.environ.get("CERTFLOW_ENV_FILE"):
search_paths.append(Path(env_file))
# 2. 当前工作目录
search_paths.append(Path.cwd() / ".env")
# 3. 打包环境：EXE 同级
if getattr(sys, "frozen", False) or hasattr(sys, "_MEIPASS2"):
exe_dir = Path(sys.executable).parent
search_paths.extend(
[
exe_dir / ".env",
exe_dir / "config" / ".env",
]
)
# 4. 项目根目录（开发环境）
try:
from certflow.utils.path_utils import get_project_root as _get_root
root = _get_root()
search_paths.extend(
[
root / ".env",
root / "config" / ".env",
]
)
except (ImportError, RuntimeError):
pass
for env_path in search_paths:
if env_path.exists():
_dotenv_load(env_path, override=False)
logger.debug(f"已加载 .env: {env_path}")
return
# 在 YAML 配置加载前执行，确保环境变量已就绪
_load_env_file()
from certflow.utils.path_utils import IS_FROZEN, get_project_root # noqa: E402
# ============================================================
# 路径辅助函数 - 适配打包环境
# ============================================================
def _get_base_dir() -> Path:
"""
获取应用程序基础目录
开发环境：项目根目录
打包环境：EXE 所在目录
Returns:
Path: 基础目录路径
"""
if IS_FROZEN:
# 打包环境：使用 exe 所在目录
return Path(sys.executable).parent
# 开发环境：使用项目根目录
return get_project_root()
BASE_DIR = _get_base_dir()
def _get_config_dir() -> Path:
"""
获取配置文件目录
开发环境：BASE_DIR/config
打包环境：BASE_DIR/config 或 BASE_DIR/../config
Returns:
Path: 配置文件目录
"""
config_dir = BASE_DIR / "config"
if config_dir.exists():
return config_dir
# 尝试上级目录的 config
parent_config = BASE_DIR.parent / "config"
if parent_config.exists():
return parent_config
return config_dir
CONFIG_DIR = _get_config_dir()
def _resolve_path(path_value: Any, base: Path = BASE_DIR) -> Path:
"""解析路径，将相对路径转换为绝对路径
支持相对路径(相对于BASE_DIR)和绝对路径两种形式.
Args:
path_value: 待解析的路径值，可以是字符串或Path对象
base: 基准目录，默认为应用程序根目录
Returns:
Path: 解析后的绝对路径
- 空值或None: 返回 base/output
- 绝对路径: 直接返回
- 相对路径: 相对于base拼接后返回
"""
if not path_value:
return base / "output"
if not isinstance(path_value, str):
path_value = str(path_value)
p = Path(path_value)
if p.is_absolute():
return p
return base / p
# ============================================================
# YAML 配置读取(惰性加载 + 缓存)
# ============================================================
_config_cache: dict | None = None
def _load_config() -> dict:
"""加载 config.yaml 为原始字典
使用惰性加载策略，仅在首次调用时加载配置并缓存.
Returns:
dict: 配置字典，包含所有YAML配置数据
"""
global _config_cache
if _config_cache is None:
from certflow.config.loader import get_config
# get_config() 返回 RootConfig 对象，转为 dict 以便 _cfg 兼容
# by_alias=True：使带 alias 的字段（如 import_placeholder→import）以别名导出，
# 其余字段无 alias 仍按字段名导出，不影响其他 _cfg 读取。
root = get_config()
_config_cache = root.model_dump(by_alias=True) if hasattr(root, "model_dump") else root
return _config_cache
def _cfg(key: str, default: Any = None) -> Any:
"""从 YAML 配置读取嵌套键值（仅 base 层，忽略覆盖层）。
覆盖层经 paths_override.cfg 读取。本函数是 base-only 的底层读取器，
供 paths_override.cfg 在无覆盖时回退调用，以及需要读"底座"的场景使用。
支持使用点号分隔的路径访问嵌套配置，如 "paths.sales_plan.primary".
Args:
key: 点号分隔的配置键路径
default: 键不存在时返回的默认值
Returns:
Any: 配置值，如果路径不存在则返回默认值
Examples:
>>> _cfg("app.name")
"CertFlow"
>>> _cfg("paths.database.dir", "database")
"database"
"""
config = _load_config()
parts = key.split(".")
value: Any = config
for part in parts:
if isinstance(value, dict):
value = value.get(part)
if value is None:
return default
else:
return default
return value
def _env(key: str) -> str:
"""读取环境变量值
Args:
key: 环境变量名称
Returns:
str: 环境变量的值，不存在时返回空字符串
"""
return os.getenv(key, "")
[文档]
def get_layout_groups(enabled_only: bool = True) -> list[dict]:
"""读取常用布局配置驱动组（output.layout_groups，D 延后仅预留接口）。
每个组是一套效果集的命名组合，UI 单选框/按钮后续加载时一键套用
target+template+layout 等到各下拉。
Args:
enabled_only: True 时仅返回 enabled 为真的组（供 UI 加载）。
Returns:
list[dict]: 每组含 name/target/template/layout/enabled 等键。
"""
groups = cfg("output.layout_groups", []) or []
if not isinstance(groups, list):
return []
if enabled_only:
return [g for g in groups if g.get("enabled", False)]
return groups
[文档]
def get_scan_layout_presets() -> list[dict]:
"""读取扫描拼版推荐值（output.targets.scan.defaults.layout_presets）。
供 ScanView「推荐拼版」下拉使用：用户选中即把 ``rows``/``cols`` 套用到拼版自旋框，
并灰显自旋框（尺寸固化思路的延伸）。列表缺失或非 list 时返回 []。
Returns:
list[dict]: 每个元素含 ``label`` / ``rows``（每列张数·纵） / ``cols``（每行张数·横）。
"""
presets = cfg("output.targets.scan.defaults.layout_presets", []) or []
return presets if isinstance(presets, list) else []
[文档]
def get_scan_default_layout() -> str:
"""读取启动默认选中的推荐拼版 label（output.targets.scan.defaults.default_layout）。
空字符串表示启动即「自定义」（拼版自旋框可编辑）；否则启动即套用该预设并灰显自旋框。
Returns:
str: 预设 label；未配置或为空时返回 ""。
"""
return str(cfg("output.targets.scan.defaults.default_layout", "") or "")
[文档]
def get_config_value(key: str, env_var: str | None = None, default: Any = None) -> Any:
"""获取配置值，优先级: 环境变量 > YAML > 传入默认值
提供统一的配置读取接口，优先使用环境变量，其次使用YAML配置，
最后使用代码中的默认值.
Args:
key: YAML配置中的键路径
env_var: 环境变量名称，可选
default: 默认值
Returns:
Any: 获取到的配置值
Examples:
>>> get_config_value("paths.sales_plan.primary", "VALVE_SALES_PLAN_PATH")
"/path/to/sales/plan.xlsx"
"""
if env_var:
value = _env(env_var)
if value:
return value
value = cfg(key)
if value is not None:
return value
return default
# ============================================================
# 应用信息
# ============================================================
APP_NAME: str = cfg("app.name", "CertFlow")
"""应用程序名称"""
APP_VERSION: str = cfg("app.version", "0.0.1")
"""应用程序版本号"""
APP_YEAR: int = cfg("app.year", 2026)
"""版权年份"""
APP_ORGANIZATION: str = cfg("app.organization", "CertFlow Team")
"""组织名称"""
APP_DEBUG: bool = cfg("app.debug", False)
"""是否启用调试模式"""
# ============================================================
# 目录路径
# ============================================================
CONFIG_DIR: Path = BASE_DIR / "config"
"""配置文件目录"""
DATABASE_DIR: Path = _resolve_path(cfg("paths.database.dir", "database"))
"""数据库文件目录"""
[文档]
def get_temp_dir() -> Path:
"""临时文件根目录。
统一走 paths.local.yaml 覆盖层（pcfg）：未配置时回退系统临时目录
（tempfile.gettempdir()，云环境正确）；本地开发可在 paths.local.yaml 写
paths.temp.base 指向项目内目录，以便在项目中观察临时文件。
"""
import tempfile
from certflow.config.paths_override import cfg
return _resolve_path(cfg("paths.temp.base", tempfile.gettempdir()))
LOGS_DIR: Path = _resolve_path(cfg("paths.logs", "logs"))
"""日志文件目录"""
RESOURCES_DIR: Path = _resolve_path(cfg("paths.resources", "resources"))
"""资源文件目录"""
UI_CONFIG_FILE: Path = CONFIG_DIR / "ui.yaml"
"""UI配置文件路径"""
# 数据库
DATABASE_PATH: Path = DATABASE_DIR / str(cfg("paths.database.name", "certflow.db"))
"""SQLite数据库文件完整路径"""
DATABASE_URL: str = f"sqlite:///{DATABASE_PATH}"
"""数据库连接URL"""
# ============================================================
# 文件路径(环境变量优先)
# ============================================================
SALES_PLAN_PATH: str | None = get_config_value("paths.sales_plan.primary", "VALVE_SALES_PLAN_PATH")
"""销售计划主文件路径(优先使用环境变量)"""
SALES_PLAN_NETWORK: str | None = get_config_value(
"paths.sales_plan.network", "VALVE_SALES_PLAN_NETWORK"
)
"""销售计划网络路径"""
SALES_PLAN_LOCAL_COPY: str | None = get_config_value(
"paths.sales_plan.local_copy", "VALVE_SALES_PLAN_LOCAL_COPY"
)
"""销售计划本地副本路径"""
SALES_PLAN_MASTER_FILE: str | None = get_config_value(
"paths.sales_plan.master_file", "VALVE_MASTER_ORDER_FILE"
)
"""主订单文件路径"""
SALES_PLAN_MASTER_SHEET: str | None = cfg("paths.sales_plan.master_sheet")
"""主订单工作表名称"""
QUALIFIED_LIST_PATH: str | None = get_config_value(
"paths.qualified_list.primary", "VALVE_QUALIFIED_LIST_PATH"
)
"""合格清单文件路径"""
QUALIFIED_LIST_XLSM_ORIGINAL: str | None = get_config_value(
"paths.qualified_list.xlsm_original", ""
)
"""原始合格清单XLSM文件路径"""
QUALIFIED_LIST_SHEET: str | None = cfg("paths.qualified_list.sheet")
"""合格清单工作表名称"""
NAMEPLATE_DB_PATH: str | None = get_config_value(
"paths.nameplate_db.primary", "VALVE_NAMEPLATE_DB_PATH"
)
"""铭牌数据库路径"""
NAMEPLATE_DB_BACKUP: str | None = cfg("paths.nameplate_db.backup")
"""铭牌数据库备份路径"""
QUALITY_CERT_FOREIGN_DIR: str | None = get_config_value(
"paths.quality_cert.foreign_dir", "VALVE_QUALITY_CERT_FOREIGN_DIR"
)
"""外协质量证书目录"""
QUALITY_CERT_SELF_MADE_DIR: str | None = cfg("paths.quality_cert.self_made_dir")
"""自制质量证书目录"""
QUALITY_CERT_REPORT_DIR: str | None = cfg("paths.quality_cert.report_dir")
"""质量证书报告目录"""
TRANSFER_DOCS_NETWORK: str | None = get_config_value(
"paths.transfer_docs.network", "VALVE_TRANSFER_DOCS_NETWORK"
)
"""交接文档网络路径"""
TRANSFER_DOCS_LOCAL: str | None = get_config_value(
"paths.transfer_docs.local", "VALVE_TRANSFER_DOCS_LOCAL"
)
"""交接文档本地路径"""
CERT_HISTORY_PATH: str | None = get_config_value(
"paths.backup.cert_history", "VALVE_CERT_BACKUP_PATH"
)
"""证书历史备份路径"""
# ============================================================
# 输出目录
# ============================================================
OUTPUT_BASE: Path = _resolve_path(cfg("paths.output.base", "output"))
"""输出文件根目录"""
OUTPUT_TEST_REPORT: Path = OUTPUT_BASE / str(cfg("paths.output.test_report", "reports"))
"""试压报告输出目录"""
OUTPUT_QUALITY_CERT: Path = OUTPUT_BASE / str(cfg("paths.output.quality_cert", "certificates"))
"""质量证书输出目录"""
OUTPUT_CERTIFICATE_IMAGES: Path = OUTPUT_BASE / str(
cfg("paths.output.certificate_images", "images")
)
"""证书图片输出目录"""
OUTPUT_NAMEPLATE: Path = OUTPUT_BASE / str(cfg("paths.output.nameplate", "nameplates"))
"""铭牌输出目录"""
OUTPUT_BACKUPS: Path = OUTPUT_BASE / str(cfg("paths.output.backups", "backups"))
"""备份文件输出目录"""
# 图片资源路径
NAMEPLATE_IMAGE_PATH: str | None = get_config_value(
"paths.images.nameplate", "VALVE_NAMEPLATE_IMAGE_PATH"
)
"""铭牌图片路径"""
NAMEPLATE_IMAGE_BACKUP: str | None = cfg("paths.images.nameplate_backup")
"""铭牌图片备份路径"""
# Excel 排版配置
ROW_HEIGHT_TITLE: Any = cfg("paths.excel_layout.row_height.title", 20)
"""Excel标题行高度"""
ROW_HEIGHT_HEADER: Any = cfg("paths.excel_layout.row_height.header", 25)
"""Excel表头行高度"""
ROW_HEIGHT_DATA: Any = cfg("paths.excel_layout.row_height.data", 18)
"""Excel数据行高度"""
ROW_HEIGHT_SUMMARY: Any = cfg("paths.excel_layout.row_height.summary", 22)
"""Excel汇总行高度"""
# ============================================================
# 编号规则配置
# ============================================================
NUMBERING_PREFIX: str = cfg("numbering.prefix", "V")
"""编号前缀(V2)"""
NUMBERING_FORMAT: str = cfg("numbering.format", "{prefix}{year}{month}{seq:03d}{suffix}")
"""编号格式模板(V2)"""
NUMBERING_SUFFIXES: dict = cfg("numbering.suffixes", {})
"""编号后缀映射(V2)"""
NUMBERING_AUTO_NUMBER_MODE: dict | None = cfg("numbering.auto_number_mode")
"""自动编号模式配置(V2)"""
NUMBERING_SUFFIX_RULES: list = cfg("numbering.suffix_rules", [])
"""编号后缀规则列表(V2)"""
# 旧接口兼容: certificate 段(V1 编号配置)
CERTIFICATE_PREFIX: str = cfg("certificate.prefix", NUMBERING_PREFIX)
"""证书编号前缀(V1兼容)"""
CERTIFICATE_DATE_FORMAT: str = cfg("certificate.date_format", "%Y%m%d")
"""证书日期格式(V1兼容)"""
CERTIFICATE_FORMAT_TEMPLATE: str = cfg("certificate.format_template", "{prefix}{date}{seq:04d}")
"""证书编号格式模板(V1兼容)"""
# ============================================================
# 颜色映射
# ============================================================
VBA_COLORS: dict = cfg("color_mapping.vba_colors", {})
"""VBA颜色索引到颜色的映射"""
FONT_COLOR_TO_SHIPPING: dict = cfg("color_mapping.font_color_to_shipping", {})
"""字体颜色到发货状态的映射"""
BG_COLOR_TO_STATUS: dict = cfg("color_mapping.bg_color_to_status", {})
"""背景颜色到状态的映射"""
THEME_FONT_COLORS: dict = cfg("color_mapping.theme_font_colors", {})
"""THEME 字体颜色到状态的映射"""
# ============================================================
# 公称压力 Lb→MPa 换算（#30 P0 DN/PN 字典 DB 化，替代 VBA 硬编码）
# 配置来源：config.yaml 的 sales_plan.pn_convert.lb_to_mpa
# （yaml 键为字符串，此处统一转为 int→float 字典供 PNService 查表）
# ============================================================
_PN_LB_TO_MPA_RAW: dict = cfg("sales_plan.pn_convert.lb_to_mpa", {}) or {}
PN_LB_TO_MPA: dict[int, float] = (
{int(k): float(v) for k, v in _PN_LB_TO_MPA_RAW.items()} if _PN_LB_TO_MPA_RAW else {}
)
"""英制 LB 磅级 → 公制 MPa 压力值换算表（配置驱动，默认空→回退 PNService 内置表）"""
# ============================================================
# 重复检查 / 唯一标识规则
# ============================================================
DUPLICATE_CHECK_FIELDS: list = cfg("duplicate_check.fields", [])
"""重复检查字段列表"""
UNIQUE_KEY_SEPARATOR: str = cfg("unique_key.separator", "|")
"""唯一键字段分隔符"""
UNIQUE_KEY_DATE_TO_SERIAL: bool = cfg("unique_key.date_to_serial", True)
"""是否将日期转换为序列号"""
UNIQUE_KEY_FIELD_FALLBACKS: dict = cfg("unique_key.key_field_fallbacks", {})
"""唯一键字段回退映射"""
REQUIRE_NUMBERED: bool = cfg("unique_key.require_numbered", False)
"""仅匹配已编号(已生成证书)行才判重（565 行语义，§9 落地待办①）"""
# ============================================================
# 排序规则
# ============================================================
SORT_RULES_KEYS: list = cfg("sort_rules.keys", [])
"""排序规则键列表"""
SORT_KEYS: list[str] = [k.get("field", "") for k in SORT_RULES_KEYS] if SORT_RULES_KEYS else []
"""排序字段名列表(向后兼容)"""
# ============================================================
# 分组规则
# ============================================================
GROUP_BY_FIELDS: list = cfg("group_by.fields", [])
"""分组字段列表"""
GROUP_BY_FIELDS_NO_PROJECT: list = cfg("group_by.fields_no_project", [])
"""非项目数据的分组字段列表"""
GROUP_BY_SUMMARY_FIELD: str = cfg("group_by.summary_field", "数量")
"""分组汇总字段名称"""
GROUP_BY_NUMBER_FORMAT: str = cfg("group_by.number_format", "编号:{:03d}")
"""组内编号格式"""
# ============================================================
# 导出配置
# ============================================================
EXPORT_HYPERLINK_COLUMN: int = cfg("export.hyperlink_column", 44)
"""超链接所在列号"""
EXPORT_HYPERLINK_REF_PREFIX: str = cfg("export.hyperlink_ref_prefix", "AR")
"""超链接引用前缀"""
EXPORT_HYPERLINK_WIDTH: int = cfg("export.hyperlink_width", 20)
"""超链接列宽度(字符数)"""
EXPORT_SUMMARY_LABEL: str = cfg("export.summary.label", "合计")
"""汇总行标签文本"""
EXPORT_SUMMARY_LABEL_COLUMN: int = cfg("export.summary.label_column", 1)
"""汇总标签所在列号"""
EXPORT_SUMMARY_BG_COLOR: str = cfg("export.summary.bg_color", "E0E0E0")
"""汇总行背景颜色"""
EXPORT_SUMMARY_COL_RANGE: int = cfg("export.summary.col_range", 25)
"""汇总列范围"""
EXPORT_SUMMARY_SUM_COLUMNS: list = cfg("export.summary.sum_columns", [])
"""需要求和的列列表"""
# ============================================================
# 销售计划配置
# ============================================================
SALE_PLAN_COLUMNS: dict = cfg("sales_plan.column_mapping.direct", {})
"""销售计划列映射字典"""
COLUMN_ALIASES: dict = cfg("sales_plan.column_mapping.aliases", {})
"""列别名映射"""
SALES_PLAN_HEADER_ROW: int = cfg("sales_plan.header_row", 2)
"""销售计划表头行号(从1开始)"""
SALES_PLAN_DATA_START_ROW: int = cfg("sales_plan.data_start_row", 3)
"""销售计划数据起始行号(从1开始)"""
SALES_PLAN_SHEET_PATTERN: str = cfg("sales_plan.sheet_pattern", "{month}月")
"""销售计划工作表名称匹配模式"""
SALES_PLAN_HEADER_KEYWORDS: list = cfg("sales_plan.header_keywords", [])
"""表头识别关键词列表"""
REQUIRED_FIELDS: list = cfg("sales_plan.required_fields", [])
"""销售计划必填字段列表"""
STYLED_IMPORT_COLUMNS: list = cfg("sales_plan.styled_import_columns", [])
"""带格式导入时保留的列列表"""
# #30 P1 导入门控：缺失要货单号的行按 mode 处理（skip / isolate / warn）
IMPORT_GATE: dict = cfg("sales_plan.import_gate", {}) or {}
"""导入门控配置：enabled / skip_when_missing / mode / match_keys（全配置驱动）"""
# 双字段映射（合同信息 ↔ 合格证打印信息），蓝图 §4.2.1
# pairs: contract→cert 成对关系；import_mode: copy / override / lazy（默认 lazy，不破坏既有行为）
DUAL_FIELD_MAPPING: dict = cfg("sales_plan.dual_field_mapping", {}) or {}
"""双字段映射配置：pairs（contract→cert）+ import_mode（copy/override/lazy）"""
# ============================================================
# 合格证清单配置
# ============================================================
QUALIFIED_LIST_HEADER_ROW: int = cfg("qualified_list.header_row", 2)
"""合格清单表头行号(从1开始)"""
QUALIFIED_LIST_DATA_START_ROW: int = cfg("qualified_list.data_start_row", 3)
"""合格清单数据起始行号(从1开始)"""
QUALIFIED_LIST_REQUIRED_FIELDS: list = cfg("qualified_list.required_fields", [])
"""合格清单必填字段列表"""
# ============================================================
# 数据导入配置
# ============================================================
IMPORT_MAPPING_PREPROCESSOR: dict | None = cfg("import_mapping.preprocessor")
"""导入映射预处理器配置"""
IMPORT_MAPPING_COLUMN_MAPPING: list = cfg("import_mapping.column_mapping", [])
"""导入列映射规则列表"""
IMPORT_MAPPING_DEFAULT_VALUES: dict = cfg("import_mapping.default_values", {})
"""导入默认值映射"""
# ============================================================
# 试压报告配置
# ============================================================
TEST_REPORT_COMPANY_NAME: str = cfg("test_report.company_name", "")
"""试压报告公司名称"""
TEST_REPORT_FORM_CODE: str = cfg("test_report.form_code", "")
"""试压报告表单代码"""
# ============================================================
# 日志配置
# ============================================================
_log_level_raw: Any = cfg("logging.level", "INFO")
LOG_LEVEL: str = str(_log_level_raw) if _log_level_raw else "INFO"
"""日志级别(DEBUG/INFO/WARNING/ERROR/CRITICAL)"""
LOG_ROTATION: str = cfg("logging.rotation", "10 MB")
"""日志轮转条件(如"10 MB"或"1 day")"""
LOG_RETENTION: str = cfg("logging.retention", "30 days")
"""日志保留时间(如"30 days")"""
LOG_COMPRESSION: str = cfg("logging.compression", "zip")
"""日志压缩格式"""
LOG_FORMAT: str = cfg("logging.format", "{time} | {level} | {name}:{function}:{line} - {message}")
"""日志格式字符串"""
# 处理环境变量占位符
if LOG_LEVEL.startswith("${"):
LOG_LEVEL = LOG_LEVEL.replace("${LOG_LEVEL:", "").rstrip("}") or "INFO"
# ============================================================
# 打印机配置
# ============================================================
DEFAULT_PRINTER: str | None = get_config_value("printer.default", "VALVE_DEFAULT_PRINTER")
"""默认打印机名称"""
SKIP_PRINTING: bool = _env("VALVE_SKIP_PRINTING") == "true"
"""是否跳过打印(用于测试)"""
# ============================================================
# UI 配置（通过 cfg() 从主配置统一读取，不再独立加载 ui.yaml）
# ============================================================
[文档]
def get_views_config() -> dict:
"""获取查询视图配置（canonical UI 源 = ui.yaml，Route A Phase 1）。
Returns:
dict: 视图配置字典（view_name → view 定义），来自 ``ui.yaml`` 的
``query_page.views``；ui.yaml 不可用时返回空字典。
"""
return get_views_config_yaml()
[文档]
def get_views_config_yaml() -> dict:
"""从 ui.yaml 读取查询视图配置（canonical UI 源，Route A Phase 1）。
返回 ``ui.yaml`` 的 ``query_page.views``；加载失败或 ui.yaml 不可用时返回空字典。
Returns:
dict: 视图配置字典（view_name → view 定义）。
"""
ui = get_ui_config()
if ui is False:
return {}
return (ui.query_page or {}).get("views", {}) or {}
# ============================================================
# UI 动态配置（ui.yaml，经 config.yaml !include 已并入 RootConfig 主路径）
# ============================================================
_ui_config_cache: Any = None # None=尚未加载; UIConfig=已加载; False=不可用
[文档]
def get_ui_config() -> Any:
"""加载并缓存 UI 动态配置（ui.yaml → UIConfig）。
供 ``query_view`` 等动态 UI 驱动层消费（§17）：用户编辑 ``ui.yaml`` 即可
增删查询右键菜单项。加载失败或文件缺失时返回 ``False``，调用方应回退到
``get_views_config()``（ui.yaml ``query_page.views``）基础菜单。
Returns:
UIConfig | False: 加载成功返回 UIConfig 实例；不可用返回 False。
"""
global _ui_config_cache
if _ui_config_cache is not None:
return _ui_config_cache
try:
from certflow.config.ui_config import UIConfig
ui_path = CONFIG_DIR / "ui.yaml"
_ui_config_cache = UIConfig.load(ui_path) if ui_path.exists() else False
except Exception as e: # noqa: BLE001
logger.warning(f"加载 UIConfig(ui.yaml) 失败，回退 VIEWS: {e}")
_ui_config_cache = False
return _ui_config_cache
# ============================================================
# 模板文件路径（通过 cfg() 从主配置统一读取）
# ============================================================
def _get_template(path_key: str) -> Path:
"""获取模板文件路径
优先从顶层 templates 配置读取，回退到 V1 的 paths.templates 配置。
Args:
path_key: 模板键名，如 "certificate_chinese"
Returns:
Path: 模板文件的绝对路径
"""
val = cfg(f"templates.{path_key}", "")
if val:
return _resolve_path(val)
v1_templates = cfg("paths.templates", {})
return _resolve_path(v1_templates.get(path_key, ""))
TEMPLATE_CERTIFICATE_CHINESE: Path = _get_template("certificate_chinese")
"""中文合格证模板路径"""
TEMPLATE_CERTIFICATE_BILINGUAL: Path = _get_template("certificate_bilingual")
"""中英文合格证模板路径"""
TEMPLATE_CERTIFICATE_RUSSIAN: Path = _get_template("certificate_russian")
"""俄文合格证模板路径"""
TEMPLATE_TEST_REPORT_SMALL: Path = _get_template("test_report_small")
"""小型试压报告模板路径"""
TEMPLATE_TEST_REPORT_MEDIUM: Path = _get_template("test_report_medium")
"""中型试压报告模板路径"""
TEMPLATE_QUALITY_CERT_FILE: Path = _get_template("quality_cert_file")
"""质量证书模板文件路径"""
# ============================================================
# 图片背景
# ============================================================
IMAGE_BG_CHINESE: str = cfg("image_backgrounds.chinese", "")
"""中文版证书背景图片路径"""
IMAGE_BG_BILINGUAL: str = cfg("image_backgrounds.bilingual", "")
"""中英文双语版证书背景图片路径"""
IMAGE_BG_RUSSIAN: str = cfg("image_backgrounds.russian", "")
"""俄文版证书背景图片路径"""
# ============================================================
# 数据库配置
# ============================================================
DATABASE_TYPE: str = cfg("database.type", "sqlite")
"""数据库类型(sqlite/mysql/postgresql)"""
DATABASE_ECHO: bool = cfg("database.echo", False)
"""是否打印SQL语句"""
DATABASE_POOL_SIZE: int = cfg("database.pool_size", 5)
"""数据库连接池大小"""
DATABASE_ACCESS_ENABLED: bool = cfg("database.access.enabled", False)
"""是否启用Access数据库"""
DATABASE_ACCESS_PRIMARY: str = cfg("database.access.primary", "")
"""Access主数据库路径"""
DATABASE_ACCESS_BACKUP: str = cfg("database.access.backup", "")
"""Access备份数据库路径"""
# ============================================================
# 环境信息
# ============================================================
ENV_COMPUTER_NAME: str | None = get_config_value("environment.computer_name", "COMPUTERNAME")
"""当前计算机名称"""
ENV_USERNAME: str | None = get_config_value("environment.username", "USERNAME")
"""当前用户名"""
# ============================================================
# 颜色值映射
# ============================================================
COLORS_YELLOW: int = cfg("colors.yellow", 65535)
"""黄色的RGB整数值"""
COLORS_BLUE: int = cfg("colors.blue", 16711680)
"""蓝色的RGB整数值"""
COLORS_RED: int = cfg("colors.red", 255)
"""红色的RGB整数值"""
COLORS_GREEN: int = cfg("colors.green", 65280)
"""绿色的RGB整数值"""
COLORS_LIGHT_BLUE: int = cfg("colors.light_blue", 15773696)
"""浅蓝色的RGB整数值"""
# ============================================================
# Excel 库选择
# ============================================================
USE_XLWINGS: bool = _env("VALVE_USE_XLWINGS") == "true"
"""是否使用xlwings库操作Excel"""
USE_OPENPYXL: bool = _env("VALVE_USE_OPENPYXL") == "true"
"""是否使用openpyxl库操作Excel"""
# ============================================================
# 导入去重配置(V1 兼容 — 现从 IDGenerator 获取唯一键字段)
# ============================================================
# UNIQUE_KEY_FIELDS 和 DEDUP_KEYS 已迁移至 IDGenerator.get_all_unique_key_fields()
# DEDUPLICATION_STRATEGY: str = cfg("import_deduplication.strategy", "skip")
# """去重策略(skip/overwrite)"""
UPDATE_ON_DUPLICATE: bool = cfg("import_deduplication.update_on_duplicate", True)
"""重复时是否检测变更并更新 SalePlan 字段"""
ON_COLLISION: str = cfg("import_deduplication.on_collision", "quarantine")
"""批内撞键(同 unique_key)处理策略：
- quarantine: 后续行隔离到待处理表（默认，保持现有行为）
- merge_max: 同键数量不同 → 取较大数量合并为一行（不隔离）
- merge_sum: 同键数量不同 → 数量求和合并为一行（不隔离）
注：仅作用于「批内内存去重」阶段的同键撞行；跨批/跨库去重仍走 _find_existing。
"""
MONITORED_FIELDS: list = cfg(
"import_deduplication.monitored_fields",
[
"product_name",
"product_model",
"product_spec",
"quantity",
"customer",
"project_name",
"plan_date",
],
)
"""导入时监控变更的字段列表"""
# ============================================================
# 确保输出目录存在
# ============================================================
for _dir in [
OUTPUT_TEST_REPORT,
OUTPUT_QUALITY_CERT,
OUTPUT_CERTIFICATE_IMAGES,
OUTPUT_NAMEPLATE,
OUTPUT_BACKUPS,
]:
_dir.mkdir(parents=True, exist_ok=True)
LOGS_DIR.mkdir(parents=True, exist_ok=True)
DATABASE_DIR.mkdir(parents=True, exist_ok=True)
# ============================================================
# 排序规则(完整版，从 YAML 动态读取)
# ============================================================
[文档]
def get_sort_rules() -> list[dict[str, str]]:
"""获取完整排序规则列表
Returns:
List[Dict[str, str]]: 排序规则列表，每个元素包含field和order字段
Examples:
>>> get_sort_rules()
[{"field": "order_number", "order": "asc"}, {"field": "amount", "order": "desc"}]
"""
keys: list = cfg("sort_rules.keys", [])
return [{"field": str(k.get("field", "")), "order": str(k.get("order", "asc"))} for k in keys]
# ============================================================
# 分组与排序配置（V3 新增）
# ============================================================
# 分组配置
GROUPING_KEYS: list = cfg("grouping.group_keys", [])
"""分组键列表（英文字段名）"""
GROUPING_KEYS_NO_PROJECT: list = cfg("grouping.group_keys_no_project", [])
"""无项目名称时的分组键列表（英文字段名）"""
# 分组序号配置
GROUPING_USE_GLOBAL_PREFIX: bool = cfg("grouping.numbering.use_global_prefix", True)
GROUPING_PREFIX_FORMAT: str = cfg("grouping.numbering.prefix_format", "G{:03d}")
GROUPING_SEQ_FORMAT: str = cfg("grouping.numbering.seq_format", "{:03d}")
GROUPING_SEPARATOR: str = cfg("grouping.numbering.separator", "-")
GROUP_INDEX_FIELD: str = cfg("grouping.numbering.group_index_field", "_group_index")
# 排序配置（组内排序）
SORTING_KEYS: list = cfg("sorting.keys", [])
"""组内排序键列表，每个元素包含 field 和 order"""
# 提取排序字段名列表（供 Sorter 使用）
SORTING_FIELD_NAMES: list = [k.get("field", "") for k in SORTING_KEYS if k.get("field")]
"""排序字段名列表（按优先级顺序）"""
# ============================================================
# 颜色映射配置
# ============================================================
FONT_COLOR_STATUS_MAP: dict = cfg("color_mapping.font_colors", {})
BG_COLOR_STATUS_MAP: dict = cfg("color_mapping.bg_colors", {})
# ============================================================
# 已完成工单回填配置
# ============================================================
COMPLETED_ORDERS_BACKFILL: dict = cfg("completed_orders_backfill", {})
# ============================================================
# 字段默认值
# ============================================================
[文档]
def get_field_default(field_name: str, default: str | None = None) -> Any:
"""获取证书字段的默认值
Args:
field_name: 字段名称
default: 默认值(字段不存在时返回)
Returns:
Any: 字段默认值
"""
defaults: dict = cfg("certificate.defaults", {})
return defaults.get(field_name, default)
# ============================================================
# 列名工具
# ============================================================
[文档]
def get_column_by_alias(excel_column: str) -> str | None:
"""通过别名查找目标列名
在列别名映射中查找给定的Excel列名，返回对应的目标字段名.
Args:
excel_column: Excel中的列名或别名
Returns:
Optional[str]: 目标字段名，未找到时返回None
Examples:
>>> get_column_by_alias("订单号")
"order_number"
>>> get_column_by_alias("客户名称")
"customer_name"
"""
aliases: dict = COLUMN_ALIASES or {}
for target, alias_list in aliases.items():
if excel_column in alias_list:
return str(target)
return None
# ============================================================
# UI 配置快捷方法
# ============================================================
[文档]
def get_ui_window_title() -> str:
"""获取UI窗口标题
Returns:
str: 窗口标题，默认为"CertFlow"
"""
return str(cfg("ui.window.title", "CertFlow"))
[文档]
def get_ui_table_alternating_colors() -> bool:
"""获取UI表格是否使用交替行颜色
Returns:
bool: 是否启用交替行颜色，默认为True
"""
return bool(cfg("ui.table.alternating_row_colors", True))
# ============================================================
# 目录确保
# ============================================================
[文档]
def ensure_directories() -> None:
"""创建所有必需的目录
遍历所有配置的目录路径，确保它们存在.
包括数据库目录、临时文件目录、日志目录、资源目录及其子目录.
"""
temp_dir = get_temp_dir()
for dir_path in [DATABASE_DIR, temp_dir, LOGS_DIR, RESOURCES_DIR]:
dir_path.mkdir(parents=True, exist_ok=True)
from certflow.config.paths_override import cfg
temp_subdirs = [
str(cfg("paths.temp.excel", "excel")),
str(cfg("paths.temp.reports", "reports")),
str(cfg("paths.temp.images", "images")),
str(cfg("paths.temp.scans", "scans")),
]
for subdir in temp_subdirs:
(temp_dir / subdir).mkdir(exist_ok=True)
template_base: str = str(cfg("paths.templates.base", "data/templates"))
template_dir = BASE_DIR / template_base
template_dir.mkdir(parents=True, exist_ok=True)
logger.info("All directories verified/created successfully")
# ============================================================
# 配置重载
# ============================================================
[文档]
def reload_config() -> None:
"""强制重新加载 YAML 配置
清除配置缓存，下次读取时会重新加载配置文件.
用于运行时动态更新配置.
"""
global _config_cache, _ucfg_bucket2_cache, _ucfg_state_cache
_config_cache = None
_ucfg_bucket2_cache = None
_ucfg_state_cache = None
_load_config()
logger.info("Configuration reloaded")
# ============================================================
# 用户配置管理(userconfig 分文件 + merge 写) —— T18
# ============================================================
#
# 落盘策略(见 docs/discussions/resource-paths-audit.md §9.5 三桶模型):
# - 桶2 稳定·每用户偏好/覆盖 → config/userconfig.yaml (被 .gitignore 忽略)
# - 桶3 易变·自动状态 → config/userconfig.state.yaml (gitignored, 运行时自动维护)
#
# 读取统一经 _get_user_config() 合并两文件; 写入可走 _save_user_config(接受合并字典并
# 按桶分流, 向后兼容旧调用点) 或 merge_user_config(按段增量, 推荐 —— 只重写对应文件,
# 避免多写者互相覆盖)。旧单文件 userconfig.yaml 首次加载时自动拆分迁移(幂等)。
_USERCONFIG_PATH: Path = CONFIG_DIR / "userconfig.yaml"
_USERCONFIG_STATE_PATH: Path = CONFIG_DIR / "userconfig.state.yaml"
# 桶3(易变·自动状态): 这些顶层键写入 userconfig.state.yaml
_STATE_KEYS: frozenset[str] = frozenset(
{
"file_sheet_configs",
"import_history",
"query_filters",
"recent_files",
"recent_import_files",
"table_state",
}
)
# 两个底层缓存(合并视图由 _get_user_config 惰性构造)
_ucfg_bucket2_cache: dict | None = None
_ucfg_state_cache: dict | None = None
def _load_user_file(path: Path) -> dict:
"""安全读取单个用户配置文件; 不存在或损坏返回空 dict。"""
if path.exists():
try:
with open(path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
except Exception as e: # pragma: no cover - 防御: 损坏文件不阻断启动
logger.warning("读取 %s 失败, 当作空: %s", path, e)
return {}
def _dump_user_file(path: Path, data: dict) -> None:
"""原子性较差但足够用的单文件写入(与历史实现行为一致)。"""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
yaml.dump(data, f, allow_unicode=True, default_flow_style=False)
def _deep_merge(base: dict, patch: dict) -> dict:
"""深合并: dict 递归合并; list/标量直接覆盖(patch 优先)。返回新 dict。"""
result = dict(base)
for k, v in patch.items():
if isinstance(v, dict) and isinstance(result.get(k), dict):
result[k] = _deep_merge(result[k], v)
else:
result[k] = v
return result
def _ensure_userconfig_split() -> None:
"""一次性迁移: 旧单文件 userconfig.yaml 中的桶3 键拆分到 userconfig.state.yaml。
仅在 userconfig.yaml 实际含桶3 键时触发; 迁移后两个文件各自干净。幂等。
"""
global _ucfg_bucket2_cache, _ucfg_state_cache
if _USERCONFIG_PATH.exists():
legacy = _load_user_file(_USERCONFIG_PATH)
state_keys_present = {k: v for k, v in legacy.items() if k in _STATE_KEYS}
if state_keys_present:
bucket2 = {k: v for k, v in legacy.items() if k not in _STATE_KEYS}
_dump_user_file(_USERCONFIG_PATH, bucket2)
existing_state = _load_user_file(_USERCONFIG_STATE_PATH)
merged_state = _deep_merge(existing_state, state_keys_present)
_dump_user_file(_USERCONFIG_STATE_PATH, merged_state)
logger.info(
"已迁移桶3 键 %s 到 %s", sorted(state_keys_present), _USERCONFIG_STATE_PATH.name
)
def _load_bucket2() -> dict:
global _ucfg_bucket2_cache
if _ucfg_bucket2_cache is None:
_ensure_userconfig_split()
_ucfg_bucket2_cache = _load_user_file(_USERCONFIG_PATH)
return _ucfg_bucket2_cache
def _load_state() -> dict:
global _ucfg_state_cache
if _ucfg_state_cache is None:
_ensure_userconfig_split()
_ucfg_state_cache = _load_user_file(_USERCONFIG_STATE_PATH)
return _ucfg_state_cache
def _get_user_config() -> dict:
"""加载并合并 userconfig.yaml + userconfig.state.yaml(惰性, 每次返回合并副本)。
桶3 键(易变状态)来自 state 文件, 桶2 键(偏好/覆盖)来自 userconfig 文件。
返回的合并字典供读取; 修改后请经 _save_user_config 或 merge_user_config 落盘。
Returns:
Dict: 合并后的用户配置字典
"""
return {**_load_bucket2(), **_load_state()}
def _save_user_config(data: dict) -> None:
"""保存用户配置(向后兼容: 接受合并字典, 按桶拆分写入两文件)。
旧调用点(视图/控制器)仍读合并字典→改→整字典回写, 本函数按顶层键归属
分流到 userconfig.yaml / userconfig.state.yaml, 不会跨桶丢失数据。
"""
global _ucfg_bucket2_cache, _ucfg_state_cache
bucket2 = {k: v for k, v in data.items() if k not in _STATE_KEYS}
state = {k: v for k, v in data.items() if k in _STATE_KEYS}
if bucket2 or not _USERCONFIG_PATH.exists():
_dump_user_file(_USERCONFIG_PATH, bucket2)
if state or not _USERCONFIG_STATE_PATH.exists():
_dump_user_file(_USERCONFIG_STATE_PATH, state)
_ucfg_bucket2_cache = bucket2
_ucfg_state_cache = state
[文档]
def merge_user_config(patch: dict, *, merge: bool = True) -> None:
"""按段增量写入用户配置(推荐, T18 核心 API)。
仅重写 patch 中涉及键所属的文件, 不再整文件重写另一桶, 消除多写者互相覆盖:
- patch 的桶2 键 → userconfig.yaml(深合并)
- patch 的桶3 键 → userconfig.state.yaml(深合并)
Args:
patch: 形如 ``{"preferences": {"theme": "dark"}}`` 或部分顶层键字典。
merge: ``True`` 深合并(默认, 保留同文件其它键); ``False`` 直接覆盖该顶层键。
"""
global _ucfg_bucket2_cache, _ucfg_state_cache
bucket2_patch = {k: v for k, v in patch.items() if k not in _STATE_KEYS}
state_patch = {k: v for k, v in patch.items() if k in _STATE_KEYS}
if bucket2_patch:
cur = _load_bucket2()
merged = _deep_merge(cur, bucket2_patch) if merge else {**cur, **bucket2_patch}
_dump_user_file(_USERCONFIG_PATH, merged)
_ucfg_bucket2_cache = merged
if state_patch:
cur = _load_state()
merged = _deep_merge(cur, state_patch) if merge else {**cur, **state_patch}
_dump_user_file(_USERCONFIG_STATE_PATH, merged)
_ucfg_state_cache = merged
[文档]
def get_recent_files() -> list[dict]:
"""获取最近打开的文件列表
Returns:
List[Dict]: 最近文件列表，每个元素包含path和last_used字段
"""
return _get_user_config().get("recent_files", [])
[文档]
def add_recent_file(file_path: str) -> None:
"""添加文件到最近打开列表
如果文件已存在，会移到列表开头.
Args:
file_path: 文件路径
"""
from datetime import datetime
max_count = get_user_preference("max_recent_files", 10)
recent_files: list = _get_user_config().get("recent_files", [])
recent_files = [f for f in recent_files if f.get("path") != file_path]
recent_files.insert(
0, {"path": file_path, "last_used": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
)
merge_user_config({"recent_files": recent_files[:max_count]})
[文档]
def get_user_preference(key: str, default: Any = None) -> Any:
"""获取用户偏好设置
Args:
key: 偏好设置的键名
default: 默认值
Returns:
Any: 偏好设置值
"""
return _get_user_config().get("preferences", {}).get(key, default)
[文档]
def should_confirm_before_delete(default: bool = True) -> bool:
"""是否在执行删除前弹出确认对话框（用户偏好，默认开启）"""
return bool(get_user_preference("confirm_before_delete", default))
# ============================================================
# 报告生成配置（config/report.yaml，经 config.yaml !include 并入）
# ============================================================
REPORT_EXPORT_PDF: bool = cfg("report.export_pdf", False)
"""生成报告时是否同时导出 PDF（基础层默认 False：仅生成 xlsx）"""
REPORT_FILENAME_FORMAT: str = cfg("report.filename_format", "{report_type}{plan_year}_{customer}")
"""报告文件名自动命名格式（占位符见 ReportOutputService._build_filename）"""
REPORT_SHEET_NAME_FORMAT: str = cfg(
"report.sheet_name_format", "{report_type}({plan_date_short}计划)_{report_date}{serial}"
)
"""报告内工作表(数据表)标签名格式：默认带计划日期、制作日期与流水号，避免千篇一律的模板变体名"""
REPORT_SAVE_PATH: str = str(cfg("report.save_path", "") or "")
"""报告保存路径覆盖（可选；为空则用 paths.report_archive.root）"""
[文档]
def should_export_report_pdf() -> bool:
"""生成报告时是否同时导出 PDF。
优先级：用户覆盖层 ``preferences.report_export_pdf`` > 基础层 ``report.export_pdf``
> ``False``。默认不导出（仅生成 xlsx）。
"""
pref = get_user_preference("report_export_pdf", None)
if pref is not None:
return bool(pref)
return bool(REPORT_EXPORT_PDF)
[文档]
def get_report_save_path() -> str:
"""报告保存路径覆盖（用户覆盖层优先，回退基础层；空字符串→用 paths.report_archive.root）。"""
override = get_override("report", "save_path")
if override:
return str(override)
return str(REPORT_SAVE_PATH)
[文档]
def get_report_page_options() -> dict[str, list[int]]:
"""报告每页记录数候选（配置可选，而非数字递增），按报告类型分组。
Returns:
dict: ``{kind: [候选值列表]}``；kind ∈ quality/pressure/material。
"""
return cfg("report.page_options", {}) or {
"quality": [5, 6, 9],
"pressure": [13, 25],
"material": [1, 2, 3],
}
[文档]
def set_user_preference(key: str, value: Any) -> None:
"""设置单个用户偏好并立即持久化
Args:
key: 偏好设置的键名
value: 偏好设置值
"""
merge_user_config({"preferences": {key: value}})
[文档]
def get_override(section: str, key: str, default: Any = None) -> Any:
"""读取 userconfig 覆盖层中指定段/键的值（优先级高于 config.yaml 基础层）。
用于系统设置中用户自定义的覆盖项（如 ``print_layout``、``system_settings``），
与 §10「基础层 + 覆盖层」架构一致。文件缺失或键不存在时返回 ``default``。
Args:
section: 覆盖段名，如 ``"print_layout"``、``"system_settings"``
key: 键名
default: 未设置时的默认值
Returns:
Any: 覆盖值；缺失时返回 default
"""
return _get_user_config().get(section, {}).get(key, default)
[文档]
def set_override(section: str, key: str, value: Any) -> None:
"""写入 userconfig 覆盖层指定段/键并立即持久化。
Args:
section: 覆盖段名，如 ``"print_layout"``、``"system_settings"``
key: 键名
value: 值
"""
merge_user_config({section: {key: value}})
[文档]
def get_override_section(section: str) -> dict:
"""读取 userconfig 覆盖层中整个段（用于一次性合并某段覆盖到基础配置）。
Args:
section: 覆盖段名
Returns:
Dict: 该段的覆盖字典；缺失时返回空字典
"""
return _get_user_config().get(section, {}) or {}
[文档]
def set_override_section(section: str, data: dict) -> None:
"""写入 userconfig 覆盖层整个段并立即持久化。
用于一次性保存某段全部覆盖字段（如 ``print_layout`` 打印版式），
与 §10「基础层 + 覆盖层」架构一致。传入空字典等效于清空该段覆盖。
Args:
section: 覆盖段名，如 ``"print_layout"``、``"system_settings"``
data: 该段的完整覆盖字典
"""
merge_user_config({section: data})
[文档]
def get_skipped_columns(file_path: str) -> list[str]:
"""获取导入时跳过的列列表
Args:
file_path: 文件路径
Returns:
List[str]: 跳过的列名列表
"""
user_config = _get_user_config()
import_history = user_config.get("import_history", {})
return import_history.get(file_path, {}).get("skipped_columns", [])
[文档]
def save_skipped_columns(file_path: str, skipped_columns: list[str]) -> None:
"""保存导入时跳过的列配置
Args:
file_path: 文件路径
skipped_columns: 跳过的列名列表
"""
merge_user_config({"import_history": {file_path: {"skipped_columns": skipped_columns}}})
logger.debug(f"已保存跳过列: {file_path} -> {skipped_columns}")
[文档]
def get_import_config(file_path: str) -> dict[str, Any]:
"""获取文件的导入配置
Args:
file_path: 文件路径
Returns:
Dict[str, Any]: 导入配置字典
"""
return _get_user_config().get("import_history", {}).get(file_path, {})
[文档]
def save_import_config(file_path: str, config: dict[str, Any]) -> None:
"""保存文件的导入配置
Args:
file_path: 文件路径
config: 导入配置字典
"""
merge_user_config({"import_history": {file_path: config}})
logger.debug(f"已保存导入配置: {file_path}")
[文档]
def get_recent_import_files(max_count: int | None = None) -> list[dict]:
"""获取最近导入的文件列表
Args:
max_count: 最大返回数量，为 None 时使用用户偏好设置
Returns:
List[Dict]: 最近导入文件列表
"""
if max_count is None:
max_count = get_user_preference("max_recent_files", 10)
recent: list = _get_user_config().get("recent_import_files", [])
return recent[:max_count]
[文档]
def add_recent_import_file(file_path: str, config: dict[str, Any] | None = None) -> None:
"""添加文件到最近导入列表
Args:
file_path: 文件路径
config: 导入配置(可选)
"""
from datetime import datetime
recent: list = _get_user_config().get("recent_import_files", [])
recent = [f for f in recent if f.get("path") != file_path]
record: dict[str, Any] = {
"path": file_path,
"name": Path(file_path).name,
"last_used": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
if config:
record["config"] = {
"sheet_name": config.get("sheet_name"),
"header_row": config.get("header_row"),
"data_start_row": config.get("data_start_row"),
"skip_rows": config.get("skip_rows", 0),
"column_mapping": config.get("column_mapping", {}),
}
recent.insert(0, record)
max_count = get_user_preference("max_recent_files", 10)
merge_user_config({"recent_import_files": recent[:max_count]})
logger.debug(f"已记录最近文件: {file_path}")
[文档]
def get_file_import_config(file_path: str) -> dict[str, Any]:
"""获取文件的导入配置(优先使用最近文件中的配置)
Args:
file_path: 文件路径
Returns:
Dict[str, Any]: 导入配置字典
"""
for item in get_recent_import_files():
if item.get("path") == file_path:
return item.get("config", {})
return _get_user_config().get("file_sheet_configs", {}).get(file_path, {})
[文档]
def clear_recent_files() -> None:
"""清空最近文件列表"""
merge_user_config({"recent_import_files": []})
logger.info("已清空最近文件列表")
[文档]
def save_sheet_config(file_path: str, sheet_configs: dict[str, Any]) -> None:
"""保存工作簿内的工作表配置
Args:
file_path: Excel文件路径
sheet_configs: 工作表配置字典，键为工作表名称，值为配置内容
"""
merge_user_config({"file_sheet_configs": {file_path: sheet_configs}})
[文档]
def ensure_ui_config() -> None:
"""确保UI配置文件存在
如果UI配置文件不存在，创建带有默认配置的ui.yaml文件.
"""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
if not UI_CONFIG_FILE.exists():
_create_default_ui_config()
logger.info(f"已创建默认UI配置文件: {UI_CONFIG_FILE}")
def _create_default_ui_config() -> None:
"""创建默认的UI配置文件
生成包含主窗口配置、导航按钮、页面配置、样式定义的默认UI配置.
"""
default_config = {
"main_window": {
"title": "{app_name} - {app_version}",
"min_width": 1280,
"min_height": 800,
"default_width": 1440,
"default_height": 900,
},
"title_bar": {
"text": "{app_name} - 合格证打印管理系统",
"font_size": 20,
"font_weight": "bold",
"padding": 15,
"background_color": "#2c7be5",
"text_color": "white",
"alignment": "center",
},
"nav_buttons": [
{
"name": "import",
"text": "导入销售计划",
"icon": "",
"tooltip": "导入Excel销售计划文件",
"page_index": 1,
"style": {"padding": "20px", "font_size": "16px", "min_width": "150px"},
},
{
"name": "print",
"text": "打印合格证",
"icon": "",
"tooltip": "打印合格证",
"page_index": None,
"style": {"padding": "20px", "font_size": "16px", "min_width": "150px"},
},
{
"name": "report",
"text": "生成报告",
"icon": "",
"tooltip": "生成统计报告",
"page_index": None,
"style": {"padding": "20px", "font_size": "16px", "min_width": "150px"},
},
{
"name": "log",
"text": "查看日志",
"icon": "",
"tooltip": "查看系统日志",
"page_index": 2,
"style": {"padding": "20px", "font_size": "16px", "min_width": "150px"},
},
],
"welcome_page": {
"title": {
"text": "欢迎使用{app_name}",
"font_size": 24,
"margin": "50px",
"alignment": "center",
},
"description": {
"text": "请选择要执行的操作",
"font_size": 14,
"margin": "20px",
"alignment": "center",
},
},
"import_page": {
"title": {
"text": "销售计划导入",
"font_size": 18,
"margin": "20px",
"font_weight": "bold",
"alignment": "center",
},
"import_button": {
"text": "选择Excel文件导入",
"style": {"padding": "15px", "font_size": "14px"},
"file_filter": "Excel文件 (*.xlsx *.xls)",
},
"result_area": {"placeholder": "导入结果将显示在这里...", "read_only": True},
"back_button": {"text": "返回首页", "style": {"padding": "10px"}},
},
"log_page": {
"title": {
"text": "系统日志",
"font_size": 18,
"margin": "20px",
"font_weight": "bold",
"alignment": "center",
},
"back_button": {"text": "返回首页", "style": {"padding": "10px"}},
"max_lines": 1000,
"auto_scroll": True,
},
"status_bar": {"default_message": "就绪", "timeout_ms": 3000},
"styles": {
"global": "QMainWindow {\n background-color: #f5f5f5;\n}\n",
"button": {
"default": (
"QPushButton {\n background-color: #2c7be5;\n color: white;\n"
" border: none;\n border-radius: 5px;\n}\n"
"QPushButton:hover {\n background-color: #1c68c5;\n}\n"
"QPushButton:pressed {\n background-color: #1557a3;\n}\n"
)
},
"text_edit": (
"QTextEdit {\n border: 1px solid #ddd;\n border-radius: 5px;\n"
" padding: 10px;\n font-family: monospace;\n}\n"
),
},
}
with open(UI_CONFIG_FILE, "w", encoding="utf-8") as f:
yaml.dump(default_config, f, allow_unicode=True, default_flow_style=False, indent=2)
# ============================================================
# 导入配置（新增，与 QueryView 风格一致）
# ============================================================
[文档]
def get_import_views() -> dict:
"""获取导入视图模板
Returns:
dict: 导入视图配置字典，键为视图键名，缺省时为空字典。
"""
return cfg("import_config.views", {})
[文档]
def get_import_view(view_key: str) -> dict:
"""获取指定视图模板
Args:
view_key: 视图键名，如 "sales_plan"。
Returns:
dict: 对应视图的配置字典，未找到时返回空字典。
"""
return get_import_views().get(view_key, {})
[文档]
def get_import_required_fields(view_key: str) -> list[str]:
"""获取视图的必填字段
Args:
view_key: 视图键名。
Returns:
list[str]: 该视图的必填字段名列表，未配置时为空列表。
"""
return get_import_view(view_key).get("required_fields", [])
[文档]
def get_import_default_strategy() -> str:
"""获取默认加载策略
Returns:
str: 默认导入策略名称，如 "smart"，缺省为 "smart"。
"""
return cfg("import_config.default_strategy", "smart")
# ============================================================
# 导入期占位目录开关（B0-4 — 批次0 B-CreateDIR/F 已定稿）
# ============================================================
# False（默认）：导入不建目录，下游写文件时由写端惰性 os.makedirs(parents=True, exist_ok=True)
# True：导入期按订单组（计划日期+订货单位+项目名称）预建 VBA 风格占位项目文件夹
CREATE_PLACEHOLDER_DIRS: bool = cfg("import.create_placeholder_dirs", False)
"""导入期是否预建占位目录（B0-4，默认 OFF）"""
# 0 冲突/0 隔离时跳过确认对话框、直接导入（蓝图 §4.1.3，默认开启）
SKIP_CONFIRM_WHEN_NO_CONFLICT: bool = cfg("import.skip_confirm_when_no_conflict", True)
"""导入无冲突时是否跳过确认对话框直接导入"""