#!/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())