55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
"""app/services/bioflow/params.py —— 参数预填(BBA_* 历史/契约文档 → 表单默认值)。
|
||
|
||
- 契约文档(gfa_core/schemas/stages/*_contract_v1.md)解析参数默认值
|
||
(parameters 小节 key=value 形式)。
|
||
- BBA_* 环境变量历史(进程内 config.bba_env 派生)。
|
||
- 存档参数预填(FR-BIO-023:表单预填存档值——由 runner._form_defaults 用
|
||
record.params 完成,本模块提供读取存档参数的工具)。
|
||
|
||
服务层纪律:纯 dataclass/dict/stdlib,无 pydantic。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 契约文档 parameters 小节匹配(`- key` 或 `key:` 列表项)
|
||
_PARAM_RE = re.compile(r"^\s*[-*]\s+([a-z_][a-z0-9_]*)\b", re.M)
|
||
|
||
|
||
def read_contract_defaults(contract_path: str | Path) -> dict[str, Any]:
|
||
"""契约文档 → 参数默认值快照(解析失败返回 {},不阻断)。"""
|
||
path = Path(contract_path)
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
except OSError:
|
||
return {}
|
||
defaults: dict[str, Any] = {}
|
||
for key in _PARAM_RE.findall(text):
|
||
defaults.setdefault(key, "")
|
||
return defaults
|
||
|
||
|
||
def contract_path_for(repo_root: str | Path, stage: str) -> Path:
|
||
"""阶段契约文档路径(仓库内 gfa_core/schemas/stages/{stage}_contract_v1.md)。"""
|
||
return Path(repo_root) / "bioflow_kernel" / "gfa_core" / "schemas" / "stages" / f"{stage}_contract_v1.md"
|
||
|
||
|
||
def bba_env_defaults(settings_bba: dict[str, str]) -> dict[str, str]:
|
||
"""BBA_* 环境 → 表单默认值标注(BBA_PFAM_DIR 等)。"""
|
||
return {key: value for key, value in settings_bba.items()}
|
||
|
||
|
||
def archive_param_prefill(archive: dict[str, Any], stage: str) -> dict[str, Any]:
|
||
"""存档参数预填(FR-BIO-023):取存档 params[stage];无则空。"""
|
||
params = archive.get("params") or {}
|
||
if not isinstance(params, dict):
|
||
return {}
|
||
return dict(params.get(stage) or {})
|