"""FlowDeck — GiteaBoardCompat: legacy board → Collection adapter (v1.3.0). Converts existing Gitea-linked boards and cards into the new collection model without breaking the existing board routes. """ from __future__ import annotations import json from app.db import get_conn class GiteaBoardCompat: """Adaptateur: board Gitea legacy → Collection. Permet de visualiser les boards Gitea existants comme des collections, avec rétrocompatibilité totale. Les anciennes routes /board/... continuent de fonctionner normalement. """ @staticmethod def from_board(board_row) -> dict: """Convertit un board existant en pseudo-collection.""" columns = json.loads(board_row.get("columns_json", "[]")) return { "collection_id": board_row["id"], "name": f"{board_row['project_owner']}/{board_row['project_name']}", "description": f"Board Gitea: {board_row['project_name']}", "icon": "📋", "gitea_owner": board_row["project_owner"], "gitea_repo": board_row["project_name"], "is_gitea_linked": True, "is_locked": False, "schema": [ {"name": "Title", "type": "title"}, {"name": "Status", "type": "select", "options": [ {"name": c, "color": "gray"} for c in columns ]}, {"name": "Priority", "type": "select", "options": [ {"name": "P1", "color": "red"}, {"name": "P2", "color": "orange"}, {"name": "P3", "color": "yellow"}, {"name": "P4", "color": "gray"}, ]}, {"name": "Due Date", "type": "date"}, {"name": "Assignee", "type": "person"}, ], } @staticmethod def from_card(card_row, gitea_issue: dict | None = None) -> dict: """Convertit une card legacy en pseudo collection_page.""" title = gitea_issue.get("title", f"Card #{card_row['id']}") if gitea_issue else f"Card #{card_row['id']}" priority = card_row.get("priority", "Medium") due_date = card_row.get("due_date", "") props = { "Status": card_row.get("column_name", "Backlog"), "Priority": priority, "Due Date": due_date if due_date else None, } if gitea_issue: assignee = gitea_issue.get("assignee") if assignee: props["Assignee"] = [{"login": assignee.get("login", ""), "avatar_url": assignee.get("avatar_url", "")}] return { "id": card_row["id"], "title": title, "position": card_row.get("position", 0), "gitea_issue_id": card_row.get("gitea_issue_id"), "gitea_issue_number": gitea_issue.get("number") if gitea_issue else None, "property_values_json": json.dumps(props), } @staticmethod def list_boards_as_collections() -> list[dict]: """Liste tous les boards comme des collections compatibles.""" with get_conn() as conn: rows = conn.execute("SELECT * FROM boards ORDER BY project_owner, project_name").fetchall() return [GiteaBoardCompat.from_board(dict(r)) for r in rows] @staticmethod def get_board_as_collection(owner: str, repo: str) -> dict | None: """Récupère un board spécifique comme collection.""" with get_conn() as conn: row = conn.execute( "SELECT * FROM boards WHERE project_owner=? AND project_name=?", (owner, repo), ).fetchone() if not row: return None return GiteaBoardCompat.from_board(dict(row)) @staticmethod def get_board_cards(owner: str, repo: str, gitea_issues: list[dict] | None = None) -> list[dict]: """Récupère les cartes d'un board comme collection_pages.""" with get_conn() as conn: board = conn.execute( "SELECT id FROM boards WHERE project_owner=? AND project_name=?", (owner, repo), ).fetchone() if not board: return [] cards = conn.execute( "SELECT * FROM cards WHERE board_id=? ORDER BY position", (board["id"],), ).fetchall() issues_by_id = {} if gitea_issues: issues_by_id = {i.get("id"): i for i in gitea_issues} return [ GiteaBoardCompat.from_card(dict(c), issues_by_id.get(c["gitea_issue_id"])) for c in cards ] @staticmethod def sync_to_collection(owner: str, repo: str, gitea_issues: list[dict]) -> int | None: """Sync un board Gitea vers une vraie collection. Crée ou met à jour une collection liée à Gitea et importe les pages. Retourne l'ID de la collection. """ with get_conn() as conn: board = conn.execute( "SELECT * FROM boards WHERE project_owner=? AND project_name=?", (owner, repo), ).fetchone() if not board: return None board_dict = dict(board) columns = json.loads(board_dict.get("columns_json", "[]")) # Upsert collection existing = conn.execute( "SELECT id FROM collections WHERE gitea_owner=? AND gitea_repo=?", (owner, repo), ).fetchone() if existing: coll_id = existing["id"] conn.execute( "UPDATE collections SET name=?, schema_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (f"{owner}/{repo}", json.dumps([ {"name": "Title", "type": "title"}, {"name": "Status", "type": "select", "options": [{"name": c, "color": "gray"} for c in columns]}, {"name": "Priority", "type": "select", "options": [{"name": "P1", "color": "red"}, {"name": "P2", "color": "orange"}, {"name": "P3", "color": "yellow"}, {"name": "P4", "color": "gray"}]}, ]), coll_id), ) else: cur = conn.execute( """INSERT INTO collections (name, icon, schema_json, gitea_owner, gitea_repo) VALUES (?, '📋', ?, ?, ?)""", (f"{owner}/{repo}", json.dumps([ {"name": "Title", "type": "title"}, {"name": "Status", "type": "select", "options": [{"name": c, "color": "gray"} for c in columns]}, {"name": "Priority", "type": "select", "options": [{"name": "P1", "color": "red"}, {"name": "P2", "color": "orange"}, {"name": "P3", "color": "yellow"}, {"name": "P4", "color": "gray"}]}, ]), owner, repo), ) coll_id = cur.lastrowid # Create default view conn.execute( "INSERT INTO collection_views (collection_id, name, view_type, config_json) VALUES (?, ?, ?, ?)", (coll_id, "Default View", "table", json.dumps({"visible_properties": ["Title", "Status", "Priority"]})), ) # Sync cards → collection_pages cards = conn.execute( "SELECT * FROM cards WHERE board_id=? ORDER BY position", (board_dict["id"],), ).fetchall() issues_by_id = {i.get("id"): i for i in gitea_issues} for card in cards: card_dict = dict(card) issue = issues_by_id.get(card_dict.get("gitea_issue_id")) title = issue.get("title", f"Issue #{card_dict.get('gitea_issue_id', card_dict['id'])}") if issue else f"Card #{card_dict['id']}" props = { "Status": card_dict.get("column_name", "Backlog"), "Priority": card_dict.get("priority", "Medium"), } # Upsert collection_page existing_page = conn.execute( "SELECT id FROM collection_pages WHERE collection_id=? AND gitea_issue_id=?", (coll_id, card_dict.get("gitea_issue_id")), ).fetchone() if existing_page: conn.execute( """UPDATE collection_pages SET title=?, position=?, property_values_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=?""", (title, card_dict.get("position", 0), json.dumps(props), existing_page["id"]), ) else: conn.execute( """INSERT INTO collection_pages (collection_id, title, position, gitea_issue_id, gitea_issue_number, property_values_json) VALUES (?, ?, ?, ?, ?, ?)""", (coll_id, title, card_dict.get("position", 0), card_dict.get("gitea_issue_id"), issue.get("number") if issue else None, json.dumps(props)), ) conn.commit() return coll_id