AFFAANhandClaude Opus 5
b4007ce825
Server to server over the CRM's integration API, deliberately not through MCP. MCP is for an agent deciding to do something; leads have to reach the CRM on a schedule whether or not anyone is talking to the agent, and a lead that arrives only when someone asks for it arrives too late. Leads are re-sent for a trailing window on every run rather than tracked as new-since-last-time. Meta's lead id is the external id, so the CRM ignores one it already has — which makes a half-failed run heal itself on the next cycle with no bookkeeping. Meta returns form answers as a list under names the form's author chose, so mapping is best-effort against aliases: full_name or first/last, phone or phone_number or mobile. Unrecognised answers are kept in metadata rather than dropped, and a lead with a phone but no name still gets through — the CRM requires a first name, and losing a real enquiry to satisfy a validator would be the wrong trade. Campaign figures come from stored insights, not Meta: this runs every few minutes and re-reading Meta would spend the quota on numbers that change hourly. They go through an upsert rather than the idempotent create used for leads, because a campaign's figures change every time they are read. One care point, learnt the hard way and now commented and tested: stored metrics are already in minor units, so summing them with the raw-row normaliser reports spend a hundredfold. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
160 lines
4.6 KiB
Python
160 lines
4.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Mapping a Meta lead-ad submission onto a CRM lead.
|
|
|
|
Form field names come from whoever built the form, so the mapping is
|
|
best-effort against aliases. The tests are mostly about the awkward cases:
|
|
a name that arrives as one field, a lead with no name at all, and answers
|
|
the mapping does not recognise.
|
|
"""
|
|
import pytest
|
|
|
|
from adclaw.meta.leads import crm_payload, field_map, split_name
|
|
|
|
|
|
def _lead(fields, **overrides):
|
|
lead = {
|
|
"id": "lead_1",
|
|
"created_time": "2026-08-01T10:00:00+0000",
|
|
"campaign_id": "meta_camp_1",
|
|
"adset_id": "meta_set_1",
|
|
"ad_id": "meta_ad_1",
|
|
"form_id": "form_1",
|
|
"field_data": [
|
|
{"name": name, "values": [value]} for name, value in fields.items()
|
|
],
|
|
}
|
|
lead.update(overrides)
|
|
return lead
|
|
|
|
|
|
# --- flattening ---
|
|
|
|
|
|
def test_field_data_is_flattened_to_a_dict():
|
|
fields = field_map(_lead({"email": "a@example.com", "phone_number": "+91 99"}))
|
|
|
|
assert fields == {"email": "a@example.com", "phone_number": "+91 99"}
|
|
|
|
|
|
def test_field_names_are_lowercased():
|
|
"""Form authors capitalise inconsistently; the aliases are lowercase."""
|
|
lead = {"field_data": [{"name": "Full_Name", "values": ["Asha"]}]}
|
|
|
|
assert field_map(lead) == {"full_name": "Asha"}
|
|
|
|
|
|
def test_answers_with_no_values_are_skipped():
|
|
lead = {
|
|
"field_data": [
|
|
{"name": "email", "values": []},
|
|
{"name": "phone", "values": ["99"]},
|
|
"not a dict",
|
|
],
|
|
}
|
|
|
|
assert field_map(lead) == {"phone": "99"}
|
|
|
|
|
|
# --- names ---
|
|
|
|
|
|
def test_explicit_first_and_last_names_win():
|
|
first, last = split_name({"first_name": "Asha", "last_name": "Menon"})
|
|
|
|
assert (first, last) == ("Asha", "Menon")
|
|
|
|
|
|
def test_a_full_name_is_split_on_the_last_space():
|
|
assert split_name({"full_name": "Asha Priya Menon"}) == ("Asha Priya", "Menon")
|
|
|
|
|
|
def test_a_one_word_name_has_no_surname():
|
|
assert split_name({"full_name": "Asha"}) == ("Asha", "")
|
|
|
|
|
|
def test_explicit_fields_beat_splitting_a_full_name():
|
|
"""A split guesses where the surname starts; the form did not."""
|
|
first, last = split_name({"full_name": "Asha Menon", "first_name": "Asha Priya"})
|
|
|
|
assert first == "Asha Priya"
|
|
|
|
|
|
def test_no_name_fields_at_all():
|
|
assert split_name({}) == ("", "")
|
|
|
|
|
|
# --- the CRM payload ---
|
|
|
|
|
|
def test_the_meta_lead_id_becomes_the_external_id():
|
|
"""It is what makes re-sending the same lead harmless."""
|
|
payload = crm_payload(_lead({"email": "a@example.com"}))
|
|
|
|
assert payload["external_id"] == "lead_1"
|
|
assert payload["provider"] == "maskanx"
|
|
|
|
|
|
def test_the_ad_that_produced_the_lead_is_carried_through():
|
|
payload = crm_payload(_lead({"email": "a@example.com"}), campaign_name="Q3")
|
|
|
|
assert payload["campaign"] == {
|
|
"campaign_id": "meta_camp_1",
|
|
"adset_id": "meta_set_1",
|
|
"ad_id": "meta_ad_1",
|
|
"form_id": "form_1",
|
|
"campaign_name": "Q3",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"alias", ["phone_number", "phone", "mobile_number", "mobile", "contact_number"],
|
|
)
|
|
def test_phone_is_found_under_any_known_alias(alias):
|
|
payload = crm_payload(_lead({alias: "+91 99"}))
|
|
|
|
assert payload["phone"] == "+91 99"
|
|
|
|
|
|
def test_a_lead_with_only_a_phone_number_still_gets_a_name():
|
|
"""The CRM requires one, and a nameless enquiry is still an enquiry."""
|
|
payload = crm_payload(_lead({"phone_number": "+91 99"}))
|
|
|
|
assert payload["first_name"] == "Unknown"
|
|
assert payload["phone"] == "+91 99"
|
|
|
|
|
|
def test_the_email_local_part_stands_in_for_a_missing_name():
|
|
payload = crm_payload(_lead({"email": "asha.menon@example.com"}))
|
|
|
|
assert payload["first_name"] == "asha.menon"
|
|
|
|
|
|
def test_unrecognised_answers_are_kept_rather_than_dropped():
|
|
"""The form's author asked for them on purpose."""
|
|
payload = crm_payload(
|
|
_lead({"full_name": "Asha Menon", "budget_range": "50-70 lakh"}),
|
|
)
|
|
|
|
assert payload["metadata"]["extra_fields"] == {"budget_range": "50-70 lakh"}
|
|
|
|
|
|
def test_mapped_answers_are_not_repeated_in_extra_fields():
|
|
payload = crm_payload(_lead({"email": "a@example.com", "full_name": "Asha Menon"}))
|
|
|
|
assert payload["metadata"]["extra_fields"] == {}
|
|
|
|
|
|
def test_the_lead_title_names_the_campaign():
|
|
payload = crm_payload(_lead({"full_name": "Asha Menon"}), campaign_name="Q3 leads")
|
|
|
|
assert payload["lead_title"] == "Asha Menon - Q3 leads"
|
|
|
|
|
|
def test_absent_optional_fields_are_none_not_empty_strings():
|
|
"""The CRM validates email; "" is not a valid address, None is absent."""
|
|
payload = crm_payload(_lead({"full_name": "Asha Menon"}))
|
|
|
|
assert payload["email"] is None
|
|
assert payload["phone"] is None
|
|
assert payload["company_name"] is None
|