"""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, }