"""per-machine 目录覆盖层 (paths.local.yaml) 读取器 + deep-merge.
设计见 docs/discussions/resource-paths-audit.md §13。与现有 ConfigLoader / _cfg /
RootConfig(extra="forbid") 严格校验共存：采用「点路径逐键生效」，不重建模型，规避
extra=forbid 限制。
覆盖层只承载 paths.* 目录类覆盖，不自动回写（区别于 userconfig.yaml 的偏好/状态）。
"""
from __future__ import annotations
import contextlib
import logging
import os
import re
from copy import deepcopy
from pathlib import Path
from typing import Any
import yaml
# 注意：CONFIG_DIR / _cfg / _resolve_path 改为在函数内惰性导入，避免与 settings 形成
# 循环导入（settings 的模块级常量需要在导入期调用 cfg）。
logger = logging.getLogger(__name__)
_ENV_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::([^}]*))?\}")
# 模块级缓存：None 表示尚未加载；{} 表示已加载但文件缺失/为空。
_paths_local_cache: dict | None = None
# ---------------------------------------------------------------------------
# 环境变量替换（自包含，不复用 loader 实例方法以避免耦合/循环导入）
# ---------------------------------------------------------------------------
def _resolve_env_vars(data: Any) -> Any:
"""递归替换 ${ENV_VAR:default} 形式的环境变量。"""
if isinstance(data, str):
def replacer(m: re.Match) -> str:
return os.environ.get(m.group(1), m.group(2) or "")
return _ENV_PATTERN.sub(replacer, data)
if isinstance(data, dict):
return {k: _resolve_env_vars(v) for k, v in data.items()}
if isinstance(data, list):
return [_resolve_env_vars(v) for v in data]
return data
# ---------------------------------------------------------------------------
# 嵌套点路径读写
# ---------------------------------------------------------------------------
def _get_nested(d: dict, parts: list[str]) -> tuple[bool, Any]:
cur: Any = d
for p in parts:
if isinstance(cur, dict) and p in cur:
cur = cur[p]
else:
return False, None
return True, cur
def _set_nested(d: dict, parts: list[str], value: Any) -> None:
cur = d
for p in parts[:-1]:
nxt = cur.get(p)
if not isinstance(nxt, dict):
nxt = {}
cur[p] = nxt
cur = nxt
cur[parts[-1]] = value
def _iter_dotted(d: dict, prefix: str = "") -> list[str]:
out: list[str] = []
for k, v in d.items():
key = f"{prefix}.{k}" if prefix else k
if isinstance(v, dict):
out.extend(_iter_dotted(v, key))
else:
out.append(key)
return out
def _get_local_path(config_dir: Path | None = None) -> Path:
if config_dir:
base = Path(config_dir)
else:
from certflow.config.settings import CONFIG_DIR
base = CONFIG_DIR
return base / "paths.local.yaml"
def _warn_unknown_keys(data: dict) -> None:
"""对 base paths 子树中不存在的覆盖键告警（防静默失效，呼应 STRICT_VALIDATION）。
仅检查 paths.<X> 顶层段是否在 base paths 中存在：dict 型字段（output/sales_plan 等）
接受任意子键，不对其子键告警（即便 base 中为 None）。
"""
ov_paths = data.get("paths")
if not isinstance(ov_paths, dict):
return
from certflow.config.settings import _cfg
base_paths = _cfg("paths", {}) or {}
if not isinstance(base_paths, dict):
return
for dotted in _iter_dotted(ov_paths, prefix="paths"):
parts = dotted.split(".")
top = parts[1] if len(parts) > 1 else None
if top is None or top not in base_paths:
logger.warning(
"paths.local.yaml 含未知路径键 %s（base paths 中无对应项），_cfg 不会读取它；"
"若为主动新增的可覆盖键请忽略。",
dotted,
)
# ---------------------------------------------------------------------------
# 公共 API
# ---------------------------------------------------------------------------
[文档]
def load_paths_local(config_dir: Path | None = None) -> dict:
"""读取本机目录覆盖层。
- 文件缺失/为空/解析失败 → 返回 {}，完全不影响 base 配置。
- 支持 ${ENV_VAR:default} 环境变量替换；不做 !include 展开（覆盖层自包含）。
- 结果懒缓存；修改后调用 reload_paths_local() 失效。
"""
global _paths_local_cache
if _paths_local_cache is not None:
return _paths_local_cache
data: dict = {}
path = _get_local_path(config_dir)
if path.exists():
try:
with open(path, encoding="utf-8") as f:
loaded = yaml.safe_load(f) or {}
data = _resolve_env_vars(loaded)
if not isinstance(data, dict):
data = {}
except Exception as e: # noqa: BLE001 — 覆盖层错误绝不阻断主配置
logger.warning("读取 %s 失败，已忽略覆盖层: %s", path, e)
data = {}
_paths_local_cache = data
_warn_unknown_keys(data)
return data
[文档]
def reload_paths_local() -> None:
"""清除覆盖层缓存，下次 load_paths_local() 重新读取。"""
global _paths_local_cache
_paths_local_cache = None
def _looks_like_path(value: str) -> bool:
"""值是否像路径：含分隔符或绝对路径（盘符/以 / 开头）。用于决定是否需要 _resolve_path。"""
if not value:
return False
return (
"/" in value
or "\\" in value
or value.startswith("/")
or (len(value) >= 2 and value[1] == ":")
)
[文档]
def cfg(key: str, default: Any = None) -> Any:
"""统一配置读取：覆盖层 > 基础层，按值类型智能解析。
- 覆盖层（paths.local.yaml）含该点路径 → 取覆盖值；否则回退 _cfg(key, default)。
**对任意键生效**（不限于 paths.*），实现全局可覆盖。
- 解析规则（仅对"像路径的字符串"走 _resolve_path，其余原样返回）：
* int / dict / list / bool / None → 原样返回（杜绝把数字/字典误当路径解析成垃圾路径）
* 普通字符串（文件名、标签、单段目录名）→ 原样返回
* 含分隔符或绝对路径的字符串 → 解析为 BASE_DIR 下的绝对路径
未配置 paths.local.yaml 时行为与 _cfg 完全一致。
这是 pcfg 的继任者：pcfg 现为其别名，新代码请直接用 cfg。
"""
from certflow.config.settings import _cfg, _resolve_path
ov = load_paths_local()
if ov:
ok, val = _get_nested(ov, key.split("."))
if ok and val is not None:
if isinstance(val, str) and _looks_like_path(val):
return _resolve_path(val)
return val
return _cfg(key, default)
# 向后兼容别名（重构过渡期保留，后续移除）。新代码请用 cfg。
pcfg = cfg
def _dump_paths_local(data: dict) -> None:
path = _get_local_path()
try:
with open(path, "w", encoding="utf-8") as f:
yaml.safe_dump(data, f, allow_unicode=True, sort_keys=False)
reload_paths_local()
except OSError as e: # noqa: BLE001
logger.warning("写入 %s 失败: %s", path, e)
[文档]
def set_path_override(key: str, value: Any) -> None:
"""写入一个路径覆盖（点路径）。独立落盘，不触碰 userconfig.yaml。"""
ov = deepcopy(load_paths_local())
_set_nested(ov, key.split("."), value)
_dump_paths_local(ov)
[文档]
def clear_path_override(key: str) -> None:
"""清除一个路径覆盖（点路径）。"""
ov = load_paths_local()
parts = key.split(".")
cur: Any = ov
for p in parts[:-1]:
if not isinstance(cur.get(p), dict):
return
cur = cur[p]
cur.pop(parts[-1], None)
_dump_paths_local(ov)
[文档]
def reset_path_overrides() -> None:
"""清空整个覆盖文件（"恢复默认"）。"""
path = _get_local_path()
if path.exists():
with contextlib.suppress(OSError):
path.unlink()
reload_paths_local()
[文档]
def list_path_overrides() -> list[str]:
"""返回所有已生效覆盖的点路径列表（配置中心"我的覆盖" tab 数据源）。"""
return _iter_dotted(load_paths_local())