Files
decision/backend/seed.py
Yvv 25437f24e3 Sprint 1 : scaffolding complet de Glibredecision
Plateforme de decisions collectives pour Duniter/G1.
Backend FastAPI async + PostgreSQL (14 tables, 8 routers, 6 services,
moteur de vote avec formule d'inertie WoT/Smith/TechComm).
Frontend Nuxt 4 + Nuxt UI v3 + Pinia (9 pages, 5 stores).
Infrastructure Docker + Woodpecker CI + Traefik.
Documentation technique et utilisateur (15 fichiers).
Seed : Licence G1, Engagement Forgeron v2.0.0, 4 protocoles de vote.
30 tests unitaires (formules, mode params, vote nuance) -- tous verts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-28 12:46:11 +01:00

531 lines
17 KiB
Python

"""Seed the database with initial FormulaConfigs, VotingProtocols, Documents, and Decisions.
Usage:
python seed.py
Idempotent: checks if data already exists before inserting.
"""
from __future__ import annotations
import asyncio
import uuid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import async_session, engine, Base
from app.models.protocol import FormulaConfig, VotingProtocol
from app.models.document import Document, DocumentItem
from app.models.decision import Decision, DecisionStep
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
async def get_or_create(
session: AsyncSession,
model,
lookup_field: str,
lookup_value,
**kwargs,
):
"""Return existing row or create a new one."""
stmt = select(model).where(getattr(model, lookup_field) == lookup_value)
result = await session.execute(stmt)
instance = result.scalar_one_or_none()
if instance is not None:
return instance, False
instance = model(**{lookup_field: lookup_value}, **kwargs)
session.add(instance)
await session.flush()
return instance, True
# ---------------------------------------------------------------------------
# Seed: FormulaConfigs
# ---------------------------------------------------------------------------
async def seed_formula_configs(session: AsyncSession) -> dict[str, FormulaConfig]:
"""Create the 4 base formula configurations."""
configs: dict[str, dict] = {
"Standard Licence G1": {
"description": "Formule standard pour la Licence G1 : vote binaire WoT.",
"duration_days": 30,
"majority_pct": 50,
"base_exponent": 0.1,
"gradient_exponent": 0.2,
"constant_base": 0.0,
},
"Forgeron avec Smith": {
"description": "Vote forgeron avec critere Smith sub-WoT.",
"duration_days": 30,
"majority_pct": 50,
"base_exponent": 0.1,
"gradient_exponent": 0.2,
"constant_base": 0.0,
"smith_exponent": 0.1,
},
"Comite Tech": {
"description": "Vote avec critere Comite Technique.",
"duration_days": 30,
"majority_pct": 50,
"base_exponent": 0.1,
"gradient_exponent": 0.2,
"constant_base": 0.0,
"techcomm_exponent": 0.1,
},
"Vote Nuance": {
"description": "Vote nuance a 6 niveaux (CONTRE..TOUT A FAIT).",
"duration_days": 30,
"majority_pct": 50,
"base_exponent": 0.1,
"gradient_exponent": 0.2,
"constant_base": 0.0,
"nuanced_min_participants": 59,
"nuanced_threshold_pct": 80,
},
}
result: dict[str, FormulaConfig] = {}
for name, params in configs.items():
instance, created = await get_or_create(
session, FormulaConfig, "name", name, **params,
)
status = "created" if created else "exists"
print(f" FormulaConfig '{name}': {status}")
result[name] = instance
return result
# ---------------------------------------------------------------------------
# Seed: VotingProtocols
# ---------------------------------------------------------------------------
async def seed_voting_protocols(
session: AsyncSession,
formulas: dict[str, FormulaConfig],
) -> dict[str, VotingProtocol]:
"""Create the 4 base voting protocols."""
protocols: dict[str, dict] = {
"Standard G1": {
"description": "Protocole binaire standard pour la Licence G1.",
"vote_type": "binary",
"formula_config_id": formulas["Standard Licence G1"].id,
"mode_params": "D30M50B.1G.2",
},
"Forgeron Smith": {
"description": "Protocole binaire avec critere Smith pour les forgerons.",
"vote_type": "binary",
"formula_config_id": formulas["Forgeron avec Smith"].id,
"mode_params": "D30M50B.1G.2S.1",
},
"Comite Tech": {
"description": "Protocole binaire avec critere Comite Technique.",
"vote_type": "binary",
"formula_config_id": formulas["Comite Tech"].id,
"mode_params": "D30M50B.1G.2T.1",
},
"Vote Nuance 6 niveaux": {
"description": "Protocole de vote nuance a 6 niveaux.",
"vote_type": "nuanced",
"formula_config_id": formulas["Vote Nuance"].id,
"mode_params": None,
},
}
result: dict[str, VotingProtocol] = {}
for name, params in protocols.items():
instance, created = await get_or_create(
session, VotingProtocol, "name", name, **params,
)
status = "created" if created else "exists"
print(f" VotingProtocol '{name}': {status}")
result[name] = instance
return result
# ---------------------------------------------------------------------------
# Seed: Document - Licence G1
# ---------------------------------------------------------------------------
LICENCE_G1_ITEMS: list[dict] = [
{
"position": "1",
"item_type": "preamble",
"title": "Preambule",
"sort_order": 1,
"current_text": (
"Licence de la monnaie libre et engagement de responsabilite. "
"La monnaie libre G1 (June) est co-produite par ses membres."
),
},
{
"position": "2",
"item_type": "section",
"title": "Avertissement TdC",
"sort_order": 2,
"current_text": (
"Certifier n'est pas uniquement s'assurer de l'identite unique "
"de la personne (son unicite). C'est aussi affirmer que vous la "
"connaissez bien et que vous saurez la joindre facilement."
),
},
{
"position": "3",
"item_type": "clause",
"title": "Conseils",
"sort_order": 3,
"current_text": (
"Connaitre la personne par plusieurs moyens de communication differents "
"(physique, electronique, etc.). Connaitre son lieu de vie principal. "
"Avoir echange avec elle en utilisant des moyens de communication "
"susceptibles d'identifier un humain vivant."
),
},
{
"position": "4",
"item_type": "verification",
"title": "Verifications",
"sort_order": 4,
"current_text": (
"De suffisamment bien connaitre la personne pour pouvoir la contacter, "
"echanger avec elle. De s'assurer que la personne a bien le controle "
"de son compte Duniter."
),
},
{
"position": "5",
"item_type": "rule",
"title": "Regles TdC",
"sort_order": 5,
"current_text": (
"Chaque membre dispose de 100 certifications possibles. "
"Il est possible de certifier 1 nouveau membre tous les 5 jours. "
"Un membre doit avoir au moins 5 certifications pour devenir membre. "
"Un membre doit renouveler son adhesion tous les 2 ans."
),
},
{
"position": "6",
"item_type": "rule",
"title": "Production DU",
"sort_order": 6,
"current_text": (
"1 DU (Dividende Universel) est produit par personne et par jour. "
"Le DU est la monnaie de base co-produite par chaque membre."
),
},
{
"position": "7",
"item_type": "rule",
"title": "Code monetaire",
"sort_order": 7,
"current_text": (
"DU formule : DU(t+1) = DU(t) + c^2 * M/N. "
"c = 4.88% / an. Le DU est re-evalue chaque equinoxe."
),
},
{
"position": "8",
"item_type": "clause",
"title": "Logiciels",
"sort_order": 8,
"current_text": (
"Les logiciels G1 doivent transmettre cette licence integralement "
"aux utilisateurs et developper un acces libre au code source."
),
},
{
"position": "9",
"item_type": "clause",
"title": "Modification",
"sort_order": 9,
"current_text": (
"Proposants, soutiens et votants doivent etre membres de la TdC. "
"Toute modification de cette licence doit etre soumise au vote "
"des membres selon le protocole en vigueur."
),
},
]
async def seed_document_licence_g1(session: AsyncSession) -> Document:
"""Create the Licence G1 document with its items."""
doc, created = await get_or_create(
session,
Document,
"slug",
"licence-g1",
title="Licence G1",
doc_type="licence",
version="0.3.0",
status="active",
description=(
"Licence de la monnaie libre G1 (June). "
"Definit les regles de la toile de confiance et du Dividende Universel."
),
)
print(f" Document 'Licence G1': {'created' if created else 'exists'}")
if created:
for item_data in LICENCE_G1_ITEMS:
item = DocumentItem(document_id=doc.id, **item_data)
session.add(item)
await session.flush()
print(f" -> {len(LICENCE_G1_ITEMS)} items created")
return doc
# ---------------------------------------------------------------------------
# Seed: Document - Engagement Forgeron v2.0.0
# ---------------------------------------------------------------------------
FORGERON_ITEMS: list[dict] = [
{
"position": "1",
"item_type": "preamble",
"title": "Intention",
"sort_order": 1,
"current_text": (
"Avec la V2, une sous-toile de confiance pour les forgerons est "
"introduite. Les forgerons (validateurs de blocs) doivent demontrer "
"leurs competences techniques et leur engagement envers le reseau."
),
},
{
"position": "2",
"item_type": "clause",
"title": "Savoirs-faire",
"sort_order": 2,
"current_text": (
"Administration systeme Linux, securite informatique, "
"cryptographie, blockchain Substrate. Le forgeron doit maitriser "
"l'ensemble de la chaine technique necessaire a la validation."
),
},
{
"position": "3",
"item_type": "clause",
"title": "Rigueur",
"sort_order": 3,
"current_text": (
"Comprendre en profondeur les configurations du runtime, "
"les parametres de consensus et les mecanismes de mise a jour "
"du reseau Duniter V2."
),
},
{
"position": "4",
"item_type": "clause",
"title": "Reactivite",
"sort_order": 4,
"current_text": (
"Reponse sous 24h aux alertes reseau. Disponibilite pour les "
"mises a jour critiques. Monitoring continu du noeud validateur."
),
},
{
"position": "5",
"item_type": "verification",
"title": "Securite aspirant",
"sort_order": 5,
"current_text": (
"Phrases aleatoires de 12+ mots, comptes separes pour identite "
"et validation, sauvegardes chiffrees des cles, infrastructure "
"securisee et a jour."
),
},
{
"position": "6",
"item_type": "verification",
"title": "Contact aspirant",
"sort_order": 6,
"current_text": (
"Le candidat forgeron doit contacter au minimum 3 forgerons "
"existants par au moins 2 canaux de communication differents "
"avant de demander ses certifications."
),
},
{
"position": "7",
"item_type": "clause",
"title": "Clauses pieges",
"sort_order": 7,
"current_text": (
"Exclusions : harcelement, abus de pouvoir, tentative "
"d'infiltration malveillante du reseau. Tout manquement "
"entraine le retrait des certifications forgeron."
),
},
{
"position": "8",
"item_type": "verification",
"title": "Securite certificateur",
"sort_order": 8,
"current_text": (
"Verification de l'intention du candidat, de ses pratiques "
"de securite, et du bon fonctionnement de son noeud validateur "
"avant de delivrer une certification forgeron."
),
},
{
"position": "9",
"item_type": "rule",
"title": "Regles TdC forgerons",
"sort_order": 9,
"current_text": (
"Etre membre de la TdC principale. Recevoir une invitation "
"d'un forgeron existant. Obtenir au minimum 3 certifications "
"de forgerons actifs. Renouvellement annuel obligatoire."
),
},
]
async def seed_document_forgeron(session: AsyncSession) -> Document:
"""Create the Engagement Forgeron v2.0.0 document with its items."""
doc, created = await get_or_create(
session,
Document,
"slug",
"engagement-forgeron",
title="Engagement Forgeron v2.0.0",
doc_type="engagement",
version="2.0.0",
status="active",
description=(
"Engagement des forgerons (validateurs) pour Duniter V2. "
"Adopte en fevrier 2026 (97 pour / 23 contre)."
),
)
print(f" Document 'Engagement Forgeron v2.0.0': {'created' if created else 'exists'}")
if created:
for item_data in FORGERON_ITEMS:
item = DocumentItem(document_id=doc.id, **item_data)
session.add(item)
await session.flush()
print(f" -> {len(FORGERON_ITEMS)} items created")
return doc
# ---------------------------------------------------------------------------
# Seed: Decision template - Processus Runtime Upgrade
# ---------------------------------------------------------------------------
RUNTIME_UPGRADE_STEPS: list[dict] = [
{
"step_order": 1,
"step_type": "qualification",
"title": "Qualification",
"description": (
"Definir le changement : specification technique, impact sur le "
"reseau, justification. Identifier les risques et dependances."
),
},
{
"step_order": 2,
"step_type": "review",
"title": "Revue",
"description": (
"Audit technique par le Comite Technique et les forgerons. "
"Revue du code, tests sur testnet, validation de la compatibilite."
),
},
{
"step_order": 3,
"step_type": "vote",
"title": "Vote",
"description": (
"Vote communautaire selon le protocole de vote en vigueur. "
"Le quorum et le seuil d'adoption dependent de la formule configuree."
),
},
{
"step_order": 4,
"step_type": "execution",
"title": "Execution",
"description": (
"Mise a jour on-chain via un extrinsic autorise. "
"Coordination avec les forgerons pour la synchronisation des noeuds."
),
},
{
"step_order": 5,
"step_type": "reporting",
"title": "Suivi",
"description": (
"Surveillance post-upgrade : monitoring des metriques reseau, "
"detection d'anomalies, rapport de stabilite sous 7 jours."
),
},
]
async def seed_decision_runtime_upgrade(session: AsyncSession) -> Decision:
"""Create the Runtime Upgrade decision template."""
decision, created = await get_or_create(
session,
Decision,
"title",
"Processus Runtime Upgrade",
description=(
"Template de decision pour les mises a jour du runtime Duniter V2. "
"5 etapes : qualification, revue, vote, execution, suivi."
),
decision_type="runtime_upgrade",
status="draft",
)
print(f" Decision 'Processus Runtime Upgrade': {'created' if created else 'exists'}")
if created:
for step_data in RUNTIME_UPGRADE_STEPS:
step = DecisionStep(decision_id=decision.id, **step_data)
session.add(step)
await session.flush()
print(f" -> {len(RUNTIME_UPGRADE_STEPS)} steps created")
return decision
# ---------------------------------------------------------------------------
# Main seed runner
# ---------------------------------------------------------------------------
async def run_seed():
"""Execute all seed functions inside a single transaction."""
print("=" * 60)
print("Glibredecision - Seed Database")
print("=" * 60)
async with async_session() as session:
async with session.begin():
print("\n[1/5] Formula Configs...")
formulas = await seed_formula_configs(session)
print("\n[2/5] Voting Protocols...")
await seed_voting_protocols(session, formulas)
print("\n[3/5] Document: Licence G1...")
await seed_document_licence_g1(session)
print("\n[4/5] Document: Engagement Forgeron v2.0.0...")
await seed_document_forgeron(session)
print("\n[5/5] Decision: Processus Runtime Upgrade...")
await seed_decision_runtime_upgrade(session)
print("\n" + "=" * 60)
print("Seed complete.")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(run_seed())