EHV/sejeteralo
7
0
Fork 1
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix seed : fixtures dev idempotentes, indépendantes de l'early-return
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source 
- DEV_FIXTURES défini au niveau module (constante partagée)
- seed principal (commune + 363 foyers + votes) sous `if commune is None`
- fixtures toujours vérifiées/insérées après, quel que soit l'état de la DB
- résout le cas prod avec DB déjà seedée

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Yvv
2026-03-24 03:58:31 +01:00
parent 19ac64c856
commit 90b069cb88
1 changed files with 156 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

305
backend/seed.py
View File

@@ -20,174 +20,181 @@ from app.engine.pricing import HouseholdData, compute_p0
XLS_PATH = os.path.join(os.path.dirname(__file__), "..", "Eau2018.xls") XLS_PATH = os.path.join(os.path.dirname(__file__), "..", "Eau2018.xls")
# Codes fixes — identiques dans le dev hint frontend
DEV_FIXTURES = [
{"identifier": "[DEV] Foyer RS 60m³", "status": "RS", "volume_m3": 60.0, "price_paid_eur": 140.0, "auth_code": "DEVTEST2"},
{"identifier": "[DEV] Foyer RP 120m³", "status": "RP", "volume_m3": 120.0, "price_paid_eur": 265.0, "auth_code": "DEVTEST3"},
{"identifier": "[DEV] Foyer PRO 350m³", "status": "PRO", "volume_m3": 350.0, "price_paid_eur": 680.0, "auth_code": "DEVTEST4"},
]
async def seed(): async def seed():
await init_db() await init_db()
async with async_session() as db: async with async_session() as db:
# Check if already seeded # ── Commune Saoû (idempotent) ──────────────────────────────────────────
result = await db.execute(select(Commune).where(Commune.slug == "saou")) result = await db.execute(select(Commune).where(Commune.slug == "saou"))
if result.scalar_one_or_none(): commune = result.scalar_one_or_none()
print("Saoû already seeded.")
return
# Create commune if commune is None:
commune = Commune( commune = Commune(
name="Saoû", name="Saoû",
slug="saou", slug="saou",
description="Commune de Saoû - Tarification progressive de l'eau", description="Commune de Saoû - Tarification progressive de l'eau",
)
db.add(commune)
await db.flush()
# Create tariff params
params = TariffParams(
commune_id=commune.id,
abop=100,
abos=100,
recettes=75000,
pmax=20,
vmax=2100,
differentiated_tariff=False,
data_year=2018,
data_imported_at=datetime.utcnow(),
)
db.add(params)
# Create super admin (manages all communes)
super_admin = AdminUser(
email="superadmin@sejeteralo.fr",
hashed_password=hash_password("superadmin"),
full_name="Super Admin",
role="super_admin",
)
db.add(super_admin)
# Create commune admin for Saoû (manages only this commune)
commune_admin = AdminUser(
email="saou@sejeteralo.fr",
hashed_password=hash_password("saou2024"),
full_name="Admin Saoû",
role="commune_admin",
)
commune_admin.communes.append(commune)
db.add(commune_admin)
# ── Dev fixture households (codes fixes, affichés dans les dev hints) ──
DEV_FIXTURES = [
{"identifier": "[DEV] Foyer RS 60m³", "status": "RS", "volume_m3": 60.0, "price_paid_eur": 140.0, "auth_code": "DEVTEST2"},
{"identifier": "[DEV] Foyer RP 120m³", "status": "RP", "volume_m3": 120.0, "price_paid_eur": 265.0, "auth_code": "DEVTEST3"},
{"identifier": "[DEV] Foyer PRO 350m³","status": "PRO", "volume_m3": 350.0, "price_paid_eur": 680.0, "auth_code": "DEVTEST4"},
]
existing_codes = {f["auth_code"] for f in DEV_FIXTURES}
for fixture in DEV_FIXTURES:
db.add(Household(commune_id=commune.id, **fixture))
# Import households from Eau2018.xls
book = xlrd.open_workbook(XLS_PATH)
sheet = book.sheet_by_name("CALCULS")
nb_hab = 363
for r in range(1, nb_hab + 1):
name = sheet.cell_value(r, 0)
status = sheet.cell_value(r, 3)
volume = sheet.cell_value(r, 4)
price = sheet.cell_value(r, 33)
code = generate_auth_code()
while code in existing_codes:
code = generate_auth_code()
existing_codes.add(code)
household = Household(
commune_id=commune.id,
identifier=str(name).strip(),
status=str(status).strip().upper(),
volume_m3=float(volume),
price_paid_eur=float(price) if price else 0.0,
auth_code=code,
) )
db.add(household) db.add(commune)
await db.flush()
await db.flush() # Create tariff params
params = TariffParams(
commune_id=commune.id,
abop=100,
abos=100,
recettes=75000,
pmax=20,
vmax=2100,
differentiated_tariff=False,
data_year=2018,
data_imported_at=datetime.utcnow(),
)
db.add(params)
# ── Publish a reference curve ── # Create super admin (manages all communes)
# Reference: vinf=400, all params=0.5 super_admin = AdminUser(
ref_vinf, ref_a, ref_b, ref_c, ref_d, ref_e = 400, 0.5, 0.5, 0.5, 0.5, 0.5 email="superadmin@sejeteralo.fr",
hashed_password=hash_password("superadmin"),
full_name="Super Admin",
role="super_admin",
)
db.add(super_admin)
hh_result = await db.execute( # Create commune admin for Saoû (manages only this commune)
select(Household).where(Household.commune_id == commune.id) commune_admin = AdminUser(
) email="saou@sejeteralo.fr",
all_households = hh_result.scalars().all() hashed_password=hash_password("saou2024"),
hh_data = [ full_name="Admin Saoû",
HouseholdData(volume_m3=h.volume_m3, status=h.status, price_paid_eur=h.price_paid_eur) role="commune_admin",
for h in all_households )
] commune_admin.communes.append(commune)
db.add(commune_admin)
ref_p0 = compute_p0( # Import households from Eau2018.xls
hh_data, book = xlrd.open_workbook(XLS_PATH)
recettes=params.recettes, abop=params.abop, abos=params.abos, sheet = book.sheet_by_name("CALCULS")
vinf=ref_vinf, vmax=params.vmax, pmax=params.pmax, nb_hab = 363
a=ref_a, b=ref_b, c=ref_c, d=ref_d, e=ref_e,
)
params.published_vinf = ref_vinf
params.published_a = ref_a
params.published_b = ref_b
params.published_c = ref_c
params.published_d = ref_d
params.published_e = ref_e
params.published_p0 = ref_p0
params.published_at = datetime.utcnow()
# ── Generate 10 votes, small variations around the reference ── existing_codes = {f["auth_code"] for f in DEV_FIXTURES}
random.seed(42) for r in range(1, nb_hab + 1):
name = sheet.cell_value(r, 0)
status = sheet.cell_value(r, 3)
volume = sheet.cell_value(r, 4)
price = sheet.cell_value(r, 33)
vote_profiles = [ code = generate_auth_code()
# 5 votes slightly below reference (eco-leaning) while code in existing_codes:
{"vinf": 350, "a": 0.45, "b": 0.48, "c": 0.40, "d": 0.52, "e": 0.55}, code = generate_auth_code()
{"vinf": 370, "a": 0.42, "b": 0.50, "c": 0.45, "d": 0.48, "e": 0.52}, existing_codes.add(code)
{"vinf": 380, "a": 0.48, "b": 0.45, "c": 0.42, "d": 0.50, "e": 0.58},
{"vinf": 360, "a": 0.50, "b": 0.52, "c": 0.38, "d": 0.55, "e": 0.50},
{"vinf": 390, "a": 0.47, "b": 0.47, "c": 0.48, "d": 0.46, "e": 0.53},
# 5 votes slightly above reference (lax-leaning)
{"vinf": 420, "a": 0.52, "b": 0.50, "c": 0.55, "d": 0.48, "e": 0.45},
{"vinf": 440, "a": 0.55, "b": 0.53, "c": 0.52, "d": 0.50, "e": 0.42},
{"vinf": 430, "a": 0.50, "b": 0.55, "c": 0.58, "d": 0.45, "e": 0.48},
{"vinf": 410, "a": 0.53, "b": 0.48, "c": 0.50, "d": 0.52, "e": 0.47},
{"vinf": 450, "a": 0.48, "b": 0.52, "c": 0.60, "d": 0.42, "e": 0.40},
]
used_households = set() db.add(Household(
vote_count = 0 commune_id=commune.id,
for prof in vote_profiles: identifier=str(name).strip(),
# Pick a unique household status=str(status).strip().upper(),
hh_pick = random.choice(all_households) volume_m3=float(volume),
while hh_pick.id in used_households: price_paid_eur=float(price) if price else 0.0,
hh_pick = random.choice(all_households) auth_code=code,
used_households.add(hh_pick.id) ))
vp0 = compute_p0( await db.flush()
# ── Publish a reference curve ──
ref_vinf, ref_a, ref_b, ref_c, ref_d, ref_e = 400, 0.5, 0.5, 0.5, 0.5, 0.5
hh_result = await db.execute(
select(Household).where(Household.commune_id == commune.id)
)
all_households = hh_result.scalars().all()
hh_data = [
HouseholdData(volume_m3=h.volume_m3, status=h.status, price_paid_eur=h.price_paid_eur)
for h in all_households
]
ref_p0 = compute_p0(
hh_data, hh_data,
recettes=params.recettes, abop=params.abop, abos=params.abos, recettes=params.recettes, abop=params.abop, abos=params.abos,
vinf=prof["vinf"], vmax=params.vmax, pmax=params.pmax, vinf=ref_vinf, vmax=params.vmax, pmax=params.pmax,
a=prof["a"], b=prof["b"], c=prof["c"], d=prof["d"], e=prof["e"], a=ref_a, b=ref_b, c=ref_c, d=ref_d, e=ref_e,
) )
vote = Vote( params.published_vinf = ref_vinf
commune_id=commune.id, params.published_a = ref_a
household_id=hh_pick.id, params.published_b = ref_b
vinf=prof["vinf"], params.published_c = ref_c
a=prof["a"], b=prof["b"], c=prof["c"], d=prof["d"], e=prof["e"], params.published_d = ref_d
computed_p0=vp0, params.published_e = ref_e
) params.published_p0 = ref_p0
db.add(vote) params.published_at = datetime.utcnow()
hh_pick.has_voted = True
vote_count += 1
await db.commit() # ── Generate 10 votes, small variations around the reference ──
print(f"Seeded: commune 'saou', {nb_hab} + 3 fixture households, {vote_count} votes") random.seed(42)
print(f" Published curve: vinf={ref_vinf}, p0={ref_p0:.3f}")
print(f" Super admin: superadmin@sejeteralo.fr / superadmin") vote_profiles = [
print(f" Commune admin Saou: saou@sejeteralo.fr / saou2024") # 5 votes slightly below reference (eco-leaning)
print(f" Dev fixtures: DEVTEST2 (RS 60m³) · DEVTEST3 (RP 120m³) · DEVTEST4 (PRO 350m³)") {"vinf": 350, "a": 0.45, "b": 0.48, "c": 0.40, "d": 0.52, "e": 0.55},
{"vinf": 370, "a": 0.42, "b": 0.50, "c": 0.45, "d": 0.48, "e": 0.52},
{"vinf": 380, "a": 0.48, "b": 0.45, "c": 0.42, "d": 0.50, "e": 0.58},
{"vinf": 360, "a": 0.50, "b": 0.52, "c": 0.38, "d": 0.55, "e": 0.50},
{"vinf": 390, "a": 0.47, "b": 0.47, "c": 0.48, "d": 0.46, "e": 0.53},
# 5 votes slightly above reference (lax-leaning)
{"vinf": 420, "a": 0.52, "b": 0.50, "c": 0.55, "d": 0.48, "e": 0.45},
{"vinf": 440, "a": 0.55, "b": 0.53, "c": 0.52, "d": 0.50, "e": 0.42},
{"vinf": 430, "a": 0.50, "b": 0.55, "c": 0.58, "d": 0.45, "e": 0.48},
{"vinf": 410, "a": 0.53, "b": 0.48, "c": 0.50, "d": 0.52, "e": 0.47},
{"vinf": 450, "a": 0.48, "b": 0.52, "c": 0.60, "d": 0.42, "e": 0.40},
]
used_households = set()
vote_count = 0
for prof in vote_profiles:
hh_pick = random.choice(all_households)
while hh_pick.id in used_households:
hh_pick = random.choice(all_households)
used_households.add(hh_pick.id)
vp0 = compute_p0(
hh_data,
recettes=params.recettes, abop=params.abop, abos=params.abos,
vinf=prof["vinf"], vmax=params.vmax, pmax=params.pmax,
a=prof["a"], b=prof["b"], c=prof["c"], d=prof["d"], e=prof["e"],
)
db.add(Vote(
commune_id=commune.id,
household_id=hh_pick.id,
vinf=prof["vinf"],
a=prof["a"], b=prof["b"], c=prof["c"], d=prof["d"], e=prof["e"],
computed_p0=vp0,
))
hh_pick.has_voted = True
vote_count += 1
await db.commit()
print(f"Seeded: commune 'saou', {nb_hab} households, {vote_count} votes")
print(f" Published curve: vinf={ref_vinf}, p0={ref_p0:.3f}")
print(f" Super admin: superadmin@sejeteralo.fr / superadmin")
print(f" Commune admin Saou: saou@sejeteralo.fr / saou2024")
else:
print("Saoû already seeded.")
# ── Dev fixtures (idempotent — insérés à chaque run si absents) ────────
fixture_added = 0
for fixture in DEV_FIXTURES:
res = await db.execute(
select(Household).where(Household.auth_code == fixture["auth_code"])
)
if res.scalar_one_or_none() is None:
db.add(Household(commune_id=commune.id, **fixture))
fixture_added += 1
if fixture_added:
await db.commit()
print(f" Dev fixtures: {fixture_added} ajoutés — DEVTEST2 (RS 60m³) · DEVTEST3 (RP 120m³) · DEVTEST4 (PRO 350m³)")
if __name__ == "__main__": if __name__ == "__main__":