Wiki/docmost/setup/seed.py

#!/usr/bin/env python3
"""Docmost seed script — utilise les endpoints internes (non-officiels mais AGPL-legal).

Cree workspace + admin + 3 spaces + page test + share link via reverse-engineering
des controllers Docmost (cf github.com/docmost/docmost/apps/server/src/core).

ATTENTION : ces endpoints ne sont pas publiquement documentes par Docmost.
Ils peuvent changer entre versions. Testez le script apres chaque bump Docmost.

Usage :
    DOCMOST_URL=http://localhost:3000 \\
    DOCMOST_ADMIN_EMAIL=admin@acadenice.fr \\
    DOCMOST_ADMIN_PASSWORD=... \\
    DOCMOST_ADMIN_NAME='Corentin JOGUET' \\
    DOCMOST_WORKSPACE_NAME=Acadenice \\
    python docmost/setup/seed.py [--schema docmost/setup/spaces.yaml]
"""

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any

import requests
import yaml


class DocmostSeed:
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})
        self.workspace_id: str | None = None

    def _post(self, path: str, body: dict[str, Any] | None = None) -> dict[str, Any]:
        url = f"{self.base_url}{path}"
        r = self.session.post(url, json=body or {})
        if r.status_code >= 300:
            print(f"    [ERROR] POST {path} → {r.status_code}\n      body={body}\n      resp={r.text[:500]}")
            r.raise_for_status()
        return r.json() if r.text else {}

    def setup_admin_and_workspace(self, name: str, email: str, password: str, workspace_name: str) -> bool:
        """POST /api/auth/setup — bootstrap workspace + admin (1ere fois seulement).
        Retourne True si setup execute, False si workspace deja initialise.
        """
        try:
            self._post(
                "/api/auth/setup",
                {"name": name, "email": email, "password": password, "workspaceName": workspace_name},
            )
            print(f"  [setup] Workspace '{workspace_name}' + admin '{email}' crees")
            return True
        except requests.HTTPError as e:
            if e.response is not None and e.response.status_code in (400, 403, 404):
                print("  [setup] Workspace deja initialise — skip")
                return False
            raise

    def login(self, email: str, password: str) -> None:
        self._post("/api/auth/login", {"email": email, "password": password})
        print(f"  [auth] Logged in as {email}")

    def get_workspace_info(self) -> dict[str, Any]:
        info = self._post("/api/workspace/info")
        self.workspace_id = info.get("id") or info.get("workspaceId")
        print(f"  [workspace] id={self.workspace_id}")
        return info

    def list_spaces(self) -> list[dict[str, Any]]:
        try:
            data = self._post("/api/spaces/", {"page": 1, "limit": 100})
            return data.get("items") or data.get("data") or []
        except requests.HTTPError:
            return []

    def create_space(self, name: str, description: str, slug: str, visibility: str = "open") -> dict[str, Any]:
        body = {
            "name": name,
            "description": description,
            "slug": slug,
            "visibility": visibility,
        }
        return self._post("/api/spaces/create", body)

    def get_or_create_space(self, name: str, description: str, slug: str, visibility: str = "open") -> str:
        existing = self.list_spaces()
        for s in existing:
            if s.get("name") == name or s.get("slug") == slug:
                sid = s.get("id")
                print(f"  [space] Reuse '{name}' id={sid}")
                return sid
        created = self.create_space(name, description, slug, visibility)
        sid = created.get("id")
        print(f"  [space] Created '{name}' id={sid}")
        return sid

    def create_page(self, space_id: str, title: str, content: dict[str, Any] | None = None) -> str:
        body: dict[str, Any] = {"spaceId": space_id, "title": title}
        if content is not None:
            body["content"] = content
        result = self._post("/api/pages/create", body)
        pid = result.get("id")
        print(f"  [page] Created '{title}' id={pid} in space={space_id}")
        return pid

    def create_share(self, page_id: str, expires_at: str | None = None) -> dict[str, Any]:
        body: dict[str, Any] = {"pageId": page_id, "includeSubPages": False}
        if expires_at:
            body["expiresAt"] = expires_at
        result = self._post("/api/shares/create", body)
        share_id = result.get("id") or result.get("key")
        share_url = result.get("url") or f"{self.base_url}/share/{share_id}"
        print(f"  [share] Created share for page={page_id} id={share_id}")
        print(f"          URL: {share_url}")
        return result


