bruno
45e59009c3
4 passes (283 → 93 routes async sur 667 = 86 % hors loop, avant 61 %) :
A. RACINE AUTH — `get_current_user` (auth/session.py) était `async def`
SANS aucun await (cookie decode = synchrone) ; idem ses clones :
`agent._current_user_id/_workspace_id/_current_admin` (34 sites) et
`sso._require_admin` (corps 0 await, 6 sites) → `def` +
47 `await` supprimés. Piège : 3 call sites passaient par l'alias `gcu`
(grep littéral aveugle) — 8 tests en échec → corrigés.
B. Re-scan : 19 routes devenues SANS await → `def` (agent 8, sso 5,
web_clipper 3, projects 2, auth 1…).
C/D. 155 routes dont les seuls awaits = `request.json()` / événements :
- try/except `body = {}` → `Body(default={})` (même tolérance)
- try/except `raise HTTPException(400)` → `Body(...)` REQUIS
(422 FastAPI — aucun test ne couvrait le 400)
- forme conditionnelle `request.json() if content-type else {}`
(54 sites) → défaut `{}` (sans corps = `{}` dans les 2 cas)
- `await fire_*` → `run_event_sync(...)` ; imports `Body` /
`run_event_sync` ajoutés aux routers convertis
Reste async (93, justifié) : form/upload/file (22), réseau gitea/llm/oidc,
`_json_body` (9), 2 JSON inline en argument, 1 fallback logique
(capture_frontend_error), 1 lecture conditionnelle (web_clipper), mixtes.
suite **1089/1089** · ruff OK · docs à jour
514 lines
23 KiB
Python
514 lines
23 KiB
Python
"""FlowDeck — teamspaces, verified pages, wiki home, collab polish (v7.3.0).
|
|
|
|
Routes under ``/api/v2/wiki`` plus the guest entry point ``/g/{token}``.
|
|
Design : ``docs/V73_Wiki_Teamspaces_Polish.md``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
|
|
from fastapi import APIRouter, Body, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
|
|
from app.auth.session import SessionManager
|
|
from app.db import get_conn
|
|
from app.services import wiki
|
|
from app.services.api_v2_helpers import audit_log
|
|
from app.services.notifications import create_notification
|
|
from app.templating import ENV
|
|
|
|
router = APIRouter(tags=["wiki"])
|
|
|
|
|
|
def _esc(value) -> str:
|
|
return html.escape(str(value))
|
|
|
|
|
|
def _user(request: Request) -> dict:
|
|
sess = SessionManager.decode_session(request.cookies.get("flowdeck_session", ""))
|
|
if not sess or not sess.get("id"):
|
|
raise HTTPException(401, "Authentication required")
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT is_admin, is_active FROM users WHERE id=?",
|
|
(sess["id"],)).fetchone()
|
|
if not row or not row["is_active"]:
|
|
raise HTTPException(403, "Account disabled")
|
|
return sess
|
|
|
|
|
|
def _workspace_id(request: Request) -> int:
|
|
wid = request.query_params.get("workspace_id")
|
|
if not wid:
|
|
raise HTTPException(400, "workspace_id required")
|
|
try:
|
|
wid = int(wid)
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "invalid workspace_id") from None
|
|
with get_conn() as conn:
|
|
if not conn.execute("SELECT 1 FROM workspaces WHERE id=?", (wid,)).fetchone():
|
|
raise HTTPException(404, "Workspace not found")
|
|
return wid
|
|
|
|
|
|
def _teamspace_or_404(teamspace_id: int, user_id: int) -> dict:
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT * FROM teamspaces WHERE id=?", (teamspace_id,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Teamspace not found")
|
|
if not wiki.can_read_teamspace(user_id, teamspace_id):
|
|
# private teamspace → 404 (not 403), same as restricted collections
|
|
raise HTTPException(404, "Teamspace not found")
|
|
return dict(row)
|
|
|
|
|
|
def _page_or_404(page_id: int) -> dict:
|
|
with get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT id, title, workspace_id, teamspace_id, deleted_at FROM pages WHERE id=?",
|
|
(page_id,)).fetchone()
|
|
if not row or row["deleted_at"]:
|
|
raise HTTPException(404, "Page not found")
|
|
return dict(row)
|
|
|
|
|
|
def _is_admin(user: dict) -> bool:
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT is_admin FROM users WHERE id=?", (user["id"],)).fetchone()
|
|
return bool(row and row["is_admin"])
|
|
|
|
|
|
def _can_verify(user: dict, page: dict) -> bool:
|
|
"""Admin, or an editor/owner of the teamspace / workspace holding the page."""
|
|
if _is_admin(user):
|
|
return True
|
|
if page.get("teamspace_id"):
|
|
return wiki.can_write_teamspace(user["id"], page["teamspace_id"])
|
|
wid = page.get("workspace_id")
|
|
if not wid:
|
|
return False
|
|
with get_conn() as conn:
|
|
owner = conn.execute("SELECT owner_id FROM workspaces WHERE id=?", (wid,)).fetchone()
|
|
if owner and owner["owner_id"] == user["id"]:
|
|
return True
|
|
member = conn.execute("SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
|
|
(wid, user["id"])).fetchone()
|
|
return bool(member and member["role"] in ("owner", "admin", "editor"))
|
|
|
|
|
|
# ── teamspaces ─────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/v2/wiki/teamspaces")
|
|
def list_teamspaces(request: Request):
|
|
user = _user(request)
|
|
wid = request.query_params.get("workspace_id")
|
|
if wid:
|
|
try:
|
|
wid = int(wid)
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "invalid workspace_id") from None
|
|
else:
|
|
wid = None
|
|
return {"teamspaces": wiki.list_teamspaces(user["id"], wid)}
|
|
|
|
|
|
@router.get("/wiki/teamspaces/{teamspace_id}", response_class=HTMLResponse)
|
|
def teamspace_page(teamspace_id: int, request: Request):
|
|
"""Teamspace detail HTML page — sidebar entry point."""
|
|
user = _user(request)
|
|
ts = _teamspace_or_404(teamspace_id, user["id"])
|
|
ts["role"] = wiki.get_teamspace_role(user["id"], teamspace_id)
|
|
with get_conn() as conn:
|
|
ws = conn.execute("SELECT name FROM workspaces WHERE id=?",
|
|
(ts["workspace_id"],)).fetchone()
|
|
pages = wiki.teamspace_pages(teamspace_id)
|
|
collections = wiki.teamspace_collections(teamspace_id)
|
|
page_rows = "\n".join(
|
|
f'<a class="ts-row" href="/pages/{p["id"]}" style="display:flex;align-items:center;gap:8px;'
|
|
f'padding:8px 10px;border-radius:6px;color:inherit;text-decoration:none;">'
|
|
f'<span>📄</span><span>{_esc(p["title"] or "Untitled")}</span></a>'
|
|
for p in pages)
|
|
coll_rows = "\n".join(
|
|
f'<a class="ts-row" href="/db/{c["id"]}" style="display:flex;align-items:center;gap:8px;'
|
|
f'padding:8px 10px;border-radius:6px;color:inherit;text-decoration:none;">'
|
|
f'<span>{_esc(c["icon"] or "🗄️")}</span><span>{_esc(c["name"] or "Untitled")}</span></a>'
|
|
for c in collections)
|
|
content_html = f"""
|
|
<div style="max-width:860px;margin:0 auto;padding:40px 24px;">
|
|
<h1 style="font-size:26px;display:flex;align-items:center;gap:10px;">
|
|
{_esc(ts['name'])}{' <span style="font-size:13px;padding:2px 8px;border-radius:10px;background:rgba(76,154,255,.15);color:#4c9aff;">🔒 private</span>' if ts['private'] else ''}
|
|
</h1>
|
|
<p style="color:var(--text-dim);">{_esc(ts.get('description') or '')}</p>
|
|
<div style="display:flex;gap:10px;font-size:12px;color:var(--text-dim);margin-bottom:24px;flex-wrap:wrap;">
|
|
<span>Workspace: {_esc((ws["name"]) if ws else '')}</span>
|
|
<span>·</span><span>Role: {_esc(ts['role'])}</span>
|
|
<span>·</span><span>{len(pages) + len(collections)} items</span>
|
|
</div>
|
|
<h2 style="font-size:16px;margin:20px 0 8px;">Pages</h2>
|
|
<div style="display:flex;flex-direction:column;gap:4px;">
|
|
{page_rows or '<p style="color:var(--text-dim);font-size:13px;">No pages yet.</p>'}
|
|
</div>
|
|
<h2 style="font-size:16px;margin:24px 0 8px;">Databases</h2>
|
|
<div style="display:flex;flex-direction:column;gap:4px;">
|
|
{coll_rows or '<p style="color:var(--text-dim);font-size:13px;">No databases yet.</p>'}
|
|
</div>
|
|
</div>
|
|
<style>
|
|
.ts-row:hover{{background:var(--bg-hover);}}
|
|
</style>"""
|
|
from app.routers.dashboard import _sidebar_data
|
|
env = ENV
|
|
sidebar = _sidebar_data(request, [])
|
|
block_tpl = env.from_string(
|
|
'{% extends "base.html" %}{% block content %}{{ content_html|safe }}{% endblock %}'
|
|
)
|
|
return block_tpl.render(
|
|
**sidebar,
|
|
request=request,
|
|
content_html=content_html,
|
|
page_title=ts["name"],
|
|
title_prefix="Teamspace",
|
|
page_icon="🏛️",
|
|
)
|
|
|
|
|
|
@router.post("/api/v2/wiki/teamspaces")
|
|
def create_teamspace(request: Request, body: dict = Body(default={})):
|
|
user = _user(request)
|
|
name = (body.get("name") or "").strip()
|
|
if not name or len(name) > 120:
|
|
raise HTTPException(400, "name required (max 120 chars)")
|
|
wid = int(body.get("workspace_id") or 0)
|
|
with get_conn() as conn:
|
|
ws = conn.execute("SELECT owner_id FROM workspaces WHERE id=?", (wid,)).fetchone()
|
|
if not ws:
|
|
raise HTTPException(404, "Workspace not found")
|
|
admin = conn.execute("SELECT is_admin FROM users WHERE id=?", (user["id"],)).fetchone()
|
|
member = conn.execute("SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
|
|
(wid, user["id"])).fetchone()
|
|
allowed = (ws["owner_id"] == user["id"] or (admin and admin["is_admin"])
|
|
or (member and member["role"] in ("admin", "editor", "owner")))
|
|
if not allowed:
|
|
raise HTTPException(403, "Editor role required in the workspace")
|
|
try:
|
|
tsid = wiki.create_teamspace(wid, name, user["id"],
|
|
description=body.get("description") or "",
|
|
private=bool(body.get("private")))
|
|
except ValueError as exc:
|
|
raise HTTPException(409, str(exc)) from None
|
|
audit_log(user, "teamspace.create", "teamspace", tsid, name, request)
|
|
return JSONResponse(status_code=201, content={"id": tsid, "name": name})
|
|
|
|
|
|
@router.get("/api/v2/wiki/teamspaces/{teamspace_id}")
|
|
def get_teamspace(teamspace_id: int, request: Request):
|
|
user = _user(request)
|
|
ts = _teamspace_or_404(teamspace_id, user["id"])
|
|
ts["role"] = wiki.get_teamspace_role(user["id"], teamspace_id)
|
|
ts["member_count"] = len(wiki.teamspace_member_ids(teamspace_id))
|
|
return ts
|
|
|
|
|
|
@router.get("/api/v2/wiki/teamspaces/{teamspace_id}/members")
|
|
def list_members(teamspace_id: int, request: Request):
|
|
user = _user(request)
|
|
_teamspace_or_404(teamspace_id, user["id"])
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT m.user_id, m.role, u.login, u.full_name FROM teamspace_members m
|
|
JOIN users u ON u.id = m.user_id WHERE m.teamspace_id=? ORDER BY u.login""",
|
|
(teamspace_id,)).fetchall()
|
|
return {"members": [dict(r) for r in rows]}
|
|
|
|
|
|
@router.put("/api/v2/wiki/teamspaces/{teamspace_id}/members/{member_id}")
|
|
def set_member(teamspace_id: int, member_id: int, request: Request, body: dict = Body(default={})):
|
|
user = _user(request)
|
|
_teamspace_or_404(teamspace_id, user["id"])
|
|
if not wiki.can_write_teamspace(user["id"], teamspace_id):
|
|
raise HTTPException(403, "Editor role required")
|
|
role = body.get("role")
|
|
if role not in wiki.TEAMSPACE_ROLES:
|
|
raise HTTPException(400, f"role must be one of {', '.join(wiki.TEAMSPACE_ROLES)}")
|
|
with get_conn() as conn:
|
|
if not conn.execute("SELECT 1 FROM users WHERE id=?", (member_id,)).fetchone():
|
|
raise HTTPException(404, "User not found")
|
|
conn.execute(
|
|
"""INSERT INTO teamspace_members (teamspace_id, user_id, role) VALUES (?,?,?)
|
|
ON CONFLICT(teamspace_id, user_id) DO UPDATE SET role=excluded.role""",
|
|
(teamspace_id, member_id, role))
|
|
conn.commit()
|
|
audit_log(user, "teamspace.member.set", "teamspace", teamspace_id,
|
|
f"u{member_id}={role}", request)
|
|
return {"status": "ok", "user_id": member_id, "role": role}
|
|
|
|
|
|
@router.delete("/api/v2/wiki/teamspaces/{teamspace_id}/members/{member_id}")
|
|
def remove_member(teamspace_id: int, member_id: int, request: Request):
|
|
user = _user(request)
|
|
_teamspace_or_404(teamspace_id, user["id"])
|
|
if not wiki.can_write_teamspace(user["id"], teamspace_id):
|
|
raise HTTPException(403, "Editor role required")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("DELETE FROM teamspace_members WHERE teamspace_id=? AND user_id=?",
|
|
(teamspace_id, member_id))
|
|
conn.commit()
|
|
if not cur.rowcount:
|
|
raise HTTPException(404, "Not a member")
|
|
return {"status": "removed", "user_id": member_id}
|
|
|
|
|
|
# ── verified pages ─────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/v2/wiki/pages/{page_id}/verification")
|
|
def get_verification(page_id: int, request: Request):
|
|
_user(request)
|
|
_page_or_404(page_id)
|
|
return {"verification": wiki.verification(page_id)}
|
|
|
|
|
|
@router.post("/api/v2/wiki/pages/{page_id}/verify")
|
|
def verify_page(page_id: int, request: Request, body: dict = Body(default={})):
|
|
user = _user(request)
|
|
page = _page_or_404(page_id)
|
|
if not _can_verify(user, page):
|
|
raise HTTPException(403, "Editor role required to verify a page")
|
|
out = wiki.verify_page(page_id, user["id"],
|
|
days=body.get("days") or wiki.VERIFICATION_DAYS_DEFAULT,
|
|
note=body.get("note") or "")
|
|
audit_log(user, "page.verify", "page", page_id, out.get("expires_at") or "", request)
|
|
return {"verification": out}
|
|
|
|
|
|
@router.delete("/api/v2/wiki/pages/{page_id}/verify")
|
|
def unverify_page(page_id: int, request: Request):
|
|
user = _user(request)
|
|
_page_or_404(page_id)
|
|
if not wiki.unverify_page(page_id):
|
|
raise HTTPException(404, "Page is not verified")
|
|
audit_log(user, "page.unverify", "page", page_id, "", request)
|
|
return {"status": "unverified", "page_id": page_id}
|
|
|
|
|
|
@router.get("/api/v2/wiki/verified")
|
|
def list_verified(request: Request):
|
|
"""Verified (non-expired) pages of a workspace — the ✅ wiki index."""
|
|
user = _user(request)
|
|
wid = _workspace_id(request)
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT p.id, p.title, p.page_icon, p.teamspace_id,
|
|
v.verified_at, v.expires_at, v.note, u.login
|
|
FROM page_verifications v
|
|
JOIN pages p ON p.id = v.page_id
|
|
LEFT JOIN users u ON u.id = v.verified_by
|
|
WHERE p.workspace_id=? AND p.deleted_at IS NULL""",
|
|
(wid,)).fetchall()
|
|
out = []
|
|
for r in rows:
|
|
item = dict(r)
|
|
if wiki.is_expired(r):
|
|
continue # badge lapsed → not listed
|
|
if item["teamspace_id"] and not wiki.can_read_teamspace(user["id"],
|
|
item["teamspace_id"]):
|
|
continue # private teamspace → hidden
|
|
item["active"] = True
|
|
out.append(item)
|
|
return {"pages": out}
|
|
|
|
|
|
# ── follows ────────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/v2/wiki/pages/{page_id}/follow")
|
|
def follow_page(page_id: int, request: Request):
|
|
user = _user(request)
|
|
_page_or_404(page_id)
|
|
now = wiki.toggle_follow(page_id, user["id"])
|
|
return {"page_id": page_id, "following": now}
|
|
|
|
|
|
@router.get("/api/v2/wiki/pages/{page_id}/followers")
|
|
def list_followers(page_id: int, request: Request):
|
|
_user(request)
|
|
_page_or_404(page_id)
|
|
ids = wiki.followers(page_id)
|
|
if not ids:
|
|
return {"followers": []}
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT id, login, full_name FROM users WHERE id IN ({','.join('?' * len(ids))})",
|
|
ids).fetchall()
|
|
return {"followers": [dict(r) for r in rows]}
|
|
|
|
|
|
# ── comment reactions ──────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/v2/wiki/comments/{comment_id}/reactions")
|
|
def react(comment_id: int, request: Request, body: dict = Body(default={})):
|
|
user = _user(request)
|
|
emoji = (body.get("emoji") or "").strip()
|
|
if not emoji:
|
|
raise HTTPException(400, "emoji required")
|
|
try:
|
|
counts = wiki.toggle_reaction(comment_id, user["id"], emoji)
|
|
except LookupError:
|
|
raise HTTPException(404, "Comment not found") from None
|
|
return {"comment_id": comment_id, "reactions": counts}
|
|
|
|
|
|
@router.get("/api/v2/wiki/comments/{comment_id}/reactions")
|
|
def list_reactions(comment_id: int, request: Request):
|
|
_user(request)
|
|
return {"comment_id": comment_id, "reactions": wiki.reactions(comment_id)}
|
|
|
|
|
|
# ── guest shares ───────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/v2/wiki/pages/{page_id}/guests")
|
|
def create_guest(page_id: int, request: Request, body: dict = Body(default={})):
|
|
user = _user(request)
|
|
page = _page_or_404(page_id)
|
|
if page.get("teamspace_id") and not wiki.can_write_teamspace(user["id"], page["teamspace_id"]):
|
|
raise HTTPException(403, "Editor role required to share")
|
|
try:
|
|
share = wiki.create_guest_share(page_id, body.get("email") or "",
|
|
body.get("role") or "viewer",
|
|
user["id"], days=body.get("days", 30))
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from None
|
|
audit_log(user, "page.guest_share", "page", page_id, share["email"], request)
|
|
return JSONResponse(status_code=201, content={
|
|
"id": share["id"], "token": share["token"], "role": share["role"],
|
|
"expires_at": share["expires_at"], "url": f"/g/{share['token']}"})
|
|
|
|
|
|
@router.get("/api/v2/wiki/pages/{page_id}/guests")
|
|
def list_guests(page_id: int, request: Request):
|
|
_user(request)
|
|
_page_or_404(page_id)
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT id, email, role, expires_at, revoked, created_at FROM guest_shares"
|
|
" WHERE page_id=? ORDER BY id DESC", (page_id,)).fetchall()
|
|
return {"guests": [dict(r) for r in rows]}
|
|
|
|
|
|
@router.delete("/api/v2/wiki/guests/{share_id}")
|
|
def revoke_guest(share_id: int, request: Request):
|
|
user = _user(request)
|
|
with get_conn() as conn:
|
|
if not conn.execute("SELECT 1 FROM guest_shares WHERE id=?", (share_id,)).fetchone():
|
|
raise HTTPException(404, "Guest share not found")
|
|
conn.execute("UPDATE guest_shares SET revoked=1 WHERE id=?", (share_id,))
|
|
conn.commit()
|
|
audit_log(user, "page.guest_revoke", "guest_share", share_id, "", request)
|
|
return {"status": "revoked", "id": share_id}
|
|
|
|
|
|
_GUEST_404 = """<!DOCTYPE html><html lang="en"><head><meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>Link unavailable — FlowDeck</title>
|
|
<style>body{font-family:-apple-system,Segoe UI,Roboto,sans-serif;max-width:520px;
|
|
margin:80px auto;padding:0 20px;color:#1f2328;text-align:center}
|
|
h1{font-size:20px} p{color:#656d76;line-height:1.6}</style></head><body>
|
|
<h1>This link is unavailable</h1>
|
|
<p>It may have expired, been revoked, or never existed.<br>
|
|
Ask the person who shared it with you for a new link.</p></body></html>"""
|
|
|
|
|
|
@router.get("/g/{token}", response_class=HTMLResponse)
|
|
def guest_page(token: str, request: Request):
|
|
"""Account-less page access (read-only or commenter). 404 if inactive."""
|
|
share = wiki.resolve_guest_share(token)
|
|
if not share:
|
|
return HTMLResponse(_GUEST_404, status_code=404)
|
|
with get_conn() as conn:
|
|
page = conn.execute("SELECT id, title, content, created_at, updated_at, deleted_at"
|
|
" FROM pages WHERE id=?", (share["page_id"],)).fetchone()
|
|
if not page or page["deleted_at"]:
|
|
return HTMLResponse(_GUEST_404, status_code=404)
|
|
wiki.record_view(share["page_id"])
|
|
return f"""<!DOCTYPE html><html lang="en"><head><meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>{page['title']} — FlowDeck guest</title>
|
|
<style>body{{font-family:-apple-system,Segoe UI,Roboto,sans-serif;max-width:760px;
|
|
margin:40px auto;padding:0 20px;line-height:1.6;color:#1f2328}}
|
|
.guest-banner{{background:#fff4e5;border:1px solid #ffd8a8;padding:10px 14px;
|
|
border-radius:8px;margin-bottom:24px;font-size:14px}}
|
|
pre{{background:#f6f8fa;padding:14px;border-radius:8px;overflow:auto;
|
|
white-space:pre-wrap;word-break:break-word}}</style></head><body>
|
|
<div class="guest-banner">You are viewing this page as a guest
|
|
({share['role']}{' — expires ' + str(share['expires_at']) if share['expires_at'] else ''}).
|
|
Editing is disabled.</div>
|
|
<h1>{page['title']}</h1><pre>{page['content'] or ''}</pre></body></html>"""
|
|
|
|
|
|
# ── page views ─────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/v2/wiki/pages/{page_id}/views")
|
|
def page_views(page_id: int, request: Request):
|
|
user = _user(request)
|
|
page = _page_or_404(page_id)
|
|
if page.get("teamspace_id") and not wiki.can_write_teamspace(user["id"], page["teamspace_id"]):
|
|
raise HTTPException(403, "Editor role required to read analytics")
|
|
return wiki.view_stats(page_id, days=request.query_params.get("days", 30))
|
|
|
|
|
|
# ── wiki home ──────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/v2/wiki/home")
|
|
def wiki_home(request: Request):
|
|
"""Aggregated knowledge home: verified pages + recents + teamspaces."""
|
|
user = _user(request)
|
|
wid = _workspace_id(request)
|
|
with get_conn() as conn:
|
|
recents = conn.execute(
|
|
"""SELECT id, title, page_icon, updated_at FROM pages
|
|
WHERE workspace_id=? AND deleted_at IS NULL
|
|
ORDER BY updated_at DESC LIMIT 20""", (wid,)).fetchall()
|
|
verified = conn.execute(
|
|
"""SELECT v.page_id, v.verified_at, v.expires_at FROM page_verifications v
|
|
JOIN pages p ON p.id = v.page_id
|
|
WHERE p.workspace_id=? AND p.deleted_at IS NULL
|
|
AND (v.expires_at IS NULL OR v.expires_at > ?)""",
|
|
(wid, __import__("datetime").datetime.now(
|
|
__import__("datetime").timezone.utc).replace(microsecond=0).isoformat()),
|
|
).fetchall()
|
|
return {"workspace_id": wid,
|
|
"teamspaces": wiki.list_teamspaces(user["id"], wid),
|
|
"verified": [dict(r) for r in verified],
|
|
"recents": [dict(r) for r in recents]}
|
|
|
|
|
|
@router.post("/api/v2/wiki/verify-expiry-sweep")
|
|
def sweep_expiry(request: Request):
|
|
"""Notify verifiers whose ✅ expires within 7 days (idempotent-ish job)."""
|
|
user = _user(request)
|
|
with get_conn() as conn:
|
|
if not conn.execute("SELECT is_admin FROM users WHERE id=?", (user["id"],)).fetchone()[0]:
|
|
raise HTTPException(403, "Admin required")
|
|
sent = 0
|
|
for row in wiki.expiring_verifications(days=7):
|
|
create_notification(row["owner_id"], None, "page.verification_expiring",
|
|
"Verification expiring soon",
|
|
f"“{row['title']}” loses its verified badge on {row['expires_at']}.",
|
|
resource_type="page", resource_id=row["page_id"])
|
|
sent += 1
|
|
return {"notified": sent}
|
|
|
|
|
|
# ── blocks (mermaid / equation_inline / progress) ───────────────────────────
|
|
|
|
@router.post("/api/v2/wiki/blocks/preview")
|
|
def preview_blocks(request: Request, body: dict = Body(default={})):
|
|
"""Render v7.3 blocks to HTML (same renderer used by the export pipeline)."""
|
|
_user(request)
|
|
blocks = body.get("blocks")
|
|
if not isinstance(blocks, list):
|
|
raise HTTPException(400, "blocks must be a list")
|
|
if len(blocks) > 200:
|
|
raise HTTPException(400, "max 200 blocks per preview")
|
|
from app.services.wiki_blocks import mmdc_available, render_block
|
|
out = [{"type": b.get("type"), "html": render_block(b)} for b in blocks
|
|
if isinstance(b, dict) and b.get("type") in ("mermaid", "equation_inline", "progress")]
|
|
return {"rendered": out, "mmdc_available": mmdc_available()}
|