Track lead-to-MQL conversion rate by source and report to Slack using code and cron

# Verify your HubSpot token works
curl -s "https://api.hubapi.com/crm/v3/objects/contacts?limit=1" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" | head -c 200

import os
import requests
from datetime import datetime, timezone, timedelta
 
HUBSPOT_ACCESS_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]
HEADERS = {"Authorization": f"Bearer {HUBSPOT_ACCESS_TOKEN}", "Content-Type": "application/json"}
 
def get_date_range(days_back=7):
    now = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
    start = now - timedelta(days=days_back)
    return str(int(start.timestamp() * 1000)), str(int(now.timestamp() * 1000))
 
def search_contacts(start_ms, end_ms, extra_filters=None):
    """Search contacts created in a date range with optional extra filters."""
    filters = [
        {"propertyName": "createdate", "operator": "GTE", "value": start_ms},
        {"propertyName": "createdate", "operator": "LT", "value": end_ms},
    ]
    if extra_filters:
        filters.extend(extra_filters)
 
    all_results = []
    after = "0"
    while True:
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts/search",
            headers=HEADERS,
            json={
                "filterGroups": [{"filters": filters}],
                "properties": ["hs_analytics_source", "lifecyclestage", "createdate"],
                "limit": 100,
                "after": after,
            },
        )
        resp.raise_for_status()
        data = resp.json()
        all_results.extend(data["results"])
        if data.get("paging", {}).get("next", {}).get("after"):
            after = data["paging"]["next"]["after"]
        else:
            break
    return all_results
 
start_ms, end_ms = get_date_range(7)
 
# All leads created in period
all_leads = search_contacts(start_ms, end_ms)
 
# MQLs created in period
mqls = search_contacts(start_ms, end_ms, extra_filters=[
    {"propertyName": "lifecyclestage", "operator": "EQ", "value": "marketingqualifiedlead"}
])
 
print(f"Last 7 days: {len(all_leads)} leads, {len(mqls)} MQLs")

from collections import defaultdict
 
def calculate_conversion(all_leads, mqls):
    leads_by_source = defaultdict(int)
    mqls_by_source = defaultdict(int)
 
    for lead in all_leads:
        source = lead["properties"].get("hs_analytics_source") or "UNKNOWN"
        leads_by_source[source] += 1
 
    for mql in mqls:
        source = mql["properties"].get("hs_analytics_source") or "UNKNOWN"
        mqls_by_source[source] += 1
 
    sources = sorted(
        set(leads_by_source) | set(mqls_by_source),
        key=lambda s: leads_by_source.get(s, 0),
        reverse=True,
    )
 
    report = []
    for source in sources:
        lead_count = leads_by_source.get(source, 0)
        mql_count = mqls_by_source.get(source, 0)
        rate = (mql_count / lead_count * 100) if lead_count > 0 else 0
        report.append({
            "source": source,
            "leads": lead_count,
            "mqls": mql_count,
            "rate": round(rate, 1),
        })
 
    total_leads = len(all_leads)
    total_mqls = len(mqls)
    overall_rate = round(total_mqls / total_leads * 100, 1) if total_leads > 0 else 0
 
    return report, total_leads, total_mqls, overall_rate

from slack_sdk import WebClient
 
def post_report(report, total_leads, total_mqls, overall_rate):
    source_lines = "\n".join(
        f"• *{r['source']}*: {r['leads']} leads → {r['mqls']} MQLs ({r['rate']}%)"
        for r in report
    )
 
    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": "📈 Weekly Lead-to-MQL Conversion Report"}},
        {"type": "section", "fields": [
            {"type": "mrkdwn", "text": f"*Total Leads*\n{total_leads}"},
            {"type": "mrkdwn", "text": f"*Total MQLs*\n{total_mqls}"},
            {"type": "mrkdwn", "text": f"*Overall Conversion*\n{overall_rate}%"},
        ]},
        {"type": "divider"},
        {"type": "section", "text": {
            "type": "mrkdwn",
            "text": f"*Conversion by Source*\n{source_lines}",
        }},
        {"type": "context", "elements": [{
            "type": "mrkdwn",
            "text": f"Last 7 days | Generated {datetime.now().strftime('%A, %B %d, %Y')}",
        }]},
    ]
 
    slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    result = slack.chat_postMessage(
        channel=os.environ["SLACK_CHANNEL_ID"],
        text="Weekly Lead-to-MQL Conversion Report",
        blocks=blocks,
        unfurl_links=False,
    )
    print(f"Posted report: {result['ts']}")
 
# --- Main ---
if __name__ == "__main__":
    start_ms, end_ms = get_date_range(7)
    all_leads = search_contacts(start_ms, end_ms)
    mqls = search_contacts(start_ms, end_ms, extra_filters=[
        {"propertyName": "lifecyclestage", "operator": "EQ", "value": "marketingqualifiedlead"}
    ])
    report, total_leads, total_mqls, overall_rate = calculate_conversion(all_leads, mqls)
    post_report(report, total_leads, total_mqls, overall_rate)