Mailchimp Unsubs to Slack #729
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| name: Mailchimp Unsubs to Slack | |
| on: | |
| schedule: | |
| - cron: "0 */12 * * *" # Every 12 hours | |
| workflow_dispatch: | |
| jobs: | |
| notify-unsubs: | |
| runs-on: ubuntu-latest | |
| steps: | |
| - name: Run unsubscribe notifier inline | |
| env: | |
| MAILCHIMP_API_KEY: ${{ secrets.MAILCHIMP_API_KEY }} | |
| MAILCHIMP_SERVER_PREFIX: ${{ secrets.MAILCHIMP_SERVER_PREFIX }} | |
| SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }} | |
| run: | | |
| python -m pip install --quiet --upgrade pip | |
| pip install --quiet --upgrade mailchimp-marketing requests | |
| python - <<'EOF' | |
| import os | |
| import json | |
| import re | |
| import requests | |
| import mailchimp_marketing as MailchimpMarketing | |
| from mailchimp_marketing.api_client import ApiClientError | |
| def escape(text): | |
| return text.replace("&", "&").replace("<", "<").replace(">", ">") | |
| try: | |
| client = MailchimpMarketing.Client() | |
| client.set_config({ | |
| "api_key": os.environ["MAILCHIMP_API_KEY"], | |
| "server": os.environ["MAILCHIMP_SERVER_PREFIX"] | |
| }) | |
| dc = os.environ["MAILCHIMP_SERVER_PREFIX"] | |
| messages = [] | |
| campaign_reports = client.reports.get_all_campaign_reports(count=10) | |
| for campaign in campaign_reports["reports"]: | |
| campaign_id = campaign["id"] | |
| raw_title = campaign.get("campaign_title", "Unknown Campaign") | |
| # Extract portion before first colon for clean short title | |
| match = re.match(r"^([^:]+)", raw_title) | |
| short_title = match.group(1).strip() if match else raw_title | |
| short_title = escape(short_title) | |
| unsubs = client.reports.get_unsubscribed_list_for_campaign(campaign_id, count=1000) | |
| reasons = unsubs.get("unsubscribes", []) | |
| for unsub in reasons: | |
| reason = unsub.get("reason", "").strip() | |
| if reason.startswith(( | |
| "None given", "No longer interested", "Spammy content", | |
| "Did not signup for list", "Inappropriate content" | |
| )): | |
| continue | |
| reason = escape(reason) | |
| email = unsub.get("email_address", "[email protected]") | |
| # ✅ Get contact_id from search_members | |
| try: | |
| search_result = client.searchMembers.search(query=email) | |
| matches = search_result.get("exact_matches", {}).get("members", []) | |
| if matches: | |
| contact_id = matches[0].get("contact_id") | |
| profile_url = f"https://{dc}.admin.mailchimp.com/audience/contact-profile?contact_id={contact_id}" | |
| email_link = f"<{profile_url}|{email}>" | |
| else: | |
| email_link = email # fallback | |
| except Exception as lookup_error: | |
| print(f"Error looking up {email}: {lookup_error}") | |
| email_link = email # fallback | |
| # Final message format: **Reason** – (<linked email> – short title) | |
| msg = f"*{reason}* – ({email_link} – {short_title})" | |
| messages.append(msg) | |
| if messages: | |
| max_messages = 50 | |
| if len(messages) > max_messages: | |
| messages = messages[:max_messages] | |
| messages.append("_…list trimmed to 50 unsubscribes._") | |
| slack_text = "\n\n📤 *New Unsubscribe Reasons:*\n\n" + "\n\n".join(messages) | |
| else: | |
| slack_text = "✅ No new meaningful unsubscribe reasons found." | |
| response = requests.post( | |
| os.environ["SLACK_WEBHOOK_URL"], | |
| data=json.dumps({"text": slack_text}), | |
| headers={"Content-Type": "application/json"} | |
| ) | |
| response.raise_for_status() | |
| except ApiClientError as error: | |
| print("Mailchimp Error:", error.text) | |
| except Exception as e: | |
| print("General Error:", e) | |
| EOF |