Spark
7eef8bb604
KANBANTE_RULES (JSON validé): conditions colonne/âge/étiquette/ WIP -> add_label / notify (garde quotidienne) / move_top; évaluation 15 min + POST /api/rules/run; actions journalisées (événement rule); page /rules. Aucune fermeture possible.
112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
"""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
|