"""FlowDeck — Web Clipper router (v6.0.0). Endpoints: GET /api/v2/web-clipper/status POST /api/v2/web-clipper/auth/verify POST /api/v2/web-clipper/clip GET /api/v2/web-clipper/devices DELETE /api/v2/web-clipper/devices/{id} GET /extensions (HTML download page) Auth: session cookie OR Bearer api_token OR Bearer extension device token. """ from __future__ import annotations import hashlib import logging from fastapi import APIRouter, HTTPException, Request from fastapi.responses import HTMLResponse from app.auth.session import SessionManager from app.db import get_conn from app.services.web_clipper import ( MAX_CLIP_BYTES, _check_rate_limit, create_page_from_clip, list_devices, log_clip, register_device, revoke_device, sanitize_html, ) logger = logging.getLogger(__name__) router = APIRouter(tags=["web-clipper"]) api_router = APIRouter(prefix="/api/v2/web-clipper", tags=["web-clipper"]) def _hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() def _user_from_request(request: Request) -> dict | None: # 1) session cookie user = SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) if user: return user # 2) Authorization Bearer auth = request.headers.get("authorization") or request.headers.get("Authorization") or "" if auth.lower().startswith("bearer "): token = auth[7:].strip() if not token: return None th = _hash(token) with get_conn() as conn: # api_tokens (Settings → API tokens) row = conn.execute( "SELECT user_id FROM api_tokens WHERE token_hash=? AND revoked=0", (th,) ).fetchone() if row: u = conn.execute("SELECT id, login, full_name, email, is_admin FROM users WHERE id=?", (row["user_id"],)).fetchone() if u: return dict(u) # extension_devices row = conn.execute( "SELECT user_id FROM extension_devices WHERE token_hash=? AND revoked=0", (th,) ).fetchone() if row: u = conn.execute("SELECT id, login, full_name, email, is_admin FROM users WHERE id=?", (row["user_id"],)).fetchone() if u: return dict(u) # legacy user_tokens row = conn.execute("SELECT gitea_user_id FROM user_tokens WHERE gitea_token=?", (token,)).fetchone() if row: u = conn.execute("SELECT id, login, full_name, email, is_admin FROM users WHERE id=?", (row["gitea_user_id"],)).fetchone() if u: return dict(u) return None def _require_user(request: Request) -> dict: user = _user_from_request(request) if not user: raise HTTPException(status_code=401, detail="Authentication required") return user # ── API: status ── @api_router.get("/status") async def clipper_status(request: Request): user = _user_from_request(request) if not user: return {"authenticated": False} with get_conn() as conn: dev_cnt = conn.execute("SELECT COUNT(*) FROM extension_devices WHERE user_id=? AND revoked=0", (user["id"],)).fetchone()[0] clip_cnt = conn.execute("SELECT COUNT(*) FROM extension_clips WHERE user_id=?", (user["id"],)).fetchone()[0] return {"authenticated": True, "user": {"id": user["id"], "login": user.get("login")}, "devices": dev_cnt, "clips": clip_cnt} # ── API: auth verify / device registration ── @api_router.post("/auth/verify") async def auth_verify(request: Request): user = _require_user(request) try: body = await request.json() except Exception: body = {} device_id = (body.get("device_id") or request.headers.get("x-device-id") or "").strip() device_name = (body.get("device_name") or body.get("deviceName") or "").strip()[:200] extension_name = (body.get("extension_name") or body.get("extensionName") or "chrome").strip()[:20].lower() if not device_id: raise HTTPException(status_code=400, detail="device_id required") if len(device_id) > 128: raise HTTPException(status_code=400, detail="device_id too long") try: res = register_device(user["id"], device_id, device_name, extension_name) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from None if res["existing"]: return {"status": "ok", "device_id": device_id, "existing": True, "message": "Device already registered"} return {"status": "ok", "device_id": device_id, "token": res["token"], "note": "Copy token now — shown once. Use as Authorization: Bearer "} @api_router.post("/clip") async def clip_page(request: Request): user = _require_user(request) # Enforce max body size early (10 MB) clen = request.headers.get("content-length") if clen: try: if int(clen) > MAX_CLIP_BYTES + 1024: raise HTTPException(status_code=413, detail="Clip too large (max 10 MB)") except ValueError: pass try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON") from None # Device identification for rate limiting and logging device_id = (body.get("device_id") or request.headers.get("x-device-id") or "web").strip()[:128] or "web" # Rate limit 50/hour per device if not _check_rate_limit(f"{user['id']}:{device_id}"): raise HTTPException(status_code=429, detail="Rate limit: max 50 clips/hour per device") url = (body.get("url") or body.get("source_url") or "").strip() title = (body.get("title") or "").strip() content = body.get("content") or body.get("html") or "" clip_type = (body.get("content_type") or body.get("clip_type") or "article").strip().lower() if clip_type not in ("article", "selection", "bookmark", "screenshot"): clip_type = "article" if not url and not title and not content: raise HTTPException(status_code=400, detail="url, title or content required") # Validate url if present if url: if not (url.startswith("http://") or url.startswith("https://")): # allow bare domain? reject javascript: if url.lower().startswith("javascript:") or url.lower().startswith("data:"): raise HTTPException(status_code=400, detail="Invalid URL") # Cap content bytes if content and len(content.encode("utf-8")) > MAX_CLIP_BYTES: raise HTTPException(status_code=413, detail="Content too large (max 10 MB)") # Sanitize HTML content if present if content and "<" in content: # sanitize but keep structure for blocks converter content = sanitize_html(content)[: MAX_CLIP_BYTES] # Prepare payload for service _img_b64 = body.get("image_base64") or body.get("screenshot") or "" if not _img_b64 and body.get("images"): try: _imgs = body.get("images") if isinstance(_imgs, list) and _imgs: _img_b64 = _imgs[0].get("base64") or _imgs[0].get("src") or "" except Exception: logger.exception("clip_page") clip_data = { "url": url, "title": title[:200], "content": content, "content_type": clip_type, "selection_html": body.get("selection_html") or body.get("selection") or "", "image_base64": _img_b64, "tags": body.get("tags") or [], "target_workspace_id": body.get("target_workspace_id") or body.get("workspace_id"), "target_page_id": body.get("target_page_id") or body.get("parent_page_id"), "metadata": body.get("metadata") or {}, } try: result = create_page_from_clip(clip_data, user["id"]) except Exception as e: logger.exception("clip creation failed: %s", e) raise HTTPException(status_code=500, detail="Failed to create page") from None # Log clip try: log_clip(user["id"], device_id, clip_type, url, result["page_id"], result["workspace_id"], result["title"]) except Exception: logger.exception("clip_page") return {"status": "ok", "page_id": result["page_id"], "title": result["title"], "workspace_id": result["workspace_id"], "url": f"/pages/{result['page_id']}"} @api_router.get("/devices") async def list_extension_devices(request: Request): user = _require_user(request) devices = list_devices(user["id"]) return {"devices": devices} @api_router.delete("/devices/{device_id}") async def revoke_extension_device(device_id: int, request: Request): user = _require_user(request) ok = revoke_device(user["id"], device_id) if not ok: raise HTTPException(status_code=404, detail="Device not found") return {"status": "revoked"} # ── HTML: /extensions download page ── @router.get("/extensions", response_class=HTMLResponse) def extensions_page(request: Request): from app.routers.dashboard import _sidebar_data from app.templating import ENV env = ENV try: sidebar = _sidebar_data(request, []) except Exception: sidebar = {} # Simple standalone page reusing base.html block_tpl = env.from_string( '{% extends "base.html" %}{% block content %}{{ content_html|safe }}{% endblock %}' ) user = _user_from_request(request) # Count for auth user devices = [] clips = 0 if user: try: devices = list_devices(user["id"]) clips = sum(d.get("clips_count", 0) for d in devices) except Exception: logger.exception("extensions_page") content_html = f"""

