"""FlowDeck — Board Kanban Notion-style."""
from __future__ import annotations
import json
import logging
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import HTMLResponse
from app.db import get_conn
from app.services.gitea_client import gitea
from app.auth.session import SessionManager
logger = logging.getLogger(__name__)
router = APIRouter(tags=["board"], prefix="/board")
def _issue_column(issue: dict, columns: list[str], board_id: int) -> str:
"""Determine which column an issue belongs to (used by api.py)."""
if issue.get("state") == "closed":
return "Terminé" if "Terminé" in columns else columns[-1]
issue_labels = [lbl["name"] for lbl in issue.get("labels", [])]
with get_conn() as conn:
for lbl in issue_labels:
row = conn.execute(
"SELECT column_name FROM col_mapping WHERE board_id=? AND gitea_label=?",
(board_id, lbl),
).fetchone()
if row and row["column_name"] in columns:
return row["column_name"]
return columns[0] if columns else "Backlog"
STATUS_MAP = {
"todo": {"label": "To-do", "color": "var(--gray)", "css": "card-status-todo"},
"progress": {"label": "In progress", "color": "var(--blue)", "css": "card-status-progress"},
"done": {"label": "Complete", "color": "var(--green)", "css": "card-status-complete"},
}
DEFAULT_GROUPS = ["Design", "Engineering", "No Team"]
def _ensure_board(owner: str, repo: str) -> dict:
with get_conn() as conn:
row = conn.execute(
"SELECT id, columns_json FROM boards WHERE project_owner=? AND project_name=?",
(owner, repo),
).fetchone()
if not row:
cur = conn.execute(
"INSERT INTO boards (project_owner, project_name) VALUES (?, ?)",
(owner, repo),
)
conn.commit()
return {"id": cur.lastrowid, "columns": ["Backlog", "À faire", "En cours", "Révision", "Terminé"]}
return {"id": row["id"], "columns": json.loads(row["columns_json"])}
def _map_issue_to_card(issue: dict) -> dict:
"""Convert a Gitea issue to a Notion-style card."""
title = issue.get("title", "Untitled")
status = "todo"
if issue.get("state") == "closed":
status = "done"
labels = issue.get("labels", [])
for lbl in labels:
name = lbl.get("name", "").lower()
if "progress" in name or "doing" in name or "in progress" in name:
status = "progress"
elif "done" in name or "complete" in name or "terminé" in name:
status = "done"
assignee = issue.get("assignee", {})
assignee_name = assignee.get("login", "") if assignee else ""
tag = labels[0].get("name", "") if labels else ""
tag_color = labels[0].get("color", "#787774") if labels else "#787774"
if tag_color and not tag_color.startswith("#"):
tag_color = f"#{tag_color}"
icon_map = {
"bug": "🐛", "feature": "✨", "enhancement": "⚡", "documentation": "📄",
"design": "🎨", "testing": "🧪", "refactor": "🔧", "security": "🔒",
}
icon = "📄"
for lbl in labels:
for keyword, emoji in icon_map.items():
if keyword in lbl.get("name", "").lower():
icon = emoji
break
return {
"id": str(issue.get("number", 0)),
"title": title,
"status": status,
"icon": icon,
"assignee": assignee_name,
"tag": tag if tag else None,
"tag_color": tag_color,
"due_date": issue.get("due_date", ""),
"url": issue.get("html_url", ""),
}
def _sidebar_data(request: Request, owner: str, repo: str) -> dict:
user = SessionManager.decode_session(request.cookies.get("flowdeck_session", ""))
workspace_name = user.get("login", "Bruno") if user else "Bruno"
workspace_initial = workspace_name[0].upper() if workspace_name else "B"
recent_pages = [{
"id": f"{owner}/{repo}",
"name": repo,
"icon": "📁",
"url": f"/board/{owner}/{repo}",
"active": True,
"indent": 0,
}]
# Sub-views as indented items
sub_views = ["Detailed board", "Table view", "Status overview", "Team Load"]
for sv in sub_views:
recent_pages.append({
"id": f"{owner}/{repo}/{sv.lower().replace(' ', '-')}",
"name": sv,
"icon": "·",
"url": f"/board/{owner}/{repo}",
"active": False,
"indent": 1,
})
return {
"workspace_name": workspace_name,
"workspace_initial": workspace_initial,
"current_page": repo,
"last_edited": "2m ago",
"recent_pages": recent_pages,
"private_pages": recent_pages[:1],
"user": user,
}
@router.get("/{owner}/{repo}", response_class=HTMLResponse)
async def board(
request: Request,
owner: str,
repo: str,
milestone: str = "",
label: str = "",
assignee: str = "",
fragment: bool = False,
):
"""Kanban board Notion-style for a Gitea project."""
try:
issues = await gitea.get_issues(owner, repo, state="all")
labels_list = await gitea.get_labels(owner, repo)
milestones = await gitea.get_milestones(owner, repo)
if milestone:
issues = [i for i in issues if i.get("milestone", {}).get("title") == milestone]
if label:
issues = [i for i in issues if any(lbl["name"] == label for lbl in i.get("labels", []))]
if assignee:
issues = [i for i in issues if i.get("assignee", {}).get("login") == assignee]
issues_only = [i for i in issues if not i.get("pull_request")]
# Convert to Notion-style cards
cards = [_map_issue_to_card(i) for i in issues_only]
# Build groups
group_names = ["Design", "Engineering", "No Team"]
groups = []
for gname in group_names:
group_cards = [c for c in cards] # For now, all cards in all groups
counts = {
"todo": len([c for c in group_cards if c["status"] == "todo"]),
"progress": len([c for c in group_cards if c["status"] == "progress"]),
"done": len([c for c in group_cards if c["status"] == "done"]),
}
groups.append({
"id": gname.lower().replace(" ", "-"),
"name": gname,
"counts": counts,
"cards": group_cards,
})
except Exception as e:
from httpx import HTTPStatusError
if isinstance(e, HTTPStatusError) and e.response is not None and e.response.status_code == 404:
return HTMLResponse(
'',
status_code=404)
logger.error("Board error: %s", e)
return HTMLResponse(f"Erreur: {e}
", status_code=500)
from jinja2 import Environment, FileSystemLoader
from datetime import datetime
env = Environment(loader=FileSystemLoader("app/templates"))
sidebar = _sidebar_data(request, owner, repo)
now_str = datetime.now().isoformat()[:10]
ctx = dict(
request=request, owner=owner, repo=repo,
groups=groups, labels=labels_list, milestones=milestones,
**sidebar,
)
template = env.get_template("board.html")
return template.render(**ctx)