wip: UpdateTags logic

mailchimp_api/processing/update_tags.py

@@ -0,0 +1,78 @@
+from collections import defaultdict
+
+import pandas as pd
+
+from ..config import Config
+from ..services.mailchimp_service import MailchimpService
+
+next_tag_map = {
+    "M1": "M2",
+    "M2": "M3",
+    "M3": None,
+}
+
+
+def _add_and_remove_tags(
+    mailchimp_service: MailchimpService,
+    list_id: str,
+    members_with_tags_df: pd.DataFrame,
+) -> None:
+    # columns: id, email, tags
+    # e.g. "first_member_id", "[email protected]", [{"id": 1, "name": "Test API Tag"}, {"id": 2, "name": "M3"}]
+
+    # create add_tag_members, remove_tag_members
+    # keys are tags, values are list of member ids
+    add_tag_members = defaultdict(list)
+    remove_tag_members = defaultdict(list)
+    # dont use crm_df anymore
+    for _, row in members_with_tags_df.iterrows():
+        member_id = row["id"]
+        tags = row["tags"]
+        for tag in tags:
+            tag_name = tag["name"]
+            if tag_name not in next_tag_map:
+                continue
+
+            next_tag = next_tag_map[tag_name]
+            if next_tag is None:
+                continue
+
+            add_tag_members[next_tag].append(member_id)
+            remove_tag_members[tag_name].append(member_id)
+
+    # print(f"add_tag_members: {add_tag_members}")
+    # print(f"remove_tag_members: {remove_tag_members}")
+
+
+def update_tags(crm_df: pd.DataFrame, config: Config, list_name: str) -> None:
+    """Update tags for members in the CRM."""
+    # Create a Mailchimp service
+    mailchimp_service = MailchimpService(config)
+
+    # Get the list ID for the list name
+    account_lists = mailchimp_service.get_account_lists()
+    list_id = None
+    for account_list in account_lists["lists"]:
+        if account_list["name"] == list_name:
+            list_id = account_list["id"]
+
+    if list_id is None:
+        raise ValueError(f"List {list_name} not found in account lists.")
+
+    # Get the members with tags
+    members_with_tags = mailchimp_service.get_members_with_tags(list_id)
+
+    members_with_tags_df = pd.DataFrame(members_with_tags["members"])
+    members_with_tags_df.rename(columns={"email_address": "email"}, inplace=True)
+
+    # filter only emails that are in the CRM
+    crm_emails = crm_df["email"].unique()
+    members_with_tags_df = members_with_tags_df[
+        members_with_tags_df["email"].isin(crm_emails)
+    ]
+
+    _add_and_remove_tags(
+        mailchimp_service=mailchimp_service,
+        list_id=list_id,
+        members_with_tags_df=members_with_tags_df,
+    )

mailchimp_api/services/mailchimp_service.py

@@ -37,6 +37,11 @@ def get_account_lists(self) -> dict[str, list[dict[str, str]]]:
 
         return self._mailchim_request_get(url)
 
+    def get_members_with_tags(self, list_id: str) -> dict[str, Any]:
+        url = f"{self.config.base_url}/lists/{list_id}/members?fields=members.id,members.email_address,members.tags"
+
+        return self._mailchim_request_get(url)
+
     def get_members(self, list_id: str) -> dict[str, list[dict[str, str]]]:
         url = f"{self.config.base_url}/lists/{list_id}/members?fields=members.email_address,members.id"
 

pyproject.toml

@@ -9,6 +9,7 @@ name = "mailchimp_api"
 
 dependencies = [
     "fastagency[autogen,mesop,openapi,server,fastapi]>=0.3.0",
+    "pandas>=2.2.3",
 ]
 
 [project.optional-dependencies]

tests/processing/test_update_tags.py

@@ -0,0 +1,78 @@
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pandas as pd
+import pytest
+
+from mailchimp_api.config import Config
+from mailchimp_api.processing.update_tags import update_tags
+from mailchimp_api.services.mailchimp_service import MailchimpService
+
+
+class TestUpdateTags:
+    @pytest.fixture(autouse=True)
+    def _setup(self) -> None:
+        self.config = Config(dc="us14", api_key="anystring")
+        # self.config = Config(dc="us14", api_key=os.getenv("MAILCHIMP_API_KEY"))
+        self.mailchimp_service = MailchimpService(config=self.config)
+        return
+
+    def _setup_mailchimp_request_method(
+        self,
+        mock_get: MagicMock,
+        json_responses: list[Any],
+        status_code: int = 200,
+    ) -> None:
+        side_effect = []
+        for json_response in json_responses:
+            mock_response = MagicMock()
+            mock_response.status_code = status_code
+            mock_response.json.return_value = json_response
+            side_effect.append(mock_response)
+
+        mock_get.side_effect = side_effect
+
+    @patch("mailchimp_api.services.mailchimp_service.requests.get")
+    def test_update_tags(self, mock_get: MagicMock) -> None:
+        json_responses = [
+            {"lists": [{"id": "list_id", "name": "airt"}]},
+            {
+                "members": [
+                    {
+                        "id": "first_member_id",
+                        "email_address": "[email protected]",
+                        "tags": [
+                            {"id": 1, "name": "Test API Tag"},
+                            {"id": 2, "name": "M3"},
+                        ],
+                    },
+                    {
+                        "id": "second_member_id",
+                        "email_address": "[email protected]",
+                        "tags": [
+                            {"id": 1, "name": "Test API Tag"},
+                            {"id": 2, "name": "M3"},
+                            {"id": 3, "name": "test-tag"},
+                        ],
+                    },
+                    {
+                        "id": "third_member_id",
+                        "email_address": "[email protected]",
+                        "tags": [
+                            {"id": 1, "name": "Test API Tag"},
+                            {"id": 2, "name": "M2"},
+                        ],
+                    },
+                ]
+            },
+        ]
+        crm_df = pd.DataFrame(
+            {
+                "email": [
+                    "[email protected]",
+                    "[email protected]",
+                ]
+            }
+        )
+        self._setup_mailchimp_request_method(mock_get, json_responses=json_responses)
+        update_tags(crm_df=crm_df, config=self.config, list_name="airt")