diff --git a/core/management/commands/export_contacts_structures.py b/core/management/commands/export_contacts_structures.py new file mode 100644 index 00000000..fcba785a --- /dev/null +++ b/core/management/commands/export_contacts_structures.py @@ -0,0 +1,18 @@ +from django.core.management.base import BaseCommand +import csv +import sys + +from core.models import Structure + + +class Command(BaseCommand): + help = """ + Exporte tous les contacts reliés à une structure. + Usage: python manage.py export_contacts_structures > contacts_structures.csv + """ + + def handle(self, *args, **options): + writer = csv.writer(sys.stdout, delimiter=";") + writer.writerow(["structure (niveau1)", "structure (niveau2)", "email"]) + for structure in Structure.objects.filter(contact__email=""): + writer.writerow([structure.niveau1, structure.niveau2, ""]) diff --git a/core/management/commands/update_contacts_structures_email.py b/core/management/commands/update_contacts_structures_email.py new file mode 100644 index 00000000..d778b191 --- /dev/null +++ b/core/management/commands/update_contacts_structures_email.py @@ -0,0 +1,31 @@ +from django.core.management.base import BaseCommand +import csv + +from core.models import Contact + + +class Command(BaseCommand): + help = """ + Ajoute les emails des contacts liés à une structure qui n'ont pas d'email renseigné. + Les adresses emails sont récupérées depuis le fichier CSV fourni en argument (fichier généré par la commande export_contacts_structures). + Usage: python manage.py update_contacts_emails contacts_structures.csv + """ + + def add_arguments(self, parser): + parser.add_argument("csv_file", type=str) + + def handle(self, *args, **options): + updates = 0 + with open(options["csv_file"], newline="") as f: + reader = csv.DictReader(f, delimiter=";") + for row in reader: + niveau1 = row["structure (niveau1)"] + niveau2 = row["structure (niveau2)"] + email = row["email"].strip() + + if email: + updates += Contact.objects.filter( + structure__niveau1=niveau1, structure__niveau2=niveau2, email="" + ).update(email=email) + + self.stdout.write(f"Mise à jour effectuée : {updates} contacts modifiés") diff --git a/core/tests/test_export_contacts_structures.py b/core/tests/test_export_contacts_structures.py new file mode 100644 index 00000000..16bbbc67 --- /dev/null +++ b/core/tests/test_export_contacts_structures.py @@ -0,0 +1,21 @@ +import pytest +from io import StringIO +from django.core.management import call_command +from core.models import Structure, Contact +from contextlib import redirect_stdout + + +@pytest.mark.django_db +def test_export_contacts_with_data(): + structure = Structure.objects.create(niveau1="Direction", niveau2="Service A") + Contact.objects.create(structure=structure, email="") + Contact.objects.create(structure=structure, email="test@example.com") + + out = StringIO() + with redirect_stdout(out): + call_command("export_contacts_structures") + + result = out.getvalue().splitlines() + assert len(result) == 2 + assert result[0] == "structure (niveau1);structure (niveau2);email" + assert result[1] == "Direction;Service A;" diff --git a/core/tests/test_update_contacts_structures_email.py b/core/tests/test_update_contacts_structures_email.py new file mode 100644 index 00000000..0354f4b9 --- /dev/null +++ b/core/tests/test_update_contacts_structures_email.py @@ -0,0 +1,42 @@ +import pytest +from io import StringIO +from contextlib import redirect_stdout +from django.core.management import call_command +from core.models import Structure, Contact + + +@pytest.fixture +def mock_csv_data(tmp_path): + data = ( + "structure (niveau1);structure (niveau2);email\n" + "Direction;Service A;new@example.com\n" + "Direction;Service B;another@example.com\n" + "Direction;Service C;" + ) + p = tmp_path / "test.csv" + p.write_text(data, encoding="utf-8") + return str(p) + + +@pytest.mark.django_db +def test_update_contacts_emails(mock_csv_data): + structure_a = Structure.objects.create(niveau1="Direction", niveau2="Service A") + structure_b = Structure.objects.create(niveau1="Direction", niveau2="Service B") + structure_c = Structure.objects.create(niveau1="Direction", niveau2="Service C") + + contact_a = Contact.objects.create(structure=structure_a, email="") + contact_b = Contact.objects.create(structure=structure_b, email="") + contact_c = Contact.objects.create(structure=structure_c, email="") + + out = StringIO() + with redirect_stdout(out): + call_command("update_contacts_structures_email", mock_csv_data) + + assert "Mise à jour effectuée : 2 contacts modifiés" in out.getvalue() + + contact_a.refresh_from_db() + contact_b.refresh_from_db() + contact_c.refresh_from_db() + assert contact_a.email == "new@example.com" + assert contact_b.email == "another@example.com" + assert contact_c.email == ""