cfdaily
bae3244e24
clear_cache() 中 _steps_cache.clear() 改为 _steps_cache = None， 避免在 get_steps() 之前调用时 AttributeError。
152 lines
4.6 KiB
Python
152 lines
4.6 KiB
Python
"""工具链事件模板引擎（Toolchain Event Hub）
|
||
|
||
加载 templates/toolchain/ 下的 Markdown 模板，提供 {variable} 占位符渲染。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
TEMPLATES_DIR = Path(__file__).parent.parent.parent / "templates" / "toolchain"
|
||
|
||
# 模板文件名映射
|
||
_TEMPLATE_MAP: Dict[str, str] = {
|
||
"review_request": "review_request.md",
|
||
"review_result": "review_result.md",
|
||
"issue_assigned": "issue_assigned.md",
|
||
"ci_failure": "ci_failure.md",
|
||
"deploy_failure": "deploy_failure.md",
|
||
"review_updated": "review_updated.md",
|
||
"review_comment": "review_comment.md",
|
||
"review_merged": "review_merged.md",
|
||
"mention": "mention.md",
|
||
}
|
||
|
||
# 模板缓存
|
||
_template_cache: Dict[str, str] = {}
|
||
|
||
|
||
def _load_template(name: str) -> str:
|
||
"""加载并缓存模板文件内容。
|
||
|
||
Args:
|
||
name: 模板名称（不含 .md 后缀）
|
||
|
||
Returns:
|
||
模板文本内容
|
||
|
||
Raises:
|
||
FileNotFoundError: 模板文件不存在
|
||
"""
|
||
if name in _template_cache:
|
||
return _template_cache[name]
|
||
|
||
filename = _TEMPLATE_MAP.get(name)
|
||
if not filename:
|
||
raise ValueError(f"Unknown template: {name}")
|
||
|
||
path = TEMPLATES_DIR / filename
|
||
if not path.exists():
|
||
raise FileNotFoundError(f"Template not found: {path}")
|
||
|
||
content = path.read_text(encoding="utf-8")
|
||
_template_cache[name] = content
|
||
logger.debug("Loaded template: %s (%d bytes)", name, len(content))
|
||
return content
|
||
|
||
|
||
def _escape_braces(value: str) -> str:
|
||
"""转义花括号防止 format_map 报错"""
|
||
return str(value).replace("{", "{{").replace("}", "}}")
|
||
|
||
|
||
def render_template(name: str, variables: Dict[str, str]) -> str:
|
||
"""渲染模板，将 {variable} 占位符替换为实际值。
|
||
|
||
使用 defaultdict(str) 确保未提供的变量替换为空字符串而非报错。
|
||
|
||
Args:
|
||
name: 模板名称
|
||
variables: 变量字典
|
||
|
||
Returns:
|
||
渲染后的文本
|
||
"""
|
||
template_text = _load_template(name)
|
||
# 先对所有变量值转义花括号，防止 format_map 报错
|
||
escaped_vars = {k: _escape_braces(v) for k, v in variables.items()}
|
||
safe_vars: Dict[str, str] = defaultdict(str, escaped_vars)
|
||
return template_text.format_map(safe_vars)
|
||
|
||
|
||
def clear_cache() -> None:
|
||
"""清空模板缓存（用于测试或热更新）"""
|
||
_template_cache.clear()
|
||
global _steps_cache
|
||
_steps_cache = None # 重置为 None，强制下次 reload
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# §21 §4.2 YAML steps 模板加载
|
||
# ---------------------------------------------------------------------------
|
||
|
||
STEPS_YAML_PATH = Path(__file__).parent.parent.parent / "config" / "toolchain-templates.yaml"
|
||
|
||
_steps_cache: Optional[dict] = None
|
||
|
||
|
||
def _load_steps_yaml() -> dict:
|
||
"""加载并缓存 toolchain-templates.yaml。"""
|
||
global _steps_cache
|
||
if _steps_cache is not None:
|
||
return _steps_cache
|
||
try:
|
||
import yaml
|
||
with open(STEPS_YAML_PATH, encoding="utf-8") as f:
|
||
_steps_cache = yaml.safe_load(f) or {}
|
||
logger.debug("Loaded steps YAML: %d action_types", len(_steps_cache))
|
||
except FileNotFoundError:
|
||
logger.warning("Steps YAML not found: %s", STEPS_YAML_PATH)
|
||
_steps_cache = {}
|
||
except Exception as e:
|
||
logger.error("Failed to load steps YAML: %s", e)
|
||
_steps_cache = {}
|
||
return _steps_cache
|
||
|
||
|
||
def get_steps(action_type: str, business_type: str = "") -> List[str]:
|
||
"""从 YAML 模板配置获取 steps。
|
||
|
||
Args:
|
||
action_type: 动作类型（issue_assigned / ci_failure / ...）
|
||
business_type: 业务子类型（feature/impl/bug/docs/refactor/test/infrastructure）
|
||
|
||
Returns:
|
||
steps 列表，找不到返回空列表
|
||
"""
|
||
templates = _load_steps_yaml()
|
||
section = templates.get(action_type, {})
|
||
if isinstance(section, dict) and business_type:
|
||
subsection = section.get(business_type, {})
|
||
return subsection.get("steps", [])
|
||
if isinstance(section, dict):
|
||
return section.get("steps", [])
|
||
return []
|
||
|
||
|
||
def get_output_template(action_type: str, business_type: str = "") -> str:
|
||
"""从 YAML 模板配置获取 output_template。"""
|
||
templates = _load_steps_yaml()
|
||
section = templates.get(action_type, {})
|
||
if isinstance(section, dict) and business_type:
|
||
subsection = section.get(business_type, {})
|
||
return subsection.get("output_template", "")
|
||
if isinstance(section, dict):
|
||
return section.get("output_template", "")
|
||
return ""
|