How to Automate Customer Communication Logging

Customer context gets lost when conversations live in separate inboxes, Slack threads, and call notes. I built a system to automate customer communication logging to your CRM so every email, call, and message gets recorded in one place. Nobody has to manually log anything.

Full conversation history. Zero manual data entry.

What You Need Before Starting

• Python 3.8+
• CRM API access (HubSpot, Salesforce, or custom)
• Email API (Gmail or Outlook)
• Call transcription service (optional)

Step 1: Capture Emails Automatically

import imaplib
import email
from datetime import datetime

def fetch_recent_emails(imap_server, email_addr, password, hours=24):
    mail = imaplib.IMAP4_SSL(imap_server)
    mail.login(email_addr, password)
    mail.select("INBOX")

    since_date = (datetime.now() - __import__('datetime').timedelta(hours=hours)).strftime("%d-%b-%Y")
    _, message_ids = mail.search(None, f'SINCE {since_date}')

    emails = []
    for msg_id in message_ids[0].split():
        _, data = mail.fetch(msg_id, "(RFC822)")
        msg = email.message_from_bytes(data[0][1])
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain":
                    body = part.get_payload(decode=True).decode()
                    break
        else:
            body = msg.get_payload(decode=True).decode()

        emails.append({
            "from": msg["From"],
            "to": msg["To"],
            "subject": msg["Subject"],
            "date": msg["Date"],
            "body": body[:2000]
        })

    mail.logout()
    return emails

Step 2: Match Communications to CRM Contacts

import re

def extract_email_address(header):
    match = re.search(r'[\w.-]+@[\w.-]+', header or "")
    return match.group(0).lower() if match else None

def match_to_crm_contact(email_address, crm_contacts):
    for contact in crm_contacts:
        if contact.get("email", "").lower() == email_address:
            return contact
    return None

def process_emails_for_crm(emails, crm_contacts):
    logs = []
    for em in emails:
        sender = extract_email_address(em["from"])
        contact = match_to_crm_contact(sender, crm_contacts)

        if contact:
            logs.append({
                "contact_id": contact["id"],
                "type": "email",
                "direction": "inbound",
                "subject": em["subject"],
                "body_preview": em["body"][:500],
                "timestamp": em["date"]
            })

    return logs

Step 3: Log to CRM via API

import requests

def log_to_hubspot(activity, api_key):
    url = "https://api.hubapi.com/engagements/v1/engagements"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}

    payload = {
        "engagement": {
            "active": True,
            "type": "EMAIL" if activity["type"] == "email" else "NOTE",
            "timestamp": int(datetime.now().timestamp() * 1000)
        },
        "associations": {
            "contactIds": [activity["contact_id"]]
        },
        "metadata": {
            "subject": activity.get("subject", ""),
            "text": activity.get("body_preview", "")
        }
    }

    response = requests.post(url, headers=headers, json=payload)
    return response.status_code == 200

def log_batch_to_crm(activities, api_key):
    logged = 0
    for activity in activities:
        if log_to_hubspot(activity, api_key):
            logged += 1
    print(f"Logged {logged}/{len(activities)} communications to CRM")
    return logged

Step 4: Add Call Logging

def log_call_to_crm(contact_id, duration_seconds, notes, api_key):
    activity = {
        "contact_id": contact_id,
        "type": "call",
        "subject": f"Call ({duration_seconds // 60} min)",
        "body_preview": notes
    }
    return log_to_hubspot(activity, api_key)

Step 5: Run on Schedule

def run_communication_sync(config):
    emails = fetch_recent_emails(
        config["imap_server"], config["email"], config["password"], hours=1
    )

    crm_contacts = fetch_crm_contacts(config["crm_api_key"])
    activities = process_emails_for_crm(emails, crm_contacts)
    logged = log_batch_to_crm(activities, config["crm_api_key"])

    print(f"Synced {logged} communications at {datetime.now().isoformat()}")

# Run every hour
# 0 * * * * python3 /path/to/comm_sync.py

What to Build Next

Add sentiment tagging that uses AI to classify each communication as positive, neutral, or negative. Surface at-risk accounts where recent communications trend negative before the client actually churns.

Related Reading

• Building Your First Automation: A Complete Guide - automation fundamentals
• The Feedback Loop That Powers Everything - CRM as a feedback system
• How to Audit Your Operations for AI Opportunities - finding CRM automation opportunities