Wiki/baserow/seed/seed_forms.py

#!/usr/bin/env python3
"""Baserow forms publics — Iteration 4 BUILD.

Cree des form views publiques sur les tables attribution et intervention.
Permet la saisie rapide via lien public sans compte Baserow (UX mobile).

Usage :
    BASEROW_URL=http://localhost:8080 \\
    BASEROW_EMAIL=admin@acadenice.fr \\
    BASEROW_PASSWORD=... \\
    python baserow/seed/seed_forms.py

Idempotent : skip si form existe deja.

ATTENTION Phase 1 : pas de validation d'identite cote form. N'importe qui
avec le lien peut soumettre. Phase 2 bridge service ajoutera l'auth SSO
+ filtre row-level (formateur saisit que ses propres attributions).
"""

import os
import sys
from typing import Any

import requests


FORMS_TO_CREATE: list[dict[str, Any]] = [
    {
        "table_name": "attribution",
        "form": {
            "name": "Saisir heures realisees (formateur)",
            "type": "form",
            "title": "Heures realisees - Acadenice formation",
            "description": "Formateur : enregistre tes heures realisees apres une session.\n\nMerci de selectionner le module concerne, ton attribution, et le nombre d'heures effectuees.",
            "submit_action": "MESSAGE",
            "submit_action_message": "Merci ! Tes heures sont enregistrees. La capacite restante est mise a jour automatiquement.",
            "public": True,
        },
        "fields_to_show": [
            "attribution_personne",
            "attribution_module",
            "attribution_heures_realisees",
            "attribution_date_fin",
            "attribution_statut",
        ],
    },
    {
        "table_name": "intervention",
        "form": {
            "name": "Saisir intervention (developpeur)",
            "type": "form",
            "title": "Intervention - Acadenice agence",
            "description": "Developpeur : log ton intervention sur une tache projet client.\n\nIndique la tache, tes heures, la date, et notes optionnelles (commit, lien PR).",
            "submit_action": "MESSAGE",
            "submit_action_message": "Merci ! L'intervention est enregistree. Capacite agence mise a jour.",
            "public": True,
        },
        "fields_to_show": [
            "intervention_personne",
            "intervention_tache",
            "intervention_heures",
            "intervention_date",
            "intervention_notes",
        ],
    },
]