def welcome_page_content() -> dict[str, Any]:
    """Tiptap doc JSON minimal pour la page Welcome."""
    return {
        "type": "doc",
        "content": [
            {"type": "heading", "attrs": {"level": 1}, "content": [
                {"type": "text", "text": "Welcome formation-hub"}]},
            {"type": "paragraph", "content": [
                {"type": "text", "text": "Wiki Docmost interne Acadenice — voir docs/ du repo wiki pour la conception complete."}]},
            {"type": "heading", "attrs": {"level": 2}, "content": [
                {"type": "text", "text": "Stack"}]},
            {"type": "bulletList", "content": [
                {"type": "listItem", "content": [
                    {"type": "paragraph", "content": [
                        {"type": "text", "marks": [{"type": "bold"}], "text": "Docmost"},
                        {"type": "text", "text": " : ce wiki (AGPL self-host)"}]}]},
                {"type": "listItem", "content": [
                    {"type": "paragraph", "content": [
                        {"type": "text", "marks": [{"type": "bold"}], "text": "Baserow"},
                        {"type": "text", "text": " : DBs structurees (MIT self-host)"}]}]},
                {"type": "listItem", "content": [
                    {"type": "paragraph", "content": [
                        {"type": "text", "marks": [{"type": "bold"}], "text": "Bridge service"},
                        {"type": "text", "text": " : Phase 2"}]}]}]},
            {"type": "paragraph", "content": [
                {"type": "text", "text": "Note : ajouter les blocks Mermaid / Drawio / Excalidraw via le slash menu une fois la page ouverte (ils ne sont pas pre-inserees par le seed)."}]},
        ],
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Seed Docmost formation-hub")
    parser.add_argument("--schema", default=str(Path(__file__).parent / "spaces.yaml"))
    parser.add_argument("--skip-setup", action="store_true", help="Skip workspace bootstrap (deja fait)")
    args = parser.parse_args()

    base_url = os.environ.get("DOCMOST_URL", "http://localhost:3000")
    email = os.environ.get("DOCMOST_ADMIN_EMAIL")
    password = os.environ.get("DOCMOST_ADMIN_PASSWORD")
    name = os.environ.get("DOCMOST_ADMIN_NAME", "Admin")
    workspace_name = os.environ.get("DOCMOST_WORKSPACE_NAME", "Acadenice")

    if not email or not password:
        print("ERROR: set DOCMOST_ADMIN_EMAIL and DOCMOST_ADMIN_PASSWORD env vars", file=sys.stderr)
        return 1

    schema = yaml.safe_load(Path(args.schema).read_text())
    seeder = DocmostSeed(base_url)

    try:
        if not args.skip_setup:
            print(f"\n[1/5] Bootstrap workspace {workspace_name}")
            seeder.setup_admin_and_workspace(name, email, password, workspace_name)

        print(f"\n[2/5] Login")
        seeder.login(email, password)

        print(f"\n[3/5] Workspace info")
        try:
            seeder.get_workspace_info()
        except requests.HTTPError:
            print("  [workspace] info endpoint a echoue — continue")

        print(f"\n[4/5] Spaces ({len(schema.get('spaces', []))})")
        space_ids: dict[str, str] = {}
        for sp in schema.get("spaces", []):
            sid = seeder.get_or_create_space(
                name=sp["name"],
                description=sp.get("description", "").strip(),
                slug=sp.get("slug", sp["name"].lower()),
                visibility=("open" if sp.get("visibility") == "workspace_members" else "private"),
            )
            if sid:
                space_ids[sp["name"]] = sid

        print(f"\n[5/5] Welcome page + share link")
        welcome = schema.get("welcome_page") or {}
        target_space = welcome.get("space", "CFA")
        if target_space in space_ids:
            page_id = seeder.create_page(
                space_id=space_ids[target_space],
                title=welcome.get("title", "Welcome formation-hub"),
                content=welcome_page_content(),
            )
            if page_id and welcome.get("share_link", {}).get("enabled"):
                seeder.create_share(page_id=page_id)
        else:
            print(f"  [skip] Space '{target_space}' introuvable, page Welcome non creee")

        print("\n=== Seed Docmost OK ===")
        print(f"  Workspace : {workspace_name}")
        print(f"  Spaces    : {len(space_ids)}")
        print(f"  Welcome   : page + share link dans space '{target_space}'")
    except requests.HTTPError as e:
        print(f"\nHTTP error: {e}", file=sys.stderr)
        if e.response is not None:
            print(f"Response: {e.response.text[:500]}", file=sys.stderr)
        return 2

    return 0


if __name__ == "__main__":
    sys.exit(main())