"""API proxy routes for SPA frontend compatibility.
The SPA expects /api/auth/* endpoints with JSON bodies.
Flask auth routes use /auth/* with form data.
This bridges the gap.
"""
from flask import Blueprint, request, jsonify, make_response, redirect, session as flask_session
from flask_login import login_user, logout_user, current_user
from werkzeug.security import check_password_hash
from datetime import datetime, timezone, timedelta
import logging
import re
import secrets
import uuid
from sqlalchemy import text
from app import db, limiter
from app.models import QuickbooksTransaction
from app.utils.pagination import paginate_query
from app.utils.csrf import require_csrf
api_proxy_bp = Blueprint('api_proxy', __name__)
logger = logging.getLogger(__name__)
def require_auth_json():
"""Decorator that returns JSON 401 instead of 302 redirect for API routes."""
def decorator(f):
from functools import wraps
@wraps(f)
def decorated(*args, **kwargs):
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
return f(*args, **kwargs)
return decorated
return decorator
def require_super_admin():
"""Decorator that returns JSON 403 if current user is not a super_admin."""
def decorator(f):
from functools import wraps
@wraps(f)
@require_auth_json()
def decorated(*args, **kwargs):
if current_user.role != 'super_admin':
return jsonify({'error': 'Forbidden — super_admin access required'}), 403
return f(*args, **kwargs)
return decorated
return decorator
# ─── Subscription tier gating ───────────────────────────────────────────────
# Tiers in ascending order. Each tier includes all features of the tiers below.
_TIER_LEVELS = {
'launch': 0,
'growth': 1,
'command': 2,
'enterprise': 3,
}
def require_tier(min_tier: str = 'growth'):
"""Decorator factory: gate a route to users whose company tier is >= min_tier.
Usage::
@api_proxy_bp.route('/api/analytics/strategic-intelligence/...')
@require_auth_json()
@require_tier(min_tier='growth')
def my_route():
...
Tiers (ascending): launch → growth → command → enterprise.
Returns 403 JSON with an upgrade prompt when the company tier is too low.
"""
def decorator(f):
from functools import wraps
@wraps(f)
def decorated(*args, **kwargs):
from app.models import UserCompany
# Determine the user's company tier
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
company_tier = 'launch'
if uc and uc.company:
company = uc.company
# Prefer settings_json['tier'] if present, otherwise Company.tier
settings = company.settings_json or {}
company_tier = settings.get('tier') or getattr(company, 'tier', 'launch') or 'launch'
required_level = _TIER_LEVELS.get(min_tier, 1)
actual_level = _TIER_LEVELS.get(company_tier, 0)
if actual_level < required_level:
return jsonify({
'error': (
f'This feature requires the {min_tier.capitalize()} plan or higher. '
f'Your current plan is {company_tier.capitalize()}. '
'Upgrade at /app/billing.'
),
'required_tier': min_tier,
'current_tier': company_tier,
'upgrade_url': '/app/billing',
}), 403
return f(*args, **kwargs)
return decorated
return decorator
# ─── Company role hierarchy ─────────────────────────────────────────────
# Higher number = more permissions
COMPANY_ROLE_LEVELS = {
'owner': 5,
'admin': 4,
'editor': 3, # SECURITY FIX: 'editor' role added between admin and member
'manager': 3, # alias — same level as editor
'rep': 2,
'member': 2, # alias for backward compat — read-only effective role
'viewer': 1,
}
# Permission → minimum role level required
COMPANY_PERMISSIONS = {
'create_org': COMPANY_ROLE_LEVELS['owner'],
'manage_team': COMPANY_ROLE_LEVELS['manager'],
'view_analytics': COMPANY_ROLE_LEVELS['manager'],
'edit_integrations': COMPANY_ROLE_LEVELS['admin'],
'view_reports': COMPANY_ROLE_LEVELS['viewer'],
'edit_settings': COMPANY_ROLE_LEVELS['admin'],
}
def _check_company_access(company_id, min_role=None, min_permission=None, required_roles=None):
"""Verify the current user has access to the given company and (optionally)
holds a sufficient company-level role.
Parameters
----------
company_id : company UUID
min_role : minimum role string (e.g. 'manager') — checked via numeric level
min_permission : permission key (e.g. 'edit_integrations') — looked up in COMPANY_PERMISSIONS
required_roles : explicit set of roles — legacy, checked via ``in``
Returns
-------
(error_response, company_membership) tuple.
If check passes, error_response is None.
"""
from app.models import db, Company, UserCompany
company = db.session.get(Company, company_id)
if not company:
return (jsonify({'error': 'Company not found'}), 404), None
membership = UserCompany.query.filter_by(
user_id=current_user.id,
company_id=company_id
).first()
if not membership:
return (jsonify({'error': 'Forbidden — no access to this company'}), 403), None
# Explicit role whitelist (legacy)
if required_roles and membership.role not in required_roles:
return (jsonify({'error': 'Forbidden — insufficient company role'}), 403), None
# Minimum role level check
if min_role:
user_level = COMPANY_ROLE_LEVELS.get(membership.role, 0)
required_level = COMPANY_ROLE_LEVELS.get(min_role, 99)
if user_level < required_level:
return (jsonify({'error': f'Forbidden — requires at least {min_role} role'}), 403), None
# Permission-based check
if min_permission:
required_level = COMPANY_PERMISSIONS.get(min_permission, 99)
user_level = COMPANY_ROLE_LEVELS.get(membership.role, 0)
if user_level < required_level:
return (jsonify({'error': f'Forbidden — requires "{min_permission}" permission'}), 403), None
return None, membership
# ============================================================================
# NOTE: Route sections have been split into focused blueprints (July 2026):
# - app/routes/auth_api.py → /api/auth/*, /api/user/*
# - app/routes/security_api.py → /api/security/*
# - app/routes/admin_api.py → /api/super-admin/*, /api/partner/*, /api/plans
# - app/routes/business_api.py → /api/company/*/revenue-leaks, forecasts, etc.
#
# This file (api_proxy.py) now serves as:
# 1. Shared utility module (require_auth_json, require_csrf, require_tier, etc.)
# 2. Connector API routes (angi, quickbooks, google_ads, facebook_ads, slack)
# ============================================================================
# =============================================================================
# Angi Lead Management
# =============================================================================
@api_proxy_bp.route('/api/company/<company_id>/angi/leads', methods=['GET'])
@require_auth_json()
def list_angi_leads(company_id):
"""List leads with filters: status, date_range, priority, project_type."""
from app.models import AngiLead
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
status = request.args.get('status', 'all')
date_range = request.args.get('date_range', '30d') # 7d, 30d, 90d, all
priority = request.args.get('priority')
project_type = request.args.get('project_type')
search = request.args.get('search', '')
query = AngiLead.query.filter_by(company_id=company_id)
if status != 'all':
query = query.filter_by(status=status)
if priority:
query = query.filter_by(priority=priority)
if project_type:
project_type = project_type[:200] # Limit length
query = query.filter(AngiLead.project_type.ilike(f'%{project_type}%'))
if search:
search = search[:200] # Limit length
query = query.filter(
db.or_(
AngiLead.first_name.ilike(f'%{search}%'),
AngiLead.last_name.ilike(f'%{search}%'),
AngiLead.email.ilike(f'%{search}%'),
AngiLead.phone.ilike(f'%{search}%'),
AngiLead.angi_lead_id.ilike(f'%{search}%'),
)
)
# Parse date range
if date_range != 'all':
days = int(date_range.replace('d', ''))
since = datetime.now(timezone.utc) - timedelta(days=days)
query = query.filter(AngiLead.received_at >= since)
pagination, meta = paginate_query(query.order_by(AngiLead.received_at.desc()))
leads = pagination.items
return jsonify({
"success": True,
"data": {
"leads": [l.to_dict() for l in leads],
"total": meta['total'],
**meta,
}
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/<lead_id>', methods=['GET'])
@require_auth_json()
def get_angi_lead(company_id, lead_id):
"""Get a single lead by ID."""
from app.models import db, AngiLead
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
return jsonify({
"success": True,
"data": lead.to_dict()
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/<lead_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_angi_lead(company_id, lead_id):
"""Update lead status, priority, notes, etc."""
from app.models import db, AngiLead
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
# Update allowed fields
updatable = {'status', 'priority', 'internal_notes', 'budget', 'budget_value',
'project_type', 'description', 'timeline', 'address', 'city',
'state', 'zip_code'}
for field in updatable:
if field in data:
setattr(lead, field, data[field])
db.session.commit()
return jsonify({
"success": True,
"data": lead.to_dict()
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/<lead_id>/respond', methods=['POST'])
@require_auth_json()
@require_csrf
def respond_to_lead_endpoint(company_id, lead_id):
"""Log a response to a lead, update status to 'responded', track response time."""
from app.models import db, AngiLead
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
data = request.get_json()
if not data or 'message' not in data:
return jsonify({'error': 'Message is required'}), 400
message = data['message']
# Calculate response time if not already responded
now = datetime.now(timezone.utc)
if lead.response_time_minutes is None and lead.received_at:
delta = now - lead.received_at
lead.response_time_minutes = int(delta.total_seconds() / 60)
lead.status = 'responded'
lead.responded_at = now
lead.last_contacted_at = now
lead.internal_notes = (lead.internal_notes or '') + f'\n[{now.isoformat()}] Response: {message}'
db.session.commit()
return jsonify({
"success": True,
"data": lead.to_dict()
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/<lead_id>/accept', methods=['POST'])
@require_auth_json()
@require_csrf
def accept_lead_endpoint(company_id, lead_id):
"""Accept a lead, update status to 'accepted'."""
from app.models import db, AngiLead
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
data = request.get_json() or {}
now = datetime.now(timezone.utc)
lead.status = 'accepted'
lead.last_contacted_at = now
if data.get('notes'):
lead.internal_notes = (lead.internal_notes or '') + f'\n[{now.isoformat()}] Accepted: {data["notes"]}'
db.session.commit()
return jsonify({
"success": True,
"data": lead.to_dict()
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/<lead_id>/reject', methods=['POST'])
@require_auth_json()
@require_csrf
def reject_lead_endpoint(company_id, lead_id):
"""Reject a lead with a reason."""
from app.models import db, AngiLead
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
data = request.get_json()
if not data or 'reason' not in data:
return jsonify({'error': 'Reason is required'}), 400
reason = data['reason']
now = datetime.now(timezone.utc)
lead.status = 'rejected'
lead.last_contacted_at = now
lead.internal_notes = (lead.internal_notes or '') + f'\n[{now.isoformat()}] Rejected: {reason}'
db.session.commit()
return jsonify({
"success": True,
"data": lead.to_dict()
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/analytics', methods=['GET'])
@require_auth_json()
def get_angi_lead_analytics(company_id):
"""Return lead analytics for a company: volume, conversion, avg response time."""
from app.models import db, AngiLead
from sqlalchemy import func
error_response, membership = _check_company_access(company_id, min_role='viewer')
if error_response:
return error_response
period_days = int(request.args.get('period_days', 30))
since = datetime.now(timezone.utc) - timedelta(days=period_days)
base_query = AngiLead.query.filter(
AngiLead.company_id == company_id,
AngiLead.received_at >= since
)
total_leads = base_query.count()
accepted = base_query.filter_by(status='accepted').count()
responded = base_query.filter_by(status='responded').count()
rejected = base_query.filter_by(status='rejected').count()
new_leads = base_query.filter_by(status='new').count()
# Average response time (only for responded leads)
responded_query = base_query.filter(
AngiLead.response_time_minutes.isnot(None)
)
avg_response_result = responded_query.with_entities(
func.avg(AngiLead.response_time_minutes)
).scalar()
avg_response_minutes = round(avg_response_result, 1) if avg_response_result else None
# Average cost per lead
cost_result = base_query.with_entities(func.avg(AngiLead.cost_per_lead)).scalar()
avg_cost = round(cost_result, 2) if cost_result else None
# Total cost
total_cost_result = base_query.with_entities(func.sum(AngiLead.cost_per_lead)).scalar()
total_cost = round(total_cost_result or 0, 2)
# Conversion rate
conversion_rate = round((accepted / total_leads * 100), 1) if total_leads > 0 else 0
response_rate = round((responded / total_leads * 100), 1) if total_leads > 0 else 0
# Premium vs standard
premium = base_query.filter_by(is_premium=True).count()
standard = total_leads - premium
# Status breakdown
status_breakdown = {
'new': new_leads,
'accepted': accepted,
'responded': responded,
'rejected': rejected,
}
# Leads by day (last N days)
daily = base_query.with_entities(
func.date(AngiLead.received_at).label('day'),
func.count(AngiLead.id).label('count')
).group_by(func.date(AngiLead.received_at)).order_by('day').all()
daily_leads = [{'day': str(d.day), 'count': d.count} for d in daily]
# Priority breakdown
priority_breakdown = {}
for priority in ['high', 'medium', 'low']:
priority_breakdown[priority] = base_query.filter_by(priority=priority).count()
return jsonify({
"success": True,
"data": {
"period_days": period_days,
"total_leads": total_leads,
"conversion_rate": conversion_rate,
"response_rate": response_rate,
"avg_response_time_minutes": avg_response_minutes,
"avg_cost_per_lead": avg_cost,
"total_cost": total_cost,
"premium_leads": premium,
"standard_leads": standard,
"status_breakdown": status_breakdown,
"daily_leads": daily_leads,
"priority_breakdown": priority_breakdown,
}
})
@api_proxy_bp.route('/api/company/<company_id>/angi/leads/sync', methods=['POST'])
@require_auth_json()
@require_csrf
def sync_angi_leads(company_id):
"""Manually trigger a sync of Angi leads."""
from app.models import db, AngiLead
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
# Get the Angi connector for this company
try:
from app.connectors import get_connector
connector = get_connector(company_id)
if not connector or connector.connector_type != 'angi':
return jsonify({'error': 'Angi connector not found for this company'}), 404
result = connector.sync()
if result.get("status") != "success":
return jsonify({
"success": False,
"error": result.get("error", "Angi sync failed"),
"data": result
}), 502
return jsonify({
"success": True,
"data": result
})
except Exception as e:
logger.exception('Sync failed')
return jsonify({'error': 'Sync failed. An error occurred.'}), 500
# =============================================================================
# QuickBooks — Company-scoped routes /api/company/<company_id>/quickbooks/*
# =============================================================================
def _map_invoice(inv, connector_id=None):
"""Map QuickbooksInvoice model fields to frontend-expected shape."""
meta = inv.metadata_json or {}
line_items = inv.line_items_json or []
subtotal = (inv.total_amount or 0) - (inv.tax_amount or 0)
return {
'id': inv.id,
'company_id': inv.company_id,
'qb_doc_id': inv.qb_doc_id,
'connector_id': connector_id or '',
'invoice_number': inv.invoice_num,
'customer_name': inv.customer_name,
'customer_email': meta.get('customer_email'),
'subtotal': round(subtotal, 2),
'tax_amount': inv.tax_amount or 0,
'total_amount': inv.total_amount or 0,
'due_date': inv.due_date.isoformat() if inv.due_date else None,
'issue_date': inv.tx_date.isoformat() if inv.tx_date else None,
'status': inv.status or 'draft',
'line_items': line_items if isinstance(line_items, list) else [],
'notes': meta.get('notes', meta.get('Memo', '')) or '',
'synced_at': inv.updated_at.isoformat() if inv.updated_at else None,
'created_at': inv.created_at.isoformat() if inv.created_at else None,
}
def _map_transaction(tx, connector_id=None):
"""Map QuickbooksTransaction model fields to frontend-expected shape."""
return {
'id': tx.id,
'company_id': tx.company_id,
'qb_doc_id': tx.qb_doc_id,
'connector_id': connector_id or '',
'transaction_date': tx.tx_date.isoformat() if tx.tx_date else None,
'description': tx.memo or '',
'amount': tx.amount or 0,
'currency': 'USD',
'transaction_type': tx.tx_type or '',
'account_name': tx.account_name,
'category': tx.class_name or '',
'synced_at': tx.updated_at.isoformat() if tx.updated_at else None,
'created_at': tx.created_at.isoformat() if tx.created_at else None,
}
def _map_customer(cust, connector_id=None):
"""Map QuickbooksCustomer model fields to frontend-expected shape."""
from app.models import QuickbooksInvoice, db
from sqlalchemy import func
addr = cust.billing_address_json or {}
invoice_count = db.session.query(func.count(QuickbooksInvoice.id)).filter(
QuickbooksInvoice.company_id == cust.company_id,
QuickbooksInvoice.customer_name == cust.display_name,
).scalar() or 0
paid_total = db.session.query(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).filter(
QuickbooksInvoice.company_id == cust.company_id,
QuickbooksInvoice.customer_name == cust.display_name,
QuickbooksInvoice.status.ilike('%paid%'),
).scalar() or 0.0
return {
'id': cust.id,
'company_id': cust.company_id,
'qb_doc_id': cust.qb_doc_id,
'connector_id': connector_id or '',
'name': cust.display_name,
'email': cust.email or None,
'phone': cust.phone or None,
'address_line1': addr.get('Line1', addr.get('line1', '')) or None,
'address_city': addr.get('City', addr.get('city', '')) or None,
'address_state': addr.get('CountrySubDivisionCode', addr.get('state', '')) or None,
'address_zip': addr.get('PostalCode', addr.get('postal_code', '')) or None,
'balance': cust.balance or 0,
'total_invoices': invoice_count,
'total_paid': round(paid_total, 2),
'synced_at': cust.updated_at.isoformat() if cust.updated_at else None,
'created_at': cust.created_at.isoformat() if cust.created_at else None,
}
def _map_expense(exp, connector_id=None):
"""Map QuickbooksExpense model fields to frontend-expected shape."""
return {
'id': exp.id,
'company_id': exp.company_id,
'qb_doc_id': exp.qb_doc_id,
'connector_id': connector_id or '',
'expense_date': exp.tx_date.isoformat() if exp.tx_date else None,
'vendor_name': exp.vendor_name or '',
'description': exp.memo or '',
'amount': exp.amount or 0,
'category': exp.category or '',
'account_name': exp.account_name or None,
'synced_at': exp.created_at.isoformat() if exp.created_at else None,
'created_at': exp.created_at.isoformat() if exp.created_at else None,
}
# ── Invoices ───────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/invoices', methods=['GET'])
@require_auth_json()
def list_quickbooks_invoices(company_id):
"""List invoices with filters and pagination."""
from app.models import QuickbooksInvoice
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
status = request.args.get('status', '')
search = request.args.get('search', '')
customer = request.args.get('customer', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = QuickbooksInvoice.query.filter_by(company_id=company_id)
if status:
query = query.filter(QuickbooksInvoice.status.ilike(f'%{status}%'))
if customer:
query = query.filter(QuickbooksInvoice.customer_name.ilike(f'%{customer}%'))
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksInvoice.tx_date >= since)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksInvoice.tx_date <= until)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
if search:
query = query.filter(
db.or_(
QuickbooksInvoice.invoice_num.ilike(f'%{search}%'),
QuickbooksInvoice.customer_name.ilike(f'%{search}%'),
QuickbooksInvoice.qb_doc_id.ilike(f'%{search}%'),
)
)
total = query.count()
invoices = query.order_by(
QuickbooksInvoice.tx_date.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_invoice(inv) for inv in invoices],
"total": total,
"page": page,
"per_page": per_page,
}
})
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/invoices/<invoice_id>', methods=['GET'])
@require_auth_json()
def get_quickbooks_invoice(company_id, invoice_id):
"""Get a single invoice by ID."""
from app.models import QuickbooksInvoice
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
invoice = db.session.get(QuickbooksInvoice, invoice_id)
if not invoice or invoice.company_id != company_id:
return jsonify({'error': 'Invoice not found'}), 404
return jsonify({
"success": True,
"data": _map_invoice(invoice)
})
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/accept-invoice/<invoice_id>', methods=['POST'])
@require_auth_json()
@require_csrf
def accept_quickbooks_invoice(company_id, invoice_id):
"""Accept/mark an invoice as paid."""
from app.models import QuickbooksInvoice
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
invoice = db.session.get(QuickbooksInvoice, invoice_id)
if not invoice or invoice.company_id != company_id:
return jsonify({'error': 'Invoice not found'}), 404
invoice.status = 'paid'
invoice.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
"success": True,
"data": _map_invoice(invoice)
})
# ── Transactions ───────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/transactions', methods=['GET'])
@require_auth_json()
def list_quickbooks_transactions(company_id):
"""List transactions with filters and pagination."""
from app.models import QuickbooksTransaction
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
tx_type = request.args.get('type', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = QuickbooksTransaction.query.filter_by(company_id=company_id)
if tx_type:
query = query.filter(QuickbooksTransaction.tx_type.ilike(f'%{tx_type}%'))
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksTransaction.tx_date >= since)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksTransaction.tx_date <= until)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
total = query.count()
transactions = query.order_by(
QuickbooksTransaction.tx_date.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_transaction(tx) for tx in transactions],
"total": total,
"page": page,
"per_page": per_page,
}
})
# ── Customers ──────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/customers', methods=['GET'])
@require_auth_json()
def list_quickbooks_customers(company_id):
"""List customers with search and pagination."""
from app.models import QuickbooksCustomer
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
search = request.args.get('search', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = QuickbooksCustomer.query.filter_by(company_id=company_id)
if search:
query = query.filter(
db.or_(
QuickbooksCustomer.display_name.ilike(f'%{search}%'),
QuickbooksCustomer.email.ilike(f'%{search}%'),
QuickbooksCustomer.phone.ilike(f'%{search}%'),
QuickbooksCustomer.qb_doc_id.ilike(f'%{search}%'),
)
)
total = query.count()
customers = query.order_by(
QuickbooksCustomer.display_name.asc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_customer(c) for c in customers],
"total": total,
"page": page,
"per_page": per_page,
}
})
# ── Expenses ───────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/expenses', methods=['GET'])
@require_auth_json()
def list_quickbooks_expenses(company_id):
"""List expenses with filters and pagination."""
from app.models import QuickbooksExpense
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
category = request.args.get('category', '')
vendor = request.args.get('vendor', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = QuickbooksExpense.query.filter_by(company_id=company_id)
if category:
query = query.filter(QuickbooksExpense.category.ilike(f'%{category}%'))
if vendor:
query = query.filter(QuickbooksExpense.vendor_name.ilike(f'%{vendor}%'))
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksExpense.tx_date >= since)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksExpense.tx_date <= until)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
total = query.count()
expenses = query.order_by(
QuickbooksExpense.tx_date.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_expense(exp) for exp in expenses],
"total": total,
"page": page,
"per_page": per_page,
}
})
# ── Analytics ──────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/analytics', methods=['GET'])
@require_auth_json()
def get_quickbooks_analytics(company_id):
"""Get QuickBooks analytics: revenue, expenses, margins, top customers."""
from app.models import QuickbooksInvoice, QuickbooksExpense, QuickbooksCustomer
from sqlalchemy import func
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
now = datetime.now(timezone.utc)
d30 = now - timedelta(days=30)
d90 = now - timedelta(days=90)
ytd = now.replace(month=1, day=1)
# ── Invoice counts & totals ─────────────────────────────────────
inv_q = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id
)
total_invoices = inv_q.count()
total_outstanding = inv_q.filter(
~QuickbooksInvoice.status.ilike('%paid%'),
QuickbooksInvoice.status.notlike('%void%'),
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
total_paid = inv_q.filter(
QuickbooksInvoice.status.ilike('%paid%')
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
total_partial = inv_q.filter(
QuickbooksInvoice.status.ilike('%partial%')
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
total_void = inv_q.filter(
QuickbooksInvoice.status.ilike('%void%')
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
# ── Revenue by period ───────────────────────────────────────────
revenue_last_30d = inv_q.filter(
QuickbooksInvoice.tx_date >= d30,
QuickbooksInvoice.status.ilike('%paid%')
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
revenue_last_90d = inv_q.filter(
QuickbooksInvoice.tx_date >= d90,
QuickbooksInvoice.status.ilike('%paid%')
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
revenue_ytd = inv_q.filter(
QuickbooksInvoice.tx_date >= ytd,
QuickbooksInvoice.status.ilike('%paid%')
).with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
# ── Expenses by period ──────────────────────────────────────────
exp_q = QuickbooksExpense.query.filter(
QuickbooksExpense.company_id == company_id
)
expenses_last_30d = exp_q.filter(
QuickbooksExpense.tx_date >= d30
).with_entities(
func.coalesce(func.sum(QuickbooksExpense.amount), 0)
).scalar() or 0.0
expenses_last_90d = exp_q.filter(
QuickbooksExpense.tx_date >= d90
).with_entities(
func.coalesce(func.sum(QuickbooksExpense.amount), 0)
).scalar() or 0.0
# ── Profit margin (based on 30-day data) ────────────────────────
if revenue_last_30d > 0:
profit_margin = round((revenue_last_30d - expenses_last_30d) / revenue_last_30d * 100, 2)
else:
profit_margin = 0.0
# ── Average invoice amount ──────────────────────────────────────
avg_invoice_amount = inv_q.with_entities(
func.coalesce(func.avg(QuickbooksInvoice.total_amount), 0)
).scalar() or 0.0
# ── Average collection days (paid invoices) ─────────────────────
paid_invoices = inv_q.filter(
QuickbooksInvoice.status.ilike('%paid%')
).all()
total_days = 0
count_days = 0
for inv in paid_invoices:
meta = inv.metadata_json or {}
paid_date = meta.get('paid_date')
if paid_date and inv.tx_date:
try:
if isinstance(paid_date, str):
pd = datetime.fromisoformat(paid_date)
if pd.tzinfo is None:
pd = pd.replace(tzinfo=timezone.utc)
else:
pd = paid_date
issue = inv.tx_date
if issue.tzinfo is None:
issue = issue.replace(tzinfo=timezone.utc)
delta = (pd - issue).days
if delta >= 0:
total_days += delta
count_days += 1
except (ValueError, TypeError):
pass
avg_collection_days = round(total_days / count_days, 1) if count_days > 0 else 0.0
# ── Invoices by status ──────────────────────────────────────────
status_groups = inv_q.with_entities(
QuickbooksInvoice.status,
func.count(QuickbooksInvoice.id).label('count'),
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0).label('total'),
).group_by(QuickbooksInvoice.status).all()
invoices_by_status = [
{'status': s.status or 'unknown', 'count': s.count, 'total': round(s.total, 2)}
for s in status_groups
]
# ── Top customers ───────────────────────────────────────────────
top_cust_raw = inv_q.with_entities(
QuickbooksInvoice.customer_name,
func.count(QuickbooksInvoice.id).label('invoice_count'),
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0).label('total_paid'),
).group_by(QuickbooksInvoice.customer_name).order_by(
text('total_paid desc')
).limit(10).all()
top_customers = [
{'name': c.customer_name, 'total_paid': round(c.total_paid, 2), 'invoice_count': c.invoice_count}
for c in top_cust_raw
]
# ── Transaction & customer counts ───────────────────────────────
total_transactions = QuickbooksTransaction.query.filter_by(
company_id=company_id
).count()
total_customers = QuickbooksCustomer.query.filter_by(
company_id=company_id
).count()
total_expenses = QuickbooksExpense.query.filter_by(
company_id=company_id
).count()
# ── Monthly revenue ─────────────────────────────────────────────
monthly_raw = inv_q.filter(
QuickbooksInvoice.status.ilike('%paid%')
).with_entities(
func.strftime('%Y-%m', QuickbooksInvoice.tx_date).label('month'),
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0).label('revenue'),
).group_by(func.strftime('%Y-%m', QuickbooksInvoice.tx_date)).order_by(
'month'
).all()
monthly_revenue = [{'month': m.month, 'revenue': round(m.revenue, 2)} for m in monthly_raw]
# ── Expense breakdown ───────────────────────────────────────────
exp_cat_raw = exp_q.with_entities(
QuickbooksExpense.category,
func.coalesce(func.sum(QuickbooksExpense.amount), 0).label('amount'),
func.count(QuickbooksExpense.id).label('count'),
).group_by(QuickbooksExpense.category).order_by(text('amount desc')).all()
expense_breakdown = [
{'category': e.category or 'uncategorized', 'amount': round(e.amount, 2), 'count': e.count}
for e in exp_cat_raw
]
return jsonify({
"success": True,
"data": {
"total_invoices": total_invoices,
"total_outstanding": round(total_outstanding, 2),
"total_paid": round(total_paid, 2),
"total_partial": round(total_partial, 2),
"total_void": round(total_void, 2),
"total_transactions": total_transactions,
"total_customers": total_customers,
"total_expenses": total_expenses,
"revenue_last_30d": round(revenue_last_30d, 2),
"revenue_last_90d": round(revenue_last_90d, 2),
"revenue_ytd": round(revenue_ytd, 2),
"expenses_last_30d": round(expenses_last_30d, 2),
"expenses_last_90d": round(expenses_last_90d, 2),
"profit_margin": profit_margin,
"avg_invoice_amount": round(avg_invoice_amount, 2),
"avg_collection_days": avg_collection_days,
"invoices_by_status": invoices_by_status,
"top_customers": top_customers,
"monthly_revenue": monthly_revenue,
"expense_breakdown": expense_breakdown,
}
})
# ── Sync ───────────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/quickbooks/sync', methods=['POST'])
@require_auth_json()
@require_csrf
def sync_quickbooks(company_id):
"""Manually trigger a QuickBooks sync."""
from app.models import Connector
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
connector = Connector.query.filter_by(
company_id=company_id,
service='quickbooks',
).filter(Connector.status.in_(['connected', 'error'])).first()
if not connector:
return jsonify({"success": False, "error": "No QuickBooks connector found"}), 404
try:
from app.connectors import build_connector
conn = build_connector(
service='quickbooks',
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
result = conn.sync()
# Update connector status
connector.last_sync_at = datetime.now(timezone.utc)
if result.get("status") == "success":
connector.status = "connected"
else:
connector.status = "error"
connector.error_message = result.get("error", "Sync failed")
db.session.commit()
# Return error to frontend if the underlying API failed
if result.get("status") != "success":
return jsonify({
"success": False,
"error": result.get("error", "QuickBooks sync failed — check connector credentials"),
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
}), 502
return jsonify({
"success": True,
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
})
except Exception as e:
logger.exception("Connector sync failed")
connector.status = "error"
connector.error_message = str(e)
db.session.commit()
return jsonify({"success": False, "error": "Sync failed. An error occurred."}), 500
# =============================================================================
# Google Ads — Company-scoped routes /api/company/<company_id>/google_ads/*
# =============================================================================
def _map_campaign(camp):
"""Map AdCampaign model fields to frontend-expected shape."""
from sqlalchemy import func
from app.models import AdMetric
meta = camp.metadata_json or {}
# Aggregate metrics for this campaign
total_spend = AdMetric.query.filter_by(
company_id=camp.company_id,
external_campaign_id=camp.external_id,
source_service='google_ads'
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
total_clicks = AdMetric.query.filter_by(
company_id=camp.company_id,
external_campaign_id=camp.external_id,
source_service='google_ads'
).with_entities(func.coalesce(func.sum(AdMetric.clicks), 0)).scalar() or 0
total_conversions = AdMetric.query.filter_by(
company_id=camp.company_id,
external_campaign_id=camp.external_id,
source_service='google_ads'
).with_entities(func.coalesce(func.sum(AdMetric.conversions), 0)).scalar() or 0.0
return {
'id': camp.id,
'company_id': camp.company_id,
'external_id': camp.external_id,
'name': camp.name or '',
'status': camp.status or 'UNKNOWN',
'budget': round(camp.budget, 2) if camp.budget else None,
'budget_type': camp.budget_type or '',
'channel_type': camp.channel_type or '',
'start_date': camp.start_date.isoformat() if camp.start_date else None,
'end_date': camp.end_date.isoformat() if camp.end_date else None,
'parent_campaign_id': camp.parent_campaign_id,
'total_spend': round(total_spend, 2),
'total_clicks': int(total_clicks),
'total_conversions': round(total_conversions, 2),
'metadata': meta,
'created_at': camp.created_at.isoformat() if camp.created_at else None,
'updated_at': camp.updated_at.isoformat() if camp.updated_at else None,
}
def _map_metric(metric):
"""Map AdMetric model fields to frontend-expected shape."""
return {
'id': metric.id,
'company_id': metric.company_id,
'external_campaign_id': metric.external_campaign_id,
'metric_date': metric.metric_date.isoformat() if metric.metric_date else None,
'spend': round(metric.spend, 2) if metric.spend else 0.0,
'impressions': int(metric.impressions) if metric.impressions else 0,
'clicks': int(metric.clicks) if metric.clicks else 0,
'conversions': round(metric.conversions, 2) if metric.conversions else 0.0,
'ctr': round(metric.ctr, 4) if metric.ctr else None,
'cpc': round(metric.cpc, 4) if metric.cpc else None,
'cpv': round(metric.cpv, 4) if metric.cpv else None,
'roas': round(metric.roas, 4) if metric.roas else None,
'created_at': metric.created_at.isoformat() if metric.created_at else None,
}
# ── Campaigns ──────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/google_ads/campaigns', methods=['GET'])
@require_auth_json()
def list_google_ads_campaigns(company_id):
"""List ad campaigns with filters and pagination."""
from app.models import db, AdCampaign
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
status = request.args.get('status', '')
search = request.args.get('search', '')
channel = request.args.get('channel', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = AdCampaign.query.filter_by(
company_id=company_id,
source_service='google_ads'
)
if status:
query = query.filter(AdCampaign.status.ilike(f'%{status}%'))
if channel:
query = query.filter(AdCampaign.channel_type.ilike(f'%{channel}%'))
if search:
query = query.filter(
db.or_(
AdCampaign.name.ilike(f'%{search}%'),
AdCampaign.external_id.ilike(f'%{search}%'),
)
)
total = query.count()
campaigns = query.order_by(
AdCampaign.updated_at.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_campaign(c) for c in campaigns],
"total": total,
"page": page,
"per_page": per_page,
}
})
@api_proxy_bp.route('/api/company/<company_id>/google_ads/campaigns/<campaign_id>', methods=['GET'])
@require_auth_json()
def get_google_ads_campaign(company_id, campaign_id):
"""Get a single campaign by ID."""
from app.models import db, AdCampaign
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
campaign = db.session.get(AdCampaign, campaign_id)
if not campaign or campaign.company_id != company_id:
return jsonify({'error': 'Campaign not found'}), 404
if campaign.source_service != 'google_ads':
return jsonify({'error': 'Campaign not found'}), 404
return jsonify({
"success": True,
"data": _map_campaign(campaign)
})
# ── Metrics ────────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/google_ads/metrics', methods=['GET'])
@require_auth_json()
def list_google_ads_metrics(company_id):
"""List daily ad metrics with filters and pagination."""
from app.models import AdMetric
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
campaign_id = request.args.get('campaign_id', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = AdMetric.query.filter_by(
company_id=company_id,
source_service='google_ads'
)
if campaign_id:
query = query.filter(AdMetric.external_campaign_id == campaign_id)
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(AdMetric.metric_date >= since.date())
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(AdMetric.metric_date <= until.date())
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
total = query.count()
metrics = query.order_by(
AdMetric.metric_date.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_metric(m) for m in metrics],
"total": total,
"page": page,
"per_page": per_page,
}
})
# ── Analytics ──────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/google_ads/analytics', methods=['GET'])
@require_auth_json()
def get_google_ads_analytics(company_id):
"""Get comprehensive Google Ads analytics."""
from sqlalchemy import func
from app.models import db, AdCampaign, AdMetric
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
now = datetime.now(timezone.utc)
d30 = now - timedelta(days=30)
d90 = now - timedelta(days=90)
ytd = now.replace(month=1, day=1)
# ── Campaign counts ──────────────────────────────────────────────
camp_q = AdCampaign.query.filter_by(
company_id=company_id,
source_service='google_ads'
)
total_campaigns = camp_q.count()
active_campaigns = camp_q.filter(AdCampaign.status == 'ENABLED').count()
paused_campaigns = camp_q.filter(AdCampaign.status == 'PAUSED').count()
# ── Metric aggregates ────────────────────────────────────────────
met_q = AdMetric.query.filter_by(
company_id=company_id,
source_service='google_ads'
)
# Spend by period
spend_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
spend_90d = met_q.filter(
AdMetric.metric_date >= d90.date()
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
spend_ytd = met_q.filter(
AdMetric.metric_date >= ytd.date()
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
# Impressions last 30 days
impressions_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.impressions), 0)).scalar() or 0
# Clicks last 30 days
clicks_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.clicks), 0)).scalar() or 0
# Conversions last 30 days
conversions_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.conversions), 0)).scalar() or 0.0
# Avg CTR last 30 days
avg_ctr = met_q.filter(
AdMetric.metric_date >= d30.date(),
AdMetric.ctr.isnot(None)
).with_entities(func.coalesce(func.avg(AdMetric.ctr), 0)).scalar() or 0.0
# Avg CPC last 30 days
avg_cpc = met_q.filter(
AdMetric.metric_date >= d30.date(),
AdMetric.cpc.isnot(None)
).with_entities(func.coalesce(func.avg(AdMetric.cpc), 0)).scalar() or 0.0
# Avg CPA = total spend / total conversions (last 30d)
spend_30d_val = spend_30d
conversions_30d_val = conversions_30d
avg_cpa = round(spend_30d_val / conversions_30d_val, 2) if conversions_30d_val > 0 else 0.0
# Total ROAS (last 30d) — average of non-null roas values
roas_values = met_q.filter(
AdMetric.metric_date >= d30.date(),
AdMetric.roas.isnot(None)
).with_entities(func.coalesce(func.avg(AdMetric.roas), 0)).scalar() or 0.0
# ── Campaigns by status ──────────────────────────────────────────
status_groups = camp_q.with_entities(
AdCampaign.status,
func.count(AdCampaign.id).label('count'),
func.coalesce(func.sum(AdCampaign.budget), 0).label('total_budget'),
).group_by(AdCampaign.status).all()
campaigns_by_status = [
{
'status': s.status or 'unknown',
'count': s.count,
'total_budget': round(s.total_budget, 2),
}
for s in status_groups
]
# ── Campaigns by channel ─────────────────────────────────────────
channel_groups = camp_q.with_entities(
AdCampaign.channel_type,
func.count(AdCampaign.id).label('count'),
).group_by(AdCampaign.channel_type).all()
# Add total spend per channel
campaigns_by_channel = []
for cg in channel_groups:
channel_spend = met_q.filter(
AdMetric.metric_date >= d30.date()
).join(
AdCampaign,
db.and_(
AdCampaign.external_id == AdMetric.external_campaign_id,
AdCampaign.source_service == 'google_ads',
)
).filter(
AdCampaign.channel_type == (cg.channel_type or 'unknown')
).with_entities(
func.coalesce(func.sum(AdMetric.spend), 0)
).scalar() or 0.0
campaigns_by_channel.append({
'channel_type': cg.channel_type or 'unknown',
'count': cg.count,
'total_spend': round(channel_spend, 2),
})
# ── Top campaigns (by spend last 30d) ────────────────────────────
top_camp_raw = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(
AdMetric.external_campaign_id,
func.coalesce(func.sum(AdMetric.spend), 0).label('spend'),
func.coalesce(func.sum(AdMetric.clicks), 0).label('clicks'),
func.coalesce(func.sum(AdMetric.conversions), 0).label('conversions'),
).group_by(AdMetric.external_campaign_id).order_by(
text('spend desc')
).limit(10).all()
# Get campaign names from AdCampaign
top_campaigns = []
for tcr in top_camp_raw:
camp = AdCampaign.query.filter_by(
company_id=company_id,
source_service='google_ads',
external_id=tcr.external_campaign_id,
).first()
top_campaigns.append({
'name': camp.name if camp else tcr.external_campaign_id or 'Unknown',
'external_id': tcr.external_campaign_id or '',
'spend': round(tcr.spend, 2),
'clicks': int(tcr.clicks),
'conversions': round(tcr.conversions, 2),
'status': camp.status if camp else 'UNKNOWN',
})
# ── Daily spend trend (last 30d) ─────────────────────────────────
daily_raw = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(
AdMetric.metric_date,
func.coalesce(func.sum(AdMetric.spend), 0).label('spend'),
func.coalesce(func.sum(AdMetric.clicks), 0).label('clicks'),
func.coalesce(func.sum(AdMetric.impressions), 0).label('impressions'),
func.coalesce(func.sum(AdMetric.conversions), 0).label('conversions'),
).group_by(AdMetric.metric_date).order_by(
AdMetric.metric_date
).all()
daily_spend_trend = [
{
'date': d.metric_date.isoformat() if d.metric_date else '',
'spend': round(d.spend, 2),
'clicks': int(d.clicks),
'impressions': int(d.impressions),
'conversions': round(d.conversions, 2),
}
for d in daily_raw
]
return jsonify({
"success": True,
"data": {
"total_campaigns": total_campaigns,
"active_campaigns": active_campaigns,
"paused_campaigns": paused_campaigns,
"total_spend_30d": round(spend_30d, 2),
"total_spend_90d": round(spend_90d, 2),
"total_spend_ytd": round(spend_ytd, 2),
"total_impressions_30d": int(impressions_30d),
"total_clicks_30d": int(clicks_30d),
"total_conversions_30d": round(conversions_30d, 2),
"avg_ctr": round(avg_ctr, 4),
"avg_cpc": round(avg_cpc, 4),
"avg_cpa": avg_cpa,
"total_roas": round(roas_values, 4),
"campaigns_by_status": campaigns_by_status,
"campaigns_by_channel": campaigns_by_channel,
"top_campaigns": top_campaigns,
"daily_spend_trend": daily_spend_trend,
}
})
# ── Sync ───────────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/google_ads/sync', methods=['POST'])
@require_auth_json()
@require_csrf
def sync_google_ads(company_id):
"""Manually trigger a Google Ads sync."""
from app.models import db, Connector
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
connector = Connector.query.filter_by(
company_id=company_id,
service='google_ads',
status='connected'
).first()
if not connector:
return jsonify({"success": False, "error": "No Google Ads connector found"}), 404
try:
from app.connectors import build_connector
conn = build_connector(
service='google_ads',
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
result = conn.sync()
# Update connector status
connector.last_sync_at = datetime.now(timezone.utc)
if result.get("status") == "success":
connector.status = "connected"
else:
connector.status = "error"
connector.error_message = result.get("error", "Sync failed")
db.session.commit()
# Return error to frontend if the underlying API failed
if result.get("status") != "success":
return jsonify({
"success": False,
"error": result.get("error", "Sync failed — check connector credentials"),
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
}), 502
return jsonify({
"success": True,
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
})
except Exception as e:
logger.exception("Connector sync failed")
connector.status = "error"
connector.error_message = str(e)
db.session.commit()
return jsonify({"success": False, "error": "Sync failed. An error occurred."}), 500
# =============================================================================
# Facebook Ads — Company-scoped routes /api/company/<company_id>/facebook_ads/*
# =============================================================================
def _map_facebook_campaign(camp):
"""Map AdCampaign model fields to frontend-expected shape for Facebook Ads."""
from sqlalchemy import func
from app.models import AdMetric
meta = camp.metadata_json or {}
# Aggregate metrics for this campaign
total_spend = AdMetric.query.filter_by(
company_id=camp.company_id,
external_campaign_id=camp.external_id,
source_service='facebook_ads'
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
total_clicks = AdMetric.query.filter_by(
company_id=camp.company_id,
external_campaign_id=camp.external_id,
source_service='facebook_ads'
).with_entities(func.coalesce(func.sum(AdMetric.clicks), 0)).scalar() or 0
total_conversions = AdMetric.query.filter_by(
company_id=camp.company_id,
external_campaign_id=camp.external_id,
source_service='facebook_ads'
).with_entities(func.coalesce(func.sum(AdMetric.conversions), 0)).scalar() or 0.0
return {
'id': camp.id,
'company_id': camp.company_id,
'external_id': camp.external_id,
'name': camp.name or '',
'status': camp.status or 'UNKNOWN',
'budget': round(camp.budget, 2) if camp.budget else None,
'budget_type': camp.budget_type or '',
'channel_type': camp.channel_type or '',
'start_date': camp.start_date.isoformat() if camp.start_date else None,
'end_date': camp.end_date.isoformat() if camp.end_date else None,
'parent_campaign_id': camp.parent_campaign_id,
'total_spend': round(total_spend, 2),
'total_clicks': int(total_clicks),
'total_conversions': round(total_conversions, 2),
'metadata': meta,
'created_at': camp.created_at.isoformat() if camp.created_at else None,
'updated_at': camp.updated_at.isoformat() if camp.updated_at else None,
}
# ── Campaigns ──────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/facebook_ads/campaigns', methods=['GET'])
@require_auth_json()
def list_facebook_ads_campaigns(company_id):
"""List ad campaigns with filters and pagination."""
from app.models import db, AdCampaign
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
status = request.args.get('status', '')
search = request.args.get('search', '')
channel = request.args.get('channel', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = AdCampaign.query.filter_by(
company_id=company_id,
source_service='facebook_ads'
)
if status:
query = query.filter(AdCampaign.status.ilike(f'%{status}%'))
if channel:
query = query.filter(AdCampaign.channel_type.ilike(f'%{channel}%'))
if search:
query = query.filter(
db.or_(
AdCampaign.name.ilike(f'%{search}%'),
AdCampaign.external_id.ilike(f'%{search}%'),
)
)
total = query.count()
campaigns = query.order_by(
AdCampaign.updated_at.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_facebook_campaign(c) for c in campaigns],
"total": total,
"page": page,
"per_page": per_page,
}
})
@api_proxy_bp.route('/api/company/<company_id>/facebook_ads/campaigns/<campaign_id>', methods=['GET'])
@require_auth_json()
def get_facebook_ads_campaign(company_id, campaign_id):
"""Get a single campaign by ID."""
from app.models import db, AdCampaign
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
campaign = db.session.get(AdCampaign, campaign_id)
if not campaign or campaign.company_id != company_id:
return jsonify({'error': 'Campaign not found'}), 404
if campaign.source_service != 'facebook_ads':
return jsonify({'error': 'Campaign not found'}), 404
return jsonify({
"success": True,
"data": _map_facebook_campaign(campaign)
})
# ── Metrics ────────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/facebook_ads/metrics', methods=['GET'])
@require_auth_json()
def list_facebook_ads_metrics(company_id):
"""List daily ad metrics with filters and pagination."""
from app.models import AdMetric
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
campaign_id = request.args.get('campaign_id', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 200)
query = AdMetric.query.filter_by(
company_id=company_id,
source_service='facebook_ads'
)
if campaign_id:
query = query.filter(AdMetric.external_campaign_id == campaign_id)
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(AdMetric.metric_date >= since.date())
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(AdMetric.metric_date <= until.date())
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
total = query.count()
metrics = query.order_by(
AdMetric.metric_date.desc()
).offset((page - 1) * per_page).limit(per_page).all()
return jsonify({
"success": True,
"data": {
"items": [_map_metric(m) for m in metrics],
"total": total,
"page": page,
"per_page": per_page,
}
})
# ── Analytics ──────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/facebook_ads/analytics', methods=['GET'])
@require_auth_json()
def get_facebook_ads_analytics(company_id):
"""Get comprehensive Facebook Ads analytics."""
from sqlalchemy import func
from app.models import db, AdCampaign, AdMetric
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
now = datetime.now(timezone.utc)
d30 = now - timedelta(days=30)
d90 = now - timedelta(days=90)
ytd = now.replace(month=1, day=1)
# ── Campaign counts ──────────────────────────────────────────────
camp_q = AdCampaign.query.filter_by(
company_id=company_id,
source_service='facebook_ads'
)
total_campaigns = camp_q.count()
active_campaigns = camp_q.filter(AdCampaign.status == 'ACTIVE').count()
paused_campaigns = camp_q.filter(AdCampaign.status == 'PAUSED').count()
# ── Metric aggregates ────────────────────────────────────────────
met_q = AdMetric.query.filter_by(
company_id=company_id,
source_service='facebook_ads'
)
# Spend by period
spend_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
spend_90d = met_q.filter(
AdMetric.metric_date >= d90.date()
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
spend_ytd = met_q.filter(
AdMetric.metric_date >= ytd.date()
).with_entities(func.coalesce(func.sum(AdMetric.spend), 0)).scalar() or 0.0
# Impressions last 30 days
impressions_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.impressions), 0)).scalar() or 0
# Clicks last 30 days
clicks_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.clicks), 0)).scalar() or 0
# Conversions last 30 days
conversions_30d = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(func.coalesce(func.sum(AdMetric.conversions), 0)).scalar() or 0.0
# Avg CTR last 30 days
avg_ctr = met_q.filter(
AdMetric.metric_date >= d30.date(),
AdMetric.ctr.isnot(None)
).with_entities(func.coalesce(func.avg(AdMetric.ctr), 0)).scalar() or 0.0
# Avg CPC last 30 days
avg_cpc = met_q.filter(
AdMetric.metric_date >= d30.date(),
AdMetric.cpc.isnot(None)
).with_entities(func.coalesce(func.avg(AdMetric.cpc), 0)).scalar() or 0.0
# Avg CPA = total spend / total conversions (last 30d)
spend_30d_val = spend_30d
conversions_30d_val = conversions_30d
avg_cpa = round(spend_30d_val / conversions_30d_val, 2) if conversions_30d_val > 0 else 0.0
# Total ROAS (last 30d) — average of non-null roas values
roas_values = met_q.filter(
AdMetric.metric_date >= d30.date(),
AdMetric.roas.isnot(None)
).with_entities(func.coalesce(func.avg(AdMetric.roas), 0)).scalar() or 0.0
# ── Campaigns by status ──────────────────────────────────────────
status_groups = camp_q.with_entities(
AdCampaign.status,
func.count(AdCampaign.id).label('count'),
func.coalesce(func.sum(AdCampaign.budget), 0).label('total_budget'),
).group_by(AdCampaign.status).all()
campaigns_by_status = [
{
'status': s.status or 'unknown',
'count': s.count,
'total_budget': round(s.total_budget, 2),
}
for s in status_groups
]
# ── Campaigns by channel ─────────────────────────────────────────
channel_groups = camp_q.with_entities(
AdCampaign.channel_type,
func.count(AdCampaign.id).label('count'),
).group_by(AdCampaign.channel_type).all()
# Add total spend per channel
campaigns_by_channel = []
for cg in channel_groups:
channel_spend = met_q.filter(
AdMetric.metric_date >= d30.date()
).join(
AdCampaign,
db.and_(
AdCampaign.external_id == AdMetric.external_campaign_id,
AdCampaign.source_service == 'facebook_ads',
)
).filter(
AdCampaign.channel_type == (cg.channel_type or 'unknown')
).with_entities(
func.coalesce(func.sum(AdMetric.spend), 0)
).scalar() or 0.0
campaigns_by_channel.append({
'channel_type': cg.channel_type or 'unknown',
'count': cg.count,
'total_spend': round(channel_spend, 2),
})
# ── Top campaigns (by spend last 30d) ────────────────────────────
top_camp_raw = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(
AdMetric.external_campaign_id,
func.coalesce(func.sum(AdMetric.spend), 0).label('spend'),
func.coalesce(func.sum(AdMetric.clicks), 0).label('clicks'),
func.coalesce(func.sum(AdMetric.conversions), 0).label('conversions'),
).group_by(AdMetric.external_campaign_id).order_by(
text('spend desc')
).limit(10).all()
# Get campaign names from AdCampaign
top_campaigns = []
for tcr in top_camp_raw:
camp = AdCampaign.query.filter_by(
company_id=company_id,
source_service='facebook_ads',
external_id=tcr.external_campaign_id,
).first()
top_campaigns.append({
'name': camp.name if camp else tcr.external_campaign_id or 'Unknown',
'external_id': tcr.external_campaign_id or '',
'spend': round(tcr.spend, 2),
'clicks': int(tcr.clicks),
'conversions': round(tcr.conversions, 2),
'status': camp.status if camp else 'UNKNOWN',
})
# ── Daily spend trend (last 30d) ─────────────────────────────────
daily_raw = met_q.filter(
AdMetric.metric_date >= d30.date()
).with_entities(
AdMetric.metric_date,
func.coalesce(func.sum(AdMetric.spend), 0).label('spend'),
func.coalesce(func.sum(AdMetric.clicks), 0).label('clicks'),
func.coalesce(func.sum(AdMetric.impressions), 0).label('impressions'),
func.coalesce(func.sum(AdMetric.conversions), 0).label('conversions'),
).group_by(AdMetric.metric_date).order_by(
AdMetric.metric_date
).all()
daily_spend_trend = [
{
'date': d.metric_date.isoformat() if d.metric_date else '',
'spend': round(d.spend, 2),
'clicks': int(d.clicks),
'impressions': int(d.impressions),
'conversions': round(d.conversions, 2),
}
for d in daily_raw
]
return jsonify({
"success": True,
"data": {
"total_campaigns": total_campaigns,
"active_campaigns": active_campaigns,
"paused_campaigns": paused_campaigns,
"total_spend_30d": round(spend_30d, 2),
"total_spend_90d": round(spend_90d, 2),
"total_spend_ytd": round(spend_ytd, 2),
"total_impressions_30d": int(impressions_30d),
"total_clicks_30d": int(clicks_30d),
"total_conversions_30d": round(conversions_30d, 2),
"avg_ctr": round(avg_ctr, 4),
"avg_cpc": round(avg_cpc, 4),
"avg_cpa": avg_cpa,
"total_roas": round(roas_values, 4),
"campaigns_by_status": campaigns_by_status,
"campaigns_by_channel": campaigns_by_channel,
"top_campaigns": top_campaigns,
"daily_spend_trend": daily_spend_trend,
}
})
# ── Sync ───────────────────────────────────────────────────────────────────
@api_proxy_bp.route('/api/company/<company_id>/facebook_ads/sync', methods=['POST'])
@require_auth_json()
@require_csrf
def sync_facebook_ads(company_id):
"""Manually trigger a Facebook Ads sync."""
from app.models import db, Connector
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
connector = Connector.query.filter_by(
company_id=company_id,
service='facebook_ads',
status='connected'
).first()
if not connector:
return jsonify({"success": False, "error": "No Facebook Ads connector found"}), 404
try:
from app.connectors import build_connector
conn = build_connector(
service='facebook_ads',
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
result = conn.sync()
# Update connector status
connector.last_sync_at = datetime.now(timezone.utc)
if result.get("status") == "success":
connector.status = "connected"
else:
connector.status = "error"
connector.error_message = result.get("error", "Sync failed")
db.session.commit()
# Return error to frontend if the underlying API failed
if result.get("status") != "success":
return jsonify({
"success": False,
"error": result.get("error", "Sync failed — check connector credentials"),
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
}), 502
return jsonify({
"success": True,
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
})
except Exception as e:
logger.exception("Connector sync failed")
connector.status = "error"
connector.error_message = str(e)
db.session.commit()
return jsonify({"success": False, "error": "Sync failed. An error occurred."}), 500
# =============================================================================
# /api/v1/partner/angi/* — Partner-style routes with Bearer user_id:company_id
# =============================================================================
def _parse_partner_auth():
"""Parse Bearer <user_id>:<company_id> auth header. Returns (user_id, company_id) or None on error."""
auth = request.headers.get('Authorization', '')
if not auth.startswith('Bearer '):
return None
token = auth[7:].strip()
parts = token.split(':', 1)
if len(parts) != 2:
return None
return parts[0], parts[1]
@api_proxy_bp.route('/api/v1/partner/angi/leads', methods=['GET'])
def partner_list_angi_leads():
"""List Angi leads for a company via partner auth."""
from app.models import db, AngiLead, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
# Validate user and company exist
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
status = request.args.get('status', 'all')
date_range = request.args.get('date_range', '30d')
priority = request.args.get('priority')
project_type = request.args.get('project_type')
search = request.args.get('search', '')
query = AngiLead.query.filter_by(company_id=company_id)
if status != 'all':
query = query.filter_by(status=status)
if priority:
query = query.filter_by(priority=priority)
if project_type:
query = query.filter(AngiLead.project_type.ilike(f'%{project_type}%'))
if search:
query = query.filter(
db.or_(
AngiLead.first_name.ilike(f'%{search}%'),
AngiLead.last_name.ilike(f'%{search}%'),
AngiLead.email.ilike(f'%{search}%'),
AngiLead.phone.ilike(f'%{search}%'),
AngiLead.angi_lead_id.ilike(f'%{search}%'),
)
)
if date_range != 'all':
days = int(date_range.replace('d', ''))
since = datetime.now(timezone.utc) - timedelta(days=days)
query = query.filter(AngiLead.received_at >= since)
pagination, meta = paginate_query(query.order_by(AngiLead.received_at.desc()))
leads = pagination.items
return jsonify({
"success": True,
"total": meta['total'],
**meta,
"leads": [l.to_dict() for l in leads],
})
@api_proxy_bp.route('/api/v1/partner/angi/leads/<lead_id>', methods=['GET'])
def partner_get_angi_lead(lead_id):
"""Get a single Angi lead via partner auth."""
from app.models import db, AngiLead, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
return jsonify({
"success": True,
"data": lead.to_dict()
})
@api_proxy_bp.route('/api/v1/partner/angi/leads/<lead_id>/actions', methods=['GET'])
def partner_get_angi_lead_actions(lead_id):
"""Get actions for a single Angi lead via partner auth."""
from app.models import db, AngiLead, AngiLeadAction, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
actions = AngiLeadAction.query.filter_by(
lead_id=lead_id,
company_id=company_id,
).order_by(AngiLeadAction.created_at.asc()).all()
return jsonify({
"success": True,
"actions": [a.to_dict() for a in actions],
})
@api_proxy_bp.route('/api/v1/partner/angi/analytics', methods=['GET'])
def partner_get_angi_analytics():
"""Get Angi lead analytics via partner auth."""
from app.models import db, AngiLead, User, Company
from sqlalchemy import func
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
period_days = int(request.args.get('period_days', 30))
since = datetime.now(timezone.utc) - timedelta(days=period_days)
base_query = AngiLead.query.filter(
AngiLead.company_id == company_id,
AngiLead.received_at >= since
)
total_leads = base_query.count()
accepted = base_query.filter_by(status='accepted').count()
responded = base_query.filter_by(status='responded').count()
rejected = base_query.filter_by(status='rejected').count()
new_leads = base_query.filter_by(status='new').count()
responded_query = base_query.filter(AngiLead.response_time_minutes.isnot(None))
avg_response_result = responded_query.with_entities(
func.avg(AngiLead.response_time_minutes)
).scalar()
avg_response_minutes = round(avg_response_result, 1) if avg_response_result else None
cost_result = base_query.with_entities(func.avg(AngiLead.cost_per_lead)).scalar()
avg_cost = round(cost_result, 2) if cost_result else None
total_cost_result = base_query.with_entities(func.sum(AngiLead.cost_per_lead)).scalar()
total_cost = round(total_cost_result or 0, 2)
conversion_rate = round((accepted / total_leads * 100), 1) if total_leads > 0 else 0
response_rate = round((responded / total_leads * 100), 1) if total_leads > 0 else 0
premium = base_query.filter_by(is_premium=True).count()
standard = total_leads - premium
status_breakdown = {
'new': new_leads,
'accepted': accepted,
'responded': responded,
'rejected': rejected,
}
daily = base_query.with_entities(
func.date(AngiLead.received_at).label('day'),
func.count(AngiLead.id).label('count')
).group_by(func.date(AngiLead.received_at)).order_by('day').all()
daily_leads = [{'day': str(d.day), 'count': d.count} for d in daily]
priority_breakdown = {}
for priority in ['high', 'medium', 'low']:
priority_breakdown[priority] = base_query.filter_by(priority=priority).count()
return jsonify({
"success": True,
"data": {
"period_days": period_days,
"total_leads": total_leads,
"conversion_rate": conversion_rate,
"response_rate": response_rate,
"avg_response_time_minutes": avg_response_minutes,
"avg_cost_per_lead": avg_cost,
"total_cost": total_cost,
"premium_leads": premium,
"standard_leads": standard,
"status_breakdown": status_breakdown,
"daily_leads": daily_leads,
"priority_breakdown": priority_breakdown,
}
})
# =============================================================================
# /api/v1/partner/quickbooks/* — Partner-style routes with Bearer user_id:company_id
# =============================================================================
@api_proxy_bp.route('/api/v1/partner/quickbooks/invoices', methods=['GET'])
def partner_list_quickbooks_invoices():
"""List QuickBooks invoices for a company via partner auth."""
from app.models import db, QuickbooksInvoice, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
status = request.args.get('status', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
search = request.args.get('search', '')
query = QuickbooksInvoice.query.filter_by(company_id=company_id)
if status:
query = query.filter(QuickbooksInvoice.status.ilike(f'%{status}%'))
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksInvoice.tx_date >= since)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksInvoice.tx_date <= until)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
if search:
query = query.filter(
db.or_(
QuickbooksInvoice.invoice_num.ilike(f'%{search}%'),
QuickbooksInvoice.customer_name.ilike(f'%{search}%'),
QuickbooksInvoice.qb_doc_id.ilike(f'%{search}%'),
)
)
pagination, meta = paginate_query(query.order_by(QuickbooksInvoice.tx_date.desc()))
invoices = pagination.items
return jsonify({
"success": True,
"total": meta['total'],
**meta,
"invoices": [inv.to_dict() for inv in invoices],
})
@api_proxy_bp.route('/api/v1/partner/quickbooks/invoices/<qb_doc_id>', methods=['GET'])
def partner_get_quickbooks_invoice(qb_doc_id):
"""Get a single QuickBooks invoice by QB document ID via partner auth."""
from app.models import QuickbooksInvoice, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
invoice = QuickbooksInvoice.query.filter_by(
company_id=company_id,
qb_doc_id=qb_doc_id,
).first()
if not invoice:
return jsonify({'error': 'Invoice not found'}), 404
return jsonify({
"success": True,
"data": invoice.to_dict()
})
@api_proxy_bp.route('/api/v1/partner/quickbooks/transactions', methods=['GET'])
def partner_list_quickbooks_transactions():
"""List QuickBooks transactions for a company via partner auth."""
from app.models import db, QuickbooksTransaction, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
tx_type = request.args.get('tx_type', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
search = request.args.get('search', '')
query = QuickbooksTransaction.query.filter_by(company_id=company_id)
if tx_type:
query = query.filter(QuickbooksTransaction.tx_type.ilike(f'%{tx_type}%'))
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksTransaction.tx_date >= since)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksTransaction.tx_date <= until)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
if search:
query = query.filter(
db.or_(
QuickbooksTransaction.qb_doc_id.ilike(f'%{search}%'),
QuickbooksTransaction.account_name.ilike(f'%{search}%'),
QuickbooksTransaction.memo.ilike(f'%{search}%'),
)
)
pagination, meta = paginate_query(query.order_by(QuickbooksTransaction.tx_date.desc()))
transactions = pagination.items
return jsonify({
"success": True,
"total": meta['total'],
**meta,
"transactions": [tx.to_dict() for tx in transactions],
})
@api_proxy_bp.route('/api/v1/partner/quickbooks/customers', methods=['GET'])
def partner_list_quickbooks_customers():
"""List QuickBooks customers for a company via partner auth."""
from app.models import db, QuickbooksCustomer, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
search = request.args.get('search', '')
query = QuickbooksCustomer.query.filter_by(company_id=company_id)
if search:
query = query.filter(
db.or_(
QuickbooksCustomer.display_name.ilike(f'%{search}%'),
QuickbooksCustomer.email.ilike(f'%{search}%'),
QuickbooksCustomer.phone.ilike(f'%{search}%'),
QuickbooksCustomer.qb_doc_id.ilike(f'%{search}%'),
)
)
pagination, meta = paginate_query(query.order_by(QuickbooksCustomer.display_name.asc()))
customers = pagination.items
return jsonify({
"success": True,
"total": meta['total'],
**meta,
"customers": [c.to_dict() for c in customers],
})
@api_proxy_bp.route('/api/v1/partner/quickbooks/expenses', methods=['GET'])
def partner_list_quickbooks_expenses():
"""List QuickBooks expenses for a company via partner auth."""
from app.models import db, QuickbooksExpense, User, Company
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
category = request.args.get('category', '')
date_from = request.args.get('date_from', '')
date_to = request.args.get('date_to', '')
search = request.args.get('search', '')
query = QuickbooksExpense.query.filter_by(company_id=company_id)
if category:
query = query.filter(QuickbooksExpense.category.ilike(f'%{category}%'))
if date_from:
try:
since = datetime.fromisoformat(date_from)
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksExpense.tx_date >= since)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_from format: {date_from}'}), 400
if date_to:
try:
until = datetime.fromisoformat(date_to)
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
query = query.filter(QuickbooksExpense.tx_date <= until)
except (ValueError, TypeError):
return jsonify({'error': f'Invalid date_to format: {date_to}'}), 400
if search:
query = query.filter(
db.or_(
QuickbooksExpense.vendor_name.ilike(f'%{search}%'),
QuickbooksExpense.account_name.ilike(f'%{search}%'),
QuickbooksExpense.memo.ilike(f'%{search}%'),
QuickbooksExpense.qb_doc_id.ilike(f'%{search}%'),
)
)
pagination, meta = paginate_query(query.order_by(QuickbooksExpense.tx_date.desc()))
expenses = pagination.items
return jsonify({
"success": True,
"total": meta['total'],
**meta,
"expenses": [exp.to_dict() for exp in expenses],
})
@api_proxy_bp.route('/api/v1/partner/quickbooks/analytics', methods=['GET'])
def partner_get_quickbooks_analytics():
"""Get QuickBooks analytics: revenue summary, expense breakdown, AR aging, top customers."""
from app.models import db, QuickbooksInvoice, QuickbooksExpense, QuickbooksCustomer, User, Company
from sqlalchemy import func
auth = _parse_partner_auth()
if not auth:
return jsonify({'error': 'Authentication required'}), 401
user_id, company_id = auth
user = User.query.get(user_id)
company = Company.query.get(company_id)
if not user or not company:
return jsonify({'error': 'User or company not found'}), 401
period_days = int(request.args.get('period_days', 90))
since = datetime.now(timezone.utc) - timedelta(days=period_days)
# ── Revenue summary ──────────────────────────────────────────────
inv_query = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.tx_date >= since,
)
total_invoices = inv_query.count()
total_revenue = inv_query.with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar()
total_revenue = round(total_revenue, 2) if total_revenue else 0.0
paid_query = inv_query.filter(QuickbooksInvoice.status.ilike('%paid%'))
paid_count = paid_query.count()
paid_revenue = paid_query.with_entities(
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0)
).scalar()
paid_revenue = round(paid_revenue, 2) if paid_revenue else 0.0
# ── AR Aging (unpaid invoices grouped by age bucket) ─────────────
unpaid_query = inv_query.filter(
~QuickbooksInvoice.status.ilike('%paid%'),
QuickbooksInvoice.status.notlike('%void%'),
)
now = datetime.now(timezone.utc)
ar_aging = {'current': 0.0, 'overdue_30': 0.0, 'overdue_60': 0.0, 'overdue_90_plus': 0.0}
unpaid_invoices = unpaid_query.all()
for inv in unpaid_invoices:
amount = inv.total_amount or 0.0
due = inv.due_date
if due is None:
ar_aging['current'] += amount
continue
if due.tzinfo is None:
due = due.replace(tzinfo=timezone.utc)
days_overdue = (now - due).days
if days_overdue <= 0:
ar_aging['current'] += amount
elif days_overdue <= 30:
ar_aging['overdue_30'] += amount
elif days_overdue <= 60:
ar_aging['overdue_60'] += amount
else:
ar_aging['overdue_90_plus'] += amount
# Round all AR aging values
ar_aging = {k: round(v, 2) for k, v in ar_aging.items()}
# Total outstanding AR
total_ar = sum(ar_aging.values())
# ── Expense breakdown by category ─────────────────────────────────
exp_query = QuickbooksExpense.query.filter(
QuickbooksExpense.company_id == company_id,
QuickbooksExpense.tx_date >= since,
)
total_expenses = exp_query.with_entities(
func.coalesce(func.sum(QuickbooksExpense.amount), 0)
).scalar()
total_expenses = round(total_expenses, 2) if total_expenses else 0.0
expense_by_category = exp_query.with_entities(
QuickbooksExpense.category,
func.coalesce(func.sum(QuickbooksExpense.amount), 0).label('total'),
).group_by(QuickbooksExpense.category).order_by(text('total desc')).all()
expense_breakdown = [{'category': c.category or 'uncategorized', 'amount': round(c.total, 2)} for c in expense_by_category]
# ── Top customers by invoice amount ──────────────────────────────
top_customers_raw = inv_query.with_entities(
QuickbooksInvoice.customer_name,
func.count(QuickbooksInvoice.id).label('invoice_count'),
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0).label('total_amount'),
).group_by(QuickbooksInvoice.customer_name).order_by(text('total_amount desc')).limit(10).all()
top_customers = [
{
'customer_name': c.customer_name,
'invoice_count': c.invoice_count,
'total_amount': round(c.total_amount, 2),
}
for c in top_customers_raw
]
# ── Daily revenue trend ──────────────────────────────────────────
daily_revenue = inv_query.with_entities(
func.date(QuickbooksInvoice.tx_date).label('day'),
func.coalesce(func.sum(QuickbooksInvoice.total_amount), 0).label('amount'),
).group_by(func.date(QuickbooksInvoice.tx_date)).order_by('day').all()
daily_revenue_list = [{'day': str(d.day), 'amount': round(d.amount, 2)} for d in daily_revenue]
# ── Net profit ───────────────────────────────────────────────────
net_profit = round(total_revenue - total_expenses, 2)
return jsonify({
"success": True,
"data": {
"period_days": period_days,
"revenue_summary": {
"total_invoices": total_invoices,
"total_revenue": total_revenue,
"paid_count": paid_count,
"paid_revenue": paid_revenue,
"unpaid_revenue": round(total_revenue - paid_revenue, 2),
},
"ar_aging": ar_aging,
"total_ar": round(total_ar, 2),
"expenses": {
"total_expenses": total_expenses,
"by_category": expense_breakdown,
},
"net_profit": net_profit,
"top_customers": top_customers,
"daily_revenue": daily_revenue_list,
}
})
# =============================================================================
# /api/company/<company_id>/slack/* — Slack workspace connector routes
# =============================================================================
from app.connectors.slack import SlackConnector
@api_proxy_bp.route('/api/company/<company_id>/slack/workspace', methods=['GET'])
@require_auth_json()
def get_slack_workspace(company_id):
"""Get Slack workspace info and channel summary."""
from app.models import db, Connector, SlackChannel
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
connector = Connector.query.filter_by(
company_id=company_id,
service='slack'
).first()
if not connector:
return jsonify({"success": False, "error": "Slack connector not configured"}), 404
try:
from app.connectors import build_connector
conn = build_connector(
service='slack',
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
workspace_info = conn._fetch_workspace_info()
except Exception as e:
logger.exception("Failed to fetch Slack workspace info")
return jsonify({"success": False, "error": "Failed to fetch workspace info. An error occurred."}), 500
channels = SlackChannel.query.filter_by(
company_id=company_id,
is_archived=False
).all()
channel_count = len(channels)
public_count = sum(1 for c in channels if not c.is_private)
private_count = sum(1 for c in channels if c.is_private)
return jsonify({
"success": True,
"data": {
"workspace": workspace_info,
"channel_count": channel_count,
"public_count": public_count,
"private_count": private_count,
"last_sync_at": connector.last_sync_at.isoformat() if connector.last_sync_at else None,
}
})
@api_proxy_bp.route('/api/company/<company_id>/slack/channels', methods=['GET'])
@require_auth_json()
def get_slack_channels(company_id):
"""List Slack channels with optional filters."""
from app.models import SlackChannel
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
search = request.args.get('search', '').strip()
is_private = request.args.get('is_private')
is_archived = request.args.get('is_archived')
query = SlackChannel.query.filter_by(company_id=company_id)
if search:
query = query.filter(SlackChannel.name.ilike(f'%{search}%'))
if is_private is not None:
query = query.filter(SlackChannel.is_private == (is_private.lower() == 'true'))
if is_archived is not None:
query = query.filter(SlackChannel.is_archived == (is_archived.lower() == 'true'))
pagination, meta = paginate_query(query.order_by(SlackChannel.name))
channels = pagination.items
return jsonify({
"success": True,
"data": {
"items": [{
"id": c.id,
"external_id": c.external_id,
"name": c.name,
"is_private": c.is_private,
"is_archived": c.is_archived,
"purpose": c.purpose,
"topic": c.topic,
"synced_at": c.created_at.isoformat() if c.created_at else None,
} for c in channels],
"total": meta['total'],
}
})
@api_proxy_bp.route('/api/company/<company_id>/slack/messages', methods=['GET'])
@require_auth_json()
def get_slack_messages(company_id):
"""List recent Slack messages."""
from app.models import db, ExternalSyncRecord
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
channel_id = request.args.get('channel_id', '').strip()
search = request.args.get('search', '').strip()
limit = request.args.get('limit', 50, type=int)
query = ExternalSyncRecord.query.filter_by(
company_id=company_id,
source_service='slack',
entity_type='message',
)
if channel_id:
query = query.filter(
ExternalSyncRecord.properties_json['channel_id'].as_string() == channel_id
)
if search:
query = query.filter(
ExternalSyncRecord.properties_json['text'].as_string().ilike(f'%{search}%')
)
messages = query.order_by(ExternalSyncRecord.created_at.desc()).limit(limit).all()
return jsonify({
"success": True,
"data": {
"items": [{
"id": m.id,
"external_id": m.external_id,
"channel_id": m.properties_json.get('channel_id', ''),
"channel_name": m.properties_json.get('channel_name', ''),
"user_id": m.properties_json.get('user_id', m.properties_json.get('user', '')),
"text": m.properties_json.get('text', ''),
"ts": m.properties_json.get('ts', ''),
"thread_ts": m.properties_json.get('thread_ts', ''),
"reactions": m.properties_json.get('reactions', []),
"bot_name": m.properties_json.get('bot_name', ''),
"type": m.properties_json.get('type', ''),
"synced_at": m.created_at.isoformat() if m.created_at else None,
} for m in messages],
"total": query.count(),
}
})
@api_proxy_bp.route('/api/company/<company_id>/slack/sync', methods=['POST'])
@require_auth_json()
@require_csrf
def sync_slack(company_id):
"""Trigger full Slack sync (workspace + channels + messages)."""
from app.models import db, Connector
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
connector = Connector.query.filter_by(
company_id=company_id,
service='slack'
).first()
if not connector:
return jsonify({"success": False, "error": "Slack connector not configured"}), 404
try:
from app.connectors import build_connector
conn = build_connector(
service='slack',
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
result = conn.sync()
# Update connector status
connector.last_sync_at = datetime.now(timezone.utc)
if result.get("status") == "success":
connector.status = "connected"
else:
connector.status = "error"
connector.error_message = result.get("error", "Sync failed")
db.session.commit()
# Return error to frontend if the underlying API failed
if result.get("status") != "success":
return jsonify({
"success": False,
"error": result.get("error", "Sync failed — check connector credentials"),
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
}), 502
return jsonify({
"success": True,
"data": {
"message": f"Synced {result.get('record_count', 0)} records",
"status": result.get("status"),
}
})
except Exception as e:
logger.exception("Connector sync failed")
connector.status = "error"
connector.error_message = str(e)
db.session.commit()
return jsonify({"success": False, "error": "Sync failed. An error occurred."}), 500
@api_proxy_bp.route('/api/company/<company_id>/slack/send-message', methods=['POST'])
@require_auth_json()
@require_csrf
def send_slack_message(company_id):
"""Send a message to a Slack channel."""
from app.models import db, Connector
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
channel_id = data.get('channel_id', '')
text = data.get('text', '')
thread_ts = data.get('thread_ts', '')
if not channel_id or not text:
return jsonify({"success": False, "error": "channel_id and text are required"}), 400
connector = Connector.query.filter_by(
company_id=company_id,
service='slack'
).first()
if not connector:
return jsonify({"success": False, "error": "Slack connector not configured"}), 404
try:
from app.connectors import build_connector
conn = build_connector(
service='slack',
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
# Build the API call
params = {'channel': channel_id, 'text': text}
if thread_ts:
params['thread_ts'] = thread_ts
result = conn._api_get('chat.postMessage', **params)
return jsonify({
"success": True,
"data": {
"status": "sent",
"channel": channel_id,
"ts": result.get('ts'),
}
})
except Exception as e:
logger.exception("Failed to send Slack message")
return jsonify({"success": False, "error": "Failed to send message. An error occurred."}), 500