"""FlowDeck — Dashboard : helpers partagés des modules de routes (A28). Les 15 helpers top-level de l'ancien dashboard.py vivent ici (état : _VERSION, WORKSPACE_COOKIE) — ré-exportés par le package. """ from __future__ import annotations import logging from fastapi import HTTPException, Request from fastapi.responses import RedirectResponse from app.auth.session import SessionManager from app.config import settings from app.db import get_conn logger = logging.getLogger(__name__) _VERSION = None WORKSPACE_COOKIE = "flowdeck_workspace" def _get_app_version() -> str: """Read version from VERSION file with caching.""" global _VERSION if _VERSION is not None: return _VERSION try: import os version_path = os.path.join(os.path.dirname(__file__), "..", "..", "VERSION") if os.path.exists(version_path): with open(version_path) as f: _VERSION = f.read().strip() else: # Docker fallback version_path = "/app/VERSION" if os.path.exists(version_path): with open(version_path) as f: _VERSION = f.read().strip() else: _VERSION = "0.0.0" except Exception: _VERSION = "0.0.0" return _VERSION def _get_user_or_redirect(request: Request): """Return decoded user or a RedirectResponse to login page. Skips redirect when DB has no users (fresh install / test env).""" user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) if not user: # Allow through if no users exist yet (fresh install / tests) try: with get_conn() as conn: count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] if count == 0: return {"id": 1, "login": "admin", "full_name": "Admin", "is_admin": True} except Exception: logger.exception("_get_user_or_redirect") return RedirectResponse("/auth/login?provider=local&expired=1", status_code=302) return user def _local_workspaces_for_user(user: dict | None, workspace_id: int = 0) -> list[dict]: """Return list of local workspaces for a user.""" if not user: return [] try: with get_conn() as conn: rows = conn.execute( "SELECT id, name FROM workspaces WHERE owner_id = ? ORDER BY name", (user["id"],) ).fetchall() return [{"id": r["id"], "name": r["name"]} for r in rows] except Exception: return [] def _sidebar_data(request: Request, repos: list[dict], include_workspace: bool = True) -> dict: user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) ws = user.get("login", "Bruno") if user else "Bruno" initial = ws[0].upper() if ws else "B" # Get avatar info from DB avatar_url = "" avatar_color = "#3A3A3A" if user: try: with get_conn() as conn: row = conn.execute("SELECT avatar_url, avatar_color FROM users WHERE id = ?", (user["id"],)).fetchone() if row: avatar_url = row["avatar_url"] or "" avatar_color = row["avatar_color"] or "#3A3A3A" except Exception: logger.exception("_sidebar_data") recent_pages = [] for repo in repos[:10]: full_name = repo.get("full_name", "") recent_pages.append({ "id": full_name, "name": repo.get("name", full_name), "icon": "folder", "url": f"/board/{full_name}", "active": False, "indent": 0, "depth": 0, "has_children": False, "children": [], }) # Active workspace from cookie (skip on pages like /workspaces where no # workspace context should be shown) from app.routers.board import _load_workspace_pages ws_cookie = request.cookies.get("flowdeck_workspace", "") active_ws_name = "Workspace" workspace_pages = [] gitea_workspace = False gitea_owner = "" gitea_repo = "" has_active_workspace = False local_ws_id = 0 if ws_cookie and ws_cookie.startswith("gitea:"): # Gitea workspace: set owner/repo for client-side tree loading # AND open the local workspace mirror of the same name in the # sidebar's top "My Workspaces" section, in parallel with the # Gitea repository tree. parts = ws_cookie.split(":", 2) if len(parts) >= 3: gitea_owner = parts[1] gitea_repo = parts[2] active_ws_name = f"{gitea_owner}/{gitea_repo}" gitea_workspace = True has_active_workspace = True # Load the local mirror workspace tree so it appears in "My # Workspaces" alongside the Gitea repository section. if user: try: with get_conn() as conn: row = conn.execute( "SELECT id FROM workspaces WHERE owner_id=? AND name=? AND settings_json LIKE ?", (user["id"], f"{gitea_owner}/{gitea_repo}", "%gitea_repo%") ).fetchone() if row: local_ws_id = row["id"] workspace_pages = _load_workspace_pages(str(local_ws_id)) except (ValueError, Exception): pass elif include_workspace and ws_cookie and user: try: wsi = int(ws_cookie) with get_conn() as conn: # Verify this workspace belongs to the current user row = conn.execute( "SELECT id, name, owner_id FROM workspaces WHERE id=? AND owner_id=?", (wsi, user["id"]) ).fetchone() if row: active_ws_name = row["name"] workspace_pages = _load_workspace_pages(ws_cookie) has_active_workspace = True # v7.46.0 : exposer l'ID de l'espace ACTIF. Avant, le sidebar # ne fournissait que son nom et le bouton Home de la sidebar # retombait sur ``local_workspaces[0]`` (premier espace TRIE # PAR NOM) → clic sur Home = changement d'espace surprise. local_ws_id = wsi # else: stale cookie from another user — ignore except (ValueError, Exception): pass # Auth method & OAuth badge data auth_method = "local" gitea_linked = False github_linked = False if user and user.get("id"): try: with get_conn() as conn: am_row = conn.execute("SELECT auth_method FROM users WHERE id=?", (user["id"],)).fetchone() if am_row and am_row["auth_method"]: auth_method = am_row["auth_method"] tokens = conn.execute( "SELECT provider FROM user_oauth_tokens WHERE user_id=?", (user["id"],) ).fetchall() for t in tokens: if t["provider"] == "gitea": gitea_linked = True elif t["provider"] == "github": github_linked = True except Exception: logger.exception("_sidebar_data") # Get local workspace ID for Gitea workspace mirror (le miroir Gitea est un # espace DISTINCT : on ne doit pas écraser ``local_ws_id`` (espace actif). gitea_mirror_ws_id = 0 if gitea_workspace and gitea_owner and gitea_repo: try: with get_conn() as conn: row = conn.execute( "SELECT id FROM workspaces WHERE owner_id=? AND name=? AND settings_json LIKE ?", (user["id"], f"{gitea_owner}/{gitea_repo}", "%gitea_repo%") ).fetchone() if row: gitea_mirror_ws_id = row["id"] except Exception: logger.exception("_sidebar_data") # Private pages for mirror workspace (when Gitea remote active) private_pages = [] if gitea_workspace and gitea_mirror_ws_id: try: with get_conn() as conn: pp_rows = conn.execute( "SELECT id, title FROM pages WHERE parent_section='Private' AND workspace_id=? AND deleted_at IS NULL ORDER BY updated_at DESC LIMIT 20", (gitea_mirror_ws_id,) ).fetchall() private_pages = [{"id": r[0], "title": r[1] or "Untitled"} for r in pp_rows] except Exception: logger.exception("_sidebar_data") # Shared / received / published pages for the sidebar (Par moi / Avec moi) shared_made_pages = [] shared_received_pages = [] published_pages = [] shared_pages = [] if user and user.get("id"): from app.routers.board import _load_shared_sidebar_pages shared_made_pages, shared_received_pages, published_pages, shared_pages = _load_shared_sidebar_pages(user["id"]) sidebar = { "workspace_name": ws, "workspace_initial": initial, "active_ws_name": active_ws_name, "workspace_key": f"{gitea_owner}/{gitea_repo}" if gitea_workspace else "", "gitea_workspace": gitea_workspace, "gitea_owner": gitea_owner, "gitea_repo": gitea_repo, "local_ws_id": local_ws_id, "workspace_pages": workspace_pages, "current_page": "Dashboard", "last_edited": "now", "recent_pages": recent_pages, "private_pages": private_pages, "favorite_pages": [], "shared_pages": shared_pages, "shared_made_pages": shared_made_pages, "shared_received_pages": shared_received_pages, "published_pages": published_pages, "user": user, "avatar_url": avatar_url, "avatar_color": avatar_color, "auth_method": auth_method, "gitea_linked": gitea_linked, "github_linked": github_linked, "has_active_workspace": has_active_workspace, "app_version": _get_app_version(), } sidebar["local_workspaces"] = _local_workspaces_for_user(user) return sidebar # ═══════════ User API endpoints ═══════════ def _get_user_id(request: Request) -> int: user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) return user["id"] if user and user.get("id") else 1 def _require_user_id(request: Request) -> int: """A3/A4 — 401 sans session (les routes qui mutent un compte ne tolèrent plus le fallback « legacy single-user » → id 1 = l'admin seedé).""" user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) if not user or not user.get("id"): raise HTTPException(401, "Authentication required") return user["id"] def _file_page_disk_path(page: dict): """Resolve the on-disk file behind a ``content_format == 'file'`` page. Returns ``(abs_path: Path, filename: str, mime: str, size: int)`` or None when the row is not a file page, references a non-textual/missing file, or the path escapes the data root (path-traversal guard). """ if (page.get("content_format") or "") != "file": return None import json as _json try: meta = _json.loads(page.get("content", "{}")) except (_json.JSONDecodeError, TypeError): meta = {} if not isinstance(meta, dict): return None rel = (meta.get("file_path") or "").replace("\\", "/").strip() if not rel or not rel.startswith("uploads/"): return None parts = rel.split("/") if ".." in parts or "." in parts: return None from pathlib import Path root = Path(settings.data_dir).resolve() full = (root / rel).resolve() try: full.relative_to(root) except ValueError: return None if not full.exists() or not full.is_file(): return None filename = parts[-1] or page.get("title", "file") mime = meta.get("mime_type") or "application/octet-stream" size = meta.get("size") or 0 return (full, filename, mime, size) def _require_page_view(request: Request, page_id: int) -> None: """A16 : lecture d'une pièce jointe = session + `can_view_page` (404 sinon).""" user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) if not user or not user.get("id"): raise HTTPException(401, "Authentication required") from app.services.permission_manager import PermissionManager if not PermissionManager(user["id"], bool(user.get("is_admin"))).can_view_page(page_id): raise HTTPException(404, "Page not found") def _build_tree_children(conn, parent_id: int | None, ws_id: int, uid: int | None = None) -> list: """Recursively build the tree of children for a node.""" if parent_id is None: rows = conn.execute( "SELECT id, title, parent_section, content_format, content, page_icon, " "(is_shared OR share_mode != 'private' OR COALESCE(published,0)) as is_shared, " "created_at, updated_at FROM pages " "WHERE parent_id IS NULL AND workspace_id=? AND deleted_at IS NULL AND collection_row_id IS NULL " "ORDER BY created_at DESC", (ws_id,), ).fetchall() else: rows = conn.execute( "SELECT id, title, parent_section, content_format, content, page_icon, " "(is_shared OR share_mode != 'private' OR COALESCE(published,0)) as is_shared, " "created_at, updated_at FROM pages " "WHERE parent_id=? AND workspace_id=? AND deleted_at IS NULL " "ORDER BY created_at DESC", (parent_id, ws_id), ).fetchall() # Get workspace owner name for author display ws_owner = conn.execute( "SELECT u.full_name, u.login FROM workspaces w JOIN users u ON u.id=w.owner_id WHERE w.id=?", (ws_id,), ).fetchone() author = ws_owner["full_name"] or ws_owner["login"] if ws_owner else "—" # Collect all page IDs to fetch tags in one query all_ids = [r["id"] for r in rows] tags_map = {} favorited_ids = set() if all_ids: placeholders = ",".join("?" for _ in all_ids) tag_rows = conn.execute( f"SELECT pt.page_id, t.id, t.name, t.color FROM page_tags pt " f"JOIN tags t ON t.id=pt.tag_id WHERE pt.page_id IN ({placeholders})", all_ids, ).fetchall() for tr in tag_rows: tags_map.setdefault(tr["page_id"], []).append({ "id": tr["id"], "name": tr["name"], "color": tr["color"], }) if uid is not None: fav_rows = conn.execute( f"SELECT page_id FROM favorites WHERE user_id=? AND page_id IN ({placeholders})", [uid, *all_ids], ).fetchall() favorited_ids = {fr["page_id"] for fr in fav_rows} tree = [] for r in rows: is_folder = r["parent_section"] == "Workspace" children = _build_tree_children(conn, r["id"], ws_id, uid) # Compute size size = 0 if r["content_format"] == "file": import json as _json try: meta = _json.loads(r["content"]) size = meta.get("size", 0) except Exception: size = len(r["content"] or "") else: size = len(r["content"] or "") tree.append({ "id": r["id"], "name": r["title"] or "Untitled", "type": "folder" if is_folder else "page", "is_folder": is_folder, "content_format": r["content_format"] if not is_folder else None, "page_icon": r["page_icon"] or "", "children": children, "has_children": len(children) > 0, "child_count": len(children), "size": size, "size_display": _format_size(size), "created_at": r["created_at"], "updated_at": r["updated_at"], "author": author, "tags": tags_map.get(r["id"], []), "is_shared": bool(r["is_shared"]), "favorited": r["id"] in favorited_ids, }) return tree def _format_size(size_bytes: int) -> str: """Human-readable file size.""" if size_bytes < 1024: return f"{size_bytes} B" elif size_bytes < 1024 * 1024: return f"{size_bytes / 1024:.1f} KB" elif size_bytes < 1024 * 1024 * 1024: return f"{size_bytes / (1024 * 1024):.1f} MB" return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB" def _build_breadcrumb(conn, folder_id: int) -> list: """Build breadcrumb trail from root to folder_id.""" breadcrumb = [] current = folder_id seen = set() while current and current not in seen: seen.add(current) row = conn.execute( "SELECT id, title, parent_id, parent_section FROM pages WHERE id=?", (current,), ).fetchone() if row: breadcrumb.insert(0, { "id": row["id"], "name": row["title"] or "Untitled", "is_folder": row["parent_section"] == "Workspace", }) current = row["parent_id"] else: break return breadcrumb def _nav_breadcrumb(conn, page_id: int) -> list: """Build a Notion-style breadcrumb chain (root -> page) for the header. Returns a list of dicts: {id, label, url, icon, menu}. The last item is the current page (url = None). Every item has ``menu: True`` so the header can open a sibling-navigation dropdown for it. """ from app.routers.board import _file_icon chain = [] current = page_id seen = set() while current and current not in seen: seen.add(current) row = conn.execute( "SELECT id, title, parent_id, parent_section, content_format " "FROM pages WHERE id=? AND deleted_at IS NULL", (current,), ).fetchone() if not row: break is_folder = row["parent_section"] == "Workspace" title = row["title"] or "Untitled" chain.insert(0, { "id": row["id"], "label": title, "url": None, "icon": "folder" if is_folder else _file_icon(title, row["content_format"]), "menu": True, }) current = row["parent_id"] # All items except the current page are navigable links. for i, item in enumerate(chain): if i < len(chain) - 1: item["url"] = f"/pages/{item['id']}" return chain def _get_active_workspace(request: Request, user_id: int = None) -> dict | None: """Get the active workspace ID from the cookie (verified for current user), or first user workspace, or None.""" ws_id = request.cookies.get(WORKSPACE_COOKIE) if ws_id: try: with get_conn() as conn: ws = conn.execute("SELECT * FROM workspaces WHERE id=?", (int(ws_id),)).fetchone() if ws: ws_dict = dict(ws) # Verify ownership — only return if it belongs to the current user if user_id is None or ws_dict.get("owner_id") == user_id: return ws_dict except (ValueError, Exception): pass # Fallback: first workspace owned by this user if user_id: with get_conn() as conn: ws = conn.execute( "SELECT * FROM workspaces WHERE owner_id=? ORDER BY id LIMIT 1", (user_id,) ).fetchone() if ws: return dict(ws) return None def _sanitize_id(block_id: str) -> str: """Sanitize a block id for use as an HTML anchor (only alnum kept).""" if not block_id: return "" return "".join(ch for ch in str(block_id) if ch.isalnum()) def _render_blocks_public(blocks: list, titles: dict | None = None) -> str: """Render FlowDeck blocks as plain HTML for public pages. v5.11.0: ``titles`` (token → label, see app.services.wiki_links) turns ``[[fdpage:ID]]`` / ``[[fddate:...]]`` tokens into chips/links. """ html_parts = [] def _wiki(c: str) -> str: if titles and ("[[fdpage:" in c or "[[fddate:" in c): from app.services.wiki_links import resolve_tokens_html return resolve_tokens_html(c, titles) return c for b in blocks: t = b.get("type", "paragraph") c = _wiki(b.get("content", "") or "") if t == "heading_1": html_parts.append(f'

