"""Lead Attribution & UTM Tracking — helper for tracking lead sources.
Generates UTM-tagged landing page URLs for ad campaigns, parses incoming
UTM parameters from webhooks/lead forms, and matches leads back to campaigns.
"""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
from urllib.parse import parse_qs, urlencode, urlparse
from ...models import AdCampaign, LeadAttribution, db
logger = logging.getLogger(__name__)
class AttributionHelper:
"""Generate UTM-tagged URLs and track lead sources."""
@staticmethod
def generate_landing_url(
base_url: str,
campaign: AdCampaign,
ad_group: Optional[str] = None,
keyword: Optional[str] = None,
ad_content: Optional[str] = None,
) -> str:
"""Generate a UTM-tagged landing page URL for an ad.
Args:
base_url: Base landing page URL (e.g., "https://example.com/landing")
campaign: AdCampaign model instance
ad_group: Optional ad group name
keyword: Optional keyword text for utm_term
ad_content: Optional ad creative identifier for utm_content
Returns:
Full URL with UTM parameters appended
"""
params: Dict[str, str] = {
"utm_source": campaign.source_service.replace("_ads", ""),
"utm_medium": "cpc",
"utm_campaign": campaign.name[:90],
}
if ad_group:
params["utm_ad_group"] = ad_group[:90]
if keyword:
params["utm_term"] = keyword[:90]
if ad_content:
params["utm_content"] = ad_content[:90]
# Parse base URL and merge with existing params
parsed = urlparse(base_url)
existing_params = dict(parse_qs(parsed.query))
# Flatten single-value lists
for key, value in existing_params.items():
if isinstance(value, list) and len(value) == 1:
existing_params[key] = value[0]
# Merge new params (overwrite)
existing_params.update(params)
# Rebuild query string
new_query = urlencode(existing_params)
new_parsed = parsed._replace(query=new_query)
return new_parsed.geturl()
@staticmethod
def extract_utm_params(url: str) -> Dict[str, str]:
"""Extract UTM parameters from a URL.
Args:
url: URL string potentially containing UTM parameters
Returns:
Dict of UTM param name -> value (without 'utm_' prefix)
"""
parsed = urlparse(url)
params = parse_qs(parsed.query)
utm_params = {}
for key, value in params.items():
if key.startswith("utm_"):
utm_params[key] = value[0] if isinstance(value, list) else value
return utm_params
@staticmethod
def extract_utm_from_lead(lead_data: Dict[str, Any]) -> Dict[str, str]:
"""Extract UTM parameters from lead data (webhook payload, form submission).
Checks these fields in order:
1. Direct UTM params in lead_data
2. Referrer URL in lead_data
3. Landing page URL in lead_data
Args:
lead_data: Dict containing lead information
Returns:
Dict of UTM param name -> value
"""
utm: Dict[str, str] = {}
# Check direct UTM params (with or without prefix)
utm_keys = [
"utm_source", "utm_medium", "utm_campaign", "utm_term",
"utm_content", "utm_ad_group", "utm_ad", "utm_id",
]
for key in utm_keys:
if key in lead_data:
utm[key] = str(lead_data[key])
# Fall back to referrer/landing page URL
if not utm:
for field in ["referrer", "landing_page", "source_url", "url"]:
if field in lead_data and lead_data[field]:
utm = AttributionHelper.extract_utm_params(str(lead_data[field]))
if utm:
break
return utm
@staticmethod
def match_campaign(
company_id: str,
utm_params: Dict[str, str],
) -> Optional[AdCampaign]:
"""Match UTM parameters to an existing campaign.
Tries to match by:
1. utm_campaign (name)
2. utm_source + partial name match
Args:
company_id: Company UUID
utm_params: Dict of UTM parameters
Returns:
AdCampaign instance if matched, None otherwise
"""
campaign = None
# Try exact match on campaign name
campaign_name = utm_params.get("utm_campaign")
if campaign_name:
campaign = AdCampaign.query.filter_by(
company_id=company_id,
name=campaign_name,
).first()
# Fallback: partial name match
if not campaign and campaign_name:
campaign = AdCampaign.query.filter(
AdCampaign.company_id == company_id,
AdCampaign.name.ilike(f"%{campaign_name}%"),
).first()
return campaign
@staticmethod
def record_attribution(
company_id: str,
external_lead_id: str,
lead_source: str,
utm_params: Optional[Dict[str, str]] = None,
click_cost: Optional[float] = None,
deal_amount: Optional[float] = None,
external_deal_id: Optional[str] = None,
attribution_window_days: int = 30,
) -> LeadAttribution:
"""Record a lead attribution event.
Maps UTM parameters directly to LeadAttribution model fields.
Args:
company_id: Company UUID
external_lead_id: Lead identifier (CRM/Angi lead ID)
lead_source: Source type (web_form, phone_call, email, angi, etc.)
utm_params: Dict of UTM parameters
click_cost: Estimated cost of that click
deal_amount: Revenue when deal closes
external_deal_id: CRM deal ID when converted
attribution_window_days: Attribution window in days
Returns:
Created LeadAttribution instance
"""
utm_params = utm_params or {}
# Determine status based on inputs
status = "lead" if external_lead_id else "click"
if external_deal_id and deal_amount:
status = "deal"
attribution = LeadAttribution(
company_id=company_id,
external_lead_id=external_lead_id,
lead_source=lead_source,
utm_source=utm_params.get("utm_source", ""),
utm_campaign=utm_params.get("utm_campaign", ""),
utm_ad_group=utm_params.get("utm_ad_group", ""),
utm_ad=utm_params.get("utm_ad", ""),
utm_content=utm_params.get("utm_content", ""),
utm_medium=utm_params.get("utm_medium", ""),
utm_term=utm_params.get("utm_term", ""),
click_date=datetime.now(timezone.utc),
lead_date=datetime.now(timezone.utc),
click_cost=click_cost,
external_deal_id=external_deal_id,
deal_amount=deal_amount,
attribution_status=status,
attribution_window_days=attribution_window_days,
metadata_json={k: v for k, v in utm_params.items() if not k.startswith("utm_")},
)
# Calculate ROAS if we have deal amount and spend
if deal_amount and click_cost and click_cost > 0:
attribution.deal_closed_date = datetime.now(timezone.utc)
attribution.roas = round(deal_amount / click_cost, 2)
db.session.add(attribution)
db.session.flush()
logger.info(
"Attribution recorded: lead=%s, source=%s, campaign=%s, status=%s",
external_lead_id, lead_source, utm_params.get("utm_campaign", ""), status,
)
return attribution
@staticmethod
def get_campaign_attribution_summary(
company_id: str,
campaign_name: str,
days: int = 30,
) -> Dict[str, Any]:
"""Get attribution summary for a campaign.
Args:
company_id: Company UUID
campaign_name: Campaign name (from utm_campaign)
days: Number of days to look back
Returns:
Dict with attribution metrics
"""
since = datetime.now(timezone.utc) - timedelta(days=days)
attributions = LeadAttribution.query.filter_by(
company_id=company_id,
utm_campaign=campaign_name,
).filter(
LeadAttribution.created_at >= since
).all()
if not attributions:
return {
"campaign_name": campaign_name,
"leads": 0,
"total_value": 0.0,
"total_cost": 0.0,
"roi": 0.0,
"avg_cost_per_lead": 0.0,
}
total_value = sum(a.deal_amount or 0 for a in attributions)
total_cost = sum(a.click_cost or 0 for a in attributions)
count = len(attributions)
roi = ((total_value - total_cost) / total_cost * 100) if total_cost > 0 else 0.0
avg_cpl = total_cost / count if count > 0 else 0.0
return {
"campaign_name": campaign_name,
"leads": count,
"total_value": round(total_value, 2),
"total_cost": round(total_cost, 2),
"roi": round(roi, 2),
"avg_cost_per_lead": round(avg_cpl, 2),
"attributions": [
{
"lead_id": a.external_lead_id,
"source": a.utm_source,
"term": a.utm_term,
"value": a.deal_amount,
"cost": a.click_cost,
"status": a.attribution_status,
"created": a.created_at.isoformat() if a.created_at else None,
}
for a in attributions
],
}
@staticmethod
def get_company_attribution_dashboard(
company_id: str,
days: int = 30,
) -> Dict[str, Any]:
"""Get attribution dashboard data for a company.
Returns aggregated attribution data across all campaigns/sources.
Args:
company_id: Company UUID
days: Number of days to look back
Returns:
Dict with dashboard metrics
"""
since = datetime.now(timezone.utc) - timedelta(days=days)
attributions = LeadAttribution.query.filter_by(
company_id=company_id,
).filter(
LeadAttribution.created_at >= since
).all()
total_leads = len(attributions)
total_value = sum(a.deal_amount or 0 for a in attributions)
total_cost = sum(a.click_cost or 0 for a in attributions)
overall_roi = ((total_value - total_cost) / total_cost * 100) if total_cost > 0 else 0.0
avg_cpl = total_cost / total_leads if total_leads > 0 else 0.0
# Group by source
by_source: Dict[str, Dict[str, Any]] = {}
for a in attributions:
source = a.utm_source or "unknown"
if source not in by_source:
by_source[source] = {
"leads": 0,
"value": 0.0,
"cost": 0.0,
}
by_source[source]["leads"] += 1
by_source[source]["value"] += a.deal_amount or 0
by_source[source]["cost"] += a.click_cost or 0
# Calculate ROI per source
for source in by_source:
cost = by_source[source]["cost"]
value = by_source[source]["value"]
by_source[source]["roi"] = round(((value - cost) / cost * 100) if cost > 0 else 0.0, 2)
by_source[source]["avg_cpl"] = round(cost / by_source[source]["leads"], 2)
by_source[source]["value"] = round(value, 2)
by_source[source]["cost"] = round(cost, 2)
# Group by campaign
by_campaign: Dict[str, Dict[str, Any]] = {}
for a in attributions:
camp = a.utm_campaign or "unknown"
if camp not in by_campaign:
by_campaign[camp] = {
"leads": 0,
"value": 0.0,
"cost": 0.0,
}
by_campaign[camp]["leads"] += 1
by_campaign[camp]["value"] += a.deal_amount or 0
by_campaign[camp]["cost"] += a.click_cost or 0
for camp in by_campaign:
cost = by_campaign[camp]["cost"]
value = by_campaign[camp]["value"]
by_campaign[camp]["roi"] = round(((value - cost) / cost * 100) if cost > 0 else 0.0, 2)
by_campaign[camp]["avg_cpl"] = round(cost / by_campaign[camp]["leads"], 2)
by_campaign[camp]["value"] = round(value, 2)
by_campaign[camp]["cost"] = round(cost, 2)
return {
"period_days": days,
"total_leads": total_leads,
"total_value": round(total_value, 2),
"total_cost": round(total_cost, 2),
"overall_roi": round(overall_roi, 2),
"avg_cpl": round(avg_cpl, 2),
"by_source": by_source,
"by_campaign": by_campaign,
}