"""Operations sysadmin de remise a zero et de donnees de demonstration."""
from __future__ import annotations
import csv
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import Base
from app.core.security import hash_pin
from app.models.affectation import Affectation
from app.models.candidature import Candidature
from app.models.collecte import Collecte
from app.models.config_repartition import ConfigRepartition
from app.models.depense import Depense
from app.models.destinataire_contribution import DestinataireContribution
from app.models.envoi_contribution import EnvoiContribution
from app.models.evenement import Evenement
from app.models.groupe import Groupe
from app.models.jeton import Jeton
from app.models.litterature import Litterature
from app.models.membre import Membre
from app.models.mouvement_reserve import MouvementReserve
from app.models.operation_bancaire import OperationBancaire
from app.models.poste import Poste
from app.models.poste_module import PosteModule
from app.models.presence import Presence
from app.models.proposition import Proposition
from app.models.pv_reunion import PvReunion
from app.models.rapport_adopte import RapportAdopte
from app.models.rapport_rsg import RapportRsg
from app.models.reserve import Reserve
from app.models.reunion import Reunion
from app.models.transaction_comptable import TransactionComptable
from app.models.vente_litterature import VenteLitterature
from app.models.vente_jeton import VenteJeton
POSTES = [
("RSG", "executif", "Représente le groupe auprès du district et rapporte les informations de service.", "2 ans de stabilité", 1, ["M3", "C2", "C5"]),
("RSG Substitut(e)", "executif", "Appuie le RSG et assure la relève lorsque nécessaire.", "1 an de stabilité", 2, ["M3", "C2", "C5"]),
("Trésorier(ière)", "executif", "Tient les registres, reçoit les sommes et présente les rapports mensuels.", "1 an de stabilité", 3, ["S1T", "S5", "M4", "G8"]),
("Secrétaire", "executif", "Prépare les procès-verbaux, les présences et la correspondance.", "6 mois de stabilité", 4, ["M0", "M1", "M2", "S0"]),
("Animateur(trice)", "service", "Anime les rencontres hebdomadaires.", "Disponibilité régulière", 5, ["S0"]),
("Responsable accueil", "physique", "Accueille les nouveaux et veille à l'atmosphère de la salle.", "Fiabilité", 6, ["C9"]),
("Responsable café", "physique", "Prépare le café et remet les petites dépenses au trésorier.", "Aucun", 7, ["S5"]),
("Responsable littérature", "service", "Maintient l'inventaire et remet les ventes au trésorier.", "Connaissance de la littérature", 8, ["S3", "C4", "M5"]),
("Responsable jetons", "service", "Prépare les jetons et les anniversaires.", "Aucun", 9, ["S2", "M6"]),
("Responsable clés", "physique", "Ouvre et ferme le local.", "Fiabilité", 10, []),
]
MEMBRES_DEMO = [
("Sysadmin", None, "sysadmin", "0000", "514-555-0000"),
("Claire", "B.", "claire.demo@example.test", "1111", "514-555-0101"),
("Marc", "L.", "marc.demo@example.test", "2222", "514-555-0102"),
("Nadia", "R.", "nadia.demo@example.test", "3333", "514-555-0103"),
("Simon", "T.", "simon.demo@example.test", "4444", "514-555-0104"),
("Julie", "P.", "julie.demo@example.test", "5555", "514-555-0105"),
("Karim", "A.", "karim.demo@example.test", "6666", "514-555-0106"),
("Mélanie", "S.", "melanie.demo@example.test", "7777", "514-555-0107"),
("Robert", "G.", "robert.demo@example.test", "8888", "514-555-0108"),
("Anne", "C.", "anne.demo@example.test", "9999", "514-555-0109"),
("Luc", "D.", "luc.demo@example.test", "1212", "514-555-0110"),
("Sophie", "M.", "sophie.demo@example.test", "3434", "514-555-0111"),
]
def _catalogue_litterature() -> list[dict]:
chemin = Path(__file__).resolve().parents[3] / "data" / "import" / "Liste-de-prix-rev.-Fevrier-2026.csv"
if chemin.exists():
with chemin.open(newline="", encoding="utf-8") as fichier:
return [
{
"code_bsg": row["code_bsg"].strip() or None,
"titre": row["titre"].strip(),
"categorie": row["categorie"].strip() or "brochure",
"prix_coutant": float(row["prix_coutant"]) if row["prix_coutant"].strip() else 0,
}
for row in csv.DictReader(fichier)
if row.get("titre", "").strip()
]
return [
{"code_bsg": "B-1", "titre": "Les Alcooliques anonymes", "categorie": "livre", "prix_coutant": 12.0},
{"code_bsg": "B-2", "titre": "Les Douze Étapes et les Douze Traditions", "categorie": "livre", "prix_coutant": 11.0},
{"code_bsg": "P-3", "titre": "Les AA sont-ils pour vous?", "categorie": "brochure", "prix_coutant": 0.0},
]
async def remettre_a_zero(db: AsyncSession) -> dict:
tables = ", ".join(table.name for table in reversed(Base.metadata.sorted_tables))
if tables:
await db.execute(text(f"TRUNCATE TABLE {tables} RESTART IDENTITY CASCADE"))
groupe, sysadmin, postes, litterature = await _creer_base(db)
await db.flush()
return {
"message": "Base remise à zéro",
"groupe_id": groupe.id,
"sysadmin_id": sysadmin.id,
"postes": len(postes),
"litterature": len(litterature),
}
async def initialiser_demo(db: AsyncSession) -> dict:
await remettre_a_zero(db)
groupe = await _premier_groupe(db)
membres = await _membres_par_courriel(db)
postes = await _postes_par_nom(db)
litterature = await _litterature(db)
await _ajouter_membres_demo(db, groupe, membres, postes)
membres = await _membres_par_courriel(db)
await _ajouter_historique_demo(db, groupe, membres, postes, litterature)
await db.flush()
return {
"message": "Base remise à zéro et données de démonstration installées",
"groupe_id": groupe.id,
"membres": len(membres),
"periode_mois": 12,
}
async def _creer_base(db: AsyncSession):
groupe = Groupe(
nom="Groupe Méditation",
district="87-16",
region="87",
adresse_local="Salle communautaire, Montréal",
jour_reunion="mardi",
heure_reunion=time(19, 30),
assemblee_affaires_ordre="premiere",
format_reunion="Discussion ouverte",
lien_aa87="https://aa87.org",
)
db.add(groupe)
await db.flush()
postes = {}
for nom, categorie, description, prerequis, ordre, modules in POSTES:
poste = Poste(
nom=nom,
categorie=categorie,
description=description,
prerequis=prerequis,
ordre_affichage=ordre,
actif=True,
)
db.add(poste)
await db.flush()
postes[nom] = poste
for module_code in modules:
db.add(PosteModule(poste_id=poste.id, module_code=module_code))
sysadmin = Membre(
groupe_id=groupe.id,
prenom="Sysadmin",
nom=None,
telephone="514-555-0000",
courriel="sysadmin",
pin_hash=hash_pin("0000"),
actif=True,
)
db.add(sysadmin)
await db.flush()
litterature = []
for item in _catalogue_litterature():
lit = Litterature(
groupe_id=groupe.id,
titre=item["titre"][:200],
categorie=item["categorie"] if item["categorie"] in {"livre", "brochure", "depliant", "pamphlet"} else "brochure",
prix_coutant=item["prix_coutant"],
quantite_stock=8,
seuil_commande=2,
code_bsg=item["code_bsg"],
)
db.add(lit)
litterature.append(lit)
db.add(DestinataireContribution(groupe_id=groupe.id, nom="District 87-16", protege=True, actif=True))
db.add(DestinataireContribution(groupe_id=groupe.id, nom="Région 87", protege=False, actif=True))
db.add(DestinataireContribution(groupe_id=groupe.id, nom="Bureau des services généraux", protege=False, actif=True))
for destination, pct in [("district", 50), ("region", 20), ("bsg", 20), ("intergroupe", 10)]:
db.add(ConfigRepartition(groupe_id=groupe.id, destination=destination, pourcentage=pct, actif_depuis=date.today().replace(day=1)))
return groupe, sysadmin, postes, litterature
async def _ajouter_membres_demo(db: AsyncSession, groupe: Groupe, membres: dict, postes: dict) -> None:
for prenom, nom, courriel, pin, telephone in MEMBRES_DEMO[1:]:
membre = Membre(
groupe_id=groupe.id,
prenom=prenom,
nom=nom,
telephone=telephone,
courriel=courriel,
date_abstinence=date.today() - timedelta(days=365 * (2 + len(membres) % 12)),
pin_hash=hash_pin(pin),
actif=True,
)
db.add(membre)
await db.flush()
membres[courriel] = membre
affectations = [
("claire.demo@example.test", "RSG"),
("marc.demo@example.test", "RSG Substitut(e)"),
("nadia.demo@example.test", "Trésorier(ière)"),
("simon.demo@example.test", "Secrétaire"),
("julie.demo@example.test", "Animateur(trice)"),
("karim.demo@example.test", "Responsable littérature"),
("melanie.demo@example.test", "Responsable jetons"),
("robert.demo@example.test", "Responsable accueil"),
("anne.demo@example.test", "Responsable café"),
]
debut = date.today() - timedelta(days=240)
for courriel, poste_nom in affectations:
membre = membres.get(courriel)
if membre and poste_nom in postes:
db.add(Affectation(groupe_id=groupe.id, membre_id=membre.id, poste_id=postes[poste_nom].id, date_debut=debut, statut="actif"))
async def _ajouter_historique_demo(db: AsyncSession, groupe: Groupe, membres: dict, postes: dict, litterature: list[Litterature]) -> None:
tresorier = membres["nadia.demo@example.test"]
secretaire = membres["simon.demo@example.test"]
rsg = membres["claire.demo@example.test"]
membres_actifs = [m for m in membres.values() if m.courriel != "sysadmin"]
reserves = {
"Loyer": Reserve(groupe_id=groupe.id, nom="Loyer", montant=0, note="Prévoir trois mois de salle"),
"Littérature": Reserve(groupe_id=groupe.id, nom="Littérature", montant=0, note="Réapprovisionnement"),
"Prudentielle": Reserve(groupe_id=groupe.id, nom="Prudentielle", montant=0, note="Réserve de sécurité"),
}
for reserve in reserves.values():
db.add(reserve)
await db.flush()
debut = _premier_mardi(date.today() - timedelta(days=365))
dates_reunions = [debut + timedelta(days=7 * i) for i in range(52)]
mois_vus = set()
banque = 0.0
encaisse = 0.0
compteurs = {"RES": 0, "NOM": 0, "DON": 0}
for index, jour in enumerate(dates_reunions):
type_reunion = "affaires" if jour.month not in mois_vus else "reguliere"
mois_vus.add(jour.month)
reunion = Reunion(
groupe_id=groupe.id,
type=type_reunion,
date_reunion=datetime.combine(jour, groupe.heure_reunion or time(19, 30)).replace(tzinfo=timezone.utc),
animateur_id=membres_actifs[index % len(membres_actifs)].id,
theme=["Acceptation", "Service", "Gratitude", "Honnêteté"][index % 4],
annonces={"district": index % 5 == 0, "anniversaire": index % 8 == 0},
note="Réunion de démonstration réaliste.",
)
db.add(reunion)
await db.flush()
for membre in membres_actifs:
db.add(Presence(reunion_id=reunion.id, membre_id=membre.id, present=(index + len(membre.prenom)) % 7 != 0))
montant_collecte = round(82 + (index % 6) * 9 + (index % 3) * 4, 2)
collecte = Collecte(
groupe_id=groupe.id,
date_collecte=jour,
montant=montant_collecte,
reunion_id=reunion.id,
detenteur_id=membres_actifs[index % len(membres_actifs)].id,
statut="recu",
saisi_par=secretaire.id,
confirme_par=secretaire.id,
date_confirmation=datetime.combine(jour, time(21, 15)).replace(tzinfo=timezone.utc),
recu_par=tresorier.id,
date_reception=jour,
)
db.add(collecte)
await db.flush()
encaisse += montant_collecte
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="collecte", compte="encaisse", montant=montant_collecte, sens="debit", source_table="collectes", source_id=collecte.id, note="Collecte 7e Tradition", cree_par=tresorier.id))
if index % 4 == 0:
don = 40 + (index % 5) * 10
banque += don
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="don", compte="banque", montant=don, sens="debit", note="Don direct au compte bancaire", cree_par=tresorier.id))
if index % 4 == 3:
depot = round(encaisse - 25, 2) if encaisse > 50 else round(encaisse, 2)
if depot > 0:
encaisse -= depot
banque += depot
db.add(OperationBancaire(groupe_id=groupe.id, type="depot", montant=depot, date_operation=jour, effectue_par=tresorier.id, note="Dépôt mensuel de l'encaisse"))
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="banque_depot", compte="banque", montant=depot, sens="debit", note="Dépôt de l'encaisse", cree_par=tresorier.id))
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="banque_depot", compte="encaisse", montant=depot, sens="credit", note="Dépôt de l'encaisse", cree_par=tresorier.id))
if type_reunion == "affaires":
await _decision(db, groupe, reunion, compteurs, "RES", jour, secretaire, rsg, "Adopter l'ordre du jour", "L'assemblée doit traiter les affaires courantes.", "L'ordre du jour est adopté tel que présenté.")
await _decision(db, groupe, reunion, compteurs, "RES", jour, rsg, secretaire, f"Orientation du mois {jour.month:02d}", "Le groupe souhaite garder une conscience de groupe active.", "Le groupe retient les priorités présentées pour le mois.")
pv = PvReunion(
reunion_id=reunion.id,
redacteur_id=secretaire.id,
contenu=f"Procès-verbal de l'assemblée du {jour.isoformat()}.\nPrésences vérifiées. Rapports reçus. Décisions consignées.",
decisions=[{"sujet": "Adoption du procès-verbal précédent", "decision": "Adopté", "proposeur": rsg.prenom, "secondeur": secretaire.prenom, "statut": "complete"}],
approuve=index > 4,
)
db.add(pv)
db.add(RapportAdopte(
groupe_id=groupe.id,
adopte_par_id=tresorier.id,
type_rapport="tresorerie",
titre=f"Rapport du trésorier — {jour.strftime('%Y-%m')}",
annee=jour.year,
mois=jour.month,
contenu_html=f"Rapport du trésorier
Rapport adopté pour {jour.strftime('%Y-%m')}.
Banque: {banque:.2f} $
Encaisse: {encaisse:.2f} $
",
donnees={"banque": round(banque, 2), "encaisse": round(encaisse, 2)},
))
if index % 6 == 1:
montant = 32 + (index % 4) * 6
methode = "COLLECTE" if encaisse >= montant else "VIREMENT"
if methode == "COLLECTE":
encaisse -= montant
compte = "encaisse"
else:
banque -= montant
compte = "banque"
depense = Depense(groupe_id=groupe.id, date_depense=jour, montant=montant, qui=membres_actifs[index % len(membres_actifs)].id, quoi=["Café et lait", "Papeterie", "Location de salle", "Fournitures accueil"][index % 4], paye=True, methode=methode, saisi_par=membres_actifs[index % len(membres_actifs)].id, confirme_par=tresorier.id, date_confirmation=datetime.combine(jour, time(20, 45)).replace(tzinfo=timezone.utc))
db.add(depense)
await db.flush()
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="depense", compte=compte, montant=montant, sens="credit", source_table="depenses", source_id=depense.id, note=depense.quoi, cree_par=tresorier.id))
if index % 8 == 2 and litterature:
item = litterature[index % len(litterature)]
montant = float(item.prix_coutant or 2) * 2
item.quantite_stock = max(0, item.quantite_stock - 2)
vente = VenteLitterature(groupe_id=groupe.id, reunion_id=reunion.id, litterature_id=item.id, quantite=2, montant=montant, date_vente=jour, vendu_par=membres["karim.demo@example.test"].id, remis_tresorier=True, recu_par=tresorier.id, date_remise=jour)
db.add(vente)
await db.flush()
encaisse += montant
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="vente_litterature", compte="encaisse", montant=montant, sens="debit", source_table="ventes_litterature", source_id=vente.id, note="Vente de littérature", cree_par=tresorier.id))
if index % 12 == 6:
montant = 6
vente_jeton = VenteJeton(
groupe_id=groupe.id,
reunion_id=reunion.id,
type_jeton=["desir", "1mois", "1an"][index % 3],
quantite=2,
montant=montant,
date_vente=jour,
vendu_par=membres["melanie.demo@example.test"].id,
remis_tresorier=True,
recu_par=tresorier.id,
date_remise=jour,
)
db.add(vente_jeton)
await db.flush()
encaisse += montant
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="vente_jeton", compte="encaisse", montant=montant, sens="debit", source_table="ventes_jetons", source_id=vente_jeton.id, note="Vente de jetons", cree_par=tresorier.id))
if index % 13 == 4:
montant = 125
if banque >= montant:
banque -= montant
decision = await _decision(db, groupe, reunion, compteurs, "DON", jour, rsg, secretaire, "Contribution au district", "La conscience de groupe prévoit soutenir les services.", "Le groupe décide d'envoyer une contribution au District 87-16.")
contribution = EnvoiContribution(groupe_id=groupe.id, date_envoi=jour, destination="District 87-16", montant=montant, methode="VIREMENT", resolution_id=decision.id, engagee=True, debitee=True, saisi_par=tresorier.id)
db.add(contribution)
await db.flush()
db.add(TransactionComptable(groupe_id=groupe.id, date_transaction=jour, type="contribution", compte="banque", montant=montant, sens="credit", source_table="envois_contributions", source_id=contribution.id, note="Contribution au district", cree_par=tresorier.id))
if index in {9, 22, 35}:
candidat = membres_actifs[index % len(membres_actifs)]
poste = postes[["Responsable accueil", "Responsable café", "Responsable clés"][index % 3]]
candidature = Candidature(groupe_id=groupe.id, poste_id=poste.id, membre_id=candidat.id, type="nomination", statut="acceptee", motivation="Disponibilité confirmée.", nomme_par=rsg.id, acceptee_par_membre=True)
db.add(candidature)
await db.flush()
decision = await _decision(db, groupe, reunion, compteurs, "NOM", jour, rsg, secretaire, f"Nomination de {candidat.prenom}", "Un poste doit être pourvu pour soutenir le groupe.", f"Le groupe nomme {candidat.prenom} au poste de {poste.nom}.")
decision.candidature_id = candidature.id
candidature.proposition_id = decision.id
if index % 10 == 5:
db.add(Jeton(groupe_id=groupe.id, reunion_id=reunion.id, membre_id=membres_actifs[index % len(membres_actifs)].id, type_jeton="multiple", date_remise=jour, annees=1 + index % 5, remis_par=membres["melanie.demo@example.test"].id))
if index % 17 == 7:
reserve = reserves[["Loyer", "Littérature", "Prudentielle"][index % 3]]
montant = 75
if banque >= montant:
reserve.montant = float(reserve.montant or 0) + montant
db.add(MouvementReserve(groupe_id=groupe.id, source_id=None, destination_id=reserve.id, montant=montant, type="reserve_virement", note="Affectation au solde réservé", cree_par=tresorier.id, cree_le=datetime.combine(jour, time(21, 0)).replace(tzinfo=timezone.utc)))
db.add(Collecte(groupe_id=groupe.id, date_collecte=dates_reunions[-1] + timedelta(days=7), montant=96, detenteur_id=membres["luc.demo@example.test"].id, statut="en_attente", saisi_par=membres["luc.demo@example.test"].id))
db.add(Depense(groupe_id=groupe.id, date_depense=date.today(), montant=48, qui=membres["anne.demo@example.test"].id, quoi="Fournitures à confirmer", paye=False, methode=None, saisi_par=membres["anne.demo@example.test"].id))
db.add(Proposition(groupe_id=groupe.id, sujet="Réviser le format de discussion", description="Proposition laissée en réflexion pour la prochaine assemblée.", proposeur_id=membres["julie.demo@example.test"].id, secondeur_id=membres["marc.demo@example.test"].id, statut="secondee"))
db.add(Proposition(groupe_id=groupe.id, sujet="Acheter un projecteur", description="La dépense n'est pas jugée prioritaire.", proposeur_id=membres["luc.demo@example.test"].id, secondeur_id=membres["sophie.demo@example.test"].id, statut="rejetee", etant_donne_que="La réserve prudente doit d'abord être consolidée."))
db.add(Evenement(groupe_id=groupe.id, titre="Atelier Étapes et Traditions", description="Atelier régional ouvert aux membres.", date_evenement=date.today() + timedelta(days=24), lieu="Centre communautaire", cree_par=rsg.id))
db.add(Evenement(groupe_id=groupe.id, titre="Anniversaire du groupe", description="Souper partage et réunion ouverte.", date_evenement=date.today() - timedelta(days=40), lieu="Local habituel", cree_par=secretaire.id))
db.add(RapportRsg(groupe_id=groupe.id, reunion_dist=date.today() - timedelta(days=18), redacteur_id=rsg.id, decisions=["Le district prépare l'inventaire annuel.", "Besoin de bénévoles pour l'accueil."], infos_region="La région sollicite les groupes pour les ateliers.", sujets_groupe="Questions transmises par la conscience de groupe."))
async def _decision(db, groupe, reunion, compteurs, prefixe, jour, proposeur, secondeur, sujet, attendu, decide):
compteurs[prefixe] += 1
numero = f"{prefixe}-{jour.year}-{compteurs[prefixe]:03d}"
proposition = Proposition(
groupe_id=groupe.id,
reunion_id=reunion.id if reunion else None,
sujet=sujet,
description=attendu,
proposeur_id=proposeur.id,
secondeur_id=secondeur.id,
statut="adoptee",
numero_resolution=numero,
date_vote=jour,
date_adoption=jour,
etant_donne_que=attendu,
groupe_a_decide_de=decide,
)
db.add(proposition)
await db.flush()
return proposition
def _premier_mardi(jour: date) -> date:
delta = (1 - jour.weekday()) % 7
return jour + timedelta(days=delta)
async def _premier_groupe(db: AsyncSession) -> Groupe:
from sqlalchemy import select
result = await db.execute(select(Groupe).limit(1))
return result.scalar_one()
async def _membres_par_courriel(db: AsyncSession) -> dict[str, Membre]:
from sqlalchemy import select
result = await db.execute(select(Membre))
return {m.courriel: m for m in result.scalars().all()}
async def _postes_par_nom(db: AsyncSession) -> dict[str, Poste]:
from sqlalchemy import select
result = await db.execute(select(Poste))
return {p.nom: p for p in result.scalars().all()}
async def _litterature(db: AsyncSession) -> list[Litterature]:
from sqlalchemy import select
result = await db.execute(select(Litterature))
return result.scalars().all()