🧩 FlowDeck Web Clipper

Capture any web page — article, selection, bookmark or screenshot — directly into FlowDeck. Install the browser extension, connect it once, then clip in one click.

🟢 Chrome / Edge

Manifest V3 — Chrome 88+, Edge 88+.

Download .zip · load unpacked in chrome://extensions

🟠 Firefox

Firefox 109+ (Manifest V2 compat).

Download .zip · load temporary add-on

⌨️ Sans extension

API directe — POST /api/v2/web-clipper/clip avec Bearer token.

Docs /help

How it works

  1. Install the extension (.zip) → enable in your browser.
  2. Open FlowDeck, go to Settings → Extensions and copy a Bearer token (or the clipper verifies via your session cookie).
  3. On any web page, click 📌 Clip to FlowDeck (floating button, right-click selection, or extension popup).
  4. Choose type: Article (full), Selection, Bookmark or Screenshot — the page is created instantly in your workspace.

Captures on this account

{len(devices)} device(s) · {clips} clip(s) total

{"".join(f'
{d.get("device_name") or d.get("extension_name")} {d.get("device_id")[:24]}{d.get("clips_count",0)} clips {d.get("last_clip_at") or ""}
' for d in devices[:10]) or '

No devices yet — clip your first page from the extension to appear here.

'}

Manage in Settings → Extensions

""" return HTMLResponse(block_tpl.render(**sidebar, request=request, page_title="Extensions", title_prefix="Extensions", page_icon="🧩", content_html=content_html))