"""Automation rules (roadmap #8): small declarative IFTTT for the board. Rules live in KANBANTE_RULES (JSON array) — deliberately boring: [{"name": "Relance révision", "when": {"column": "review", "min_age_days": 3}, "then": {"notify": true}}, {"name": "Urgent qui vieillit", "when": {"column": "in_progress", "min_age_days": 7}, "then": {"add_label": "urgent"}}] Conditions: column, min_age_days (time in the current column), label (present on the card), wip_over (column holds MORE cards than N — then every card of the column matches). Actions: add_label, notify (ntfy), move_top (of its column). Guardrail, by design: there is NO close/delete/move action — a rule can never close an issue or move a card between columns. """ import json from app.models import KANBAN_COLUMNS CONDITION_KEYS = {"column", "min_age_days", "label", "wip_over"} ACTION_KEYS = {"add_label", "notify", "move_top"} def parse_rules(raw: str) -> tuple[list[dict], list[str]]: """Validate KANBANTE_RULES; returns (rules, human errors).""" if not raw or not raw.strip(): return [], [] try: data = json.loads(raw) except ValueError as exc: return [], [f"JSON invalide : {exc}"] if not isinstance(data, list): return [], ["KANBANTE_RULES doit être un tableau JSON de règles"] rules: list[dict] = [] errors: list[str] = [] for i, item in enumerate(data): label = f"règle #{i + 1}" if not isinstance(item, dict): errors.append(f"{label} : pas un objet") continue name = str(item.get("name") or label) when = item.get("when") then = item.get("then") if not isinstance(when, dict) or not isinstance(then, dict): errors.append(f"{name} : 'when' et 'then' (objets) sont requis") continue bad_when = set(when) - CONDITION_KEYS bad_then = set(then) - ACTION_KEYS if bad_when: errors.append(f"{name} : conditions inconnues {sorted(bad_when)}") continue if bad_then: errors.append( f"{name} : actions inconnues {sorted(bad_then)} " "(fermer/déplacer une carte par règle est interdit en v1)" ) continue if "column" in when and when["column"] not in KANBAN_COLUMNS: errors.append(f"{name} : colonne inconnue « {when['column']} »") continue if "min_age_days" in when and ( not isinstance(when["min_age_days"], int) or when["min_age_days"] < 0 ): errors.append(f"{name} : min_age_days doit être un entier ≥ 0") continue if "wip_over" in when and ( not isinstance(when["wip_over"], int) or when["wip_over"] < 1 ): errors.append(f"{name} : wip_over doit être un entier ≥ 1") continue if "add_label" in then and ( not isinstance(then["add_label"], str) or not then["add_label"].strip() ): errors.append(f"{name} : add_label doit être un nom d'étiquette") continue if not then: errors.append(f"{name} : aucune action") continue rules.append({"name": name, "when": when, "then": then}) return rules, errors def matching_cards( rule: dict, columns: dict[str, list[dict]], ages: dict[int, dict], ) -> list[tuple[dict, str]]: """Cards of `columns` (merged board) matched by the rule.""" when = rule["when"] targets = [when["column"]] if "column" in when else list(columns) matches: list[tuple[dict, str]] = [] for column in targets: cards = columns.get(column, []) if "wip_over" in when and len(cards) <= when["wip_over"]: continue for issue in cards: if "label" in when and when["label"] not in [ (lab or {}).get("name") for lab in issue.get("labels") or [] ]: continue if "min_age_days" in when: age = ages.get(issue["number"]) or {} if (age.get("days") or 0) < when["min_age_days"]: continue matches.append((issue, column)) return matches