"""FlowDeck β€” Board Kanban Notion-style + multi-vues.""" from __future__ import annotations import json import logging from fastapi import APIRouter, Request, HTTPException, Query 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") STATUS_COLORS = { "todo": "var(--gray)", "progress": "var(--blue)", "done": "var(--green)", } STATUS_LABELS = { "todo": "To-do", "progress": "In progress", "done": "Complete", } def _issue_column(issue: dict, columns: list[str], board_id: int) -> str: 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" def _map_issue_to_card(issue: dict) -> dict: 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: status = "progress" elif "done" in name or "complete" in name or "terminΓ©" in name: status = "done" assignee = issue.get("assignee", {}) or {} assignee_name = assignee.get("login", "") 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 kw, emoji in icon_map.items(): if kw in lbl.get("name", "").lower(): icon = emoji break return { "id": str(issue.get("number", 0)), "title": title, "status": status, "status_color": STATUS_COLORS.get(status, "var(--gray)"), "status_label": STATUS_LABELS.get(status, "To-do"), "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", ""), "keywords": [lbl.get("name", "") for lbl in labels if lbl.get("name") != tag], } 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, }] for sv in ["Detailed board", "Table view", "Status overview", "Team Load"]: 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, } def _apply_filters(cards: list[dict], status_filter: str, filters: str) -> list[dict]: if status_filter: allowed = set(status_filter.split(",")) cards = [c for c in cards if c["status"] in allowed] if filters: for f in filters.split(","): if ":" in f: prop, val = f.split(":", 1) if prop == "assignee": cards = [c for c in cards if c.get("assignee", "").lower() == val.lower()] elif prop == "tag": cards = [c for c in cards if c.get("tag", "").lower() == val.lower()] return cards def _apply_sorts(cards: list[dict], sorts: str) -> list[dict]: if not sorts: return cards for sort_spec in reversed(sorts.split(",")): if ":" not in sort_spec: continue field, direction = sort_spec.split(":", 1) reverse = direction == "desc" if field == "name": cards.sort(key=lambda c: c["title"].lower(), reverse=reverse) elif field == "status": order = {"todo": 0, "progress": 1, "done": 2} cards.sort(key=lambda c: order.get(c["status"], 0), reverse=reverse) elif field == "assignee": cards.sort(key=lambda c: c.get("assignee", "").lower(), reverse=reverse) elif field == "deadline": cards.sort(key=lambda c: c.get("due_date", ""), reverse=reverse) return cards # ── Board page ── @router.get("/{owner}/{repo}", response_class=HTMLResponse) async def board(request: Request, owner: str, repo: str): from jinja2 import Environment, FileSystemLoader env = Environment(loader=FileSystemLoader("app/templates")) sidebar = _sidebar_data(request, owner, repo) template = env.get_template("board.html") return template.render(request=request, owner=owner, repo=repo, groups=[], **sidebar) # ── View fragments ── @router.get("/{owner}/{repo}/view/{view}", response_class=HTMLResponse) async def board_view( request: Request, owner: str, repo: str, view: str, status: str = Query(default=""), filter: str = Query(default=""), sort: str = Query(default=""), ): """Return a view fragment (Kanban, Table, Status Overview, Team Load, Detailed).""" try: issues = await gitea.get_issues(owner, repo, state="all") issues_only = [i for i in issues if not i.get("pull_request")] cards = [_map_issue_to_card(i) for i in issues_only] cards = _apply_filters(cards, status, filter) cards = _apply_sorts(cards, sort) except Exception as e: logger.error("Board view error: %s", e) cards = [] from jinja2 import Environment, FileSystemLoader env = Environment(loader=FileSystemLoader("app/templates")) group_names = ["Design", "Engineering", "No Team"] groups = [] for gname in group_names: gid = gname.lower().replace(" ", "-") gcards = cards groups.append({ "id": gid, "name": gname, "counts": { "todo": len([c for c in gcards if c["status"] == "todo"]), "progress": len([c for c in gcards if c["status"] == "progress"]), "done": len([c for c in gcards if c["status"] == "done"]), }, "cards": gcards, }) ctx = {"owner": owner, "repo": repo, "groups": groups, "cards": cards} if view == "table": grouped = {} for g in groups: grouped[g["name"]] = g["cards"] ctx["grouped_cards"] = grouped template = env.get_template("table_view.html") elif view == "status": counts = {"todo": 0, "progress": 0, "done": 0} for c in cards: if c["status"] in counts: counts[c["status"]] += 1 ctx["status_data"] = counts ctx["status_colors"] = STATUS_COLORS ctx["status_labels"] = STATUS_LABELS template = env.get_template("status_overview.html") elif view == "teamload": members = {} for c in cards: name = c.get("assignee") or "Unassigned" if name not in members: members[name] = {"name": name, "initial": name[0].upper(), "todo": 0, "progress": 0, "complete": 0, "total": 0} status_key = c["status"] if c["status"] in ("todo", "progress") else "complete" members[name][status_key] += 1 members[name]["total"] += 1 ctx["team_data"] = list(members.values()) template = env.get_template("team_load.html") elif view == "detailed": template = env.get_template("detailed_board.html") else: template = env.get_template("board_fragment.html") return template.render(**ctx) # ── Kanban fragment (default) ── @router.get("/{owner}/{repo}/view/kanban", response_class=HTMLResponse) async def board_kanban(request: Request, owner: str, repo: str): return await board_view(request, owner, repo, "kanban")