merge-config.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441
  1. #!/usr/bin/env python3
  2. # /// script
  3. # requires-python = ">=3.9"
  4. # dependencies = ["pyyaml"]
  5. # ///
  6. """Merge module configuration into shared _bmad/config.yaml and config.user.yaml.
  7. Reads a module.yaml definition and a JSON answers file, then writes or updates
  8. the shared config.yaml (core values at root + module section) and config.user.yaml
  9. (user_name, communication_language, plus any module variable with user_setting: true).
  10. Uses an anti-zombie pattern for the module section in config.yaml.
  11. Legacy migration: when --legacy-dir is provided, reads old per-module config files
  12. from {legacy-dir}/{module-code}/config.yaml and {legacy-dir}/core/config.yaml.
  13. Matching values serve as fallback defaults (answers override them). After a
  14. successful merge, the legacy config.yaml files are deleted. Only the current
  15. module and core directories are touched — other module directories are left alone.
  16. Exit codes: 0=success, 1=validation error, 2=runtime error
  17. """
  18. import argparse
  19. import json
  20. import sys
  21. from pathlib import Path
  22. try:
  23. import yaml
  24. except ImportError:
  25. print("Error: pyyaml is required (PEP 723 dependency)", file=sys.stderr)
  26. sys.exit(2)
  27. def parse_args():
  28. parser = argparse.ArgumentParser(
  29. description="Merge module config into shared _bmad/config.yaml with anti-zombie pattern."
  30. )
  31. parser.add_argument(
  32. "--config-path",
  33. required=True,
  34. help="Path to the target _bmad/config.yaml file",
  35. )
  36. parser.add_argument(
  37. "--module-yaml",
  38. required=True,
  39. help="Path to the module.yaml definition file",
  40. )
  41. parser.add_argument(
  42. "--answers",
  43. required=True,
  44. help="Path to JSON file with collected answers",
  45. )
  46. parser.add_argument(
  47. "--user-config-path",
  48. required=True,
  49. help="Path to the target _bmad/config.user.yaml file",
  50. )
  51. parser.add_argument(
  52. "--legacy-dir",
  53. help="Path to _bmad/ directory to check for legacy per-module config files. "
  54. "Matching values are used as fallback defaults, then legacy files are deleted.",
  55. )
  56. parser.add_argument(
  57. "--verbose",
  58. action="store_true",
  59. help="Print detailed progress to stderr",
  60. )
  61. return parser.parse_args()
  62. def load_yaml_file(path: str) -> dict:
  63. """Load a YAML file, returning empty dict if file doesn't exist."""
  64. file_path = Path(path)
  65. if not file_path.exists():
  66. return {}
  67. with open(file_path, "r", encoding="utf-8") as f:
  68. content = yaml.safe_load(f)
  69. return content if content else {}
  70. def load_json_file(path: str) -> dict:
  71. """Load a JSON file."""
  72. with open(path, "r", encoding="utf-8") as f:
  73. return json.load(f)
  74. # Keys that live at config root (shared across all modules)
  75. _CORE_KEYS = frozenset(
  76. {"user_name", "communication_language", "document_output_language", "output_folder"}
  77. )
  78. def load_legacy_values(
  79. legacy_dir: str, module_code: str, module_yaml: dict, verbose: bool = False
  80. ) -> tuple[dict, dict, list]:
  81. """Read legacy per-module config files and return core/module value dicts.
  82. Reads {legacy_dir}/core/config.yaml and {legacy_dir}/{module_code}/config.yaml.
  83. Only returns values whose keys match the current schema (core keys or module.yaml
  84. variable definitions). Other modules' directories are not touched.
  85. Returns:
  86. (legacy_core, legacy_module, files_found) where files_found lists paths read.
  87. """
  88. legacy_core: dict = {}
  89. legacy_module: dict = {}
  90. files_found: list = []
  91. # Read core legacy config
  92. core_path = Path(legacy_dir) / "core" / "config.yaml"
  93. if core_path.exists():
  94. core_data = load_yaml_file(str(core_path))
  95. files_found.append(str(core_path))
  96. for k, v in core_data.items():
  97. if k in _CORE_KEYS:
  98. legacy_core[k] = v
  99. if verbose:
  100. print(f"Legacy core config: {list(legacy_core.keys())}", file=sys.stderr)
  101. # Read module legacy config
  102. mod_path = Path(legacy_dir) / module_code / "config.yaml"
  103. if mod_path.exists():
  104. mod_data = load_yaml_file(str(mod_path))
  105. files_found.append(str(mod_path))
  106. for k, v in mod_data.items():
  107. if k in _CORE_KEYS:
  108. # Core keys duplicated in module config — only use if not already set
  109. if k not in legacy_core:
  110. legacy_core[k] = v
  111. elif k in module_yaml and isinstance(module_yaml[k], dict):
  112. # Module-specific key that matches a current variable definition
  113. legacy_module[k] = v
  114. if verbose:
  115. print(
  116. f"Legacy module config: {list(legacy_module.keys())}", file=sys.stderr
  117. )
  118. return legacy_core, legacy_module, files_found
  119. def apply_legacy_defaults(answers: dict, legacy_core: dict, legacy_module: dict) -> dict:
  120. """Apply legacy values as fallback defaults under the answers.
  121. Legacy values fill in any key not already present in answers.
  122. Explicit answers always win.
  123. """
  124. merged = dict(answers)
  125. if legacy_core:
  126. core = merged.get("core", {})
  127. filled_core = dict(legacy_core) # legacy as base
  128. filled_core.update(core) # answers override
  129. merged["core"] = filled_core
  130. if legacy_module:
  131. mod = merged.get("module", {})
  132. filled_mod = dict(legacy_module) # legacy as base
  133. filled_mod.update(mod) # answers override
  134. merged["module"] = filled_mod
  135. return merged
  136. def cleanup_legacy_configs(
  137. legacy_dir: str, module_code: str, verbose: bool = False
  138. ) -> list:
  139. """Delete legacy config.yaml files for this module and core only.
  140. Returns list of deleted file paths.
  141. """
  142. deleted = []
  143. for subdir in (module_code, "core"):
  144. legacy_path = Path(legacy_dir) / subdir / "config.yaml"
  145. if legacy_path.exists():
  146. if verbose:
  147. print(f"Deleting legacy config: {legacy_path}", file=sys.stderr)
  148. legacy_path.unlink()
  149. deleted.append(str(legacy_path))
  150. return deleted
  151. def extract_module_metadata(module_yaml: dict) -> dict:
  152. """Extract non-variable metadata fields from module.yaml."""
  153. meta = {}
  154. for k in ("name", "description"):
  155. if k in module_yaml:
  156. meta[k] = module_yaml[k]
  157. meta["version"] = module_yaml.get("module_version") # null if absent
  158. if "default_selected" in module_yaml:
  159. meta["default_selected"] = module_yaml["default_selected"]
  160. return meta
  161. def apply_result_templates(
  162. module_yaml: dict, module_answers: dict, verbose: bool = False
  163. ) -> dict:
  164. """Apply result templates from module.yaml to transform raw answer values.
  165. For each answer, if the corresponding variable definition in module.yaml has
  166. a 'result' field, replaces {value} in that template with the answer. Skips
  167. the template if the answer already contains '{project-root}' to prevent
  168. double-prefixing.
  169. """
  170. transformed = {}
  171. for key, value in module_answers.items():
  172. var_def = module_yaml.get(key)
  173. if (
  174. isinstance(var_def, dict)
  175. and "result" in var_def
  176. and "{project-root}" not in str(value)
  177. ):
  178. template = var_def["result"]
  179. transformed[key] = template.replace("{value}", str(value))
  180. if verbose:
  181. print(
  182. f"Applied result template for '{key}': {value} → {transformed[key]}",
  183. file=sys.stderr,
  184. )
  185. else:
  186. transformed[key] = value
  187. return transformed
  188. def merge_config(
  189. existing_config: dict,
  190. module_yaml: dict,
  191. answers: dict,
  192. verbose: bool = False,
  193. ) -> dict:
  194. """Merge answers into config, applying anti-zombie pattern.
  195. Args:
  196. existing_config: Current config.yaml contents (may be empty)
  197. module_yaml: The module definition
  198. answers: JSON with 'core' and/or 'module' keys
  199. verbose: Print progress to stderr
  200. Returns:
  201. Updated config dict ready to write
  202. """
  203. config = dict(existing_config)
  204. module_code = module_yaml.get("code")
  205. if not module_code:
  206. print("Error: module.yaml must have a 'code' field", file=sys.stderr)
  207. sys.exit(1)
  208. # Migrate legacy core: section to root
  209. if "core" in config and isinstance(config["core"], dict):
  210. if verbose:
  211. print("Migrating legacy 'core' section to root", file=sys.stderr)
  212. config.update(config.pop("core"))
  213. # Strip user-only keys from config — they belong exclusively in config.user.yaml
  214. for key in _CORE_USER_KEYS:
  215. if key in config:
  216. if verbose:
  217. print(f"Removing user-only key '{key}' from config (belongs in config.user.yaml)", file=sys.stderr)
  218. del config[key]
  219. # Write core values at root (global properties, not nested under "core")
  220. # Exclude user-only keys — those belong exclusively in config.user.yaml
  221. core_answers = answers.get("core")
  222. if core_answers:
  223. shared_core = {k: v for k, v in core_answers.items() if k not in _CORE_USER_KEYS}
  224. if shared_core:
  225. if verbose:
  226. print(f"Writing core config at root: {list(shared_core.keys())}", file=sys.stderr)
  227. config.update(shared_core)
  228. # Anti-zombie: remove existing module section
  229. if module_code in config:
  230. if verbose:
  231. print(
  232. f"Removing existing '{module_code}' section (anti-zombie)",
  233. file=sys.stderr,
  234. )
  235. del config[module_code]
  236. # Build module section: metadata + variable values
  237. module_section = extract_module_metadata(module_yaml)
  238. module_answers = apply_result_templates(
  239. module_yaml, answers.get("module", {}), verbose
  240. )
  241. module_section.update(module_answers)
  242. if verbose:
  243. print(
  244. f"Writing '{module_code}' section with keys: {list(module_section.keys())}",
  245. file=sys.stderr,
  246. )
  247. config[module_code] = module_section
  248. return config
  249. # Core keys that are always written to config.user.yaml
  250. _CORE_USER_KEYS = ("user_name", "communication_language")
  251. def extract_user_settings(module_yaml: dict, answers: dict) -> dict:
  252. """Collect settings that belong in config.user.yaml.
  253. Includes user_name and communication_language from core answers, plus any
  254. module variable whose definition contains user_setting: true.
  255. """
  256. user_settings = {}
  257. core_answers = answers.get("core", {})
  258. for key in _CORE_USER_KEYS:
  259. if key in core_answers:
  260. user_settings[key] = core_answers[key]
  261. module_answers = answers.get("module", {})
  262. for var_name, var_def in module_yaml.items():
  263. if isinstance(var_def, dict) and var_def.get("user_setting") is True:
  264. if var_name in module_answers:
  265. user_settings[var_name] = module_answers[var_name]
  266. return user_settings
  267. def write_config(config: dict, config_path: str, verbose: bool = False) -> None:
  268. """Write config dict to YAML file, creating parent dirs as needed."""
  269. path = Path(config_path)
  270. path.parent.mkdir(parents=True, exist_ok=True)
  271. if verbose:
  272. print(f"Writing config to {path}", file=sys.stderr)
  273. with open(path, "w", encoding="utf-8") as f:
  274. yaml.dump(
  275. config,
  276. f,
  277. default_flow_style=False,
  278. allow_unicode=True,
  279. sort_keys=False,
  280. )
  281. def reject_unresolved_paths(named_paths: list[tuple[str, str]]) -> None:
  282. """Exit with a clear error if any path argument still contains the literal
  283. ``{project-root}`` token. That token is meaningful only inside config
  284. values; filesystem path arguments must be resolved by the caller. Failing
  285. loudly here prevents silently creating a junk ``{project-root}/`` directory.
  286. """
  287. for name, value in named_paths:
  288. if value and "{project-root}" in value:
  289. print(
  290. json.dumps(
  291. {
  292. "status": "error",
  293. "error": (
  294. f"Unresolved '{{project-root}}' token in {name} path: {value!r}. "
  295. "Resolve '{project-root}' to the actual project root before running "
  296. "this script — it is a filesystem path, not a config value."
  297. ),
  298. },
  299. indent=2,
  300. ),
  301. file=sys.stderr,
  302. )
  303. sys.exit(1)
  304. def main():
  305. args = parse_args()
  306. reject_unresolved_paths(
  307. [
  308. ("--config-path", args.config_path),
  309. ("--user-config-path", args.user_config_path),
  310. ("--legacy-dir", args.legacy_dir),
  311. ]
  312. )
  313. # Load inputs
  314. module_yaml = load_yaml_file(args.module_yaml)
  315. if not module_yaml:
  316. print(f"Error: Could not load module.yaml from {args.module_yaml}", file=sys.stderr)
  317. sys.exit(1)
  318. answers = load_json_file(args.answers)
  319. existing_config = load_yaml_file(args.config_path)
  320. if args.verbose:
  321. exists = Path(args.config_path).exists()
  322. print(f"Config file exists: {exists}", file=sys.stderr)
  323. if exists:
  324. print(f"Existing sections: {list(existing_config.keys())}", file=sys.stderr)
  325. # Legacy migration: read old per-module configs as fallback defaults
  326. legacy_files_found = []
  327. if args.legacy_dir:
  328. module_code = module_yaml.get("code", "")
  329. legacy_core, legacy_module, legacy_files_found = load_legacy_values(
  330. args.legacy_dir, module_code, module_yaml, args.verbose
  331. )
  332. if legacy_core or legacy_module:
  333. answers = apply_legacy_defaults(answers, legacy_core, legacy_module)
  334. if args.verbose:
  335. print("Applied legacy values as fallback defaults", file=sys.stderr)
  336. # Merge and write config.yaml
  337. updated_config = merge_config(existing_config, module_yaml, answers, args.verbose)
  338. write_config(updated_config, args.config_path, args.verbose)
  339. # Merge and write config.user.yaml
  340. user_settings = extract_user_settings(module_yaml, answers)
  341. existing_user_config = load_yaml_file(args.user_config_path)
  342. updated_user_config = dict(existing_user_config)
  343. updated_user_config.update(user_settings)
  344. if user_settings:
  345. write_config(updated_user_config, args.user_config_path, args.verbose)
  346. # Legacy cleanup: delete old per-module config files
  347. legacy_deleted = []
  348. if args.legacy_dir:
  349. legacy_deleted = cleanup_legacy_configs(
  350. args.legacy_dir, module_yaml["code"], args.verbose
  351. )
  352. # Output result summary as JSON
  353. module_code = module_yaml["code"]
  354. result = {
  355. "status": "success",
  356. "config_path": str(Path(args.config_path).resolve()),
  357. "user_config_path": str(Path(args.user_config_path).resolve()),
  358. "module_code": module_code,
  359. "core_updated": bool(answers.get("core")),
  360. "module_keys": list(updated_config.get(module_code, {}).keys()),
  361. "user_keys": list(user_settings.keys()),
  362. "legacy_configs_found": legacy_files_found,
  363. "legacy_configs_deleted": legacy_deleted,
  364. }
  365. print(json.dumps(result, indent=2))
  366. if __name__ == "__main__":
  367. main()