101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
scripts/sync_enums.py
|
|
|
|
Génère une migration SQL non destructive pour ajouter les valeurs d'ENUM
|
|
manquantes en se basant sur les CSV de `bdd/data_test`.
|
|
|
|
Usage:
|
|
python3 scripts/sync_enums.py > migrations/00_sync_enums.sql
|
|
|
|
Relire le SQL généré avant application en production.
|
|
"""
|
|
from pathlib import Path
|
|
import csv
|
|
from collections import defaultdict
|
|
import sys
|
|
|
|
# Mapping colonne CSV -> nom du type enum en base
|
|
# NOTE: mapping is now based on filename (more precise)
|
|
CSV_DIR = Path('bdd/data_test')
|
|
|
|
# Define per-file mappings: filename stem -> {column_name: enum_type}
|
|
FILE_COLUMN_ENUM_MAP = {
|
|
'utilisateurs': {
|
|
'role': 'role_type',
|
|
'genre': 'genre_type',
|
|
'statut': 'statut_utilisateur_type',
|
|
},
|
|
'enfants': {
|
|
'statut': 'statut_enfant_type',
|
|
'genre': 'genre_type',
|
|
},
|
|
'contrats': {
|
|
'statut': 'statut_contrat_type',
|
|
},
|
|
'dossiers': {
|
|
'statut': 'statut_dossier_type',
|
|
},
|
|
'evenements': {
|
|
'type': 'type_evenement_type',
|
|
'statut': 'statut_evenement_type',
|
|
},
|
|
'validations': {
|
|
'statut': 'statut_validation_type',
|
|
}
|
|
}
|
|
|
|
def quote_sql_literal(s: str) -> str:
|
|
return "'" + s.replace("'", "''") + "'"
|
|
|
|
def discover_values():
|
|
# heuristique : map common column names to enums
|
|
enum_values = defaultdict(set)
|
|
for p in sorted(CSV_DIR.glob('*.csv')):
|
|
try:
|
|
with p.open(newline='', encoding='utf-8') as fh:
|
|
reader = csv.DictReader(fh)
|
|
fname = p.stem
|
|
per_file_map = FILE_COLUMN_ENUM_MAP.get(fname, {})
|
|
for row in reader:
|
|
for col, enum in per_file_map.items():
|
|
if col in row and row[col] is not None:
|
|
v = row[col].strip()
|
|
if v != '':
|
|
enum_values[enum].add(v)
|
|
except Exception as e:
|
|
print(f"-- Error reading {p}: {e}", file=sys.stderr)
|
|
return enum_values
|
|
|
|
def emit_sql(enum_values):
|
|
lines = []
|
|
lines.append('-- Generated by scripts/sync_enums.py — review before applying')
|
|
for enum_type, vals in enum_values.items():
|
|
if not vals:
|
|
continue
|
|
for v in sorted(vals):
|
|
lit = quote_sql_literal(v)
|
|
block = f"""
|
|
DO $$
|
|
BEGIN
|
|
IF NOT EXISTS (
|
|
SELECT 1 FROM pg_enum e
|
|
JOIN pg_type t ON e.enumtypid = t.oid
|
|
WHERE t.typname = {quote_sql_literal(enum_type)}
|
|
AND e.enumlabel = {lit}
|
|
) THEN
|
|
ALTER TYPE {enum_type} ADD VALUE {lit};
|
|
END IF;
|
|
END$$;
|
|
"""
|
|
lines.append(block)
|
|
return '\n'.join(lines)
|
|
|
|
def main():
|
|
enum_values = discover_values()
|
|
sql = emit_sql(enum_values)
|
|
print(sql)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|