"""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 (premier pack de modalites) # --------------------------------------------------------------------------- async def seed_voting_protocols( session: AsyncSession, formulas: dict[str, FormulaConfig], ) -> dict[str, VotingProtocol]: """Create the first pack of voting modalities (3 protocols).""" protocols: dict[str, dict] = { "Vote majoritaire": { "description": ( "Vote binaire a majorite simple. Le seuil d'adoption " "s'adapte dynamiquement au taux de participation via " "la formule d'inertie WoT." ), "vote_type": "binary", "formula_config_id": formulas["Standard Licence G1"].id, "mode_params": "D30M50B.1G.2", }, "Vote quadratique": { "description": ( "Vote pondere par la racine carree des certifications. " "Reduit l'influence des gros certificateurs et favorise " "une participation large et diversifiee." ), "vote_type": "binary", "formula_config_id": formulas["Forgeron avec Smith"].id, "mode_params": "D30M50B.1G.2S.1", }, "Vote permanent": { "description": ( "Vote continu sans date de fin. Le resultat evolue en " "temps reel avec chaque nouveau vote. Adapte aux documents " "de reference sous revision permanente." ), "vote_type": "binary", "formula_config_id": formulas["Comite Tech"].id, "mode_params": "D30M50B.1G.2T.1", }, } 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 - Acte d'engagement certification # --------------------------------------------------------------------------- ENGAGEMENT_CERTIFICATION_ITEMS: list[dict] = [ { "position": "1", "item_type": "preamble", "title": "Objet", "sort_order": 1, "current_text": ( "Le present acte definit les engagements de tout membre de la " "toile de confiance qui certifie l'identite d'une autre personne " "dans le reseau Duniter." ), }, { "position": "2", "item_type": "clause", "title": "Connaissance personnelle", "sort_order": 2, "current_text": ( "Je certifie connaitre personnellement la personne que je " "certifie, l'avoir rencontree physiquement a plusieurs reprises, " "et pouvoir la contacter par au moins deux moyens de communication " "differents." ), }, { "position": "3", "item_type": "clause", "title": "Verification d'identite", "sort_order": 3, "current_text": ( "Je certifie avoir verifie que la personne n'a qu'un seul compte " "membre dans la toile de confiance, et que l'identite declaree " "correspond a une personne humaine vivante." ), }, { "position": "4", "item_type": "clause", "title": "Engagement de suivi", "sort_order": 4, "current_text": ( "Je m'engage a surveiller l'activite de mes certifies et a " "signaler tout comportement suspect (comptes multiples, " "usurpation d'identite, comptes abandonnes)." ), }, { "position": "5", "item_type": "verification", "title": "Delai entre certifications", "sort_order": 5, "current_text": ( "Je respecte un delai minimum de reflexion de 5 jours entre " "chaque nouvelle certification emise." ), }, { "position": "6", "item_type": "rule", "title": "Renouvellement", "sort_order": 6, "current_text": ( "Je renouvelle mes certifications avant leur expiration pour " "maintenir la cohesion de la toile de confiance." ), }, { "position": "7", "item_type": "clause", "title": "Responsabilite", "sort_order": 7, "current_text": ( "Je suis conscient que la certification engage ma responsabilite " "vis-a-vis de la communaute. Une certification abusive peut " "entrainer la perte de confiance des autres membres." ), }, { "position": "8", "item_type": "rule", "title": "Revocation", "sort_order": 8, "current_text": ( "Une certification peut etre revoquee si les conditions de " "l'engagement ne sont plus remplies. La revocation est soumise " "au protocole de vote en vigueur." ), }, ] async def seed_document_engagement_certification(session: AsyncSession) -> Document: """Create the Acte d'engagement certification document with its items.""" doc, created = await get_or_create( session, Document, "slug", "engagement-certification", title="Acte d'engagement certification", doc_type="engagement", version="1.0.0", status="active", description=( "Acte d'engagement pour les certificateurs de la toile de confiance " "Duniter. Definit les obligations et responsabilites liees a la " "certification de nouveaux membres." ), ) print(f" Document 'Acte d'engagement certification': {'created' if created else 'exists'}") if created: for item_data in ENGAGEMENT_CERTIFICATION_ITEMS: item = DocumentItem(document_id=doc.id, **item_data) session.add(item) await session.flush() print(f" -> {len(ENGAGEMENT_CERTIFICATION_ITEMS)} items created") return doc # --------------------------------------------------------------------------- # Seed: Document - Acte d'engagement forgeron v2.0.0 # --------------------------------------------------------------------------- ENGAGEMENT_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_engagement_forgeron(session: AsyncSession) -> Document: """Create the Acte d'engagement forgeron v2.0.0 document with its items.""" doc, created = await get_or_create( session, Document, "slug", "engagement-forgeron", title="Acte d'engagement forgeron", doc_type="engagement", version="2.0.0", status="active", description=( "Acte d'engagement des forgerons (validateurs de blocs) pour " "Duniter V2. Adopte en fevrier 2026 (97 pour / 23 contre)." ), ) print(f" Document 'Acte d'engagement forgeron': {'created' if created else 'exists'}") if created: for item_data in ENGAGEMENT_FORGERON_ITEMS: item = DocumentItem(document_id=doc.id, **item_data) session.add(item) await session.flush() print(f" -> {len(ENGAGEMENT_FORGERON_ITEMS)} items created") return doc # --------------------------------------------------------------------------- # Seed: Decision template - 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", "Runtime Upgrade", description=( "Processus de mise a jour du runtime Duniter V2 on-chain. " "Chaque upgrade suit un protocole strict en 5 etapes. " "(Decision on-chain)" ), decision_type="runtime_upgrade", status="draft", ) print(f" Decision '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 (premier pack de modalites)...") await seed_voting_protocols(session, formulas) print("\n[3/5] Document: Acte d'engagement certification...") await seed_document_engagement_certification(session) print("\n[4/5] Document: Acte d'engagement forgeron...") await seed_document_engagement_forgeron(session) print("\n[5/5] Decision: Runtime Upgrade...") await seed_decision_runtime_upgrade(session) print("\n" + "=" * 60) print("Seed complete.") print("=" * 60) if __name__ == "__main__": asyncio.run(run_seed())