"""Lead Generation Engine API routes.
Campaign templates, creatives, keywords, optimization rules, and attribution.
Gated to growth+ tiers since paid ads management is a premium feature.
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone, timedelta
from flask import Blueprint, request, jsonify, g
from flask_login import current_user
from ..models import (
db, Company, UserCompany,
AdTemplate, AdCreative, AdKeyword, LeadAttribution,
OptimizationRule, OptimizationLog,
AdCampaign, AdMetric,
)
from .api_proxy import require_auth_json, require_tier, _check_company_access
from app.utils.csrf import require_csrf
logger = logging.getLogger(__name__)
lead_gen_bp = Blueprint("lead_gen", __name__)
# =============================================================================
# Campaign Templates
# =============================================================================
@lead_gen_bp.route('/api/lead-gen/templates', methods=['GET'])
@require_auth_json()
def list_templates():
"""List available campaign templates, optionally filtered by vertical."""
vertical = request.args.get('vertical')
query = AdTemplate.query.filter_by(is_active=True)
if vertical:
query = query.filter_by(vertical=vertical)
templates = query.order_by(AdTemplate.vertical, AdTemplate.name).all()
return jsonify({
"success": True,
"data": {
"templates": [t.to_dict() for t in templates],
"total": len(templates),
}
})
@lead_gen_bp.route('/api/lead-gen/templates/<template_id>', methods=['GET'])
@require_auth_json()
def get_template(template_id):
"""Get a single campaign template by ID."""
template = db.session.get(AdTemplate, template_id)
if not template:
return jsonify({'error': 'Template not found'}), 404
return jsonify({
"success": True,
"data": template.to_dict()
})
@lead_gen_bp.route('/api/lead-gen/templates', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='command')
def create_template():
"""Create a new campaign template. Super admin only or company admin."""
# Only super_admins can create system-wide templates via this endpoint
# Company admins can create company-specific templates (handled via company route)
if current_user.role != 'super_admin':
return jsonify({'error': 'Only super_admin can create system templates. Use /api/company/<id>/lead-gen/templates for company-specific.'}), 403
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
required = ['name', 'slug', 'vertical', 'structure']
missing = [f for f in required if not data.get(f)]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
# Check slug uniqueness
existing = AdTemplate.query.filter_by(slug=data['slug']).first()
if existing:
return jsonify({'error': f'Template slug "{data["slug"]}" already exists'}), 409
template = AdTemplate(
name=data['name'],
slug=data['slug'],
vertical=data['vertical'],
description=data.get('description', ''),
budget_recommendation=data.get('budget_recommendation'),
target_cpa=data.get('target_cpa'),
bidding_strategy=data.get('bidding_strategy', 'target_cpa'),
structure_json=data['structure'],
is_active=data.get('is_active', True),
)
db.session.add(template)
db.session.commit()
logger.info("Template created: %s by user %s", template.slug, current_user.id)
return jsonify({
"success": True,
"data": template.to_dict()
}), 201
@lead_gen_bp.route('/api/lead-gen/templates/<template_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='command')
def update_template(template_id):
"""Update an existing campaign template."""
if current_user.role != 'super_admin':
return jsonify({'error': 'Only super_admin can modify system templates'}), 403
template = db.session.get(AdTemplate, template_id)
if not template:
return jsonify({'error': 'Template not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
updatable = ['name', 'description', 'budget_recommendation', 'target_cpa',
'bidding_strategy', 'structure_json', 'is_active']
for field in updatable:
if field in data:
setattr(template, field, data[field])
template.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
"success": True,
"data": template.to_dict()
})
@lead_gen_bp.route('/api/lead-gen/templates/<template_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def delete_template(template_id):
"""Soft-delete a campaign template."""
if current_user.role != 'super_admin':
return jsonify({'error': 'Only super_admin can delete templates'}), 403
template = db.session.get(AdTemplate, template_id)
if not template:
return jsonify({'error': 'Template not found'}), 404
template.is_active = False
db.session.commit()
return jsonify({"success": True, "message": "Template deactivated"})
# =============================================================================
# Ad Creatives
# =============================================================================
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/creatives', methods=['GET'])
@require_auth_json()
def list_creatives(company_id):
"""List ad creatives for a company, optionally filtered by vertical/service."""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
vertical = request.args.get('vertical')
service_type = request.args.get('service_type')
ad_type = request.args.get('ad_type')
query = AdCreative.query.filter_by(company_id=company_id, is_active=True)
if vertical:
query = query.filter_by(vertical=vertical)
if service_type:
query = query.filter_by(service_type=service_type)
if ad_type:
query = query.filter_by(ad_type=ad_type)
creatives = query.order_by(AdCreative.updated_at.desc()).all()
return jsonify({
"success": True,
"data": {
"creatives": [
{
'id': c.id,
'vertical': c.vertical,
'service_type': c.service_type,
'ad_type': c.ad_type,
'headlines': c.headlines,
'descriptions': c.descriptions,
'total_impressions': c.total_impressions,
'total_clicks': c.total_clicks,
'total_conversions': c.total_conversions,
'avg_ctr': c.avg_ctr,
'is_winner': c.is_winner,
'is_active': c.is_active,
'created_at': c.created_at.isoformat() if c.created_at else None,
}
for c in creatives
],
"total": len(creatives),
}
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/creatives', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def create_creative(company_id):
"""Create a new ad creative variant."""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
required = ['vertical', 'service_type', 'ad_type', 'headlines', 'descriptions']
missing = [f for f in required if not data.get(f)]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
creative = AdCreative(
company_id=company_id,
vertical=data['vertical'],
service_type=data['service_type'],
ad_type=data['ad_type'],
headlines=data['headlines'],
descriptions=data['descriptions'],
display_url=data.get('display_url', ''),
final_url=data.get('final_url', ''),
image_url=data.get('image_url', ''),
call_to_action=data.get('call_to_action', ''),
)
db.session.add(creative)
db.session.commit()
return jsonify({
"success": True,
"data": {
'id': creative.id,
'vertical': creative.vertical,
'service_type': creative.service_type,
'ad_type': creative.ad_type,
'headlines': creative.headlines,
'descriptions': creative.descriptions,
}
}), 201
# =============================================================================
# Optimization Rules
# =============================================================================
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/rules', methods=['GET'])
@require_auth_json()
def list_rules(company_id):
"""List optimization rules for a company."""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
rules = OptimizationRule.query.filter_by(company_id=company_id).all()
return jsonify({
"success": True,
"data": {
"rules": [
{
'id': r.id,
'name': r.name,
'rule_type': r.rule_type,
'target_level': r.target_level,
'params': r.params_json,
'is_active': r.is_active,
'last_run_at': r.last_run_at.isoformat() if r.last_run_at else None,
'total_actions': r.total_actions,
}
for r in rules
],
"total": len(rules),
}
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/rules', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def create_rule(company_id):
"""Create an optimization rule.
Default rules applied if none specified:
- kill_high_cpa: Pause campaigns with CPA > 2x target
- pause_zero_conv: Pause campaigns with 0 conversions after $100 spend
"""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
required = ['name', 'rule_type', 'params']
missing = [f for f in required if not data.get(f)]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
# Safety validation: ensure budget caps are set
params = data['params']
if data['rule_type'] in ('kill_high_cpa', 'scale_low_cpa', 'budget_reallocation'):
if 'max_budget_increase_pct' not in params:
params['max_budget_increase_pct'] = 50 # Default: max 50% budget increase
rule = OptimizationRule(
company_id=company_id,
name=data['name'],
rule_type=data['rule_type'],
target_level=data.get('target_level', 'campaign'),
params_json=params,
is_active=data.get('is_active', True),
)
db.session.add(rule)
db.session.commit()
return jsonify({
"success": True,
"data": {
'id': rule.id,
'name': rule.name,
'rule_type': rule.rule_type,
'target_level': rule.target_level,
'params': rule.params_json,
'is_active': rule.is_active,
}
}), 201
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/rules/<rule_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def update_rule(company_id, rule_id):
"""Update an optimization rule."""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
rule = db.session.get(OptimizationRule, rule_id)
if not rule or rule.company_id != company_id:
return jsonify({'error': 'Rule not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
updatable = ['name', 'rule_type', 'target_level', 'params_json', 'is_active']
for field in updatable:
if field in data:
setattr(rule, field, data[field])
rule.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
"success": True,
"data": {
'id': rule.id,
'name': rule.name,
'rule_type': rule.rule_type,
'is_active': rule.is_active,
'params': rule.params_json,
}
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/rules/<rule_id>/run', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def run_rule(company_id, rule_id):
"""Manually trigger an optimization rule run."""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
rule = db.session.get(OptimizationRule, rule_id)
if not rule or rule.company_id != company_id:
return jsonify({'error': 'Rule not found'}), 404
if not rule.is_active:
return jsonify({'error': 'Rule is inactive. Activate before running.'}), 400
# Import the optimization engine
from app.services.lead_gen.optimization_engine import OptimizationEngine
engine = OptimizationEngine(company_id)
result = engine.run_rule(rule)
return jsonify({
"success": True,
"data": result
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/optimization-log', methods=['GET'])
@require_auth_json()
def list_optimization_logs(company_id):
"""List optimization action history."""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
limit = int(request.args.get('limit', 50))
days = int(request.args.get('days', 30))
since = datetime.now(timezone.utc) - timedelta(days=days)
logs = (
OptimizationLog.query
.filter_by(company_id=company_id)
.filter(OptimizationLog.created_at >= since)
.order_by(OptimizationLog.created_at.desc())
.limit(limit)
.all()
)
return jsonify({
"success": True,
"data": {
"logs": [
{
'id': l.id,
'rule_id': l.rule_id,
'source_service': l.source_service,
'target_type': l.target_type,
'target_name': l.target_name,
'action': l.action,
'reason': l.reason,
'status': l.status,
'created_at': l.created_at.isoformat() if l.created_at else None,
}
for l in logs
],
"total": len(logs),
}
})
# =============================================================================
# Lead Attribution
# =============================================================================
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/attribution', methods=['GET'])
@require_auth_json()
def list_attribution(company_id):
"""List lead attributions with filters."""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
utm_source = request.args.get('utm_source')
status = request.args.get('status')
days = int(request.args.get('days', 30))
since = datetime.now(timezone.utc) - timedelta(days=days)
query = LeadAttribution.query.filter_by(company_id=company_id)
query = query.filter(LeadAttribution.created_at >= since)
if utm_source:
query = query.filter_by(utm_source=utm_source)
if status:
query = query.filter_by(attribution_status=status)
attributions = query.order_by(LeadAttribution.created_at.desc()).limit(100).all()
# Calculate summary stats
from sqlalchemy import func
total_spend = db.session.query(func.sum(LeadAttribution.total_ad_spend)).filter(
LeadAttribution.company_id == company_id,
LeadAttribution.created_at >= since,
).scalar() or 0
total_revenue = db.session.query(func.sum(LeadAttribution.deal_amount)).filter(
LeadAttribution.company_id == company_id,
LeadAttribution.attribution_status == 'deal',
LeadAttribution.created_at >= since,
).scalar() or 0
return jsonify({
"success": True,
"data": {
"attributions": [
{
'id': a.id,
'utm_source': a.utm_source,
'utm_campaign': a.utm_campaign,
'utm_term': a.utm_term,
'lead_source': a.lead_source,
'attribution_status': a.attribution_status,
'click_cost': a.click_cost,
'total_ad_spend': a.total_ad_spend,
'deal_amount': a.deal_amount,
'roas': a.roas,
'created_at': a.created_at.isoformat() if a.created_at else None,
}
for a in attributions
],
"summary": {
"total_spend": round(total_spend, 2),
"total_revenue": round(total_revenue, 2),
"overall_roas": round(total_revenue / total_spend, 2) if total_spend > 0 else 0,
"attribution_count": len(attributions),
}
}
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/attribution', methods=['POST'])
@require_auth_json()
@require_csrf
def create_attribution(company_id):
"""Capture a lead attribution from UTM params.
Called when a form is submitted with UTM tracking params.
"""
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
attribution = LeadAttribution(
company_id=company_id,
external_lead_id=data.get('external_lead_id'),
lead_source=data.get('lead_source', 'web_form'),
utm_source=data.get('utm_source', ''),
utm_campaign=data.get('utm_campaign', ''),
utm_ad_group=data.get('utm_ad_group', ''),
utm_ad=data.get('utm_ad', ''),
utm_content=data.get('utm_content', ''),
utm_medium=data.get('utm_medium', ''),
utm_term=data.get('utm_term', ''),
click_date=data.get('click_date'),
lead_date=datetime.now(timezone.utc),
click_cost=data.get('click_cost'),
total_ad_spend=data.get('total_ad_spend'),
attribution_status=data.get('attribution_status', 'lead'),
)
db.session.add(attribution)
db.session.commit()
return jsonify({
"success": True,
"data": {
'id': attribution.id,
'utm_campaign': attribution.utm_campaign,
'attribution_status': attribution.attribution_status,
}
}), 201
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/attribution/<attribution_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_attribution(company_id, attribution_id):
"""Update attribution with deal info when a lead converts."""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
attribution = db.session.get(LeadAttribution, attribution_id)
if not attribution or attribution.company_id != company_id:
return jsonify({'error': 'Attribution not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
# Update deal info
if 'external_deal_id' in data:
attribution.external_deal_id = data['external_deal_id']
if 'deal_amount' in data:
attribution.deal_amount = data['deal_amount']
attribution.deal_closed_date = datetime.now(timezone.utc)
attribution.attribution_status = 'deal'
# Calculate ROAS
if attribution.total_ad_spend and attribution.total_ad_spend > 0:
attribution.roas = round(attribution.deal_amount / attribution.total_ad_spend, 2)
attribution.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
"success": True,
"data": {
'id': attribution.id,
'attribution_status': attribution.attribution_status,
'deal_amount': attribution.deal_amount,
'roas': attribution.roas,
}
})
# =============================================================================
# Campaign Generation (from template)
# =============================================================================
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/campaigns/generate', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def generate_campaign(company_id):
"""Generate a campaign on Google/Facebook Ads from a template.
Takes a template ID and campaign parameters (city, radius, budget) and
creates the actual campaign on the connected ad platform.
"""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
required = ['template_id', 'source_service', 'city', 'radius_miles', 'daily_budget']
missing = [f for f in required if not data.get(f)]
if missing:
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
# Get template
template = db.session.get(AdTemplate, data['template_id'])
if not template or not template.is_active:
return jsonify({'error': 'Template not found or inactive'}), 404
# Check for connected ad account
from ..models import Connector
connector = Connector.query.filter_by(
company_id=company_id,
service=data['source_service'],
status='connected',
).first()
if not connector:
return jsonify({
'error': f'No connected {data["source_service"]} account. Connect your ad account first.',
'connect_url': f'/app/connectors/{data["source_service"]}/connect',
}), 400
# Use the campaign generation service
from app.services.lead_gen.campaign_generator import CampaignGenerator
generator = CampaignGenerator(company_id)
result = generator.generate_campaign(
template=template,
source_service=data['source_service'],
connector=connector,
city=data['city'],
radius_miles=data['radius_miles'],
daily_budget=data['daily_budget'],
state=data.get('state', ''),
zip_code=data.get('zip_code', ''),
landing_page_url=data.get('landing_page_url', ''),
tracking_phone=data.get('tracking_phone', ''),
dry_run=data.get('dry_run', False),
)
return jsonify({
"success": True,
"data": result
}), 201
# =============================================================================
# Lead Gen Dashboard Summary
# =============================================================================
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/dashboard', methods=['GET'])
@require_auth_json()
def lead_gen_dashboard(company_id):
"""Return a summary dashboard for the lead gen module."""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
days = int(request.args.get('days', 30))
since = datetime.now(timezone.utc) - timedelta(days=days)
# Active campaigns
active_campaigns = AdCampaign.query.filter(
AdCampaign.company_id == company_id,
AdCampaign.status == 'ENABLED',
).count()
# Ad metrics summary
from sqlalchemy import func
metrics_query = AdMetric.query.filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= since.date(),
)
total_spend = metrics_query.with_entities(func.sum(AdMetric.spend)).scalar() or 0
total_impressions = metrics_query.with_entities(func.sum(AdMetric.impressions)).scalar() or 0
total_clicks = metrics_query.with_entities(func.sum(AdMetric.clicks)).scalar() or 0
total_conversions = metrics_query.with_entities(func.sum(AdMetric.conversions)).scalar() or 0
# Attribution summary
attr_query = LeadAttribution.query.filter(
LeadAttribution.company_id == company_id,
LeadAttribution.created_at >= since,
)
total_revenue = attr_query.with_entities(func.sum(LeadAttribution.deal_amount)).filter(
LeadAttribution.attribution_status == 'deal',
).scalar() or 0
active_rules = OptimizationRule.query.filter_by(
company_id=company_id,
is_active=True,
).count()
recent_actions = OptimizationLog.query.filter_by(
company_id=company_id,
).filter(
OptimizationLog.created_at >= since
).count()
return jsonify({
"success": True,
"data": {
"campaigns": {
"active": active_campaigns,
"total_spend": round(total_spend, 2),
"total_impressions": int(total_impressions),
"total_clicks": int(total_clicks),
"total_conversions": float(total_conversions),
},
"attribution": {
"total_revenue": round(total_revenue, 2),
"overall_roas": round(total_revenue / total_spend, 2) if total_spend > 0 else 0,
},
"optimization": {
"active_rules": active_rules,
"recent_actions": recent_actions,
},
}
})
# =============================================================================
# Keywords
# =============================================================================
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/keywords', methods=['GET'])
@require_auth_json()
def list_keywords(company_id):
"""List keywords for a company, optionally filtered by campaign/ad group/status.
Query params:
campaign_id — filter by external_campaign_id
ad_group_id — filter by external_ad_group_id
status — filter by status (enabled/paused/removed)
source_service — filter by source (google_ads/facebook_ads)
search — text search on keyword text (ILIKE)
sort — sort field (spend, clicks, impressions, conversions, ctr, cpc)
order — asc or desc (default: desc)
limit — max results (default: 100, max: 500)
"""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
query = AdKeyword.query.filter_by(company_id=company_id)
# Filters
campaign_id = request.args.get('campaign_id')
if campaign_id:
query = query.filter_by(external_campaign_id=campaign_id)
ad_group_id = request.args.get('ad_group_id')
if ad_group_id:
query = query.filter_by(external_ad_group_id=ad_group_id)
status_filter = request.args.get('status')
if status_filter:
query = query.filter_by(status=status_filter)
source = request.args.get('source_service')
if source:
query = query.filter_by(source_service=source)
search = request.args.get('search')
if search:
query = query.filter(AdKeyword.text.ilike(f"%{search}%"))
# Sorting
sort_field = request.args.get('sort', 'total_spend')
sort_order = request.args.get('order', 'desc')
sort_map = {
'spend': AdKeyword.total_spend,
'clicks': AdKeyword.total_clicks,
'impressions': AdKeyword.total_impressions,
'conversions': AdKeyword.total_conversions,
'ctr': AdKeyword.avg_ctr,
'cpc': AdKeyword.max_cpc,
'updated': AdKeyword.updated_at,
}
col = sort_map.get(sort_field, AdKeyword.total_spend)
query = query.order_by(col.desc() if sort_order == 'desc' else col.asc())
limit = min(int(request.args.get('limit', 100)), 500)
keywords = query.limit(limit).all()
# Calculate aggregate stats
from sqlalchemy import func
stats_query = AdKeyword.query.filter_by(company_id=company_id)
stats = {
'total': stats_query.count(),
'enabled': stats_query.filter_by(status='enabled').count(),
'paused': stats_query.filter_by(status='paused').count(),
'total_spend': round((stats_query.with_entities(func.sum(AdKeyword.total_spend)).scalar() or 0), 2),
'total_clicks': int(stats_query.with_entities(func.sum(AdKeyword.total_clicks)).scalar() or 0),
'total_impressions': int(stats_query.with_entities(func.sum(AdKeyword.total_impressions)).scalar() or 0),
'total_conversions': float(stats_query.with_entities(func.sum(AdKeyword.total_conversions)).scalar() or 0),
}
# Calculate aggregate CPA and CTR
total_spend_val = stats['total_spend']
total_conv_val = stats['total_conversions']
stats['aggregate_cpa'] = round(total_spend_val / total_conv_val, 2) if total_conv_val > 0 else 0
return jsonify({
"success": True,
"data": {
"keywords": [
{
'id': kw.id,
'text': kw.text,
'match_type': kw.match_type,
'source_service': kw.source_service,
'campaign_id': kw.external_campaign_id,
'ad_group_id': kw.external_ad_group_id,
'status': kw.status,
'max_cpc': kw.max_cpc,
'impressions': kw.total_impressions,
'clicks': kw.total_clicks,
'spend': round(kw.total_spend, 2),
'conversions': kw.total_conversions,
'ctr': round(kw.avg_ctr, 2) if kw.avg_ctr else None,
'avg_position': round(kw.avg_position, 1) if kw.avg_position else None,
'cpa': round(kw.total_spend / kw.total_conversions, 2) if kw.total_conversions > 0 else None,
'search_volume': kw.search_volume,
'competition_level': kw.competition_level,
'cpc_suggestion': kw.cpc_suggestion,
'trend_data': kw.trend_data,
'created_at': kw.created_at.isoformat() if kw.created_at else None,
'updated_at': kw.updated_at.isoformat() if kw.updated_at else None,
'metadata': kw.metadata_json,
}
for kw in keywords
],
"stats": stats,
"total_returned": len(keywords),
}
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/keywords/<keyword_id>', methods=['GET'])
@require_auth_json()
def get_keyword(company_id, keyword_id):
"""Get a single keyword by ID with full details."""
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
kw = db.session.get(AdKeyword, keyword_id)
if not kw or kw.company_id != company_id:
return jsonify({'error': 'Keyword not found'}), 404
return jsonify({
"success": True,
"data": {
'id': kw.id,
'text': kw.text,
'match_type': kw.match_type,
'source_service': kw.source_service,
'campaign_id': kw.external_campaign_id,
'ad_group_id': kw.external_ad_group_id,
'status': kw.status,
'max_cpc': kw.max_cpc,
'target_cpa': kw.target_cpa,
'impressions': kw.total_impressions,
'clicks': kw.total_clicks,
'spend': round(kw.total_spend, 2),
'conversions': kw.total_conversions,
'ctr': round(kw.avg_ctr, 2) if kw.avg_ctr else None,
'avg_position': round(kw.avg_position, 1) if kw.avg_position else None,
'cpa': round(kw.total_spend / kw.total_conversions, 2) if kw.total_conversions > 0 else None,
'search_volume': kw.search_volume,
'competition_level': kw.competition_level,
'cpc_suggestion': kw.cpc_suggestion,
'trend_data': kw.trend_data,
'metadata': kw.metadata_json,
'created_at': kw.created_at.isoformat() if kw.created_at else None,
'updated_at': kw.updated_at.isoformat() if kw.updated_at else None,
}
})
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/keywords/<keyword_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def update_keyword(company_id, keyword_id):
"""Update keyword bid, status, or target CPA.
Supports pausing/enabling keywords and adjusting bids.
Changes are applied locally AND pushed to Google Ads if connected.
"""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
kw = db.session.get(AdKeyword, keyword_id)
if not kw or kw.company_id != company_id:
return jsonify({'error': 'Keyword not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
updatable = ['status', 'max_cpc', 'target_cpa']
changes = {}
for field in updatable:
if field in data:
old_val = getattr(kw, field)
new_val = data[field]
if old_val != new_val:
changes[field] = {'old': old_val, 'new': new_val}
setattr(kw, field, new_val)
kw.updated_at = datetime.now(timezone.utc)
# Push changes to Google Ads if connected
connector_error = None
if kw.source_service == 'google_ads' and changes:
try:
from app.connectors import build_connector
from ..models import Connector
connector_record = Connector.query.filter_by(
company_id=company_id,
service='google_ads',
status='connected',
).first()
if not connector_record:
logger.warning("Google Ads connector not connected for company %s", company_id)
connector_error = "Google Ads not connected"
else:
connector = build_connector(
service='google_ads',
company_id=company_id,
config=connector_record.config_json,
connector_id=connector_record.id,
)
if connector and connector.config.get('access_token'):
criterion_id = (kw.metadata_json or {}).get('criterion_id')
ad_group_id = kw.external_ad_group_id
campaign_id = kw.external_campaign_id
if criterion_id:
# Build resource name
customer_id = connector.config.get('customer_id', '')
resource_name = f"customers/{customer_id}/adGroupCriteria/{criterion_id}"
# Handle status change
if 'status' in changes:
ga_status_map = {
'enabled': 'ENABLED',
'paused': 'PAUSED',
'removed': 'REMOVED',
}
ga_status = ga_status_map.get(changes['status']['new'], 'ENABLED')
operation = [{
'update': {
'resourceName': resource_name,
'adGroupCriterion': {
'status': ga_status,
},
'updateMask': 'status',
}
}]
connector._check_rate_limit()
connector._retry(
connector._mutate_resource,
resource="adGroupCriteria",
operations=operation,
customer_id=customer_id,
access_token=connector.config['access_token'],
developer_token=connector.config.get('developer_token', ''),
)
# Handle bid change
if 'max_cpc' in changes:
new_bid_micros = str(int(changes['max_cpc']['new'] * 1_000_000))
operation = [{
'update': {
'resourceName': resource_name,
'adGroupCriterion': {
'cpcBid': int(new_bid_micros),
},
'updateMask': 'cpcBid',
}
}]
connector._check_rate_limit()
connector._retry(
connector._mutate_resource,
resource="adGroupCriteria",
operations=operation,
customer_id=customer_id,
access_token=connector.config['access_token'],
developer_token=connector.config.get('developer_token', ''),
)
else:
logger.warning(
"Cannot push keyword change to Google Ads: "
"missing criterion_id for keyword %s", keyword_id
)
except Exception as e:
logger.warning("Failed to push keyword change to Google Ads: %s", e)
connector_error = str(e)
db.session.commit()
result = {
"success": True,
"data": {
'id': kw.id,
'text': kw.text,
'status': kw.status,
'max_cpc': kw.max_cpc,
'changes': changes,
},
}
if connector_error:
result['warnings'] = [f"Local update succeeded but Google Ads push failed: {connector_error}"]
return jsonify(result)
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/keywords/<keyword_id>/sync', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def sync_keyword(company_id, keyword_id):
"""Force-sync a single keyword's metrics from Google Ads."""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
kw = db.session.get(AdKeyword, keyword_id)
if not kw or kw.company_id != company_id:
return jsonify({'error': 'Keyword not found'}), 404
if kw.source_service != 'google_ads':
return jsonify({'error': 'Only Google Ads keywords can be synced'}), 400
try:
from app.connectors import build_connector
from ..models import Connector
connector_record = Connector.query.filter_by(
company_id=company_id,
service='google_ads',
status='connected',
).first()
if not connector_record:
return jsonify({'error': 'Google Ads not connected'}), 400
connector = build_connector(
service='google_ads',
company_id=company_id,
config=connector_record.config_json,
connector_id=connector_record.id,
)
if not connector or not connector.config.get('access_token'):
return jsonify({'error': 'Google Ads not connected'}), 400
criterion_id = (kw.metadata_json or {}).get('criterion_id')
if not criterion_id:
return jsonify({'error': 'Missing criterion_id — run a full sync first'}), 400
customer_id = connector.config['customer_id']
criterion_resource = f"customers/{customer_id}/adGroupCriteria/{criterion_id}"
# Fetch current keyword metrics
query = (
f"SELECT "
f"ad_group_criterion.keyword.text, "
f"ad_group_criterion.keyword.match_type, "
f"ad_group_criterion.status, "
f"ad_group_criterion.cpc_bid_micros, "
f"metrics.impressions, "
f"metrics.clicks, "
f"metrics.cost_micros, "
f"metrics.conversions, "
f"metrics.average_position, "
f"metrics.ctr "
f"FROM ad_group_criterion "
f"WHERE ad_group_criterion.resource_name = '{criterion_resource}'"
)
response = connector._retry(
connector._gaql_query,
query=query,
customer_id=customer_id,
access_token=connector.config['access_token'],
developer_token=connector.config.get('developer_token', ''),
)
results = response.get('results', [])
if not results:
return jsonify({'error': 'No data returned from Google Ads'}), 404
fields = results[0].get('fields', {})
metrics = fields.get('metrics', {})
criterion = fields.get('ad_group_criterion', {})
keyword_info = criterion.get('keyword', {})
cost_micros = metrics.get('cost_micros', 0)
spend = float(cost_micros) / 1_000_000 if cost_micros else 0.0
cpc_micros = criterion.get('cpc_bid_micros')
max_cpc = float(cpc_micros) / 1_000_000 if cpc_micros else None
ctr_val = float(metrics.get('ctr', 0) or 0)
ctr_val = ctr_val * 100 if ctr_val > 0 else None
avg_pos = float(metrics.get('average_position', 0) or 0)
avg_pos = avg_pos if avg_pos > 0 else None
status_map = {
'ENABLED': 'enabled',
'PAUSED': 'paused',
'REMOVED': 'removed',
'ELIGIBLE': 'enabled',
'NOT_ELIGIBLE': 'paused',
}
kw.total_impressions = int(metrics.get('impressions', 0) or 0)
kw.total_clicks = int(metrics.get('clicks', 0) or 0)
kw.total_spend = spend
kw.total_conversions = float(metrics.get('conversions', 0) or 0)
kw.avg_ctr = ctr_val
kw.avg_position = avg_pos
kw.max_cpc = max_cpc
kw.status = status_map.get(criterion.get('status', ''), 'enabled')
kw.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
"success": True,
"data": {
'id': kw.id,
'text': kw.text,
'spend': round(kw.total_spend, 2),
'clicks': kw.total_clicks,
'conversions': kw.total_conversions,
'ctr': round(kw.avg_ctr, 2) if kw.avg_ctr else None,
'synced_at': datetime.now(timezone.utc).isoformat(),
}
})
except Exception as e:
logger.error("Keyword sync failed for %s: %s", keyword_id, e, exc_info=True)
return jsonify({'error': f'Sync failed: {str(e)}'}), 500
# -- Keyword research sync -------------------------------------------------
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/keywords/research-sync', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def sync_keyword_research(company_id):
"""Pull keyword research data (search volume, competition, CPC) via KeywordPlanIdeaService.
Updates all tracked keywords for the company with research data from
Google Ads Keyword Planner. This is a separate sync from performance metrics.
"""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
try:
from app.connectors import build_connector
from ..models import Connector
connector_record = Connector.query.filter_by(
company_id=company_id,
service='google_ads',
status='connected',
).first()
if not connector_record:
return jsonify({'error': 'Google Ads not connected'}), 400
connector = build_connector(
service='google_ads',
company_id=company_id,
config=connector_record.config_json,
connector_id=connector_record.id,
)
if not connector or not connector.config.get('access_token'):
return jsonify({'error': 'Google Ads not connected'}), 400
updated = connector._sync_keyword_research()
return jsonify({
'success': True,
'data': {
'keywords_updated': updated,
'synced_at': datetime.now(timezone.utc).isoformat(),
}
})
except Exception as e:
logger.error("Keyword research sync failed for %s: %s", company_id, e, exc_info=True)
return jsonify({'error': f'Research sync failed: {str(e)}'}), 500
# -- Bulk keyword research lookup -------------------------------------------
@lead_gen_bp.route('/api/company/<company_id>/lead-gen/keyword-research', methods=['POST'])
@require_auth_json()
@require_csrf
@require_tier(min_tier='growth')
def lookup_keyword_research(company_id):
"""Look up research data for a list of keyword texts (not yet tracked).
Accepts a JSON body with a `keywords` array of strings.
Returns research data without persisting to AdKeyword table.
Useful for pre-campaign keyword planning.
"""
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
data = request.get_json()
if not data or 'keywords' not in data:
return jsonify({'error': 'Missing keywords array in request body'}), 400
keywords = data['keywords']
if not isinstance(keywords, list) or not keywords:
return jsonify({'error': 'keywords must be a non-empty array'}), 400
if len(keywords) > 100:
return jsonify({'error': 'Maximum 100 keywords per request'}), 400
try:
from app.connectors import build_connector
from ..models import Connector
connector_record = Connector.query.filter_by(
company_id=company_id,
service='google_ads',
status='connected',
).first()
if not connector_record:
return jsonify({'error': 'Google Ads not connected'}), 400
connector = build_connector(
service='google_ads',
company_id=company_id,
config=connector_record.config_json,
connector_id=connector_record.id,
)
if not connector or not connector.config.get('access_token'):
return jsonify({'error': 'Google Ads not connected'}), 400
ideas = connector._get_keyword_ideas(
keywords=keywords,
customer_id=connector.config['customer_id'],
access_token=connector.config['access_token'],
developer_token=connector.config.get('developer_token', ''),
)
return jsonify({
'success': True,
'data': {
'keyword_ideas': ideas,
'count': len(ideas),
'queried_at': datetime.now(timezone.utc).isoformat(),
}
})
except Exception as e:
logger.error("Keyword research lookup failed for %s: %s", company_id, e, exc_info=True)
return jsonify({'error': f'Keyword research lookup failed: {str(e)}'}), 500