Spark
ec3590d25f
La diapositive ne grandit plus au-delà de l'écran: plus de barre de défilement de page, « Ouvertes »/« Fermées »/« EN DIRECT » toujours affichés au-dessus de la navigation des diapos; cartes tronquées en fondu + ligne « +N autres cartes en vol »; mode compact sous 760px de hauteur (TV 720p).
3264 lines
114 KiB
Python
3264 lines
114 KiB
Python
"""
|
||
Kanbante — Kanban project manager integrated with Gitea.
|
||
FastAPI backend serving HTML + JSON API.
|
||
|
||
0.2.0 highlights (see CHANGELOG.md):
|
||
- Optional HTTP Basic auth (KANBANTE_AUTH_USER / _PASSWORD) + CSRF
|
||
double-submit protection on every POST.
|
||
- Two-way sync: moving a card to "done" closes the Gitea issue,
|
||
moving it back reopens it (KANBANTE_SYNC_GITEA=0 disables).
|
||
- Repos are classified with a rolling activity window and Gitea's real
|
||
"archived" flag, not a hardcoded year.
|
||
- Board state is reconciled with Gitea on every board load, drag order
|
||
is persisted, and Gitea failures render a proper 502 error page
|
||
instead of a bare 500.
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import re
|
||
import secrets
|
||
from contextlib import asynccontextmanager
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
from fastapi import FastAPI, Form, Request
|
||
from fastapi.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
||
from starlette.middleware.base import BaseHTTPMiddleware
|
||
|
||
from app.config import APP_VERSION, get_settings
|
||
from app.gitea_client import GiteaClient
|
||
from app import code_links as code_links_service
|
||
from app import digest as digest_service
|
||
from app import flow as flow_service
|
||
from app import forecast as forecast_service
|
||
from app import rules as rules_service
|
||
from app import sprint as sprint_service
|
||
from app.briefing import (
|
||
format_briefing_text,
|
||
format_standup_text,
|
||
ntfy_configured,
|
||
send_ntfy,
|
||
)
|
||
from app.capture import ntfy_capture_loop, parse_capture_text
|
||
from app.models import (
|
||
KANBAN_COLUMNS,
|
||
add_pick,
|
||
column_entry_times,
|
||
get_board_state,
|
||
get_events,
|
||
get_kv,
|
||
get_metrics,
|
||
get_notes,
|
||
get_picks,
|
||
get_stats,
|
||
init_db,
|
||
log_event,
|
||
move_issue,
|
||
ping_db,
|
||
reconcile_board_state,
|
||
remove_pick,
|
||
set_column_order,
|
||
set_kv,
|
||
set_notes,
|
||
)
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger("kanbante")
|
||
|
||
BASE = Path(__file__).resolve().parent.parent
|
||
CSRF_COOKIE = "kanbante_csrf"
|
||
|
||
COLUMN_LABELS = {
|
||
"backlog": "📥 Backlog",
|
||
"todo": "📋 À faire",
|
||
"in_progress": "🚧 En cours",
|
||
"qa": "🧪 QA",
|
||
"review": "🔍 Révision",
|
||
"done": "✅ Terminé",
|
||
}
|
||
|
||
COLUMN_COLORS = {
|
||
"backlog": "#6b7280",
|
||
"todo": "#3b82f6",
|
||
"in_progress": "#f59e0b",
|
||
"qa": "#06b6d4",
|
||
"review": "#8b5cf6",
|
||
"done": "#10b981",
|
||
}
|
||
|
||
FOCUS_COLUMNS = ["in_progress", "qa", "review"]
|
||
|
||
# Columns mirrored to Gitea as kanban:<column> labels (backlog and
|
||
# done carry none: done is the closed state itself).
|
||
LABELLED_COLUMNS = ("todo", "in_progress", "qa", "review")
|
||
|
||
EVENT_ICONS = {
|
||
"created": "➕",
|
||
"moved": "➡️",
|
||
"closed": "✅",
|
||
"reopened": "🔁",
|
||
"commented": "💬",
|
||
}
|
||
|
||
|
||
# ── App setup ──────────────────────────────────────────────
|
||
|
||
async def _briefing_scheduler() -> None:
|
||
"""Send the daily briefing to ntfy at KANBANTE_BRIEFING_TIME.
|
||
|
||
Checks once a minute; the kv store remembers the last day sent so
|
||
restarts never double-send.
|
||
"""
|
||
while True:
|
||
try:
|
||
settings = get_settings()
|
||
if settings.briefing_time and ntfy_configured(settings):
|
||
now = datetime.now()
|
||
today = now.strftime("%Y-%m-%d")
|
||
if (
|
||
now.strftime("%H:%M") >= settings.briefing_time
|
||
and get_kv("briefing_last_sent") != today
|
||
):
|
||
data = await _briefing_data()
|
||
if await send_ntfy(settings, format_briefing_text(data)):
|
||
set_kv("briefing_last_sent", today)
|
||
logger.info("Daily briefing sent to ntfy (%s)", today)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception as exc: # never let the scheduler die silently
|
||
logger.warning("Briefing scheduler error: %s", exc)
|
||
await asyncio.sleep(60)
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
init_db()
|
||
settings = get_settings()
|
||
if not settings.auth_enabled:
|
||
logger.warning(
|
||
"Authentication is DISABLED (KANBANTE_AUTH_USER / "
|
||
"KANBANTE_AUTH_PASSWORD not set). Anyone who can reach this "
|
||
"service can read every repo visible to the Gitea token and "
|
||
"modify boards. Set the credentials or put Kanbante behind "
|
||
"an authenticating reverse proxy."
|
||
)
|
||
if not settings.gitea_token:
|
||
logger.warning(
|
||
"GITEA_TOKEN is not set: only public Gitea content will be visible."
|
||
)
|
||
scheduler = asyncio.create_task(_briefing_scheduler())
|
||
digest_task = asyncio.create_task(_digest_scheduler())
|
||
rules_task = asyncio.create_task(_rules_scheduler())
|
||
capture_task = None
|
||
if settings.capture_ntfy_topic and settings.ntfy_url:
|
||
async def _capture_create(title: str, body: str) -> None:
|
||
repo = get_settings().capture_repo
|
||
if not repo:
|
||
logger.warning("ntfy capture: no KANBANTE_CAPTURE_REPO, skipped")
|
||
return
|
||
await _do_capture(title, body, repo, "backlog")
|
||
|
||
capture_task = asyncio.create_task(
|
||
ntfy_capture_loop(
|
||
_capture_create,
|
||
lambda: get_kv("capture_ntfy_last_id"),
|
||
lambda value: set_kv("capture_ntfy_last_id", value),
|
||
)
|
||
)
|
||
logger.info(
|
||
"ntfy capture listener started (topic %s)",
|
||
settings.capture_ntfy_topic,
|
||
)
|
||
yield
|
||
scheduler.cancel()
|
||
digest_task.cancel()
|
||
rules_task.cancel()
|
||
if capture_task is not None:
|
||
capture_task.cancel()
|
||
await gitea.aclose()
|
||
|
||
|
||
app = FastAPI(title="Kanbante", version=APP_VERSION, lifespan=lifespan)
|
||
app.mount("/static", StaticFiles(directory=str(BASE / "static")), name="static")
|
||
gitea = GiteaClient()
|
||
|
||
|
||
# ── Auth middleware ────────────────────────────────────────
|
||
|
||
class AuthMiddleware(BaseHTTPMiddleware):
|
||
"""HTTP Basic auth, enabled only when credentials are configured.
|
||
|
||
/api/health stays public so Docker's HEALTHCHECK (liveness) works.
|
||
When KANBANTE_REQUIRE_AUTH=1 and no credentials are configured, every
|
||
other route answers 503: failing closed beats failing open.
|
||
"""
|
||
|
||
async def dispatch(self, request: Request, call_next):
|
||
# /api/health: liveness for Docker. /webhooks/gitea: authenticated
|
||
# by its HMAC signature, not by Basic auth (Gitea cannot log in).
|
||
if request.url.path in ("/api/health", "/webhooks/gitea", "/api/capture"):
|
||
return await call_next(request)
|
||
settings = get_settings()
|
||
if not settings.auth_enabled:
|
||
if settings.require_auth:
|
||
return JSONResponse(
|
||
{
|
||
"ok": False,
|
||
"error": "Authentication required but "
|
||
"KANBANTE_AUTH_USER / KANBANTE_AUTH_PASSWORD "
|
||
"are not configured.",
|
||
},
|
||
status_code=503,
|
||
)
|
||
return await call_next(request)
|
||
|
||
ok = False
|
||
header = request.headers.get("authorization", "")
|
||
if header.lower().startswith("basic "):
|
||
try:
|
||
decoded = base64.b64decode(header[6:]).decode("utf-8")
|
||
except Exception:
|
||
decoded = ""
|
||
user, _, password = decoded.partition(":")
|
||
ok = secrets.compare_digest(
|
||
user, settings.auth_user
|
||
) and secrets.compare_digest(password, settings.auth_password)
|
||
if not ok:
|
||
return Response(
|
||
"Authentication required",
|
||
status_code=401,
|
||
headers={"WWW-Authenticate": 'Basic realm="Kanbante"'},
|
||
)
|
||
return await call_next(request)
|
||
|
||
|
||
app.add_middleware(AuthMiddleware)
|
||
|
||
|
||
# ── Live updates (SSE hub) ─────────────────────────────────
|
||
|
||
class LiveHub:
|
||
"""Fan-out of Gitea webhook events to connected browsers (SSE)."""
|
||
|
||
def __init__(self) -> None:
|
||
self._subscribers: set[asyncio.Queue] = set()
|
||
|
||
def subscribe(self) -> asyncio.Queue:
|
||
queue: asyncio.Queue = asyncio.Queue(maxsize=50)
|
||
self._subscribers.add(queue)
|
||
return queue
|
||
|
||
def unsubscribe(self, queue: asyncio.Queue) -> None:
|
||
self._subscribers.discard(queue)
|
||
|
||
async def broadcast(self, payload: dict) -> None:
|
||
for queue in list(self._subscribers):
|
||
try:
|
||
queue.put_nowait(payload)
|
||
except asyncio.QueueFull:
|
||
pass # slow client: it will resync on its next load
|
||
|
||
@property
|
||
def subscriber_count(self) -> int:
|
||
return len(self._subscribers)
|
||
|
||
|
||
hub = LiveHub()
|
||
|
||
|
||
async def _publish_board_changed(repo_full_name: str) -> None:
|
||
"""Tell every connected client this repo's board changed (#15).
|
||
|
||
Called after each mutation Kanbante knows about (move, create,
|
||
capture, label toggle). Webhooks broadcast separately with a
|
||
richer payload; board clients accept both shapes.
|
||
"""
|
||
await hub.broadcast({"repo": repo_full_name, "type": "board-changed"})
|
||
|
||
|
||
# ── Templates ──────────────────────────────────────────────
|
||
|
||
_tpl_env = Environment(
|
||
loader=FileSystemLoader(str(BASE / "app" / "templates")),
|
||
autoescape=select_autoescape(["html"]),
|
||
)
|
||
|
||
|
||
# Which main-nav entry a template represents (issue #16): the
|
||
# shared _header.html partial highlights it from this one place
|
||
# instead of 19 hand-maintained copies.
|
||
_NAV_BY_TEMPLATE = {
|
||
"dashboard.html": "home",
|
||
"focus.html": "focus",
|
||
"today.html": "today",
|
||
"my.html": "my",
|
||
"activity.html": "activity",
|
||
"briefing.html": "briefing",
|
||
"standup.html": "standup",
|
||
"search.html": "search",
|
||
"capture.html": "capture",
|
||
"digest.html": "digest",
|
||
"forecast.html": "forecast",
|
||
"rules.html": "rules",
|
||
"health.html": "health",
|
||
"health_project.html": "health",
|
||
}
|
||
|
||
|
||
def render(
|
||
request: Request, name: str, status_code: int = 200, **context
|
||
) -> HTMLResponse:
|
||
"""Render a Jinja2 template and manage the CSRF cookie.
|
||
|
||
The CSRF token lives in an HttpOnly cookie; templates expose it in a
|
||
meta tag / hidden fields (double-submit pattern) and POST handlers
|
||
compare the submitted value with the cookie.
|
||
"""
|
||
token = request.cookies.get(CSRF_COOKIE)
|
||
generated = token is None
|
||
if generated:
|
||
token = secrets.token_urlsafe(32)
|
||
context.setdefault("csrf_token", token)
|
||
context.setdefault("version", APP_VERSION)
|
||
context.setdefault("active_nav", _NAV_BY_TEMPLATE.get(name, ""))
|
||
html = _tpl_env.get_template(name).render(**context)
|
||
response = HTMLResponse(html, status_code=status_code)
|
||
if generated:
|
||
response.set_cookie(CSRF_COOKIE, token, samesite="lax", httponly=True)
|
||
return response
|
||
|
||
|
||
def _csrf_valid(request: Request, form_token: str) -> bool:
|
||
cookie = request.cookies.get(CSRF_COOKIE, "")
|
||
provided = form_token or request.headers.get("x-csrf-token", "")
|
||
return bool(cookie) and bool(provided) and secrets.compare_digest(cookie, provided)
|
||
|
||
|
||
def _csrf_error() -> JSONResponse:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid or missing CSRF token"}, status_code=403
|
||
)
|
||
|
||
|
||
@app.exception_handler(httpx.HTTPError)
|
||
async def gitea_error_handler(request: Request, exc: httpx.HTTPError):
|
||
"""Gitea failures become a clean 502, never a bare 500."""
|
||
logger.warning("Gitea error on %s: %s", request.url.path, exc)
|
||
if request.url.path.startswith("/api/"):
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Gitea is unreachable or returned an error"},
|
||
status_code=502,
|
||
)
|
||
return render(
|
||
request,
|
||
"error.html",
|
||
status_code=502,
|
||
message="Kanbante n'a pas pu joindre Gitea. Vérifie que l'instance "
|
||
"est en ligne et que le jeton est valide, puis réessaie.",
|
||
)
|
||
|
||
|
||
# ── Helpers ────────────────────────────────────────────────
|
||
|
||
def _parse_datetime(value: str | None) -> datetime | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def classify_repos(
|
||
repos: list[dict], now: datetime | None = None
|
||
) -> tuple[list[dict], list[dict], list[dict]]:
|
||
"""Split repos into (active, inactive, archived).
|
||
|
||
"Archived" is Gitea's own flag. Among the rest, a repo is active if
|
||
it was updated within the configured rolling window — no hardcoded
|
||
calendar year.
|
||
"""
|
||
settings = get_settings()
|
||
now = now or datetime.now(timezone.utc)
|
||
cutoff = now - timedelta(days=settings.active_window_days)
|
||
active, inactive, archived = [], [], []
|
||
for repo in repos:
|
||
if repo.get("archived"):
|
||
archived.append(repo)
|
||
continue
|
||
updated = _parse_datetime(repo.get("updated_at"))
|
||
if updated is not None and updated >= cutoff:
|
||
active.append(repo)
|
||
else:
|
||
inactive.append(repo)
|
||
return active, inactive, archived
|
||
|
||
|
||
def _is_issue(item: dict) -> bool:
|
||
"""True for issues, False for pull requests.
|
||
|
||
Gitea's issues endpoint also returns PRs. A PR carries a
|
||
``pull_request`` *object*; a plain issue carries the same key
|
||
with value ``null`` (verified against Gitea 1.27.3). Testing key
|
||
presence therefore classifies every real issue as a PR and
|
||
empties the board — test the value's truthiness instead.
|
||
"""
|
||
return not item.get("pull_request")
|
||
|
||
|
||
def _merge_issues_with_board(
|
||
issues: list[dict], board: dict[str, list[dict]]
|
||
) -> dict[str, list[dict]]:
|
||
"""Merge Gitea issues into kanban columns, honouring saved order."""
|
||
placement: dict[int, tuple[str, float]] = {}
|
||
for col_name, cards in board.items():
|
||
for card in cards:
|
||
placement[card["issue_number"]] = (col_name, card["position"])
|
||
|
||
columns: dict[str, list[dict]] = {col: [] for col in KANBAN_COLUMNS}
|
||
for index, issue in enumerate(issues):
|
||
col, pos = placement.get(issue["number"], ("backlog", None))
|
||
issue["_sort"] = (0, pos, index) if pos is not None else (1, 0.0, index)
|
||
columns[col].append(issue)
|
||
for col in columns.values():
|
||
col.sort(key=lambda i: i["_sort"])
|
||
return columns
|
||
|
||
|
||
def _adopt_label_columns(
|
||
full_name: str, open_issues: list[dict], board_state: dict
|
||
) -> None:
|
||
"""Adopt the column named by an issue's kanban:* label, if any.
|
||
|
||
Labels are the Gitea-visible mirror of columns (written on every
|
||
drag). If the label changed outside Kanbante — edited in Gitea,
|
||
or set by another tool — the board follows it and the move is
|
||
journaled (source "gitea"). Rules: exactly one kanban:* label
|
||
(zero or several = no opinion), target must be a labelled column,
|
||
and label synchronisation must be enabled.
|
||
"""
|
||
if not get_settings().sync_labels:
|
||
return
|
||
placement: dict[int, str] = {}
|
||
for col_name, cards in board_state.items():
|
||
for card in cards:
|
||
placement[card["issue_number"]] = col_name
|
||
for issue in open_issues:
|
||
names = [
|
||
str(label.get("name", "")) for label in issue.get("labels") or []
|
||
]
|
||
kanban = [n for n in names if n.startswith("kanban:")]
|
||
if len(kanban) != 1:
|
||
continue
|
||
target = kanban[0].split(":", 1)[1]
|
||
if target not in LABELLED_COLUMNS:
|
||
continue
|
||
current = placement.get(issue["number"], "backlog")
|
||
if current != target:
|
||
move_issue(full_name, issue["number"], target)
|
||
log_event(
|
||
full_name, issue["number"], "moved",
|
||
from_column=current, to_column=target, source="gitea",
|
||
)
|
||
placement[issue["number"]] = target
|
||
|
||
|
||
def _split_full_name(repo_full_name: str) -> tuple[str, str] | None:
|
||
owner, sep, repo = repo_full_name.partition("/")
|
||
if not sep or not owner or not repo or "/" in repo:
|
||
return None
|
||
return owner, repo
|
||
|
||
|
||
def _parse_order(raw: str) -> list[int]:
|
||
numbers = []
|
||
for chunk in (raw or "").split(","):
|
||
chunk = chunk.strip()
|
||
if chunk.isdigit():
|
||
numbers.append(int(chunk))
|
||
return numbers
|
||
|
||
|
||
# ── Routes: Pages ─────────────────────────────────────────
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
async def dashboard(request: Request):
|
||
"""Main dashboard — all projects at a glance."""
|
||
repos = await gitea.search_repos()
|
||
active, inactive, archived = classify_repos(repos)
|
||
for repo_info in active:
|
||
repo_info["forecast_label"] = _repo_forecast_label(
|
||
repo_info["full_name"], repo_info.get("open_issues_count") or 0
|
||
)
|
||
return render(
|
||
request,
|
||
"dashboard.html",
|
||
active_repos=active,
|
||
inactive_repos=inactive,
|
||
archived_repos=archived,
|
||
stats=get_stats(),
|
||
COLUMN_LABELS=COLUMN_LABELS,
|
||
COLUMN_COLORS=COLUMN_COLORS,
|
||
active_window_days=get_settings().active_window_days,
|
||
metrics=get_metrics(window_days=30),
|
||
)
|
||
|
||
|
||
@app.get("/board/{owner}/{repo}", response_class=HTMLResponse)
|
||
async def board(request: Request, owner: str, repo: str, milestone: str = ""):
|
||
"""Kanban board for a specific repo (optionally one milestone)."""
|
||
full_name = f"{owner}/{repo}"
|
||
all_items = await gitea.get_repo_issues(owner, repo, state="all")
|
||
issues = [i for i in all_items if _is_issue(i)]
|
||
open_issues = [i for i in issues if i.get("state") == "open"]
|
||
closed_numbers = {i["number"] for i in issues if i.get("state") != "open"}
|
||
open_numbers = {i["number"] for i in open_issues}
|
||
|
||
# Keep local state truthful: closed in Gitea -> done, deleted -> purged.
|
||
changes = reconcile_board_state(full_name, open_numbers, closed_numbers)
|
||
for number in changes["closed_numbers"]:
|
||
log_event(full_name, number, "closed", to_column="done", source="gitea")
|
||
|
||
# A kanban:* label changed outside Kanbante wins over local state.
|
||
_adopt_label_columns(full_name, open_issues, get_board_state(full_name))
|
||
|
||
# Milestone view: filter the board to one milestone (sprint-style),
|
||
# with its deadline and completion progress.
|
||
milestones: list[dict] = []
|
||
selected_milestone = None
|
||
milestone_progress = None
|
||
milestone_forecast = None
|
||
display_issues = open_issues
|
||
try:
|
||
milestones = await gitea.get_milestones(owner, repo)
|
||
except Exception as exc: # milestones are decoration, never fatal
|
||
logger.debug("milestones failed for %s: %s", full_name, exc)
|
||
milestones = []
|
||
if milestone:
|
||
def _in_milestone(issue: dict) -> bool:
|
||
ms = issue.get("milestone") or {}
|
||
return bool(ms) and (
|
||
str(ms.get("id")) == milestone or ms.get("title") == milestone
|
||
)
|
||
|
||
display_issues = [i for i in open_issues if _in_milestone(i)]
|
||
selected_milestone = next(
|
||
(
|
||
m for m in milestones
|
||
if str(m.get("id")) == milestone or m.get("title") == milestone
|
||
),
|
||
None,
|
||
)
|
||
if selected_milestone is None and display_issues:
|
||
selected_milestone = display_issues[0].get("milestone")
|
||
scoped = [i for i in issues if _in_milestone(i)]
|
||
total = len(scoped)
|
||
closed_count = sum(1 for i in scoped if i.get("state") == "closed")
|
||
milestone_progress = {
|
||
"total": total,
|
||
"closed": closed_count,
|
||
"pct": round(100 * closed_count / total) if total else 0,
|
||
"due_on": (selected_milestone or {}).get("due_on"),
|
||
"title": (selected_milestone or {}).get("title", milestone),
|
||
}
|
||
# Forecast (issue #7): ETA of the remaining milestone work
|
||
# at this repo's real closure pace.
|
||
remaining = total - closed_count
|
||
weekly = forecast_service.weekly_closures(
|
||
get_events(repo_full_name=full_name, limit=2000)
|
||
)
|
||
est = forecast_service.estimate(remaining, weekly)
|
||
if est is None:
|
||
milestone_forecast = "pas assez d'historique pour prévoir"
|
||
else:
|
||
milestone_forecast = forecast_service.eta_label(est)
|
||
due_raw = milestone_progress["due_on"]
|
||
if due_raw and not est.get("done"):
|
||
due_date = _parse_datetime(str(due_raw))
|
||
if due_date and est["central"] > due_date.date():
|
||
milestone_forecast += " — ⚠️ après l'échéance"
|
||
|
||
board_state = get_board_state(full_name)
|
||
columns = _merge_issues_with_board(display_issues, board_state)
|
||
await _annotate_code(owner, repo, columns)
|
||
|
||
labels: dict[str, dict] = {}
|
||
for issue in open_issues:
|
||
for label in issue.get("labels") or []:
|
||
labels[label["name"]] = label
|
||
|
||
repo_info = {}
|
||
try:
|
||
repo_info = await gitea.get_repo(owner, repo)
|
||
except Exception as exc: # non-critical decoration
|
||
logger.debug("repo info failed for %s: %s", full_name, exc)
|
||
|
||
settings = get_settings()
|
||
return render(
|
||
request,
|
||
"board.html",
|
||
owner=owner,
|
||
repo=repo,
|
||
full_name=full_name,
|
||
columns=columns,
|
||
column_labels=COLUMN_LABELS,
|
||
column_colors=COLUMN_COLORS,
|
||
labels=sorted(labels.values(), key=lambda l: l["name"].lower()),
|
||
notes=get_notes(full_name) or "",
|
||
repo_info=repo_info,
|
||
ages=_card_ages(full_name, display_issues, board_state),
|
||
wip_limits=settings.wip_limits or {},
|
||
aging_days=settings.aging_days,
|
||
milestones=milestones,
|
||
selected_milestone=selected_milestone,
|
||
milestone_progress=milestone_progress,
|
||
milestone_forecast=milestone_forecast,
|
||
milestone_param=milestone,
|
||
)
|
||
|
||
|
||
def _board_columns_for(
|
||
owner: str, repo: str, all_items: list[dict], milestone: str = ""
|
||
) -> tuple[dict[str, list[dict]], list[dict], dict]:
|
||
"""Reconcile + merge pipeline shared by the board page and its
|
||
JSON twin (issue #15). Returns (columns, display_issues, state)."""
|
||
full_name = f"{owner}/{repo}"
|
||
issues = [i for i in all_items if _is_issue(i)]
|
||
open_issues = [i for i in issues if i.get("state") == "open"]
|
||
closed_numbers = {i["number"] for i in issues if i.get("state") != "open"}
|
||
open_numbers = {i["number"] for i in open_issues}
|
||
changes = reconcile_board_state(full_name, open_numbers, closed_numbers)
|
||
for number in changes["closed_numbers"]:
|
||
log_event(full_name, number, "closed", to_column="done", source="gitea")
|
||
_adopt_label_columns(full_name, open_issues, get_board_state(full_name))
|
||
display_issues = open_issues
|
||
if milestone:
|
||
def _in_ms(issue: dict) -> bool:
|
||
ms = issue.get("milestone") or {}
|
||
return bool(ms) and (
|
||
str(ms.get("id")) == milestone or ms.get("title") == milestone
|
||
)
|
||
|
||
display_issues = [i for i in open_issues if _in_ms(i)]
|
||
board_state = get_board_state(full_name)
|
||
columns = _merge_issues_with_board(display_issues, board_state)
|
||
return columns, display_issues, board_state
|
||
|
||
|
||
@app.get("/api/board/{owner}/{repo}")
|
||
async def api_board(owner: str, repo: str, milestone: str = ""):
|
||
"""The board as JSON + server-rendered card fragments (issue #15).
|
||
|
||
The board page swaps these fragments in place when the board
|
||
changes elsewhere — same merge/reconcile pipeline as the HTML
|
||
page, so label adoptions and closures are reflected identically.
|
||
"""
|
||
full_name = f"{owner}/{repo}"
|
||
all_items = await gitea.get_repo_issues(owner, repo, state="all")
|
||
columns, display_issues, board_state = _board_columns_for(
|
||
owner, repo, all_items, milestone
|
||
)
|
||
await _annotate_code(owner, repo, columns)
|
||
settings = get_settings()
|
||
ages = _card_ages(full_name, display_issues, board_state)
|
||
card_template = _tpl_env.get_template("_card.html")
|
||
payload: dict[str, dict] = {}
|
||
for col in KANBAN_COLUMNS:
|
||
cards = columns[col]
|
||
if cards:
|
||
html = "".join(
|
||
card_template.render(
|
||
issue=issue, ages=ages,
|
||
aging_days=settings.aging_days, full_name=full_name,
|
||
)
|
||
for issue in cards
|
||
)
|
||
else:
|
||
html = '<p class="column-empty">Glissez une carte ici</p>'
|
||
payload[col] = {"count": len(cards), "html": html}
|
||
return {"repo": full_name, "columns": payload}
|
||
|
||
|
||
def _card_ages(
|
||
full_name: str, open_issues: list[dict], board: dict[str, list[dict]]
|
||
) -> dict[int, dict]:
|
||
"""Days each open card has spent in its current column.
|
||
|
||
Source of truth, in order: the event journal (latest entry),
|
||
board_state.updated_at (pre-journal data), the issue's own
|
||
creation date.
|
||
"""
|
||
entered = column_entry_times(full_name)
|
||
updated = {
|
||
card["issue_number"]: card["updated_at"]
|
||
for cards in board.values()
|
||
for card in cards
|
||
}
|
||
now = datetime.now(timezone.utc)
|
||
ages: dict[int, dict] = {}
|
||
for issue in open_issues:
|
||
number = issue["number"]
|
||
raw = entered.get(number) or updated.get(number) or issue.get("created_at")
|
||
since = _parse_datetime(raw)
|
||
if since is None:
|
||
continue
|
||
if since.tzinfo is None:
|
||
since = since.replace(tzinfo=timezone.utc)
|
||
ages[number] = {"days": max(0, (now - since).days), "since": raw}
|
||
return ages
|
||
|
||
|
||
async def _collect_focus_groups() -> list[dict]:
|
||
"""In-progress / in-review cards across all active repos."""
|
||
repos = await gitea.search_repos()
|
||
active, _inactive, _archived = classify_repos(repos)
|
||
active = active[:100]
|
||
|
||
async def _open_issues(repo: dict):
|
||
owner, _, name = repo["full_name"].partition("/")
|
||
items = await gitea.get_repo_issues(owner, name, state="open")
|
||
return repo["full_name"], [i for i in items if _is_issue(i)]
|
||
|
||
results = await asyncio.gather(
|
||
*(_open_issues(r) for r in active), return_exceptions=True
|
||
)
|
||
groups = []
|
||
for result in results:
|
||
if isinstance(result, Exception):
|
||
logger.warning("focus: repo fetch failed: %s", result)
|
||
continue
|
||
full_name, open_issues = result
|
||
_adopt_label_columns(full_name, open_issues, get_board_state(full_name))
|
||
columns = _merge_issues_with_board(open_issues, get_board_state(full_name))
|
||
cards = [
|
||
{"issue": issue, "column": col}
|
||
for col in FOCUS_COLUMNS
|
||
for issue in columns[col]
|
||
]
|
||
if cards:
|
||
groups.append({"full_name": full_name, "cards": cards})
|
||
return groups
|
||
|
||
|
||
@app.post("/api/issues/{owner}/{repo}/{number}/assign-self")
|
||
async def api_assign_self(
|
||
request: Request,
|
||
owner: str,
|
||
repo: str,
|
||
number: int,
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Assign the token's user to an issue (triage key A, issue #11)."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
user = await gitea.get_user()
|
||
login = user.get("login", "")
|
||
if not login:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Could not resolve the current user"},
|
||
status_code=502,
|
||
)
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
current = [
|
||
a.get("login", "")
|
||
for a in (issue.get("assignees") or [])
|
||
if a.get("login")
|
||
]
|
||
if issue.get("assignee") and issue["assignee"].get("login"):
|
||
if issue["assignee"]["login"] not in current:
|
||
current.append(issue["assignee"]["login"])
|
||
if login not in current:
|
||
current.append(login)
|
||
await gitea.set_issue_assignees(owner, repo, number, current)
|
||
await _publish_board_changed(f"{owner}/{repo}")
|
||
return {"ok": True, "assignees": current}
|
||
|
||
|
||
def _excerpt(body: str | None, limit: int = 280) -> str:
|
||
"""Plain-text excerpt of an issue body for the triage card."""
|
||
text = re.sub(r"[#>*`_~\-]{1,}", " ", body or "")
|
||
text = re.sub(r"\[([^\]]*)\]\([^)]*\)", r"\1", text)
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
return text[:limit] + ("…" if len(text) > limit else "")
|
||
|
||
|
||
@app.get("/triage", response_class=HTMLResponse)
|
||
async def triage_picker(request: Request):
|
||
"""Triage mode (issue #11): pick a project by its backlog size."""
|
||
repos = await gitea.search_repos()
|
||
active, _inactive, _archived = classify_repos(repos)
|
||
projects = []
|
||
for repo_info in active:
|
||
owner, _, name = repo_info["full_name"].partition("/")
|
||
try:
|
||
all_items = await gitea.get_repo_issues(owner, name, state="all")
|
||
except Exception:
|
||
continue
|
||
columns, _display, _state = _board_columns_for(owner, name, all_items)
|
||
backlog = len(columns["backlog"])
|
||
if backlog:
|
||
projects.append(
|
||
{"full_name": repo_info["full_name"], "backlog": backlog}
|
||
)
|
||
projects.sort(key=lambda p: -p["backlog"])
|
||
return render(request, "triage.html", projects=projects)
|
||
|
||
|
||
@app.get("/triage/{owner}/{repo}", response_class=HTMLResponse)
|
||
async def triage_session(request: Request, owner: str, repo: str):
|
||
"""One backlog card at a time, decided from the keyboard."""
|
||
full_name = f"{owner}/{repo}"
|
||
all_items = await gitea.get_repo_issues(owner, repo, state="all")
|
||
columns, display_issues, board_state = _board_columns_for(
|
||
owner, repo, all_items
|
||
)
|
||
ages = _card_ages(full_name, display_issues, board_state)
|
||
queue = [
|
||
{
|
||
"number": issue["number"],
|
||
"title": issue.get("title", ""),
|
||
"excerpt": _excerpt(issue.get("body")),
|
||
"labels": [
|
||
{"name": l.get("name", ""), "color": l.get("color", "")}
|
||
for l in issue.get("labels") or []
|
||
],
|
||
"created_at": (issue.get("created_at") or "")[:10],
|
||
"age_days": (ages.get(issue["number"]) or {}).get("days"),
|
||
"html_url": issue.get("html_url", ""),
|
||
"assignees": [
|
||
a.get("login", "") for a in issue.get("assignees") or []
|
||
],
|
||
}
|
||
for issue in columns["backlog"]
|
||
]
|
||
return render(
|
||
request,
|
||
"triage_session.html",
|
||
full_name=full_name,
|
||
queue=queue,
|
||
queue_json=json.dumps(queue, ensure_ascii=False),
|
||
column_labels=COLUMN_LABELS,
|
||
)
|
||
|
||
|
||
def _today_str(offset_days: int = 0) -> str:
|
||
return (datetime.now() + timedelta(days=offset_days)).strftime("%Y-%m-%d")
|
||
|
||
|
||
async def _enrich_pick(pick: dict) -> dict | None:
|
||
"""A today pick + its live issue data and current column."""
|
||
owner, _, name = pick["repo_full_name"].partition("/")
|
||
try:
|
||
issue = await gitea.get_issue(owner, name, pick["issue_number"])
|
||
except Exception:
|
||
return None
|
||
if issue.get("state") == "closed":
|
||
column = "done"
|
||
else:
|
||
placement = {
|
||
card["issue_number"]: col
|
||
for col, cards in get_board_state(pick["repo_full_name"]).items()
|
||
for card in cards
|
||
}
|
||
column = placement.get(pick["issue_number"], "backlog")
|
||
return {
|
||
"repo": pick["repo_full_name"],
|
||
"number": pick["issue_number"],
|
||
"title": issue.get("title", ""),
|
||
"state": issue.get("state", "open"),
|
||
"column": column,
|
||
"column_label": COLUMN_LABELS[column],
|
||
"html_url": issue.get("html_url", ""),
|
||
}
|
||
|
||
|
||
@app.post("/api/today/pin")
|
||
async def api_today_pin(
|
||
request: Request,
|
||
repo_full_name: str = Form(...),
|
||
issue_number: int = Form(...),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Pin a card in today's focus (max 3, issue #5)."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
if _split_full_name(repo_full_name) is None:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid repo_full_name"}, status_code=400
|
||
)
|
||
today = _today_str()
|
||
picks = get_picks(today)
|
||
if any(
|
||
p["repo_full_name"] == repo_full_name and p["issue_number"] == issue_number
|
||
for p in picks
|
||
):
|
||
return {"ok": True, "already": True, "count": len(picks)}
|
||
if len(picks) >= 3:
|
||
return JSONResponse(
|
||
{
|
||
"ok": False,
|
||
"error": "Focus du jour complet (3 cartes max) — "
|
||
"retire d'abord une carte sur /today.",
|
||
},
|
||
status_code=400,
|
||
)
|
||
add_pick(today, repo_full_name, issue_number)
|
||
return {"ok": True, "count": len(picks) + 1}
|
||
|
||
|
||
@app.post("/api/today/unpin")
|
||
async def api_today_unpin(
|
||
request: Request,
|
||
repo_full_name: str = Form(...),
|
||
issue_number: int = Form(...),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Remove a card from today's focus."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
remove_pick(_today_str(), repo_full_name, issue_number)
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/today", response_class=HTMLResponse)
|
||
async def today_page(request: Request):
|
||
"""Focus of the day (issue #5): up to 3 pinned cards, calmly."""
|
||
today = _today_str()
|
||
picks = []
|
||
for pick in get_picks(today):
|
||
enriched = await _enrich_pick(pick)
|
||
if enriched:
|
||
picks.append(enriched)
|
||
picked_keys = {(p["repo"], p["number"]) for p in picks}
|
||
|
||
carry_over = []
|
||
for pick in get_picks(_today_str(-1)):
|
||
if (pick["repo_full_name"], pick["issue_number"]) in picked_keys:
|
||
continue
|
||
enriched = await _enrich_pick(pick)
|
||
if enriched and enriched["state"] != "closed":
|
||
carry_over.append(enriched)
|
||
|
||
candidates = []
|
||
if len(picks) < 3:
|
||
for group in await _collect_focus_groups():
|
||
for card in group["cards"]:
|
||
key = (group["full_name"], card["issue"]["number"])
|
||
if key in picked_keys:
|
||
continue
|
||
candidates.append(
|
||
{
|
||
"repo": group["full_name"],
|
||
"number": card["issue"]["number"],
|
||
"title": card["issue"].get("title", ""),
|
||
"column_label": COLUMN_LABELS[card["column"]],
|
||
}
|
||
)
|
||
candidates = candidates[:8]
|
||
|
||
recap = []
|
||
picked_numbers = {(p["repo"], p["number"]) for p in picks}
|
||
for event in get_events(limit=300):
|
||
if not str(event.get("created_at") or "").startswith(today):
|
||
continue
|
||
if (event["repo_full_name"], event["issue_number"]) in picked_numbers:
|
||
recap.append(event)
|
||
return render(
|
||
request,
|
||
"today.html",
|
||
date_label=datetime.now().strftime("%A %d %B %Y"),
|
||
picks=picks,
|
||
carry_over=carry_over,
|
||
candidates=candidates,
|
||
recap=recap,
|
||
column_labels=COLUMN_LABELS,
|
||
)
|
||
|
||
|
||
async def _sprint_markdown(owner: str, repo: str, milestone: str):
|
||
"""Assemble the sprint story; (title, markdown) or None."""
|
||
full_name = f"{owner}/{repo}"
|
||
try:
|
||
milestones = await gitea.get_milestones(owner, repo)
|
||
except Exception:
|
||
milestones = []
|
||
selected = next(
|
||
(
|
||
m for m in milestones
|
||
if str(m.get("id")) == milestone or m.get("title") == milestone
|
||
),
|
||
None,
|
||
)
|
||
if selected is None:
|
||
return None
|
||
|
||
def _in_ms(issue: dict) -> bool:
|
||
ms = issue.get("milestone") or {}
|
||
return bool(ms) and (
|
||
str(ms.get("id")) == str(selected.get("id"))
|
||
or ms.get("title") == selected.get("title")
|
||
)
|
||
|
||
all_items = await gitea.get_repo_issues(owner, repo, state="all")
|
||
scoped = [i for i in all_items if _is_issue(i) and _in_ms(i)]
|
||
numbers = {i["number"] for i in scoped}
|
||
|
||
events = [
|
||
e for e in get_events(repo_full_name=full_name, limit=2000)
|
||
if e["issue_number"] in numbers
|
||
]
|
||
first_in_progress: dict[int, object] = {}
|
||
closed_at: dict[int, object] = {}
|
||
counts = {"moved": 0, "commented": 0, "reopened": 0, "created": 0}
|
||
event_dates = []
|
||
for event in events:
|
||
at = _parse_datetime(str(event.get("created_at") or ""))
|
||
if at is not None:
|
||
event_dates.append(at)
|
||
if event["event_type"] in counts:
|
||
counts[event["event_type"]] += 1
|
||
if (
|
||
event["event_type"] == "moved"
|
||
and event.get("to_column") == "in_progress"
|
||
):
|
||
first_in_progress.setdefault(event["issue_number"], at)
|
||
if event["event_type"] == "closed":
|
||
closed_at[event["issue_number"]] = at
|
||
|
||
columns, _display, _state = _board_columns_for(owner, repo, all_items)
|
||
placement = {
|
||
card["number"]: col for col, cards in columns.items() for card in cards
|
||
}
|
||
|
||
delivered, carried, cycles = [], [], []
|
||
for issue in scoped:
|
||
if issue.get("state") == "closed":
|
||
cycle = None
|
||
start = first_in_progress.get(issue["number"])
|
||
end = closed_at.get(issue["number"])
|
||
if start is None:
|
||
start = _parse_datetime(str(issue.get("created_at") or ""))
|
||
if end is None:
|
||
end = _parse_datetime(str(issue.get("closed_at") or ""))
|
||
if start is not None and end is not None:
|
||
cycle = round((end - start).total_seconds() / 86400, 1)
|
||
if cycle < 0:
|
||
cycle = None
|
||
else:
|
||
cycles.append(cycle)
|
||
delivered.append(
|
||
{
|
||
"number": issue["number"],
|
||
"title": issue.get("title", ""),
|
||
"html_url": issue.get("html_url", ""),
|
||
"cycle_days": cycle,
|
||
}
|
||
)
|
||
else:
|
||
col = placement.get(issue["number"], "backlog")
|
||
carried.append(
|
||
{
|
||
"number": issue["number"],
|
||
"title": issue.get("title", ""),
|
||
"column_label": COLUMN_LABELS[col],
|
||
}
|
||
)
|
||
delivered.sort(key=lambda d: d["number"])
|
||
carried.sort(key=lambda d: d["number"])
|
||
|
||
if event_dates:
|
||
period = (
|
||
f"du {min(event_dates).strftime('%Y-%m-%d')} "
|
||
f"au {max(event_dates).strftime('%Y-%m-%d')}"
|
||
)
|
||
else:
|
||
period = "aucun événement journalisé pour ce milestone"
|
||
avg_cycle = round(sum(cycles) / len(cycles), 1) if cycles else None
|
||
markdown = sprint_service.format_sprint_markdown(
|
||
full_name,
|
||
selected.get("title", milestone),
|
||
selected.get("due_on"),
|
||
period,
|
||
delivered,
|
||
carried,
|
||
counts,
|
||
avg_cycle,
|
||
)
|
||
return selected.get("title", milestone), markdown
|
||
|
||
|
||
@app.get("/api/sprint/{owner}/{repo}")
|
||
async def api_sprint(owner: str, repo: str, milestone: str = ""):
|
||
"""The sprint story as JSON markdown (issue #4)."""
|
||
result = await _sprint_markdown(owner, repo, milestone)
|
||
if result is None:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Milestone introuvable"}, status_code=404
|
||
)
|
||
title, markdown = result
|
||
return {"ok": True, "milestone": title, "markdown": markdown}
|
||
|
||
|
||
@app.get("/sprint/{owner}/{repo}", response_class=HTMLResponse)
|
||
async def sprint_page(request: Request, owner: str, repo: str, milestone: str = ""):
|
||
"""Sprint story page: rendered narrative + copy/download."""
|
||
full_name = f"{owner}/{repo}"
|
||
result = await _sprint_markdown(owner, repo, milestone) if milestone else None
|
||
milestones = []
|
||
if result is None:
|
||
try:
|
||
milestones = await gitea.get_milestones(owner, repo)
|
||
except Exception:
|
||
milestones = []
|
||
title, markdown = result if result else ("", "")
|
||
return render(
|
||
request,
|
||
"sprint.html",
|
||
full_name=full_name,
|
||
milestone_title=title,
|
||
markdown=markdown,
|
||
milestones=milestones,
|
||
milestone_param=milestone,
|
||
)
|
||
|
||
|
||
async def _apply_rule_actions(
|
||
summary: dict, rule: dict, full_name: str, owner: str, repo: str,
|
||
issue: dict, column: str, columns: dict, ages: dict,
|
||
) -> None:
|
||
"""Execute one rule's actions on one card (issue #8)."""
|
||
then = rule["then"]
|
||
number = issue["number"]
|
||
acted: list[str] = []
|
||
if then.get("add_label"):
|
||
label = then["add_label"]
|
||
present = [
|
||
(lab or {}).get("name") for lab in issue.get("labels") or []
|
||
]
|
||
if label not in present:
|
||
lab = await gitea.ensure_label(owner, repo, label, "fb8500")
|
||
await gitea.add_issue_label(owner, repo, number, lab["id"])
|
||
acted.append(f"étiquette « {label} » ajoutée")
|
||
if then.get("move_top"):
|
||
order = [c["number"] for c in columns.get(column, [])]
|
||
if order and order[0] != number:
|
||
set_column_order(
|
||
full_name, column,
|
||
[number] + [n for n in order if n != number],
|
||
)
|
||
acted.append("remontée en tête de colonne")
|
||
if then.get("notify"):
|
||
guard = (
|
||
f"rule_notified:{rule['name']}:{full_name}:{number}:{_today_str()}"
|
||
)
|
||
if not get_kv(guard):
|
||
age = (ages.get(number) or {}).get("days")
|
||
age_txt = f" depuis {age} j" if age is not None else ""
|
||
text = (
|
||
f"⚙️ Règle « {rule['name']} »\n"
|
||
f"{full_name} #{number} « {issue.get('title', '')} » — "
|
||
f"{COLUMN_LABELS[column]}{age_txt}"
|
||
)
|
||
if await send_ntfy(get_settings(), text):
|
||
set_kv(guard, "1")
|
||
acted.append("notification ntfy envoyée")
|
||
if acted:
|
||
log_event(
|
||
full_name, number, "rule",
|
||
from_column=column, to_column=column, source="rule",
|
||
)
|
||
summary["actions"].append(
|
||
{
|
||
"rule": rule["name"],
|
||
"repo": full_name,
|
||
"number": number,
|
||
"column": column,
|
||
"did": acted,
|
||
}
|
||
)
|
||
|
||
|
||
async def _run_rules_once() -> dict:
|
||
"""Evaluate every configured rule once (scheduler + manual run)."""
|
||
settings = get_settings()
|
||
rules, errors = rules_service.parse_rules(settings.rules_raw)
|
||
summary: dict = {
|
||
"at": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"rules": len(rules),
|
||
"errors": errors,
|
||
"actions": [],
|
||
}
|
||
if rules:
|
||
repos = await gitea.search_repos()
|
||
active, _inactive, _archived = classify_repos(repos)
|
||
for repo_info in active[:50]:
|
||
full_name = repo_info["full_name"]
|
||
owner, _, name = full_name.partition("/")
|
||
try:
|
||
all_items = await gitea.get_repo_issues(
|
||
owner, name, state="all"
|
||
)
|
||
except Exception:
|
||
continue
|
||
columns, display_issues, board_state = _board_columns_for(
|
||
owner, name, all_items
|
||
)
|
||
ages = _card_ages(full_name, display_issues, board_state)
|
||
for rule in rules:
|
||
matches = rules_service.matching_cards(rule, columns, ages)
|
||
for issue, column in matches:
|
||
if len(summary["actions"]) >= 50:
|
||
break
|
||
try:
|
||
await _apply_rule_actions(
|
||
summary, rule, full_name, owner, name,
|
||
issue, column, columns, ages,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"Rule %s failed on %s#%s: %s",
|
||
rule["name"], full_name, issue["number"], exc,
|
||
)
|
||
set_kv("rules_last_run", json.dumps(summary, ensure_ascii=False))
|
||
return summary
|
||
|
||
|
||
async def _rules_scheduler() -> None:
|
||
"""Evaluate the automation rules every 15 minutes (issue #8)."""
|
||
while True:
|
||
try:
|
||
if get_settings().rules_raw.strip():
|
||
await _run_rules_once()
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
logger.exception("Rules evaluation failed")
|
||
await asyncio.sleep(900)
|
||
|
||
|
||
@app.get("/rules", response_class=HTMLResponse)
|
||
async def rules_page(request: Request):
|
||
"""Automation rules: configuration status + last evaluation."""
|
||
settings = get_settings()
|
||
rules, errors = rules_service.parse_rules(settings.rules_raw)
|
||
last_run = None
|
||
raw = get_kv("rules_last_run")
|
||
if raw:
|
||
try:
|
||
last_run = json.loads(raw)
|
||
except ValueError:
|
||
last_run = None
|
||
return render(
|
||
request,
|
||
"rules.html",
|
||
rules=rules,
|
||
errors=errors,
|
||
configured=bool(settings.rules_raw.strip()),
|
||
last_run=last_run,
|
||
)
|
||
|
||
|
||
@app.post("/api/rules/run")
|
||
async def api_rules_run(request: Request, csrf_token: str = Form("")):
|
||
"""Evaluate the rules right now (manual trigger)."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
summary = await _run_rules_once()
|
||
return {"ok": True, "summary": summary}
|
||
|
||
|
||
async def _project_health(owner: str, repo: str) -> dict:
|
||
"""Bottlenecks + zombies for one project (issue #9)."""
|
||
full_name = f"{owner}/{repo}"
|
||
settings = get_settings()
|
||
all_items = await gitea.get_repo_issues(owner, repo, state="all")
|
||
columns, display_issues, _bs = _board_columns_for(
|
||
owner, repo, all_items
|
||
)
|
||
events = get_events(repo_full_name=full_name, limit=2000)
|
||
dwell = flow_service.summarize_dwell(flow_service.column_dwell(events))
|
||
slowest = flow_service.slowest_column(dwell)
|
||
activity = flow_service.last_activity(events)
|
||
placement = {
|
||
issue["number"]: col
|
||
for col, cards in columns.items()
|
||
for issue in cards
|
||
}
|
||
now = datetime.now()
|
||
zombies = []
|
||
for issue in display_issues:
|
||
number = issue["number"]
|
||
candidates = [
|
||
ts for ts in (
|
||
activity.get((full_name, number)),
|
||
flow_service.parse_ts(issue.get("updated_at")),
|
||
flow_service.parse_ts(issue.get("created_at")),
|
||
)
|
||
if ts is not None
|
||
]
|
||
if not candidates:
|
||
continue
|
||
days = (now - max(candidates)).days
|
||
if days >= settings.zombie_days:
|
||
column = placement.get(number, "backlog")
|
||
zombies.append(
|
||
{
|
||
"number": number,
|
||
"title": issue.get("title", ""),
|
||
"column": column,
|
||
"column_label": COLUMN_LABELS[column],
|
||
"days_inactive": days,
|
||
"html_url": issue.get("html_url", ""),
|
||
}
|
||
)
|
||
zombies.sort(key=lambda z: -z["days_inactive"])
|
||
return {
|
||
"full_name": full_name,
|
||
"dwell": dwell,
|
||
"slowest": slowest,
|
||
"zombies": zombies,
|
||
"zombie_days": settings.zombie_days,
|
||
"open_count": len(display_issues),
|
||
"column_labels": COLUMN_LABELS,
|
||
}
|
||
|
||
|
||
@app.get("/health", response_class=HTMLResponse)
|
||
async def health_index(request: Request):
|
||
"""Flow health, global summary (issue #9)."""
|
||
repos = await gitea.search_repos()
|
||
active, _inactive, _archived = classify_repos(repos)
|
||
projects = []
|
||
for repo_info in active:
|
||
owner, _, name = repo_info["full_name"].partition("/")
|
||
try:
|
||
projects.append(await _project_health(owner, name))
|
||
except Exception as exc:
|
||
logger.warning("Health failed for %s: %s", repo_info["full_name"], exc)
|
||
return render(request, "health.html", projects=projects)
|
||
|
||
|
||
@app.get("/health/{owner}/{repo}", response_class=HTMLResponse)
|
||
async def health_project(request: Request, owner: str, repo: str):
|
||
"""Flow health of one project: dwell per column + zombies."""
|
||
repos = await gitea.search_repos()
|
||
known = {r["full_name"] for r in repos}
|
||
if f"{owner}/{repo}" not in known:
|
||
return render(
|
||
request, "health_project.html",
|
||
health=None, full_name=f"{owner}/{repo}",
|
||
)
|
||
health = await _project_health(owner, repo)
|
||
return render(
|
||
request, "health_project.html",
|
||
health=health, full_name=f"{owner}/{repo}",
|
||
)
|
||
|
||
|
||
@app.get("/api/standup")
|
||
async def api_standup():
|
||
"""The automatic standup as JSON text (issue #6)."""
|
||
data = await _briefing_data()
|
||
text = format_standup_text(await _standup_data(data))
|
||
return {"ok": True, "text": text}
|
||
|
||
|
||
@app.get("/standup", response_class=HTMLResponse)
|
||
async def standup_page(request: Request):
|
||
"""Standup page: yesterday / today / blockers, copy in one click."""
|
||
data = await _briefing_data()
|
||
text = format_standup_text(await _standup_data(data))
|
||
return render(
|
||
request,
|
||
"standup.html",
|
||
text=text,
|
||
include_in_briefing=get_settings().briefing_include_standup,
|
||
)
|
||
|
||
|
||
@app.get("/forecast", response_class=HTMLResponse)
|
||
async def forecast_page(request: Request):
|
||
"""Forecasts (issue #7): real-pace ETAs per project + milestone."""
|
||
repos = await gitea.search_repos()
|
||
active, _inactive, _archived = classify_repos(repos)
|
||
projects = []
|
||
for repo_info in active:
|
||
full = repo_info["full_name"]
|
||
owner, _, name = full.partition("/")
|
||
weekly = forecast_service.weekly_closures(
|
||
get_events(repo_full_name=full, limit=2000)
|
||
)
|
||
max_count = max((b["count"] for b in weekly), default=0)
|
||
bars = [
|
||
{
|
||
"label": b["start"].strftime("S%V"),
|
||
"count": b["count"],
|
||
"h": round(100 * b["count"] / max_count) if max_count else 0,
|
||
}
|
||
for b in weekly
|
||
]
|
||
remaining = repo_info.get("open_issues_count") or 0
|
||
est = forecast_service.estimate(remaining, weekly)
|
||
milestones = []
|
||
try:
|
||
ms_list = await gitea.get_milestones(owner, name)
|
||
except Exception:
|
||
ms_list = []
|
||
for ms in ms_list:
|
||
if ms.get("state") not in (None, "open"):
|
||
continue
|
||
rem = ms.get("open_issues") or 0
|
||
if rem == 0 and not (ms.get("closed_issues") or 0):
|
||
continue
|
||
m_est = forecast_service.estimate(rem, weekly)
|
||
verdict = None
|
||
due_raw = ms.get("due_on")
|
||
if m_est and not m_est.get("done") and due_raw:
|
||
due = _parse_datetime(str(due_raw))
|
||
if due is not None:
|
||
verdict = (
|
||
"⚠️ en retard à ce rythme"
|
||
if m_est["central"] > due.date()
|
||
else "✅ dans les temps à ce rythme"
|
||
)
|
||
milestones.append(
|
||
{
|
||
"title": ms.get("title", ""),
|
||
"due_on": (due_raw or "")[:10],
|
||
"remaining": rem,
|
||
"eta": forecast_service.eta_label(m_est),
|
||
"verdict": verdict,
|
||
}
|
||
)
|
||
projects.append(
|
||
{
|
||
"full_name": full,
|
||
"bars": bars,
|
||
"stats": forecast_service.throughput_stats(weekly),
|
||
"remaining": remaining,
|
||
"eta_label": forecast_service.eta_label(est),
|
||
"milestones": milestones,
|
||
}
|
||
)
|
||
return render(request, "forecast.html", projects=projects)
|
||
|
||
|
||
@app.get("/focus", response_class=HTMLResponse)
|
||
async def focus(request: Request):
|
||
"""Cross-project view: everything in progress or in review."""
|
||
return render(
|
||
request,
|
||
"focus.html",
|
||
groups=await _collect_focus_groups(),
|
||
column_labels=COLUMN_LABELS,
|
||
column_colors=COLUMN_COLORS,
|
||
)
|
||
|
||
|
||
def _group_issue_results(issues: list[dict]) -> list[dict]:
|
||
"""Group search results by repo, annotated with the board column."""
|
||
by_repo: dict[str, list[dict]] = {}
|
||
for issue in issues:
|
||
full_name = (issue.get("repository") or {}).get("full_name", "?")
|
||
by_repo.setdefault(full_name, []).append(issue)
|
||
groups = []
|
||
for full_name in sorted(by_repo):
|
||
state = get_board_state(full_name) if "/" in full_name else {}
|
||
column_of = {
|
||
card["issue_number"]: column
|
||
for column, cards in state.items()
|
||
for card in cards
|
||
}
|
||
rows = []
|
||
for issue in by_repo[full_name]:
|
||
if issue.get("state") == "open":
|
||
column = column_of.get(issue["number"], "backlog")
|
||
else:
|
||
column = "done"
|
||
rows.append({"issue": issue, "column": column})
|
||
groups.append({"full_name": full_name, "rows": rows})
|
||
return groups
|
||
|
||
|
||
WALL_COLUMNS = ["in_progress", "qa", "review", "todo"]
|
||
|
||
|
||
async def _collect_wall_data(only_repo: str | None) -> list[dict]:
|
||
"""Slides for the wallboard: one per active project.
|
||
|
||
Each slide carries the column counts, the cards in flight
|
||
(in progress / QA / review / todo, in flow order) with their age
|
||
in column, and how many cards were closed today (journal).
|
||
"""
|
||
if only_repo:
|
||
if _split_full_name(only_repo) is None:
|
||
return []
|
||
known = await gitea.search_repos()
|
||
match = [r for r in known if r.get("full_name") == only_repo]
|
||
if not match:
|
||
return []
|
||
repos = match
|
||
else:
|
||
all_repos = await gitea.search_repos()
|
||
active, _inactive, _archived = classify_repos(all_repos)
|
||
repos = active[:12]
|
||
|
||
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||
closed_today: dict[str, int] = {}
|
||
for event in get_events(limit=500):
|
||
if event["event_type"] == "closed" and str(
|
||
event["created_at"]
|
||
).startswith(today):
|
||
closed_today[event["repo_full_name"]] = (
|
||
closed_today.get(event["repo_full_name"], 0) + 1
|
||
)
|
||
|
||
slides = []
|
||
for repo in repos:
|
||
full_name = repo["full_name"]
|
||
owner, _, name = full_name.partition("/")
|
||
try:
|
||
items = await gitea.get_repo_issues(owner, name, state="open")
|
||
except Exception as exc:
|
||
logger.debug("wall: issues failed for %s: %s", full_name, exc)
|
||
continue
|
||
open_issues = [i for i in items if _is_issue(i)]
|
||
_adopt_label_columns(full_name, open_issues, get_board_state(full_name))
|
||
board_state = get_board_state(full_name)
|
||
columns = _merge_issues_with_board(open_issues, board_state)
|
||
ages = _card_ages(full_name, open_issues, board_state)
|
||
aging_days = get_settings().aging_days
|
||
cards = []
|
||
for col in WALL_COLUMNS:
|
||
for issue in columns[col]:
|
||
age = ages.get(issue["number"]) or {}
|
||
days = age.get("days")
|
||
cards.append(
|
||
{
|
||
"number": issue["number"],
|
||
"title": issue.get("title", ""),
|
||
"column": col,
|
||
"column_label": COLUMN_LABELS[col],
|
||
"age_days": days,
|
||
"aging": days is not None and days >= aging_days,
|
||
}
|
||
)
|
||
slides.append(
|
||
{
|
||
"full_name": full_name,
|
||
"counts": {col: len(columns[col]) for col in KANBAN_COLUMNS},
|
||
"open_total": len(open_issues),
|
||
"cards": cards[:9],
|
||
"cards_total": len(cards),
|
||
"closed_today": closed_today.get(full_name, 0),
|
||
}
|
||
)
|
||
return slides
|
||
|
||
|
||
@app.get("/wall", response_class=HTMLResponse)
|
||
async def wall(request: Request, repo: str = ""):
|
||
"""Wallboard: full-screen, auto-rotating, live project display."""
|
||
slides = await _collect_wall_data(repo or None)
|
||
return render(
|
||
request,
|
||
"wall.html",
|
||
slides=slides,
|
||
single=bool(repo),
|
||
column_labels=COLUMN_LABELS,
|
||
column_colors=COLUMN_COLORS,
|
||
columns=KANBAN_COLUMNS,
|
||
)
|
||
|
||
|
||
@app.get("/search", response_class=HTMLResponse)
|
||
async def search_page(request: Request, q: str = ""):
|
||
"""Global issue search across every repo (Gitea search API)."""
|
||
q = q.strip()
|
||
groups: list[dict] = []
|
||
total = 0
|
||
if q:
|
||
results = await gitea.search_issues(q=q, state="all")
|
||
total = len(results)
|
||
groups = _group_issue_results(results)
|
||
return render(
|
||
request,
|
||
"search.html",
|
||
q=q,
|
||
groups=groups,
|
||
total=total,
|
||
column_labels=COLUMN_LABELS,
|
||
column_colors=COLUMN_COLORS,
|
||
)
|
||
|
||
|
||
@app.get("/my", response_class=HTMLResponse)
|
||
async def my_issues(request: Request):
|
||
"""Open issues assigned to the Gitea token's owner, all repos."""
|
||
user = None
|
||
groups: list[dict] = []
|
||
total = 0
|
||
try:
|
||
user = await gitea.get_user()
|
||
except Exception as exc:
|
||
logger.debug("get_user failed for /my: %s", exc)
|
||
if user and user.get("login"):
|
||
results = await gitea.search_issues(
|
||
state="open", assigned_by=user["login"]
|
||
)
|
||
total = len(results)
|
||
groups = _group_issue_results(results)
|
||
return render(
|
||
request,
|
||
"my.html",
|
||
user=user,
|
||
groups=groups,
|
||
total=total,
|
||
column_labels=COLUMN_LABELS,
|
||
column_colors=COLUMN_COLORS,
|
||
)
|
||
|
||
|
||
# ── Briefing ───────────────────────────────────────────────
|
||
|
||
async def _briefing_data() -> dict:
|
||
"""Everything the daily briefing needs, gathered in one dict."""
|
||
settings = get_settings()
|
||
focus_groups = await _collect_focus_groups()
|
||
|
||
repos = await gitea.search_repos()
|
||
active, _i, _a = classify_repos(repos)
|
||
soon = datetime.now(timezone.utc) + timedelta(days=7)
|
||
aging: list[dict] = []
|
||
milestones_due: list[dict] = []
|
||
|
||
async def _per_repo(repo: dict):
|
||
full_name = repo["full_name"]
|
||
owner, _, name = full_name.partition("/")
|
||
items = await gitea.get_repo_issues(owner, name, state="open")
|
||
open_issues = [i for i in items if _is_issue(i)]
|
||
ages = _card_ages(full_name, open_issues, get_board_state(full_name))
|
||
board = get_board_state(full_name)
|
||
column_of = {
|
||
card["issue_number"]: column
|
||
for column, cards in board.items()
|
||
for card in cards
|
||
}
|
||
repo_aging = [
|
||
{
|
||
"repo": full_name,
|
||
"number": issue["number"],
|
||
"title": issue["title"],
|
||
"days": ages[issue["number"]]["days"],
|
||
"column_label": COLUMN_LABELS[
|
||
column_of.get(issue["number"], "backlog")
|
||
],
|
||
}
|
||
for issue in open_issues
|
||
if issue["number"] in ages
|
||
and ages[issue["number"]]["days"] >= settings.aging_days
|
||
]
|
||
repo_milestones = []
|
||
try:
|
||
for ms in await gitea.get_milestones(owner, name):
|
||
due = _parse_datetime(ms.get("due_on"))
|
||
if due and due <= soon and (ms.get("open_issues") or 0) > 0:
|
||
repo_milestones.append(
|
||
{
|
||
"repo": full_name,
|
||
"title": ms.get("title", ""),
|
||
"due_on": str(ms["due_on"])[:10],
|
||
"open": ms.get("open_issues") or 0,
|
||
}
|
||
)
|
||
except Exception as exc:
|
||
logger.debug("briefing milestones failed for %s: %s", full_name, exc)
|
||
return repo_aging, repo_milestones
|
||
|
||
results = await asyncio.gather(
|
||
*(_per_repo(r) for r in active[:100]), return_exceptions=True
|
||
)
|
||
for result in results:
|
||
if isinstance(result, Exception):
|
||
logger.warning("briefing: repo fetch failed: %s", result)
|
||
continue
|
||
repo_aging, repo_milestones = result
|
||
aging.extend(repo_aging)
|
||
milestones_due.extend(repo_milestones)
|
||
aging.sort(key=lambda c: -c["days"])
|
||
|
||
metrics_24h = get_metrics(window_days=1)
|
||
metrics_30d = get_metrics(window_days=30)
|
||
|
||
# Focus of the day (issue #5): today's pins + yesterday's
|
||
# unfinished picks, proposed again in the morning briefing.
|
||
today_picks: list[dict] = []
|
||
for pick in get_picks(_today_str()):
|
||
enriched = await _enrich_pick(pick)
|
||
if enriched:
|
||
today_picks.append(enriched)
|
||
picked_keys = {(p["repo"], p["number"]) for p in today_picks}
|
||
carry_over: list[dict] = []
|
||
for pick in get_picks(_today_str(-1)):
|
||
if (pick["repo_full_name"], pick["issue_number"]) in picked_keys:
|
||
continue
|
||
enriched = await _enrich_pick(pick)
|
||
if enriched and enriched["state"] != "closed":
|
||
carry_over.append(enriched)
|
||
|
||
data = {
|
||
"date": datetime.now().strftime("%d/%m/%Y"),
|
||
"today_picks": today_picks,
|
||
"carry_over": carry_over,
|
||
"focus_total": sum(len(g["cards"]) for g in focus_groups),
|
||
"focus_groups": [
|
||
{
|
||
"full_name": g["full_name"],
|
||
"count": len(g["cards"]),
|
||
"titles": [c["issue"]["title"] for c in g["cards"]],
|
||
}
|
||
for g in focus_groups
|
||
],
|
||
"aging": aging[:10],
|
||
"milestones_due": milestones_due[:5],
|
||
"activity_24h": {
|
||
**metrics_24h,
|
||
"avg_cycle_days": metrics_30d["avg_cycle_days"],
|
||
},
|
||
}
|
||
if get_settings().briefing_include_standup:
|
||
data["standup"] = format_standup_text(await _standup_data(data))
|
||
return data
|
||
|
||
|
||
async def _standup_data(briefing_data: dict) -> dict:
|
||
"""Yesterday / today / blockers for the automatic standup (#6)."""
|
||
yesterday = datetime.now() - timedelta(days=1)
|
||
y_str = yesterday.strftime("%Y-%m-%d")
|
||
events = [
|
||
e for e in get_events(limit=1000)
|
||
if str(e.get("created_at") or "").startswith(y_str)
|
||
]
|
||
closed_keys: list[tuple[str, int]] = []
|
||
seen: set[tuple[str, int]] = set()
|
||
moved = 0
|
||
for event in events:
|
||
if event["event_type"] == "closed":
|
||
key = (event["repo_full_name"], event["issue_number"])
|
||
if key not in seen:
|
||
seen.add(key)
|
||
closed_keys.append(key)
|
||
elif event["event_type"] == "moved":
|
||
moved += 1
|
||
titles: dict[tuple[str, int], str] = {}
|
||
for repo_name in {repo for repo, _ in closed_keys}:
|
||
owner, _, name = repo_name.partition("/")
|
||
try:
|
||
items = await gitea.get_repo_issues(owner, name, state="all")
|
||
except Exception:
|
||
continue
|
||
for item in items:
|
||
if _is_issue(item):
|
||
titles[(repo_name, item["number"])] = item.get("title", "")
|
||
blockers = [
|
||
card for card in briefing_data.get("aging", [])
|
||
if "Révision" in card.get("column_label", "")
|
||
or "QA" in card.get("column_label", "")
|
||
]
|
||
focus_titles = [
|
||
f"{g['full_name']} — {title}"
|
||
for g in briefing_data.get("focus_groups", [])
|
||
for title in g.get("titles", [])[:2]
|
||
]
|
||
return {
|
||
"today_label": datetime.now().strftime("%A %d %B %Y"),
|
||
"yesterday_label": yesterday.strftime("%A %d %B"),
|
||
"closed_yesterday": [
|
||
{
|
||
"repo": repo,
|
||
"number": number,
|
||
"title": titles.get((repo, number), ""),
|
||
}
|
||
for repo, number in closed_keys
|
||
],
|
||
"moved_yesterday": moved,
|
||
"picks": briefing_data.get("today_picks", []),
|
||
"focus_titles": focus_titles,
|
||
"blockers": blockers,
|
||
}
|
||
|
||
|
||
@app.get("/briefing", response_class=HTMLResponse)
|
||
async def briefing_page(request: Request):
|
||
"""The daily briefing, readable on the web before it is pushed."""
|
||
settings = get_settings()
|
||
data = await _briefing_data()
|
||
return render(
|
||
request,
|
||
"briefing.html",
|
||
data=data,
|
||
text=format_briefing_text(data),
|
||
ntfy_configured=ntfy_configured(settings),
|
||
briefing_time=settings.briefing_time,
|
||
last_sent=get_kv("briefing_last_sent"),
|
||
)
|
||
|
||
|
||
@app.get("/api/briefing")
|
||
async def api_briefing():
|
||
data = await _briefing_data()
|
||
return {"ok": True, "briefing": data, "text": format_briefing_text(data)}
|
||
|
||
|
||
@app.post("/api/briefing/send")
|
||
async def api_briefing_send(request: Request, csrf_token: str = Form("")):
|
||
"""Send the briefing to the configured ntfy topic, right now."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
settings = get_settings()
|
||
if not ntfy_configured(settings):
|
||
return JSONResponse(
|
||
{
|
||
"ok": False,
|
||
"error": "ntfy is not configured "
|
||
"(KANBANTE_NTFY_URL / KANBANTE_NTFY_TOPIC)",
|
||
},
|
||
status_code=400,
|
||
)
|
||
data = await _briefing_data()
|
||
sent = await send_ntfy(settings, format_briefing_text(data))
|
||
if not sent:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "ntfy server rejected or unreachable"},
|
||
status_code=502,
|
||
)
|
||
set_kv("briefing_last_sent", datetime.now().strftime("%Y-%m-%d"))
|
||
return {"ok": True, "sent": True}
|
||
|
||
|
||
# ── Webhooks + live stream ─────────────────────────────────
|
||
|
||
def _apply_external_transition(
|
||
full_name: str, number: int, target_column: str, event_type: str
|
||
) -> str:
|
||
"""Apply a close/reopen that happened in Gitea itself.
|
||
|
||
Idempotent by construction: when Kanbante's own API made the
|
||
change, the local row is already in the target state and this is
|
||
a no-op — so our own webhook echo never double-logs.
|
||
"""
|
||
board = get_board_state(full_name)
|
||
current = None
|
||
for column, cards in board.items():
|
||
if any(c["issue_number"] == number for c in cards):
|
||
current = column
|
||
break
|
||
if current is None or current == target_column:
|
||
return "noop"
|
||
if event_type == "reopened" and current != "done":
|
||
return "noop" # locally open already: nothing truthful to apply
|
||
move_issue(full_name, number, target_column)
|
||
if event_type == "closed":
|
||
log_event(full_name, number, "closed", to_column="done", source="webhook")
|
||
else:
|
||
log_event(
|
||
full_name, number, "reopened",
|
||
from_column="done", to_column=target_column, source="webhook",
|
||
)
|
||
return event_type
|
||
|
||
|
||
async def _apply_pr_merged(full_name: str, pr: dict) -> list[str]:
|
||
"""A merged PR sends its linked cards to done (issue #2).
|
||
|
||
Links come from #n references in the PR title/body and from the
|
||
head branch name (issue-n-*). The issue is closed in Gitea when
|
||
it is not already (a plain reference does not auto-close), the
|
||
card is moved idempotently, and the journal records the webhook
|
||
as the source.
|
||
"""
|
||
refs = code_links_service.extract_issue_refs(
|
||
pr.get("title")
|
||
) | code_links_service.extract_issue_refs(pr.get("body"))
|
||
branch_number = code_links_service.branch_issue_number(
|
||
(pr.get("head") or {}).get("ref")
|
||
)
|
||
if branch_number is not None:
|
||
refs.add(branch_number)
|
||
owner, _, repo = full_name.partition("/")
|
||
results: list[str] = []
|
||
for number in sorted(refs):
|
||
try:
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
except Exception:
|
||
results.append(f"#{number}:unknown")
|
||
continue
|
||
if not _is_issue(issue):
|
||
results.append(f"#{number}:not-issue")
|
||
continue
|
||
if issue.get("state") != "closed":
|
||
try:
|
||
await gitea.set_issue_state(owner, repo, number, "closed")
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"PR merge: closing %s#%s failed: %s", full_name, number, exc
|
||
)
|
||
outcome = _apply_external_transition(full_name, number, "done", "closed")
|
||
if outcome == "noop":
|
||
board = get_board_state(full_name)
|
||
placed = any(
|
||
c["issue_number"] == number
|
||
for cards in board.values()
|
||
for c in cards
|
||
)
|
||
if not placed:
|
||
move_issue(full_name, number, "done")
|
||
log_event(
|
||
full_name, number, "closed",
|
||
to_column="done", source="webhook",
|
||
)
|
||
outcome = "closed"
|
||
results.append(f"#{number}:{outcome}")
|
||
return results
|
||
|
||
|
||
@app.post("/webhooks/gitea")
|
||
async def gitea_webhook(request: Request):
|
||
"""Receive Gitea webhooks: verify HMAC, apply external close/reopen,
|
||
invalidate the read cache and push a live event to browsers."""
|
||
settings = get_settings()
|
||
raw = await request.body()
|
||
if settings.webhook_secret:
|
||
signature = request.headers.get("x-gitea-signature", "")
|
||
expected = hmac.new(
|
||
settings.webhook_secret.encode(), raw, hashlib.sha256
|
||
).hexdigest()
|
||
if not signature or not hmac.compare_digest(signature, expected):
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid webhook signature"},
|
||
status_code=401,
|
||
)
|
||
try:
|
||
payload = json.loads(raw or b"{}")
|
||
except ValueError:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid JSON payload"}, status_code=400
|
||
)
|
||
|
||
event = request.headers.get("x-gitea-event", "")
|
||
action = payload.get("action", "")
|
||
full_name = (payload.get("repository") or {}).get("full_name", "")
|
||
issue = payload.get("issue") or {}
|
||
number = issue.get("number")
|
||
|
||
handled = "ignored"
|
||
if event == "issues" and full_name and number:
|
||
if action == "closed":
|
||
handled = _apply_external_transition(full_name, number, "done", "closed")
|
||
elif action == "reopened":
|
||
handled = _apply_external_transition(
|
||
full_name, number, "backlog", "reopened"
|
||
)
|
||
elif event == "pull_request" and action == "closed" and full_name:
|
||
pr = payload.get("pull_request") or {}
|
||
if pr.get("merged"):
|
||
results = await _apply_pr_merged(full_name, pr)
|
||
handled = "pr-merged:" + (",".join(results) if results else "none")
|
||
|
||
gitea.invalidate_cache()
|
||
await hub.broadcast(
|
||
{"repo": full_name, "event": event, "action": action, "issue": number}
|
||
)
|
||
return {"ok": True, "handled": handled}
|
||
|
||
|
||
async def sse_event_stream(request: Request):
|
||
"""Yield SSE frames for one subscriber until it disconnects.
|
||
|
||
Split out of the route so it can be unit-tested directly:
|
||
test clients cannot reliably tear down an infinite HTTP stream.
|
||
"""
|
||
queue = hub.subscribe()
|
||
try:
|
||
yield "retry: 5000\n\n"
|
||
while True:
|
||
try:
|
||
payload = await asyncio.wait_for(queue.get(), timeout=25)
|
||
except asyncio.TimeoutError:
|
||
if await request.is_disconnected():
|
||
break
|
||
yield ": heartbeat\n\n"
|
||
continue
|
||
if await request.is_disconnected():
|
||
break
|
||
yield f"event: gitea\ndata: {json.dumps(payload)}\n\n"
|
||
finally:
|
||
hub.unsubscribe(queue)
|
||
|
||
|
||
async def _do_capture(
|
||
title: str, body: str, repo_full_name: str, column: str = "backlog"
|
||
) -> dict:
|
||
"""Create a Gitea issue from a capture and place its card.
|
||
|
||
Shared by POST /api/capture, the /capture page and the ntfy
|
||
listener. Raises ValueError for bad input (callers map to 400).
|
||
"""
|
||
title = (title or "").strip()
|
||
if not title or len(title) > 255:
|
||
raise ValueError("Title required (max 255 chars)")
|
||
if column not in KANBAN_COLUMNS:
|
||
raise ValueError("Invalid column")
|
||
parts = _split_full_name(repo_full_name)
|
||
if parts is None:
|
||
raise ValueError("Invalid repo (expected owner/name)")
|
||
owner, repo = parts
|
||
issue = await gitea.create_issue(owner, repo, title, body or "")
|
||
move_issue(repo_full_name, issue["number"], column)
|
||
log_event(
|
||
repo_full_name, issue["number"], "created",
|
||
to_column=column, source="capture",
|
||
)
|
||
await _publish_board_changed(repo_full_name)
|
||
return {
|
||
"number": issue["number"],
|
||
"title": issue.get("title", title),
|
||
"html_url": issue.get("html_url", ""),
|
||
"repo_full_name": repo_full_name,
|
||
"column": column,
|
||
}
|
||
|
||
|
||
@app.post("/api/capture")
|
||
async def api_capture(request: Request):
|
||
"""Quick capture endpoint (roadmap #12).
|
||
|
||
Auth: dedicated capture token (X-Capture-Token header or
|
||
Authorization: Bearer), distinct from the Gitea token. Exempt
|
||
from Basic Auth and CSRF — it is a machine-to-machine endpoint.
|
||
Disabled (503) until KANBANTE_CAPTURE_TOKEN is set.
|
||
"""
|
||
settings = get_settings()
|
||
if not settings.capture_token:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Capture is not configured "
|
||
"(set KANBANTE_CAPTURE_TOKEN)"},
|
||
status_code=503,
|
||
)
|
||
provided = request.headers.get("x-capture-token", "")
|
||
auth = request.headers.get("authorization", "")
|
||
if not provided and auth.lower().startswith("bearer "):
|
||
provided = auth[7:].strip()
|
||
if not provided or not secrets.compare_digest(provided, settings.capture_token):
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid capture token"}, status_code=401
|
||
)
|
||
|
||
content_type = request.headers.get("content-type", "")
|
||
title = body = text = repo = ""
|
||
column = "backlog"
|
||
if "application/json" in content_type:
|
||
try:
|
||
payload = await request.json()
|
||
except Exception:
|
||
payload = {}
|
||
if isinstance(payload, dict):
|
||
title = str(payload.get("title") or "")
|
||
body = str(payload.get("body") or "")
|
||
text = str(payload.get("text") or "")
|
||
repo = str(payload.get("repo") or "")
|
||
column = str(payload.get("column") or "backlog")
|
||
elif "application/x-www-form-urlencoded" in content_type or (
|
||
"multipart/form-data" in content_type
|
||
):
|
||
form = await request.form()
|
||
title = str(form.get("title") or "")
|
||
body = str(form.get("body") or "")
|
||
text = str(form.get("text") or "")
|
||
repo = str(form.get("repo") or "")
|
||
column = str(form.get("column") or "backlog")
|
||
else: # text/plain and friends: the whole body is the capture
|
||
text = (await request.body()).decode("utf-8", errors="replace")
|
||
|
||
if not title and text:
|
||
title, rest = parse_capture_text(text)
|
||
body = body or rest
|
||
repo = repo or settings.capture_repo
|
||
if not repo:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "No repo given and KANBANTE_CAPTURE_REPO "
|
||
"is not set"},
|
||
status_code=400,
|
||
)
|
||
try:
|
||
created = await _do_capture(title, body, repo, column)
|
||
except ValueError as exc:
|
||
return JSONResponse({"ok": False, "error": str(exc)}, status_code=400)
|
||
return JSONResponse({"ok": True, "issue": created}, status_code=201)
|
||
|
||
|
||
@app.get("/capture", response_class=HTMLResponse)
|
||
async def capture_page(request: Request):
|
||
"""Quick-capture page: form, status and usage instructions."""
|
||
settings = get_settings()
|
||
return render(
|
||
request,
|
||
"capture.html",
|
||
capture_repo=settings.capture_repo,
|
||
api_enabled=bool(settings.capture_token),
|
||
ntfy_topic=settings.capture_ntfy_topic,
|
||
ntfy_url=settings.ntfy_url,
|
||
created=None,
|
||
error=None,
|
||
)
|
||
|
||
|
||
@app.post("/capture", response_class=HTMLResponse)
|
||
async def capture_submit(
|
||
request: Request,
|
||
repo_full_name: str = Form(""),
|
||
title: str = Form(""),
|
||
body: str = Form(""),
|
||
column: str = Form("backlog"),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Web form capture (session + CSRF, like every other form)."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
settings = get_settings()
|
||
repo_full_name = repo_full_name.strip() or settings.capture_repo
|
||
created, error = None, None
|
||
if not repo_full_name:
|
||
error = "Aucun dépôt : renseigne le dépôt ou définis KANBANTE_CAPTURE_REPO."
|
||
else:
|
||
try:
|
||
created = await _do_capture(title, body, repo_full_name, column)
|
||
except ValueError as exc:
|
||
error = str(exc)
|
||
return render(
|
||
request,
|
||
"capture.html",
|
||
capture_repo=settings.capture_repo,
|
||
api_enabled=bool(settings.capture_token),
|
||
ntfy_topic=settings.capture_ntfy_topic,
|
||
ntfy_url=settings.ntfy_url,
|
||
created=created,
|
||
error=error,
|
||
)
|
||
|
||
|
||
async def _weekly_digest() -> dict:
|
||
"""Compose the weekly digest (journal of the last 7 days + focus)."""
|
||
now = datetime.now(timezone.utc)
|
||
since = (now - timedelta(days=7)).replace(tzinfo=None)
|
||
events = []
|
||
for event in get_events(limit=1000):
|
||
at = _parse_datetime(str(event.get("created_at") or ""))
|
||
if at is not None and at.replace(tzinfo=None) >= since:
|
||
events.append(event)
|
||
|
||
closed_keys: list[tuple[str, int]] = []
|
||
seen: set[tuple[str, int]] = set()
|
||
for event in events:
|
||
if event["event_type"] == "closed":
|
||
key = (event["repo_full_name"], event["issue_number"])
|
||
if key not in seen:
|
||
seen.add(key)
|
||
closed_keys.append(key)
|
||
|
||
titles: dict[tuple[str, int], dict] = {}
|
||
for repo_name in {repo for repo, _ in closed_keys}:
|
||
owner, _, name = repo_name.partition("/")
|
||
try:
|
||
items = await gitea.get_repo_issues(owner, name, state="all")
|
||
except Exception:
|
||
continue
|
||
for item in items:
|
||
if _is_issue(item):
|
||
titles[(repo_name, item["number"])] = item
|
||
closed_items = [
|
||
{
|
||
"repo": repo,
|
||
"number": number,
|
||
"title": (titles.get((repo, number)) or {}).get("title", ""),
|
||
"html_url": (titles.get((repo, number)) or {}).get("html_url", ""),
|
||
}
|
||
for repo, number in closed_keys
|
||
]
|
||
|
||
focus_groups = []
|
||
for group in await _collect_focus_groups():
|
||
focus_groups.append(
|
||
{
|
||
"full_name": group["full_name"],
|
||
"cards": [
|
||
{
|
||
"issue": card["issue"],
|
||
"column": COLUMN_LABELS[card["column"]],
|
||
}
|
||
for card in group["cards"]
|
||
],
|
||
}
|
||
)
|
||
|
||
week_label = (
|
||
f"{(now - timedelta(days=6)).strftime('%Y-%m-%d')}"
|
||
f" au {now.strftime('%Y-%m-%d')}"
|
||
)
|
||
markdown = digest_service.format_digest_markdown(
|
||
week_label,
|
||
closed_items,
|
||
created_count=sum(1 for e in events if e["event_type"] == "created"),
|
||
moved_count=sum(1 for e in events if e["event_type"] == "moved"),
|
||
focus_groups=focus_groups,
|
||
)
|
||
return {
|
||
"title": f"Kanbante — semaine du {week_label}",
|
||
"week": now.strftime("%G-W%V"),
|
||
"markdown": markdown,
|
||
"closed_count": len(closed_items),
|
||
}
|
||
|
||
|
||
def _render_digest(request: Request, banner: str | None = None,
|
||
tools: list[str] | None = None):
|
||
settings = get_settings()
|
||
return render(
|
||
request,
|
||
"digest.html",
|
||
pending=get_kv("digest_pending"),
|
||
last_result=get_kv("digest_last_result"),
|
||
obsigate_configured=bool(settings.obsigate_mcp_url),
|
||
obsigate_url=settings.obsigate_mcp_url,
|
||
obsigate_tool=settings.obsigate_mcp_tool,
|
||
webhook_configured=bool(settings.digest_webhook_url),
|
||
digest_time=settings.digest_time,
|
||
digest_weekday=settings.digest_weekday,
|
||
banner=banner,
|
||
tools=tools,
|
||
)
|
||
|
||
|
||
@app.get("/digest", response_class=HTMLResponse)
|
||
async def digest_page(request: Request):
|
||
"""Weekly digest: preview, approval and delivery status."""
|
||
return _render_digest(request)
|
||
|
||
|
||
@app.post("/digest/probe", response_class=HTMLResponse)
|
||
async def digest_probe(request: Request, csrf_token: str = Form("")):
|
||
"""List the tools of the configured ObsiGate MCP server."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
settings = get_settings()
|
||
if not settings.obsigate_mcp_url:
|
||
return _render_digest(request, banner="❌ Aucun serveur MCP ObsiGate configuré (KANBANTE_OBSIGATE_MCP_URL).")
|
||
ok, result = await digest_service.mcp_list_tools(
|
||
settings.obsigate_mcp_url, settings.obsigate_mcp_token
|
||
)
|
||
if ok:
|
||
return _render_digest(
|
||
request,
|
||
banner=f"✅ Connexion MCP réussie — {len(result)} outil(s) disponible(s).",
|
||
tools=result,
|
||
)
|
||
return _render_digest(request, banner=f"❌ {result}")
|
||
|
||
|
||
@app.post("/digest/send", response_class=HTMLResponse)
|
||
async def digest_send(
|
||
request: Request,
|
||
target: str = Form(...),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Deliver the digest — only ever from this explicit approval."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
settings = get_settings()
|
||
digest = await _weekly_digest()
|
||
if target == "webhook":
|
||
if not settings.digest_webhook_url:
|
||
return _render_digest(request, banner="❌ Webhook non configuré (KANBANTE_DIGEST_WEBHOOK_URL).")
|
||
ok, message = await digest_service.send_webhook(
|
||
settings.digest_webhook_url,
|
||
digest["title"], digest["markdown"], digest["week"],
|
||
)
|
||
elif target == "obsigate":
|
||
if not settings.obsigate_mcp_url or not settings.obsigate_mcp_tool:
|
||
return _render_digest(request, banner="❌ ObsiGate incomplet : KANBANTE_OBSIGATE_MCP_URL et KANBANTE_OBSIGATE_MCP_TOOL sont requis (utilise « Tester la connexion » pour voir les outils).")
|
||
try:
|
||
arguments = digest_service.render_mcp_args(
|
||
settings.obsigate_mcp_args, digest["title"], digest["markdown"]
|
||
)
|
||
except ValueError as exc:
|
||
return _render_digest(request, banner=f"❌ {exc}")
|
||
ok, message = await digest_service.mcp_call_tool(
|
||
settings.obsigate_mcp_url, settings.obsigate_mcp_token,
|
||
settings.obsigate_mcp_tool, arguments,
|
||
)
|
||
else:
|
||
return _render_digest(request, banner="❌ Cible inconnue.")
|
||
stamp = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
set_kv("digest_last_result", f"{stamp} · {target} · {message}")
|
||
if ok:
|
||
set_kv("digest_pending", "")
|
||
return _render_digest(request, banner=f"✅ Digest envoyé ({target}) : {message}")
|
||
return _render_digest(request, banner=f"❌ Échec de l'envoi ({target}) : {message}")
|
||
|
||
|
||
async def _digest_scheduler() -> None:
|
||
"""Compose the weekly digest and notify — never deliver it.
|
||
|
||
Delivery to ObsiGate is a cross-app write: it waits for the
|
||
one-click approval on /digest (standing rule).
|
||
"""
|
||
while True:
|
||
try:
|
||
settings = get_settings()
|
||
if settings.digest_time:
|
||
now = datetime.now()
|
||
week = now.strftime("%G-W%V")
|
||
if (
|
||
now.weekday() == settings.digest_weekday
|
||
and now.strftime("%H:%M") >= settings.digest_time
|
||
and get_kv("digest_pending_week") != week
|
||
):
|
||
digest = await _weekly_digest()
|
||
set_kv("digest_pending", digest["markdown"])
|
||
set_kv("digest_pending_week", week)
|
||
logger.info("Weekly digest composed for %s", week)
|
||
if ntfy_configured(settings):
|
||
await send_ntfy(
|
||
settings,
|
||
"🧠 Digest hebdo Kanbante prêt — "
|
||
"en attente d'approbation sur la page /digest.",
|
||
)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception as exc:
|
||
logger.warning("Digest scheduler error: %s", exc)
|
||
await asyncio.sleep(60)
|
||
|
||
|
||
@app.get("/api/stream")
|
||
async def live_stream(request: Request):
|
||
"""Server-Sent Events: browsers reload softly on Gitea activity."""
|
||
return StreamingResponse(
|
||
sse_event_stream(request),
|
||
media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
||
)
|
||
|
||
|
||
@app.get("/activity", response_class=HTMLResponse)
|
||
async def activity(request: Request):
|
||
"""Activity feed: the latest entries of the event journal."""
|
||
events = get_events(limit=100)
|
||
return render(
|
||
request,
|
||
"activity.html",
|
||
events=events,
|
||
event_icons=EVENT_ICONS,
|
||
column_labels=COLUMN_LABELS,
|
||
)
|
||
|
||
|
||
# ── Routes: API ────────────────────────────────────────────
|
||
|
||
@app.post("/api/move")
|
||
async def api_move(
|
||
request: Request,
|
||
repo_full_name: str = Form(...),
|
||
issue_number: int = Form(...),
|
||
to_column: str = Form(...),
|
||
from_column: str = Form(""),
|
||
to_order: str = Form(""),
|
||
from_order: str = Form(""),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Move an issue to a new column and persist the resulting order.
|
||
|
||
With KANBANTE_SYNC_GITEA enabled (default), moving to "done" closes
|
||
the issue in Gitea and moving out of "done" reopens it. A failed
|
||
write-back is reported in the response (gitea_sync="failed") instead
|
||
of failing the move itself.
|
||
"""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
if to_column not in KANBAN_COLUMNS:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid column"}, status_code=400
|
||
)
|
||
parts = _split_full_name(repo_full_name)
|
||
if parts is None:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid repo_full_name"}, status_code=400
|
||
)
|
||
owner, repo = parts
|
||
|
||
move_issue(repo_full_name, issue_number, to_column)
|
||
if from_column in KANBAN_COLUMNS and from_column != to_column:
|
||
remaining = [
|
||
n for n in _parse_order(from_order) if n != issue_number
|
||
]
|
||
if remaining:
|
||
set_column_order(repo_full_name, from_column, remaining)
|
||
order = _parse_order(to_order)
|
||
if order:
|
||
if issue_number not in order:
|
||
order.append(issue_number)
|
||
set_column_order(repo_full_name, to_column, order)
|
||
|
||
if from_column != to_column:
|
||
log_event(
|
||
repo_full_name, issue_number, "moved",
|
||
from_column or None, to_column,
|
||
)
|
||
|
||
result: dict = {"ok": True, "gitea_sync": "skipped"}
|
||
settings = get_settings()
|
||
if settings.sync_gitea:
|
||
try:
|
||
if to_column == "done":
|
||
await gitea.set_issue_state(owner, repo, issue_number, "closed")
|
||
result["gitea_sync"] = "closed"
|
||
log_event(repo_full_name, issue_number, "closed", to_column="done")
|
||
elif from_column == "done":
|
||
await gitea.set_issue_state(owner, repo, issue_number, "open")
|
||
result["gitea_sync"] = "reopened"
|
||
log_event(
|
||
repo_full_name, issue_number, "reopened",
|
||
from_column="done", to_column=to_column,
|
||
)
|
||
except httpx.HTTPError as exc:
|
||
logger.warning(
|
||
"Gitea write-back failed for %s#%s: %s",
|
||
repo_full_name,
|
||
issue_number,
|
||
exc,
|
||
)
|
||
result["gitea_sync"] = "failed"
|
||
|
||
# Mirror intermediate columns to Gitea labels (kanban:<column>) so
|
||
# the workflow state is visible in Gitea itself, not just here.
|
||
result["label_sync"] = "skipped"
|
||
if settings.sync_labels and from_column != to_column:
|
||
try:
|
||
desired = (
|
||
f"kanban:{to_column}" if to_column in LABELLED_COLUMNS else None
|
||
)
|
||
issue = await gitea.get_issue(owner, repo, issue_number)
|
||
current = [
|
||
label for label in issue.get("labels") or []
|
||
if str(label.get("name", "")).startswith("kanban:")
|
||
]
|
||
for label in current:
|
||
if label.get("name") != desired:
|
||
await gitea.remove_issue_label(
|
||
owner, repo, issue_number, label["id"]
|
||
)
|
||
if desired and all(l.get("name") != desired for l in current):
|
||
label = await gitea.ensure_label(
|
||
owner, repo, desired, COLUMN_COLORS[to_column]
|
||
)
|
||
await gitea.add_issue_label(owner, repo, issue_number, label["id"])
|
||
result["label_sync"] = "ok"
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"Label sync failed for %s#%s: %s", repo_full_name, issue_number, exc
|
||
)
|
||
result["label_sync"] = "failed"
|
||
await _publish_board_changed(repo_full_name)
|
||
return result
|
||
|
||
|
||
@app.post("/api/issues")
|
||
async def api_create_issue(
|
||
request: Request,
|
||
repo_full_name: str = Form(...),
|
||
title: str = Form(...),
|
||
body: str = Form(""),
|
||
column: str = Form("backlog"),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Create a Gitea issue from the board and place its card."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
parts = _split_full_name(repo_full_name)
|
||
if parts is None:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid repo_full_name"}, status_code=400
|
||
)
|
||
title = title.strip()
|
||
if not title or len(title) > 255:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Title required (max 255 chars)"},
|
||
status_code=400,
|
||
)
|
||
if column not in KANBAN_COLUMNS:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Invalid column"}, status_code=400
|
||
)
|
||
owner, repo = parts
|
||
issue = await gitea.create_issue(owner, repo, title, body)
|
||
move_issue(repo_full_name, issue["number"], column)
|
||
log_event(repo_full_name, issue["number"], "created", to_column=column)
|
||
await _publish_board_changed(repo_full_name)
|
||
return JSONResponse(
|
||
{
|
||
"ok": True,
|
||
"issue": {
|
||
"number": issue["number"],
|
||
"title": issue.get("title", title),
|
||
"html_url": issue.get("html_url", ""),
|
||
},
|
||
},
|
||
status_code=201,
|
||
)
|
||
|
||
|
||
def _issue_payload(issue: dict) -> dict:
|
||
assignees = issue.get("assignees") or []
|
||
if not assignees and issue.get("assignee"):
|
||
assignees = [issue["assignee"]]
|
||
milestone = issue.get("milestone") or {}
|
||
return {
|
||
"number": issue.get("number"),
|
||
"title": issue.get("title", ""),
|
||
"body": issue.get("body") or "",
|
||
"state": issue.get("state", "open"),
|
||
"html_url": issue.get("html_url", ""),
|
||
"created_at": issue.get("created_at", ""),
|
||
"updated_at": issue.get("updated_at", ""),
|
||
"due_date": issue.get("due_date"),
|
||
"labels": [
|
||
{"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
|
||
for l in issue.get("labels") or []
|
||
],
|
||
"assignees": [
|
||
{"login": a.get("login", ""), "avatar_url": a.get("avatar_url", "")}
|
||
for a in assignees
|
||
],
|
||
"milestone": milestone.get("title"),
|
||
}
|
||
|
||
|
||
def _comment_payload(comment: dict) -> dict:
|
||
user = comment.get("user") or {}
|
||
return {
|
||
"id": comment.get("id"),
|
||
"body": comment.get("body") or "",
|
||
"created_at": comment.get("created_at", ""),
|
||
"updated_at": comment.get("updated_at", ""),
|
||
"html_url": comment.get("html_url", ""),
|
||
"user": {
|
||
"login": user.get("login", "?"),
|
||
"avatar_url": user.get("avatar_url", ""),
|
||
},
|
||
}
|
||
|
||
|
||
async def _fetch_code_sources(owner: str, repo: str):
|
||
"""Branches + PRs + recent commits of a repo (best effort each)."""
|
||
async def _safe(coro):
|
||
try:
|
||
return await coro
|
||
except Exception as exc:
|
||
logger.debug("code links source failed for %s/%s: %s", owner, repo, exc)
|
||
return []
|
||
|
||
branches, pulls, commits = await asyncio.gather(
|
||
_safe(gitea.list_branches(owner, repo)),
|
||
_safe(gitea.list_pulls(owner, repo)),
|
||
_safe(gitea.list_commits(owner, repo)),
|
||
)
|
||
return branches, pulls, commits
|
||
|
||
|
||
async def _annotate_code(owner: str, repo: str, columns: dict) -> None:
|
||
"""Attach a compact _code summary to each card issue (issue #1)."""
|
||
branches, pulls, commits = await _fetch_code_sources(owner, repo)
|
||
if not (branches or pulls or commits):
|
||
return
|
||
for cards in columns.values():
|
||
for issue in cards:
|
||
links = code_links_service.find_code_links(
|
||
branches, pulls, commits, issue["number"]
|
||
)
|
||
if links["branches"] or links["pulls"] or links["commits"]:
|
||
issue["_code"] = links["summary"]
|
||
|
||
|
||
@app.get("/api/issues/{owner}/{repo}/{number}/code")
|
||
async def api_issue_code(owner: str, repo: str, number: int):
|
||
"""Branches, PRs and commits linked to a card (issue #1)."""
|
||
branches, pulls, commits = await _fetch_code_sources(owner, repo)
|
||
links = code_links_service.find_code_links(branches, pulls, commits, number)
|
||
suggested = f"issue-{number}"
|
||
try:
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
suggested = code_links_service.branch_name_for(
|
||
number, issue.get("title", "")
|
||
)
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "suggested_branch": suggested, **links}
|
||
|
||
|
||
_BRANCH_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
|
||
|
||
|
||
async def _sync_kanban_label(
|
||
owner: str, repo: str, number: int, column: str
|
||
) -> None:
|
||
"""Mirror a column on the issue's kanban:* label (best effort)."""
|
||
if not get_settings().sync_labels:
|
||
return
|
||
try:
|
||
desired = (
|
||
f"kanban:{column}" if column in LABELLED_COLUMNS else None
|
||
)
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
current = [
|
||
label for label in issue.get("labels") or []
|
||
if str(label.get("name", "")).startswith("kanban:")
|
||
]
|
||
for label in current:
|
||
if label.get("name") != desired:
|
||
await gitea.remove_issue_label(
|
||
owner, repo, number, label["id"]
|
||
)
|
||
if desired and all(l.get("name") != desired for l in current):
|
||
label = await gitea.ensure_label(
|
||
owner, repo, desired, COLUMN_COLORS[column]
|
||
)
|
||
await gitea.add_issue_label(owner, repo, number, label["id"])
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"Label sync failed for %s/%s#%s: %s", owner, repo, number, exc
|
||
)
|
||
|
||
|
||
@app.post("/api/issues/{owner}/{repo}/{number}/branch")
|
||
async def api_create_branch(
|
||
owner: str,
|
||
repo: str,
|
||
number: int,
|
||
request: Request,
|
||
name: str = Form(""),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Create the card's branch in Gitea (issue #3), then start the card.
|
||
|
||
Default name: issue-<n>-<slug of the title> — the convention the
|
||
merged-PR automation (issue #2) recognizes. Creating the branch
|
||
moves the card to In Progress and journals it.
|
||
"""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
try:
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
except Exception:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Issue introuvable"}, status_code=404
|
||
)
|
||
if not _is_issue(issue):
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Pas une issue"}, status_code=400
|
||
)
|
||
branch = (name or "").strip() or code_links_service.branch_name_for(
|
||
number, issue.get("title", "")
|
||
)
|
||
if (
|
||
not _BRANCH_NAME_RE.match(branch)
|
||
or ".." in branch
|
||
or branch.endswith("/")
|
||
or branch.endswith(".lock")
|
||
):
|
||
return JSONResponse(
|
||
{"ok": False, "error": f"Nom de branche invalide : {branch}"},
|
||
status_code=400,
|
||
)
|
||
repo_info = {}
|
||
try:
|
||
repo_info = await gitea.get_repo(owner, repo)
|
||
except Exception:
|
||
pass
|
||
base_url = repo_info.get("html_url", "")
|
||
branch_url = f"{base_url}/src/branch/{branch}" if base_url else ""
|
||
existing = await _safe_list_branches(owner, repo)
|
||
if branch in existing:
|
||
return {
|
||
"ok": True, "branch": branch, "exists": True,
|
||
"url": branch_url, "moved": False,
|
||
}
|
||
default_branch = repo_info.get("default_branch") or "main"
|
||
try:
|
||
await gitea.create_branch(owner, repo, branch, default_branch)
|
||
except Exception as exc:
|
||
return JSONResponse(
|
||
{"ok": False, "error": f"Création impossible : {exc}"},
|
||
status_code=502,
|
||
)
|
||
gitea.invalidate_cache()
|
||
|
||
full_name = f"{owner}/{repo}"
|
||
moved = False
|
||
if issue.get("state") == "open":
|
||
board = get_board_state(full_name)
|
||
current = next(
|
||
(
|
||
col for col, cards in board.items()
|
||
if any(c["issue_number"] == number for c in cards)
|
||
),
|
||
"backlog",
|
||
)
|
||
if current != "in_progress":
|
||
move_issue(full_name, number, "in_progress")
|
||
log_event(
|
||
full_name, number, "moved",
|
||
from_column=current, to_column="in_progress",
|
||
source="branch",
|
||
)
|
||
await _sync_kanban_label(owner, repo, number, "in_progress")
|
||
moved = True
|
||
await _publish_board_changed(full_name)
|
||
return {
|
||
"ok": True, "branch": branch, "from": default_branch,
|
||
"url": branch_url, "moved": moved,
|
||
}
|
||
|
||
|
||
async def _safe_list_branches(owner: str, repo: str) -> list[str]:
|
||
try:
|
||
return [b.get("name", "") for b in await gitea.list_branches(owner, repo)]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
@app.get("/api/issues/{owner}/{repo}/{number}")
|
||
async def api_issue_detail(owner: str, repo: str, number: int):
|
||
"""Issue detail + comments for the card panel (proxied from Gitea)."""
|
||
try:
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
comments = await gitea.get_issue_comments(owner, repo, number)
|
||
except httpx.HTTPStatusError as exc:
|
||
if exc.response.status_code == 404:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Issue not found"}, status_code=404
|
||
)
|
||
raise
|
||
return {
|
||
"ok": True,
|
||
"issue": _issue_payload(issue),
|
||
"comments": [_comment_payload(c) for c in comments],
|
||
}
|
||
|
||
|
||
@app.post("/api/issues/{owner}/{repo}/{number}/comments")
|
||
async def api_issue_comment(
|
||
request: Request,
|
||
owner: str,
|
||
repo: str,
|
||
number: int,
|
||
body: str = Form(...),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Post a comment on the issue in Gitea."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
body = body.strip()
|
||
if not body:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Comment must not be empty"}, status_code=400
|
||
)
|
||
if len(body) > 10_000:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Comment too large (max 10000 chars)"},
|
||
status_code=413,
|
||
)
|
||
comment = await gitea.create_comment(owner, repo, number, body)
|
||
log_event(f"{owner}/{repo}", number, "commented")
|
||
return JSONResponse(
|
||
{"ok": True, "comment": _comment_payload(comment)}, status_code=201
|
||
)
|
||
|
||
|
||
@app.get("/api/events")
|
||
async def api_events(repo: str = "", limit: int = 100):
|
||
"""Latest journal entries (all repos, or one via ?repo=owner/name)."""
|
||
events = get_events(repo or None, max(1, min(limit, 500)))
|
||
return {"ok": True, "events": events}
|
||
|
||
|
||
@app.get("/api/metrics")
|
||
async def api_metrics(repo: str = "", window: int = 30):
|
||
"""Throughput / cycle-time metrics from the journal."""
|
||
metrics = get_metrics(repo or None, window)
|
||
weekly = forecast_service.weekly_closures(
|
||
get_events(repo_full_name=repo or None, limit=2000)
|
||
)
|
||
metrics["weekly_closures"] = [b["count"] for b in weekly]
|
||
stats = forecast_service.throughput_stats(weekly)
|
||
metrics["throughput_4w"] = stats
|
||
return {"ok": True, "metrics": metrics}
|
||
|
||
|
||
def _repo_forecast_label(repo_full_name: str, remaining: int) -> str | None:
|
||
"""Backlog burn-down ETA label for one repo (journal-based)."""
|
||
weekly = forecast_service.weekly_closures(
|
||
get_events(repo_full_name=repo_full_name, limit=2000)
|
||
)
|
||
est = forecast_service.estimate(remaining, weekly)
|
||
if est is None:
|
||
return None
|
||
if est.get("done"):
|
||
return "✅ rien d'ouvert"
|
||
label = forecast_service.eta_label(est)
|
||
return f"backlog épuisé {label}" if label else None
|
||
|
||
|
||
# ── Label management ───────────────────────────────────────
|
||
|
||
@app.get("/api/repos/{owner}/{repo}/labels")
|
||
async def api_list_labels(owner: str, repo: str):
|
||
"""All label definitions of a repo."""
|
||
labels = await gitea.get_labels(owner, repo)
|
||
return {
|
||
"ok": True,
|
||
"labels": [
|
||
{"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
|
||
for l in labels
|
||
],
|
||
}
|
||
|
||
|
||
@app.post("/api/repos/{owner}/{repo}/labels")
|
||
async def api_create_label(
|
||
request: Request,
|
||
owner: str,
|
||
repo: str,
|
||
name: str = Form(...),
|
||
color: str = Form("7c5cfc"),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Create a label definition in the repo."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
name = name.strip()
|
||
color = color.strip().lstrip("#")
|
||
if not name or len(name) > 50:
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Label name must be 1-50 chars"},
|
||
status_code=400,
|
||
)
|
||
if len(color) != 6 or any(c not in "0123456789abcdefABCDEF" for c in color):
|
||
return JSONResponse(
|
||
{"ok": False, "error": "Color must be a 6-digit hex value"},
|
||
status_code=400,
|
||
)
|
||
label = await gitea.create_label(owner, repo, name, color)
|
||
gitea.forget_label_cache(owner, repo)
|
||
return JSONResponse(
|
||
{
|
||
"ok": True,
|
||
"label": {
|
||
"id": label.get("id"),
|
||
"name": label.get("name", name),
|
||
"color": label.get("color", color),
|
||
},
|
||
},
|
||
status_code=201,
|
||
)
|
||
|
||
|
||
@app.delete("/api/repos/{owner}/{repo}/labels/{label_id}")
|
||
async def api_delete_label(
|
||
request: Request, owner: str, repo: str, label_id: int
|
||
):
|
||
"""Delete a label definition (CSRF via the X-CSRF-Token header)."""
|
||
if not _csrf_valid(request, ""):
|
||
return _csrf_error()
|
||
await gitea.delete_label(owner, repo, label_id)
|
||
return {"ok": True, "deleted": label_id}
|
||
|
||
|
||
@app.post("/api/issues/{owner}/{repo}/{number}/labels")
|
||
async def api_toggle_issue_label(
|
||
request: Request,
|
||
owner: str,
|
||
repo: str,
|
||
number: int,
|
||
label_id: int = Form(...),
|
||
action: str = Form(...),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Add or remove one label on one issue; returns the new label set."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
if action not in ("add", "remove"):
|
||
return JSONResponse(
|
||
{"ok": False, "error": "action must be 'add' or 'remove'"},
|
||
status_code=400,
|
||
)
|
||
if action == "add":
|
||
await gitea.add_issue_label(owner, repo, number, label_id)
|
||
else:
|
||
await gitea.remove_issue_label(owner, repo, number, label_id)
|
||
await _publish_board_changed(f"{owner}/{repo}")
|
||
issue = await gitea.get_issue(owner, repo, number)
|
||
return {
|
||
"ok": True,
|
||
"labels": [
|
||
{"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
|
||
for l in issue.get("labels") or []
|
||
],
|
||
}
|
||
|
||
|
||
# ── MCP server (read-only tools over JSON-RPC) ─────────────
|
||
|
||
MCP_TOOLS = [
|
||
{
|
||
"name": "list_projects",
|
||
"description": "List the projects (Gitea repos) Kanbante tracks, grouped by activity.",
|
||
"inputSchema": {"type": "object", "properties": {}},
|
||
},
|
||
{
|
||
"name": "get_board",
|
||
"description": "Get the kanban board of one repo: columns with their cards.",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"owner": {"type": "string"},
|
||
"repo": {"type": "string"},
|
||
},
|
||
"required": ["owner", "repo"],
|
||
},
|
||
},
|
||
{
|
||
"name": "get_issue",
|
||
"description": "Get one issue with its body and comments.",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"owner": {"type": "string"},
|
||
"repo": {"type": "string"},
|
||
"number": {"type": "integer"},
|
||
},
|
||
"required": ["owner", "repo", "number"],
|
||
},
|
||
},
|
||
{
|
||
"name": "get_focus",
|
||
"description": "Cards in progress or in review across all active projects.",
|
||
"inputSchema": {"type": "object", "properties": {}},
|
||
},
|
||
{
|
||
"name": "search_issues",
|
||
"description": "Search issues by text across all repos.",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"query": {"type": "string"},
|
||
"state": {"type": "string", "enum": ["open", "closed", "all"]},
|
||
},
|
||
"required": ["query"],
|
||
},
|
||
},
|
||
{
|
||
"name": "get_activity",
|
||
"description": "Latest entries of the board event journal.",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {"limit": {"type": "integer"}},
|
||
},
|
||
},
|
||
{
|
||
"name": "get_metrics",
|
||
"description": "Throughput and cycle-time metrics from the journal.",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {"repo": {"type": "string"}},
|
||
},
|
||
},
|
||
]
|
||
|
||
|
||
async def _mcp_call_tool(name: str, args: dict) -> str:
|
||
"""Execute one read-only MCP tool and return text content."""
|
||
if name == "list_projects":
|
||
repos = await gitea.search_repos()
|
||
active, inactive, archived = classify_repos(repos)
|
||
lines = [f"Active projects ({len(active)}):"]
|
||
lines += [
|
||
f"- {r['full_name']} ({r.get('open_issues_count') or 0} open issues)"
|
||
for r in active
|
||
]
|
||
lines.append(
|
||
f"Inactive: {len(inactive)} - Archived: {len(archived)}"
|
||
)
|
||
return "\n".join(lines)
|
||
if name == "get_board":
|
||
owner, repo = args["owner"], args["repo"]
|
||
items = await gitea.get_repo_issues(owner, repo, state="all")
|
||
issues = [i for i in items if _is_issue(i)]
|
||
open_issues = [i for i in issues if i.get("state") == "open"]
|
||
columns = _merge_issues_with_board(open_issues, get_board_state(f"{owner}/{repo}"))
|
||
return json.dumps(
|
||
{
|
||
column: [
|
||
{
|
||
"number": i["number"],
|
||
"title": i["title"],
|
||
"labels": [l.get("name") for l in i.get("labels") or []],
|
||
}
|
||
for i in cards
|
||
]
|
||
for column, cards in columns.items()
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
if name == "get_issue":
|
||
detail = await api_issue_detail(args["owner"], args["repo"], int(args["number"]))
|
||
return json.dumps(detail, ensure_ascii=False, default=str)
|
||
if name == "get_focus":
|
||
groups = await _collect_focus_groups()
|
||
return json.dumps(
|
||
[
|
||
{
|
||
"repo": g["full_name"],
|
||
"cards": [
|
||
{
|
||
"number": c["issue"]["number"],
|
||
"title": c["issue"]["title"],
|
||
"column": c["column"],
|
||
}
|
||
for c in g["cards"]
|
||
],
|
||
}
|
||
for g in groups
|
||
],
|
||
ensure_ascii=False,
|
||
)
|
||
if name == "search_issues":
|
||
results = await gitea.search_issues(
|
||
q=args["query"], state=args.get("state", "open")
|
||
)
|
||
return json.dumps(
|
||
[
|
||
{
|
||
"repo": (i.get("repository") or {}).get("full_name"),
|
||
"number": i["number"],
|
||
"title": i["title"],
|
||
"state": i["state"],
|
||
}
|
||
for i in results
|
||
],
|
||
ensure_ascii=False,
|
||
)
|
||
if name == "get_activity":
|
||
return json.dumps(
|
||
get_events(limit=int(args.get("limit", 50))), ensure_ascii=False
|
||
)
|
||
if name == "get_metrics":
|
||
return json.dumps(get_metrics(args.get("repo") or None), ensure_ascii=False)
|
||
raise ValueError(f"Unknown tool: {name}")
|
||
|
||
|
||
@app.get("/mcp")
|
||
async def mcp_info():
|
||
"""Discoverability for the MCP endpoint (the protocol is POST)."""
|
||
return {
|
||
"name": "kanbante",
|
||
"version": APP_VERSION,
|
||
"transport": "streamable-http (JSON-RPC 2.0 over POST)",
|
||
"read_only": True,
|
||
"tools": [t["name"] for t in MCP_TOOLS],
|
||
}
|
||
|
||
|
||
@app.post("/mcp")
|
||
async def mcp_endpoint(request: Request):
|
||
"""Minimal MCP server: initialize / tools/list / tools/call / ping.
|
||
|
||
Read-only by design (Bruno's rule: cross-platform reads first,
|
||
writes go through approval flows elsewhere).
|
||
"""
|
||
try:
|
||
payload = await request.json()
|
||
except ValueError:
|
||
return JSONResponse(
|
||
{"jsonrpc": "2.0", "id": None,
|
||
"error": {"code": -32700, "message": "Parse error"}},
|
||
status_code=400,
|
||
)
|
||
if isinstance(payload, list): # batch: not supported by this server
|
||
return JSONResponse(
|
||
{"jsonrpc": "2.0", "id": None,
|
||
"error": {"code": -32600, "message": "Batch requests not supported"}},
|
||
status_code=400,
|
||
)
|
||
method = payload.get("method", "")
|
||
rpc_id = payload.get("id")
|
||
if rpc_id is None: # notification: acknowledge silently
|
||
return Response(status_code=202)
|
||
params = payload.get("params") or {}
|
||
|
||
def _result(result):
|
||
return {"jsonrpc": "2.0", "id": rpc_id, "result": result}
|
||
|
||
def _error(code, message):
|
||
return {"jsonrpc": "2.0", "id": rpc_id,
|
||
"error": {"code": code, "message": message}}
|
||
|
||
if method == "initialize":
|
||
return _result(
|
||
{
|
||
"protocolVersion": params.get("protocolVersion", "2024-11-05"),
|
||
"capabilities": {"tools": {}},
|
||
"serverInfo": {"name": "kanbante", "version": APP_VERSION},
|
||
}
|
||
)
|
||
if method == "ping":
|
||
return _result({})
|
||
if method == "tools/list":
|
||
return _result({"tools": MCP_TOOLS})
|
||
if method == "tools/call":
|
||
tool_name = params.get("name", "")
|
||
arguments = params.get("arguments") or {}
|
||
try:
|
||
text = await _mcp_call_tool(tool_name, arguments)
|
||
except KeyError as exc:
|
||
return _result(
|
||
{"content": [{"type": "text", "text": f"Missing argument: {exc}"}],
|
||
"isError": True}
|
||
)
|
||
except Exception as exc:
|
||
return _result(
|
||
{"content": [{"type": "text", "text": f"Tool error: {exc}"}],
|
||
"isError": True}
|
||
)
|
||
return _result({"content": [{"type": "text", "text": text}], "isError": False})
|
||
return _error(-32601, f"Method not found: {method}")
|
||
|
||
|
||
@app.post("/api/notes")
|
||
async def api_notes(
|
||
request: Request,
|
||
repo_full_name: str = Form(...),
|
||
note: str = Form(""),
|
||
csrf_token: str = Form(""),
|
||
):
|
||
"""Save project notes."""
|
||
if not _csrf_valid(request, csrf_token):
|
||
return _csrf_error()
|
||
settings = get_settings()
|
||
if len(note) > settings.max_note_chars:
|
||
return JSONResponse(
|
||
{
|
||
"ok": False,
|
||
"error": f"Note too large (max {settings.max_note_chars} chars)",
|
||
},
|
||
status_code=413,
|
||
)
|
||
set_notes(repo_full_name, note)
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/api/stats")
|
||
async def api_stats():
|
||
"""Get project stats."""
|
||
return get_stats()
|
||
|
||
|
||
@app.get("/api/palette")
|
||
async def api_palette(q: str = ""):
|
||
"""Data for the Ctrl+K command palette (roadmap #10).
|
||
|
||
Projects are always returned (the palette filters them locally);
|
||
issues are searched globally once the query has 2+ characters,
|
||
annotated with their current board column.
|
||
"""
|
||
repos = await gitea.search_repos()
|
||
active, inactive, _archived = classify_repos(repos)
|
||
projects = [
|
||
{"full_name": r["full_name"]} for r in (active + inactive)[:60]
|
||
]
|
||
issues: list[dict] = []
|
||
query = q.strip()
|
||
if len(query) >= 2:
|
||
try:
|
||
found = await gitea.search_issues(query, state="all")
|
||
except Exception as exc:
|
||
logger.debug("palette search failed: %s", exc)
|
||
found = []
|
||
placements: dict[str, dict[int, str]] = {}
|
||
for item in found[:40]:
|
||
if not _is_issue(item):
|
||
continue
|
||
full = (item.get("repository") or {}).get("full_name", "")
|
||
if not full:
|
||
continue
|
||
if full not in placements:
|
||
placements[full] = {
|
||
card["issue_number"]: col
|
||
for col, cards in get_board_state(full).items()
|
||
for card in cards
|
||
}
|
||
if item.get("state") == "closed":
|
||
column = "done"
|
||
else:
|
||
column = placements[full].get(item["number"], "backlog")
|
||
issues.append(
|
||
{
|
||
"repo": full,
|
||
"number": item["number"],
|
||
"title": item.get("title", ""),
|
||
"state": item.get("state", "open"),
|
||
"column": column,
|
||
"column_label": COLUMN_LABELS[column],
|
||
}
|
||
)
|
||
if len(issues) >= 8:
|
||
break
|
||
return {"projects": projects, "issues": issues}
|
||
|
||
|
||
@app.get("/api/projects")
|
||
async def api_projects():
|
||
"""Get all projects (JSON), classified like the dashboard."""
|
||
repos = await gitea.search_repos()
|
||
active, inactive, archived = classify_repos(repos)
|
||
return {"active": active, "inactive": inactive, "archived": archived}
|
||
|
||
|
||
@app.get("/api/health")
|
||
async def health():
|
||
"""Liveness probe: the process is up. Deliberately dependency-free."""
|
||
return {"status": "ok", "version": APP_VERSION}
|
||
|
||
|
||
@app.get("/api/readiness")
|
||
async def readiness():
|
||
"""Readiness probe: database writable and Gitea reachable."""
|
||
db_ok = ping_db()
|
||
gitea_ok = await gitea.ping()
|
||
ready = db_ok and gitea_ok
|
||
return JSONResponse(
|
||
{
|
||
"status": "ready" if ready else "not_ready",
|
||
"version": APP_VERSION,
|
||
"checks": {"database": db_ok, "gitea": gitea_ok},
|
||
},
|
||
status_code=200 if ready else 503,
|
||
)
|