feat(crm): Management commande de Synchronisation des utilisateurs avec Brevo

lemarche/crm/management/commands/crm_brevo_sync_contacts.py

@@ -0,0 +1,78 @@
+import time
+
+from django.conf import settings
+
+from lemarche.users.models import User
+
+# from lemarche.users.constants import User
+from lemarche.utils.apis import api_brevo
+from lemarche.utils.commands import BaseCommand
+
+
+class Command(BaseCommand):
+    """
+    Command script to send Users to Brevo CRM (companies) or set Brevo CRM IDs to Users models
+
+    Usage:
+    python manage.py crm_brevo_sync_contacts --dry-run
+    python manage.py crm_brevo_sync_contacts --brevo-list-id=23 --kind-users=SIAE
+    python manage.py crm_brevo_sync_contacts --brevo-list-id=10 --kind-users=SIAE --dry-run
+    """
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--kind-users", dest="kind_users", type=str, default=User.KIND_SIAE, help="set kind of users"
+        )
+        parser.add_argument(
+            "--brevo-list-id",
+            dest="brevo_list_id",
+            type=int,
+            default=settings.BREVO_CL_SIGNUP_BUYER_ID,
+            help="set brevo list id",
+        )
+        parser.add_argument("--dry-run", dest="dry_run", action="store_true", help="Dry run (no changes to the DB)")
+        parser.add_argument(
+            "--with-existing-contacts",
+            dest="with_existing_contacts",
+            type=bool,
+            default=True,
+            help="make it with existing contacts in brevo",
+        )
+
+    def handle(self, dry_run: bool, kind_users: str, brevo_list_id: int, with_existing_contacts: bool, **options):
+        self.stdout.write("-" * 80)
+        self.stdout.write("Script to sync with Contact Brevo CRM...")
+
+        users_qs = User.objects.filter(kind=kind_users)
+        progress = 0
+
+        self.stdout.write(f"User: find {users_qs.count()} users {kind_users}.")
+        existing_contacts = None
+        if with_existing_contacts:
+            existing_contacts = api_brevo.get_all_users_from_list(list_id=brevo_list_id)
+            self.stdout.write(f"Contacts in brevo list: find {len(existing_contacts)} contacts.")
+
+        if not dry_run:
+            for user in users_qs:
+                brevo_contact_id = None
+                # if we have existing_contacts in brevo
+                if existing_contacts:
+                    # try to get id by dictionnary of existing contacts
+                    brevo_contact_id = existing_contacts.get(user.email)
+                    if brevo_contact_id == user.brevo_contact_id:
+                        # if brevo contact id and user brevo contact id we skip user
+                        self.stdout.write(f"Contact {user.email} already in Brevo.")
+                        continue
+                # if we still not have contact id
+                if not brevo_contact_id:
+                    self.stdout.write(f"Create and save contact {user.email} in Brevo.")
+                    api_brevo.create_contact(user=user, list_id=brevo_list_id, with_user_save=True)
+                # if we already have the brevo_contact_id, we can simply save it
+                else:
+                    self.stdout.write(f"Save existing contact {user.email}.")
+                    user.brevo_contact_id = brevo_contact_id
+                    user.save()
+
+                progress += 1
+                if (progress % 10) == 0:  # avoid API rate-limiting
+                    time.sleep(1)

lemarche/utils/apis/api_brevo.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import time
 
 import sib_api_v3_sdk
 from django.conf import settings
@@ -27,7 +28,7 @@ def get_api_client():
     return sib_api_v3_sdk.ApiClient(config)
 
 
-def create_contact(user, list_id: int):
+def create_contact(user, list_id: int, with_user_save=True):
     api_client = get_api_client()
     api_instance = sib_api_v3_sdk.ContactsApi(api_client)
     new_contact = sib_api_v3_sdk.CreateContact(
@@ -45,7 +46,10 @@ def create_contact(user, list_id: int):
     )
 
     try:
-        api_response = api_instance.create_contact(new_contact)
+        api_response = api_instance.create_contact(new_contact).to_dict()
+        if with_user_save:
+            user.brevo_contact_id = api_response.get("id")
+            user.save()
         logger.info(f"Success Brevo->ContactsApi->create_contact: {api_response}")
     except ApiException as e:
         logger.error(f"Exception when calling Brevo->ContactsApi->create_contact: {e}")
@@ -221,3 +225,56 @@ def link_deal_with_list_contact(tender, contact_list: list = None):
 
     except ApiException as e:
         logger.error("Exception when calling Brevo->DealApi->crm_deals_link_unlink_id_patch: %s\n" % e)
+
+
+def get_all_users_from_list(
+    list_id: int = settings.BREVO_CL_SIGNUP_BUYER_ID, limit=500, offset=0, max_retries=3, verbose=False
+):
+    """
+    Fetches all users from a specified Brevo CRM list, using pagination and retry strategies.
+
+    Args:
+        list_id (int): ID of the list to fetch users from. Defaults to BREVO_CL_SIGNUP_BUYER_ID.
+        limit (int): Number of users to fetch per request. Defaults to 500.
+        offset (int): Initial offset for fetching users. Defaults to 0.
+        max_retries (int): Maximum number of retries on API failure. Defaults to 3.
+
+    Returns:
+        dict: Maps user emails to their IDs.
+
+    Raises:
+        ApiException: On API failures exceeding retry limit.
+        Exception: On unexpected errors.
+
+    This function attempts to retrieve all contacts from the given list and handles API errors
+    by retrying up to `max_retries` times with exponential backoff.
+    """
+    api_client = get_api_client()
+    api_instance = sib_api_v3_sdk.ContactsApi(api_client)
+    result = {}
+    is_finished = False
+    retry_count = 0
+    while not is_finished:
+        try:
+            api_response = api_instance.get_contacts_from_list(list_id=list_id, limit=limit, offset=offset).to_dict()
+            contacts = api_response.get("contacts", [])
+            if verbose:
+                logger.info(f"Contacts fetched: {len(contacts)} at offset {offset}")
+            for contact in contacts:
+                result[contact.get("email")] = contact.get("id")
+            # Update the loop exit condition
+            if len(contacts) < limit:
+                is_finished = True
+            else:
+                offset += limit
+        except ApiException as e:
+            logger.error(f"Exception when calling ContactsApi->get_contacts_from_list: {e}")
+            retry_count += 1
+            if retry_count > max_retries:
+                logger.error("Max retries exceeded. Exiting function.")
+                break
+            time.sleep(2**retry_count)  # Exponential backoff
+        except Exception as e:
+            logger.error(f"Unexpected error: {e}")
+            break
+    return result