config.py

07-Automation/partner_pipeline/config.py

1. """partner-thresholds.md config and the arming guard."""
3. from __future__ import annotations
5. import re
6. import sys
8. import yaml
10. from .paths import CONFIG_MD
12. # Any value of this literal in the active mode means the pipeline is not armed.
13. SENTINEL = "TBD"
14. # The two classify outcomes that are not a partner tier.
15. NON_PARTNER = ("Not a partner client", "UNCLASSIFIED: thresholds not set")
18. def load_config() -> dict:
19. """Extract and parse the first ```yaml fenced block from partner-thresholds.md."""
20. if not CONFIG_MD.exists():
21. sys.exit(f"Config not found: {CONFIG_MD}")
22. text = CONFIG_MD.read_text(encoding="utf-8")
23. m = re.search(r"```yaml\s*\n(.*?)\n```", text, re.DOTALL)
24. if not m:
25. sys.exit("No ```yaml block found in partner-thresholds.md")
26. return yaml.safe_load(m.group(1)) or {}
29. def _has_sentinel(node) -> bool:
30. if isinstance(node, str):
31. return node.strip() == SENTINEL
32. if isinstance(node, dict):
33. return any(_has_sentinel(v) for v in node.values())
34. if isinstance(node, list):
35. return any(_has_sentinel(v) for v in node)
36. return False
39. def is_armed(cfg: dict) -> tuple[bool, list[str]]:
40. """A run is armed only when the active mode has no TBDs and min_orders is set."""
41. reasons = []
42. mode = cfg.get("mode", "ltv_bands")
43. active = cfg.get(mode)
44. if _has_sentinel(active):
45. reasons.append(f"'{mode}' block still has TBD placeholder(s)")
46. if cfg.get("min_orders_to_qualify") == SENTINEL:
47. reasons.append("min_orders_to_qualify is TBD")
48. return (len(reasons) == 0, reasons)
51. def guard_or_exit():
52. """The write stages refuse to run until the thresholds are set."""
53. cfg = load_config()
54. armed, reasons = is_armed(cfg)
55. if not armed:
56. sys.exit("REFUSED: thresholds not set (" + "; ".join(reasons) +
57. "). Fill partner-thresholds.md before write stages.")
58. return cfg