"""Seed the database with real data from Duniter community documents. Sources: - Engagement Forgeron v2.0.0: https://forum.monnaie-libre.fr/t/33165 - Engagement Certification (Licence G1): monnaie-libre.fr/licence-g1/ - Runtime Upgrade process template Idempotent: checks if data already exists before inserting. """ from __future__ import annotations import asyncio import hashlib import uuid from datetime import datetime, timedelta, timezone 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 from app.models.user import DuniterIdentity from app.models.vote import VoteSession, Vote # --------------------------------------------------------------------------- # 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 def fake_signature(payload: str) -> str: """Generate a deterministic fake Ed25519-like signature for seed data.""" return hashlib.sha256(payload.encode()).hexdigest() # --------------------------------------------------------------------------- # Seed: FormulaConfigs # --------------------------------------------------------------------------- async def seed_formula_configs(session: AsyncSession) -> dict[str, FormulaConfig]: 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]: 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: Engagement Certification (Licence G1 - obligations certificateur) # --------------------------------------------------------------------------- ENGAGEMENT_CERTIFICATION_ITEMS: list[dict] = [ { "position": "C1", "item_type": "clause", "title": "Transmission de la licence", "sort_order": 1, "current_text": ( "Toute operation de certification d'un nouveau membre de la monnaie libre G1 " "doit prealablement s'accompagner de la transmission de cette licence " "de la monnaie libre G1 au certifie." ), }, { "position": "C2", "item_type": "clause", "title": "Connaissance suffisante du certifie", "sort_order": 2, "current_text": ( "Certifier n'est pas uniquement s'assurer que vous avez rencontre la personne, " "c'est assurer a la communaute G1 que vous connaissez suffisamment bien la " "personne que vous vous appretez a certifier." ), }, { "position": "C3", "item_type": "clause", "title": "Contact par plusieurs moyens", "sort_order": 3, "current_text": ( "Connaitre la personne par plusieurs moyens de contact differents " "(physique, electronique, etc.) permettant de verifier son identite " "et de maintenir un lien de confiance dans la duree." ), }, { "position": "C4", "item_type": "clause", "title": "Ne jamais certifier seul", "sort_order": 4, "current_text": ( "Ne certifiez jamais seul, mais accompagne d'au moins un autre membre " "de la TdC G1, pour garantir un double controle et eviter les " "certifications abusives." ), }, { "position": "C5", "item_type": "verification", "title": "Verification des certifications existantes", "sort_order": 5, "current_text": ( "Avant toute certification, assurez-vous de verifier si le compte du " "certifie a deja recu une ou plusieurs certifications, et de qui " "elles proviennent." ), }, { "position": "C6", "item_type": "verification", "title": "Verification de maitrise du compte", "sort_order": 6, "current_text": ( "Verifier la maitrise du compte par un transfert test : envoyer " "quelques G1 et demander un renvoi, afin de s'assurer que la personne " "controle bien sa cle privee." ), }, { "position": "C7", "item_type": "verification", "title": "Verification de la licence", "sort_order": 7, "current_text": ( "Verifiez que vos contacts ont bien etudie et compris la licence G1 " "a jour avant de proceder a la certification." ), }, { "position": "C8", "item_type": "verification", "title": "Document de revocation", "sort_order": 8, "current_text": ( "D'avoir bien verifie avec la personne concernee qu'elle a bien genere " "son document Duniter de revocation de compte, et qu'elle le conserve " "en lieu sur." ), }, { "position": "C9", "item_type": "clause", "title": "Rencontre physique ou multi-canal", "sort_order": 9, "current_text": ( "De rencontrer la personne physiquement, OU de verifier a distance " "le lien personne / cle publique par plusieurs moyens de communication " "differents et independants." ), }, ] async def seed_document_engagement_certification(session: AsyncSession) -> Document: 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=( "Obligations des certificateurs de la toile de confiance Duniter. " "Chaque clause est soumise au vote permanent de la communaute." ), ) 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: Engagement Forgeron v2.0.0 (real content from forum topic 33165) # --------------------------------------------------------------------------- ENGAGEMENT_FORGERON_ITEMS: list[dict] = [ # --- Aspirant Forgeron : Securite et conformite --- { "position": "A1", "item_type": "clause", "title": "Intention et motivation", "sort_order": 1, "current_text": ( "J'ai clarifie ce qui me motive a devenir forgeron, " "j'en assume les raisons." ), }, { "position": "A2", "item_type": "clause", "title": "Veille securite", "sort_order": 2, "current_text": ( "Je fais de la veille pour maintenir mes pratiques de securite " "systeme et reseau a jour." ), }, { "position": "A3", "item_type": "clause", "title": "Notifications forum", "sort_order": 3, "current_text": ( "J'ai active les notifications sur forum.duniter.org pour etre " "alerte des discussions importantes concernant le reseau." ), }, { "position": "A4", "item_type": "verification", "title": "Phrase de recuperation aleatoire", "sort_order": 4, "current_text": ( "Je confirme que ma phrase de recuperation a ete generee " "aleatoirement et n'est pas une phrase choisie par moi." ), }, { "position": "A5", "item_type": "verification", "title": "Compte separe", "sort_order": 5, "current_text": ( "J'utilise un autre compte pour mes transactions courantes ; " "le compte forgeron est strictement reserve a la validation." ), }, { "position": "A6", "item_type": "verification", "title": "Sauvegarde phrase de recuperation", "sort_order": 6, "current_text": ( "J'ai stocke ma phrase de recuperation sur plusieurs supports " "physiques distincts et securises." ), }, { "position": "A7", "item_type": "verification", "title": "Noeud a jour et synchronise", "sort_order": 7, "current_text": ( "Je gere deja un noeud a jour, correctement synchronise et " "joignable par les autres noeuds du reseau." ), }, { "position": "A8", "item_type": "verification", "title": "API unsafe non exposee", "sort_order": 8, "current_text": ( "J'ai veille a ne pas exposer publiquement l'api unsafe " "de mon noeud validateur." ), }, { "position": "A9", "item_type": "clause", "title": "Transparence technique", "sort_order": 9, "current_text": ( "Je fournis a la demande d'un autre forgeron, mes choix " "techniques (materiel, OS, configuration reseau)." ), }, { "position": "A10", "item_type": "clause", "title": "Declaration offline en cas de doute", "sort_order": 10, "current_text": ( "Je me declare offline en cas de doute sur la securite " "de mon noeud ou de mon infrastructure." ), }, { "position": "A11", "item_type": "clause", "title": "Reactivite 24h", "sort_order": 11, "current_text": ( "Je m'engage a repondre en moins de 24h aux forgerons " "quand je suis declare online." ), }, # --- Aspirant Forgeron : Contact --- { "position": "A12", "item_type": "clause", "title": "Contact multi-canal", "sort_order": 12, "current_text": ( "Je sais joindre efficacement et rapidement 3 des forgerons " "par au moins 2 canaux de communication differents." ), }, # --- Aspirant Forgeron : Connaissances --- { "position": "A13", "item_type": "clause", "title": "Acceptation des engagements", "sort_order": 13, "current_text": ( "J'ai lu et j'accepte de respecter l'ensemble des " "engagements forgerons en vigueur." ), }, { "position": "A14", "item_type": "clause", "title": "Regles de la TdC forgeron", "sort_order": 14, "current_text": ( "J'ai pris connaissance des regles et delais associes au " "fonctionnement de la TdC forgeron." ), }, { "position": "A15", "item_type": "clause", "title": "Fonctionnement blockchain", "sort_order": 15, "current_text": ( "J'ai bien compris le fonctionnement d'un reseau blockchain " "Duniter et le role du validateur." ), }, # --- Aspirant Forgeron : Pieges (expected: NON) --- { "position": "A16", "item_type": "rule", "title": "Piege : harcelement", "sort_order": 16, "current_text": ( "[Piege - reponse attendue : NON] " "J'insiste, harcele ou fais pression pour etre certifie forgeron." ), }, { "position": "A17", "item_type": "rule", "title": "Piege : gloire et pouvoir", "sort_order": 17, "current_text": ( "[Piege - reponse attendue : NON] " "Je veux etre forgeron pour la gloire et le pouvoir." ), }, { "position": "A18", "item_type": "rule", "title": "Piege : nuisance", "sort_order": 18, "current_text": ( "[Piege - reponse attendue : NON] " "Je cherche a nuire a l'ecosysteme G1." ), }, # --- Certificateur Forgeron : Securite et conformite --- { "position": "C1", "item_type": "clause", "title": "Intention du certifie questionnee", "sort_order": 19, "current_text": ( "J'ai questionne l'intention du certifie a rejoindre " "les forgerons et verifie sa motivation." ), }, { "position": "C2", "item_type": "verification", "title": "Pratiques de securite du certifie", "sort_order": 20, "current_text": ( "J'ai demande au certifie quelles etaient ses pratiques de " "securite systeme et reseau." ), }, { "position": "C3", "item_type": "verification", "title": "Phrase aleatoire du certifie", "sort_order": 21, "current_text": ( "Le certifie m'assure que son compte forgeron est issu d'une " "phrase generee aleatoirement." ), }, { "position": "C4", "item_type": "verification", "title": "Sauvegarde du certifie", "sort_order": 22, "current_text": ( "Le certifie m'assure avoir stocke sa phrase de recuperation " "sur plusieurs supports physiques." ), }, { "position": "C5", "item_type": "verification", "title": "Noeud du certifie verifie", "sort_order": 23, "current_text": ( "J'ai verifie que le certifie gere deja un noeud a jour, " "correctement synchronise et joignable." ), }, { "position": "C6", "item_type": "clause", "title": "Configuration du certifie notee", "sort_order": 24, "current_text": ( "J'ai note le style de configuration du noeud du certifie " "(materiel, OS, hebergement)." ), }, { "position": "C7", "item_type": "clause", "title": "Engagement d'information du certifie", "sort_order": 25, "current_text": ( "Le certifie s'est engage a m'informer de tout changement " "significatif de sa configuration." ), }, { "position": "C8", "item_type": "verification", "title": "Risques offline connus du certifie", "sort_order": 26, "current_text": ( "J'ai verifie avec le certifie qu'il connait les risques " "d'etre declare offline et les consequences." ), }, # --- Certificateur Forgeron : Contact --- { "position": "C9", "item_type": "clause", "title": "Joindre les certifies", "sort_order": 27, "current_text": ( "Je sais joindre efficacement les forgerons que j'ai certifies." ), }, { "position": "C10", "item_type": "clause", "title": "Deux canaux de contact", "sort_order": 28, "current_text": ( "Je peux les joindre par au moins 2 canaux differents." ), }, { "position": "C11", "item_type": "clause", "title": "Contact sous 24h en cas de defaut", "sort_order": 29, "current_text": ( "Je m'engage a contacter sous 24h ce forgeron si un defaut " "concerne son noeud." ), }, # --- Certificateur Forgeron : Connaissances --- { "position": "C12", "item_type": "verification", "title": "Engagements acceptes par le certifie", "sort_order": 30, "current_text": ( "J'ai verifie que le certifie a accepte les engagements " "forgerons integralement." ), }, { "position": "C13", "item_type": "verification", "title": "Regles consultables par le certifie", "sort_order": 31, "current_text": ( "J'ai verifie que le certifie sait ou consulter les regles " "detaillees de la TdC forgeron." ), }, { "position": "C14", "item_type": "verification", "title": "Delais connus du certifie", "sort_order": 32, "current_text": ( "J'ai verifie que le certifie connait les delais de passage " "en ligne et hors ligne." ), }, # --- Certificateur Forgeron : Pieges (expected: NON) --- { "position": "C15", "item_type": "rule", "title": "Piege : certification sous pression", "sort_order": 33, "current_text": ( "[Piege - reponse attendue : NON] " "Je certifie sous la menace ou autre forme de pression." ), }, { "position": "C16", "item_type": "rule", "title": "Piege : avantage personnel", "sort_order": 34, "current_text": ( "[Piege - reponse attendue : NON] " "Je tire un avantage personnel en echange de ma certification." ), }, ] async def seed_document_engagement_forgeron(session: AsyncSession) -> Document: 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) Duniter V2. " "Adopte en fevrier 2026 (97 pour / 23 contre). " "34 clauses : aspirant (18) + certificateur (16)." ), ) 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 - 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: 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 # --------------------------------------------------------------------------- # Seed: Simulated voters + votes on first 3 engagement items # --------------------------------------------------------------------------- VOTER_NAMES = [ "Moul", "Poka", "Hugo", "Elois", "Cgeek", "Galuel", "Tortue", "Inso", "Tuxmain", "Matograine", "Maaltir", ] async def seed_voters(session: AsyncSession) -> list[DuniterIdentity]: """Create 11 simulated Duniter identities for voting demo.""" voters: list[DuniterIdentity] = [] for i, name in enumerate(VOTER_NAMES): # Deterministic address from name addr_hash = hashlib.sha256(name.encode()).hexdigest()[:32] address = f"5{addr_hash[:47]}" voter, created = await get_or_create( session, DuniterIdentity, "address", address, display_name=name, wot_status="member", is_smith=(i < 5), # First 5 are smiths ) if created: print(f" Voter '{name}': created") voters.append(voter) return voters async def seed_votes_on_items( session: AsyncSession, doc: Document, protocol: VotingProtocol, voters: list[DuniterIdentity], ): """Create vote sessions on first 3 items with 10 for + 1 against.""" # Fetch items for this document stmt = select(DocumentItem).where( DocumentItem.document_id == doc.id ).order_by(DocumentItem.sort_order).limit(3) result = await session.execute(stmt) items = result.scalars().all() if not items: print(" No items found to vote on") return now = datetime.now(timezone.utc) for item in items: # Check if a session already exists for this item existing_stmt = select(VoteSession).where( VoteSession.item_version_id == None, VoteSession.voting_protocol_id == protocol.id, ) # We'll use a simpler idempotency check session_id = uuid.uuid5(uuid.NAMESPACE_URL, f"seed-vote-{item.id}") check_stmt = select(VoteSession).where(VoteSession.id == session_id) check_result = await session.execute(check_stmt) if check_result.scalar_one_or_none() is not None: print(f" VoteSession for '{item.title}': exists") continue vote_session = VoteSession( id=session_id, decision_id=None, item_version_id=None, voting_protocol_id=protocol.id, wot_size=7224, smith_size=23, techcomm_size=5, starts_at=now - timedelta(days=15), ends_at=now + timedelta(days=15), status="open", votes_for=10, votes_against=1, votes_total=11, threshold_required=97.0, ) session.add(vote_session) await session.flush() # 10 votes "for" for voter in voters[:10]: payload = f"vote:{vote_session.id}:{voter.id}:for" vote = Vote( session_id=vote_session.id, voter_id=voter.id, vote_value="for", comment="Oui c'est mieux que l'existant", signature=fake_signature(payload), signed_payload=payload, voter_wot_status="member", voter_is_smith=voter.is_smith, ) session.add(vote) # 1 vote "against" against_voter = voters[10] payload = f"vote:{vote_session.id}:{against_voter.id}:against" vote = Vote( session_id=vote_session.id, voter_id=against_voter.id, vote_value="against", comment="Non, on ne remplace pas tel quel", signature=fake_signature(payload), signed_payload=payload, voter_wot_status="member", voter_is_smith=False, ) session.add(vote) await session.flush() print(f" VoteSession for '{item.title}': created (10 pour, 1 contre)") # --------------------------------------------------------------------------- # Main seed runner # --------------------------------------------------------------------------- async def run_seed(): print("=" * 60) print("Glibredecision - Seed Database") print("=" * 60) async with async_session() as session: async with session.begin(): print("\n[1/7] Formula Configs...") formulas = await seed_formula_configs(session) print("\n[2/7] Voting Protocols...") protocols = await seed_voting_protocols(session, formulas) print("\n[3/7] Document: Acte d'engagement certification...") await seed_document_engagement_certification(session) print("\n[4/7] Document: Acte d'engagement forgeron v2.0.0...") doc_forgeron = await seed_document_engagement_forgeron(session) print("\n[5/7] Decision: Runtime Upgrade...") await seed_decision_runtime_upgrade(session) print("\n[6/7] Simulated voters...") voters = await seed_voters(session) print("\n[7/7] Votes on first 3 engagements forgeron...") await seed_votes_on_items( session, doc_forgeron, protocols["Vote majoritaire"], voters, ) print("\n" + "=" * 60) print("Seed complete.") print("=" * 60) if __name__ == "__main__": asyncio.run(run_seed())