class BaserowForms:
    def __init__(self, base_url: str, email: str, password: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.email = email
        self.password = password
        self.token: str | None = None
        self.session = requests.Session()

    def _headers(self) -> dict[str, str]:
        return {"Authorization": f"JWT {self.token}", "Content-Type": "application/json"}

    def login(self) -> None:
        r = self.session.post(
            f"{self.base_url}/api/user/token-auth/",
            json={"email": self.email, "password": self.password},
        )
        r.raise_for_status()
        self.token = r.json()["token"]
        print(f"  [auth] Logged in as {self.email}")

    def list_workspaces(self) -> list[dict[str, Any]]:
        r = self.session.get(f"{self.base_url}/api/workspaces/", headers=self._headers())
        r.raise_for_status()
        return r.json()

    def list_databases(self, workspace_id: int) -> list[dict[str, Any]]:
        r = self.session.get(
            f"{self.base_url}/api/applications/workspace/{workspace_id}/", headers=self._headers()
        )
        r.raise_for_status()
        return [a for a in r.json() if a["type"] == "database"]

    def list_tables(self, database_id: int) -> dict[str, dict[str, Any]]:
        r = self.session.get(
            f"{self.base_url}/api/database/tables/database/{database_id}/", headers=self._headers()
        )
        r.raise_for_status()
        return {t["name"]: t for t in r.json()}

    def list_views(self, table_id: int) -> list[dict[str, Any]]:
        r = self.session.get(
            f"{self.base_url}/api/database/views/table/{table_id}/", headers=self._headers()
        )
        r.raise_for_status()
        return r.json()

    def list_fields(self, table_id: int) -> dict[str, dict[str, Any]]:
        r = self.session.get(
            f"{self.base_url}/api/database/fields/table/{table_id}/", headers=self._headers()
        )
        r.raise_for_status()
        return {f["name"]: f for f in r.json()}

    def create_form(self, table_id: int, form_def: dict[str, Any]) -> dict[str, Any]:
        r = self.session.post(
            f"{self.base_url}/api/database/views/table/{table_id}/",
            headers=self._headers(),
            json=form_def,
        )
        if r.status_code >= 300:
            print(f"    [ERROR] create form payload={form_def} resp={r.text[:300]}")
            r.raise_for_status()
        return r.json()

    def update_field_options(
        self, view_id: int, field_options: dict[int, dict[str, Any]]
    ) -> None:
        r = self.session.patch(
            f"{self.base_url}/api/database/views/form/{view_id}/field-options/",
            headers=self._headers(),
            json={"field_options": field_options},
        )
        if r.status_code >= 300:
            print(f"    [ERROR] update field options resp={r.text[:300]}")
            r.raise_for_status()

    def run(self) -> None:
        print(f"\n[1/4] Login")
        self.login()

        print(f"\n[2/4] Find workspace 'Acadenice' + database 'formation-hub'")
        ws = next((w for w in self.list_workspaces() if w["name"] == "Acadenice"), None)
        if not ws:
            raise RuntimeError("Workspace 'Acadenice' not found")
        db = next((d for d in self.list_databases(ws["id"]) if d["name"] == "formation-hub"), None)
        if not db:
            raise RuntimeError("Database 'formation-hub' not found")
        print(f"  Workspace id={ws['id']}, Database id={db['id']}")

        print(f"\n[3/4] Resolve tables")
        tables = self.list_tables(db["id"])

        print(f"\n[4/4] Create forms")
        for spec in FORMS_TO_CREATE:
            tname = spec["table_name"]
            if tname not in tables:
                print(f"  [skip] table '{tname}' absent")
                continue
            tid = tables[tname]["id"]
            existing_forms = [v for v in self.list_views(tid) if v["type"] == "form"]
            existing_names = {v["name"] for v in existing_forms}
            if spec["form"]["name"] in existing_names:
                form = next(v for v in existing_forms if v["name"] == spec["form"]["name"])
                print(f"  [form] Reuse '{spec['form']['name']}' on {tname} id={form['id']}")
            else:
                form = self.create_form(tid, spec["form"])
                print(f"  [form] Created '{form['name']}' on {tname} id={form['id']}")

            # Configure visible fields
            fields = self.list_fields(tid)
            field_options: dict[int, dict[str, Any]] = {}
            for fname in spec["fields_to_show"]:
                if fname in fields:
                    field_options[fields[fname]["id"]] = {
                        "enabled": True,
                        "required": fname in ("attribution_personne", "attribution_module", "attribution_heures_realisees", "intervention_personne", "intervention_tache", "intervention_heures"),
                    }
            if field_options:
                self.update_field_options(form["id"], field_options)
                print(f"    [form] {len(field_options)} fields visible/required configures")

            slug = form.get("slug")
            if slug:
                print(f"    [public URL] {self.base_url}/form/{slug}")
            else:
                print(f"    [note] slug pas trouve dans la response (cf UI Baserow pour le lien public)")

        print("\n=== Forms OK ===")


def main() -> int:
    base_url = os.environ.get("BASEROW_URL", "http://localhost:8080")
    email = os.environ.get("BASEROW_EMAIL")
    password = os.environ.get("BASEROW_PASSWORD")
    if not email or not password:
        print("ERROR: set BASEROW_EMAIL et BASEROW_PASSWORD env vars", file=sys.stderr)
        return 1
    try:
        BaserowForms(base_url, email, password).run()
    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
    except Exception as e:
        print(f"\nError: {e}", file=sys.stderr)
        return 3
    return 0


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