"""FlowDeck — Board : board_views. Découpe A28 de l'ancien app/routers/board.py (2 101 lignes, 53 routes) — un module par concern, contrat inchangé. """ from __future__ import annotations import json import logging from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import HTMLResponse from app.db import get_conn from app.services.gitea_client import gitea from ._common import ( STATUS_COLORS, STATUS_LABELS, _apply_filters, _apply_sorts, _extract_ai_keywords, _get_project_properties, _map_issue_to_card, _sidebar_data, ) logger = logging.getLogger(__name__) router = APIRouter(tags=["board"], prefix="/board") # ═══════════ Board page ═══════════ @router.get("/{owner}/{repo}", response_class=HTMLResponse) def board(request: Request, owner: str, repo: str, view: str = Query(default="")): from app.auth.session import SessionManager from app.routers.dashboard._common import _cleanup_empty_mirrors, _ensure_forge_mirror from app.templating import ENV env = ENV # Selecting a repository also opens a workspace-local of the same name: # ensure the mirror exists, prune other empty mirrors. user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) if user and owner and repo: try: mid = _ensure_forge_mirror(user["id"], owner, repo, forge="gitea") _cleanup_empty_mirrors(user["id"], keep_name=f"{owner}/{repo}", keep_id=mid or 0) except Exception: logger.exception("board mirror ensure") sidebar = _sidebar_data(request, owner, repo) template = env.get_template("board.html") return template.render(request=request, owner=owner, repo=repo, groups=[], initial_view=view, **sidebar) # ═══════════ View fragments ═══════════ # ═══════════ 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=""), ): 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, owner, repo) 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 app.templating import ENV env = ENV # Dynamic groups from Gitea labels (fallback to hardcoded) 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} template_map = { "table": "table_view.html", "status": "status_overview.html", "teamload": "team_load.html", "detailed": "detailed_board.html", } if view == "table": grouped = {g["name"]: g["cards"] for g in groups} ctx["grouped_cards"] = grouped elif view == "status": counts = {"todo": 0, "progress": 0, "done": 0} for c in cards: if c["status"] in counts: counts[c["status"]] += 1 ctx.update(status_data=counts, status_colors=STATUS_COLORS, status_labels=STATUS_LABELS) 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} sk = c["status"] if c["status"] in ("todo", "progress") else "complete" members[name][sk] += 1 members[name]["total"] += 1 ctx["team_data"] = list(members.values()) elif view == "detailed": pass else: template_map["kanban"] = "board_fragment.html" template_name = template_map.get(view, "board_fragment.html") template = env.get_template(template_name) return template.render(**ctx) # ═══════════ v0.9.0: Custom Properties API ═══════════ # ═══════════ v0.9.0: Custom Properties API ═══════════ @router.get("/api/properties/{owner}/{repo}") def get_properties(owner: str, repo: str): return {"properties": _get_project_properties(owner, repo)} @router.post("/api/properties/{owner}/{repo}") def create_property(owner: str, repo: str, name: str = Query(...), prop_type: str = Query(default="select"), options: str = Query(default="")): opts = json.dumps([o.strip() for o in options.split(",") if o.strip()]) with get_conn() as conn: try: conn.execute( "INSERT INTO project_properties (project_owner, project_name, name, prop_type, options_json) VALUES (?,?,?,?,?)", (owner, repo, name, prop_type, opts), ) conn.commit() except Exception as e: raise HTTPException(409, f"Property already exists: {e}") from e return {"status": "ok", "name": name, "type": prop_type} @router.delete("/api/properties/{owner}/{repo}") def delete_property(owner: str, repo: str, name: str = Query(...)): with get_conn() as conn: conn.execute( "DELETE FROM project_properties WHERE project_owner=? AND project_name=? AND name=?", (owner, repo, name), ) conn.commit() return {"status": "ok"} @router.post("/api/properties/{owner}/{repo}/values") def set_property_value(owner: str, repo: str, issue_id: int = Query(...), name: str = Query(...), value: str = Query(default="")): with get_conn() as conn: prop = conn.execute( "SELECT id FROM project_properties WHERE project_owner=? AND project_name=? AND name=?", (owner, repo, name), ).fetchone() if not prop: raise HTTPException(404, f"Property '{name}' not found") conn.execute( "INSERT OR REPLACE INTO property_values (property_id, gitea_issue_id, value) VALUES (?,?,?)", (prop["id"], issue_id, value), ) conn.commit() return {"status": "ok"} # ═══════════ v0.9.0: AI Keywords API ═══════════ # ═══════════ v0.9.0: AI Keywords API ═══════════ @router.get("/api/ai-keywords/{owner}/{repo}") def get_ai_keywords(owner: str, repo: str): with get_conn() as conn: rows = conn.execute( "SELECT keyword, color, usage_count FROM ai_keywords WHERE project_owner=? AND project_name=? ORDER BY usage_count DESC LIMIT 30", (owner, repo), ).fetchall() return {"keywords": [dict(r) for r in rows]} @router.post("/api/ai-keywords/{owner}/{repo}/extract") async def extract_ai_keywords(owner: str, repo: str): """Re-extract keywords from all issues in the repo.""" try: issues = await gitea.get_issues(owner, repo, state="all") for issue in issues: if not issue.get("pull_request"): _extract_ai_keywords(owner, repo, issue.get("labels", []), issue.get("body", "")) except Exception as e: raise HTTPException(500, str(e)) from e return {"status": "ok", "issues_scanned": len(issues)} # ═══════════ Pages Markdown ═══════════