feat(health): santé du flux — goulots et cartes zombies (issue #9)
app/flow.py: séjours par colonne reconstruits du journal,
médianes, colonne la plus lente; zombies = aucune activité
(journal/Gitea) depuis KANBANTE_ZOMBIE_DAYS (30). Pages /health
et /health/{o}/{r} + actions rapides (commenter/backlog/fermer).
This commit is contained in:
+123
@@ -0,0 +1,123 @@
|
||||
"""Flow health (roadmap #9): bottlenecks and zombie cards.
|
||||
|
||||
Boards show state; this module reads the event journal for health:
|
||||
how long cards actually stay in each column (dwell samples are the
|
||||
intervals between an entry and the next move/close), and which
|
||||
open cards have had no activity for a long time (zombies).
|
||||
"""
|
||||
|
||||
import statistics
|
||||
from datetime import datetime
|
||||
|
||||
from app.models import KANBAN_COLUMNS
|
||||
|
||||
|
||||
def parse_ts(raw) -> datetime | None:
|
||||
"""Parse journal ('YYYY-MM-DD HH:MM:SS') and Gitea ISO stamps."""
|
||||
text = str(raw or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
|
||||
try:
|
||||
return datetime.strptime(text[:19], fmt)
|
||||
except ValueError:
|
||||
continue
|
||||
try:
|
||||
return datetime.strptime(text[:10], "%Y-%m-%d")
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def column_dwell(events: list[dict]) -> dict[str, list[float]]:
|
||||
"""Dwell samples (in days) per column, rebuilt from the journal.
|
||||
|
||||
A stay starts when a card enters a column (created / moved /
|
||||
reopened) and ends at its next move or close. The first stay of
|
||||
a card whose journal starts mid-flight is unknowable and is
|
||||
simply not sampled.
|
||||
"""
|
||||
by_issue: dict[tuple[str, int], list[dict]] = {}
|
||||
for event in events:
|
||||
by_issue.setdefault(
|
||||
(event["repo_full_name"], event["issue_number"]), []
|
||||
).append(event)
|
||||
dwell: dict[str, list[float]] = {}
|
||||
for issue_events in by_issue.values():
|
||||
ordered = sorted(
|
||||
issue_events, key=lambda e: str(e.get("created_at") or "")
|
||||
)
|
||||
current: str | None = None
|
||||
since: datetime | None = None
|
||||
for event in ordered:
|
||||
at = parse_ts(event.get("created_at"))
|
||||
if at is None:
|
||||
continue
|
||||
kind = event["event_type"]
|
||||
if kind == "created":
|
||||
if current is None:
|
||||
current = event.get("to_column") or "backlog"
|
||||
since = at
|
||||
elif kind == "moved":
|
||||
if current is not None and since is not None:
|
||||
dwell.setdefault(current, []).append(
|
||||
(at - since).total_seconds() / 86400
|
||||
)
|
||||
elif current is None and event.get("from_column"):
|
||||
# The stay in from_column began before the journal
|
||||
# window: only the new stay can be timed.
|
||||
pass
|
||||
current = event.get("to_column") or current
|
||||
since = at
|
||||
elif kind == "closed":
|
||||
if current is not None and since is not None:
|
||||
dwell.setdefault(current, []).append(
|
||||
(at - since).total_seconds() / 86400
|
||||
)
|
||||
current = "done"
|
||||
since = at
|
||||
elif kind == "reopened":
|
||||
current = event.get("to_column") or "backlog"
|
||||
since = at
|
||||
return dwell
|
||||
|
||||
|
||||
def summarize_dwell(dwell: dict[str, list[float]]) -> list[dict]:
|
||||
"""Median dwell per column, in board order."""
|
||||
summary = []
|
||||
for column in KANBAN_COLUMNS:
|
||||
samples = dwell.get(column) or []
|
||||
if not samples:
|
||||
continue
|
||||
summary.append(
|
||||
{
|
||||
"column": column,
|
||||
"count": len(samples),
|
||||
"median_days": round(statistics.median(samples), 1),
|
||||
"avg_days": round(sum(samples) / len(samples), 1),
|
||||
}
|
||||
)
|
||||
return summary
|
||||
|
||||
|
||||
def slowest_column(summary: list[dict], min_samples: int = 2) -> dict | None:
|
||||
"""The bottleneck: slowest column (done excluded) with data."""
|
||||
candidates = [
|
||||
row for row in summary
|
||||
if row["column"] != "done" and row["count"] >= min_samples
|
||||
]
|
||||
if not candidates:
|
||||
return None
|
||||
return max(candidates, key=lambda r: r["median_days"])
|
||||
|
||||
|
||||
def last_activity(events: list[dict]) -> dict[tuple[str, int], datetime]:
|
||||
"""Most recent journal timestamp per (repo, issue)."""
|
||||
latest: dict[tuple[str, int], datetime] = {}
|
||||
for event in events:
|
||||
at = parse_ts(event.get("created_at"))
|
||||
if at is None:
|
||||
continue
|
||||
key = (event["repo_full_name"], event["issue_number"])
|
||||
if key not in latest or at > latest[key]:
|
||||
latest[key] = at
|
||||
return latest
|
||||
Reference in New Issue
Block a user