{c}

') elif t == "heading_2": html_parts.append(f'

{c}

') elif t == "heading_3": html_parts.append(f'

{c}

') elif t == "heading_4": html_parts.append(f'

{c}

') elif t == "bulleted_list": html_parts.append(f'
  • {c}
  • ') elif t == "numbered_list": html_parts.append(f'
  • {c}
  • ') elif t == "to_do": checked = "checked" if b.get("checked") else "" todo_style = "text-decoration:line-through;opacity:.5" if b.get("checked") else "" html_parts.append( f'
    ' f'' f'{c}' f'
    ' ) elif t == "toggle": children_html = "" if b.get("children"): children_html = '
    ' children_html += _render_blocks_public(b["children"], titles) children_html += "
    " html_parts.append( f'
    {c}{children_html}
    ' ) elif t == "quote": html_parts.append( f'
    {c}
    ' ) elif t == "table_of_contents": toc = [ x for x in blocks if x.get("type", "").startswith("heading_") and (x.get("content") or "").strip() ] if toc: items = [] for h in toc: lvl = int(h["type"].split("_")[-1]) items.append( f'
    ' f'{h.get("content","")}
    ' ) html_parts.append( '
    ' '
    On this page
    ' + "".join(items) + "
    " ) elif t == "math": tex = c.replace("&", "&").replace("<", "<").replace(">", ">") html_parts.append( f'
    ' ) elif t == "columns": cols_html = "" for child in b.get("children") or []: cols_html += ( '
    ' + _render_blocks_public([child], titles) + "
    " ) html_parts.append( f'
    {cols_html}
    ' ) elif t == "callout": icon = b.get("icon", "💡") bg = (b.get("style") or {}).get("bgColor", "rgba(76,154,255,.1)") html_parts.append( f'
    ' f'{icon}' f'{c}
    ' ) elif t == "code": lang = b.get("language", "") lang_label = f"
    {lang}
    " if lang else "" html_parts.append( f'
    '
                    f'{lang_label}'
                    f'{c}
    ' ) elif t == "divider": html_parts.append('
    ') elif t == "image": src = b.get("src", "") alt = b.get("alt", "") html_parts.append( f'
    ' f'{alt}' f'
    ' ) elif t == "video": src = b.get("src", "") if src: html_parts.append( f'' ) elif t == "audio": src = b.get("src", "") if src: html_parts.append( f'' ) elif t == "bookmark": url = b.get("url") or b.get("src") or "" title = b.get("title") or url desc = b.get("description") or "" img = b.get("image") or "" site = b.get("site_name") or "" img_html = ( f'' if img else "" ) desc_html = f'
    {desc}
    ' if desc else "" site_html = f'
    {site}
    ' if site else "" html_parts.append( f'' f'
    ' f'
    {title}
    ' f'{desc_html}{site_html}
    {img_html}
    ' ) elif t == "embed": url = b.get("src", "") emb = b.get("embed_type") or "" if emb in ("inline_dbs", "collection"): html_parts.append('
    [Embedded content]
    ') elif emb == "download": html_parts.append( f'⬇ {b.get("file_name") or "Download"}' ) elif emb == "pdf" and url: html_parts.append( f'' ) elif url: from app.services.embeds import embed_src src = b.get("embed_src") or embed_src(url) or url height = b.get("height") or 520 try: height = int(height) except (ValueError, TypeError): height = 520 html_parts.append( f'
    ' ) elif t == "synced": # v6.5.0: render synced block instances (resolved server-side). if b.get("_synced_deleted"): html_parts.append( '
    ' 'Deleted synced block
    ' ) else: inner = b.get("_synced_content") if not isinstance(inner, list) or not inner: try: import json as _sj parsed = _sj.loads(b.get("content") or "[]") inner = parsed if isinstance(parsed, list) else [] except (ValueError, TypeError): inner = [] inner = [{"type": "paragraph", "content": str(x)} if not isinstance(x, dict) else x for x in inner] if inner: html_parts.append( '
    ' + _render_blocks_public(inner, titles) + '
    ' ) else: html_parts.append(f'

    {c}

    ') return "\n".join(html_parts) # ═══════════ Library page actions API ═══════════