"""Admin blueprint — plans, super-admin, partner, demo-request, CSRF token, company mgmt.
Split from api_proxy.py monolith for maintainability (July 2026).
Imports shared utilities from api_proxy.
"""
from flask import Blueprint, request, jsonify
from flask_login import current_user
from datetime import datetime, timezone, timedelta
import re
import secrets
from app import db, limiter
from app.models import Company
from app.utils.pagination import paginate_query
from app.routes.api_proxy import (
require_auth_json,
require_csrf,
require_super_admin,
_check_company_access,
COMPANY_ROLE_LEVELS,
COMPANY_PERMISSIONS,
)
admin_api_bp = Blueprint('admin_api', __name__)
@admin_api_bp.route('/api/plans', methods=['GET'])
def api_plans():
"""Return available plans for the SPA - includes current tier if authenticated."""
plans = [
{
'id': 'launch',
'name': 'Launch',
'priceCents': 4900,
'tagline': 'Start small, prove the number.',
'highlights': ['Single market', '1 seat', 'Core forecast', 'Basic goals'],
'limits': {'markets': 1, 'seats': 1, 'integrations': 2, 'storageGb': 5},
'trialDays': 7,
'popular': False,
'trialEligible': True
},
{
'id': 'growth',
'name': 'Growth',
'priceCents': 9900,
'tagline': 'Scale with confidence across markets.',
'highlights': ['5 markets', '5 seats', 'Advanced forecast', 'Coaching', 'Optimizations'],
'limits': {'markets': 5, 'seats': 5, 'integrations': 5, 'storageGb': 25},
'trialDays': 7,
'popular': True,
'trialEligible': True
},
{
'id': 'command',
'name': 'Command',
'priceCents': 19900,
'tagline': 'Full portfolio command and control.',
'highlights': ['Unlimited markets', 'Unlimited seats', 'All features', 'API access'],
'limits': {'markets': 999, 'seats': 999, 'integrations': 99, 'storageGb': 100},
'trialDays': 7,
'popular': False,
'trialEligible': True
},
{
'id': 'enterprise',
'name': 'Enterprise',
'priceCents': None,
'tagline': 'Custom for PE-backed portfolios.',
'highlights': ['Custom everything', 'Dedicated support', 'SLA', 'SSO/SAML'],
'limits': {'markets': 999, 'seats': 999, 'integrations': 99, 'storageGb': 999},
'trialDays': 0,
'popular': False,
'trialEligible': False
}
]
response = {'plans': plans, 'trialDays': 7}
# Include current tier if user is authenticated
if current_user.is_authenticated:
from app.models import db, Company, UserCompany
user_company = UserCompany.query.filter_by(user_id=current_user.id).first()
if user_company:
company = db.session.get(Company, user_company.company_id)
if company:
response['tier'] = (company.settings_json or {}).get('tier', 'starter')
else:
response['tier'] = 'starter'
else:
response['tier'] = 'starter'
return jsonify(response)
@admin_api_bp.route('/api/partner/me', methods=['GET'])
@require_auth_json()
def api_partner_me():
"""Return partner profile info for SPA."""
from app.models import Company, UserCompany
companies = Company.query.join(
UserCompany, Company.id == UserCompany.company_id
).filter(
UserCompany.user_id == current_user.id
).all()
total_revenue = sum(c.annual_revenue or 0 for c in companies)
return jsonify({
'partner': {
'id': current_user.id,
'email': current_user.email,
'full_name': current_user.full_name or current_user.email.split('@')[0],
'company': current_user.company or 'Command Center Partner',
'role': 'partner',
'organization_count': len(companies),
'total_portfolio_revenue': total_revenue,
'is_active': True
}
})
@admin_api_bp.route('/api/partner/company-role', methods=['GET'])
@require_auth_json()
def api_partner_company_role():
"""Return the current user's role within their primary (first) company.
Returns:
{
"companyRole": "owner" | "admin" | "manager" | "rep" | "viewer",
"companyId": "...",
"companyName": "...",
"level": 5,
"permissions": ["create_org", "manage_team", ...]
}
"""
from app.models import UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({
'companyRole': None,
'companyId': None,
'companyName': None,
'level': 0,
'permissions': []
})
role = uc.role
level = COMPANY_ROLE_LEVELS.get(role, 0)
# Compute which permissions the user holds
permissions = [
perm for perm, min_level in COMPANY_PERMISSIONS.items()
if level >= min_level
]
return jsonify({
'companyRole': role,
'companyId': uc.company_id,
'companyName': uc.company.name if uc.company else None,
'level': level,
'permissions': permissions,
})
@admin_api_bp.route('/api/partner/overview', methods=['GET'])
@require_auth_json()
def api_partner_overview():
"""Partner overview — scoped to current user's companies only.
Returns the same DashboardStats shape as /api/super-admin/overview
but filtered to the authenticated user's portfolio. New signups get
empty data (1 org = their own company, 0 revenue).
"""
from app.models import UserCompany
user_companies = UserCompany.query.filter_by(user_id=current_user.id).all()
company_ids = [uc.company_id for uc in user_companies]
companies = Company.query.filter(Company.id.in_(company_ids)).all() if company_ids else []
total_revenue = sum(c.annual_revenue or 0 for c in companies)
avg_revenue = total_revenue / len(companies) if companies else 0
active = len([c for c in companies if getattr(c, 'status', None) != 'suspended'])
return jsonify({
'totalOrganizations': len(companies),
'totalUsers': len(user_companies),
'totalActive': active,
'totalSuspended': len(companies) - active,
'totalRevenue': total_revenue,
'avgRevenue': round(avg_revenue, 2),
'recentActivity': [],
'systemHealth': {
'emailFailures': 0,
'integrationErrors': 0,
'activeAlerts': 0,
'lastSync': None
}
})
@admin_api_bp.route('/api/super-admin/groups', methods=['GET'])
@require_super_admin()
@limiter.limit("20/hour")
def api_super_admin_groups():
"""Return user groups with member counts."""
from app.models import db, UserGroup, GroupMember, UserCompany
from sqlalchemy.orm import joinedload
groups = (
UserGroup.query
.order_by(UserGroup.created_at.desc())
.all()
)
# Audit M2 fix: avoid N+1 — batch-load members and user emails in 2 queries
group_ids = [g.id for g in groups]
members_by_group = {}
if group_ids:
all_members = GroupMember.query.filter(GroupMember.group_id.in_(group_ids)).all()
user_ids = {m.user_id for m in all_members}
ucs = UserCompany.query.options(joinedload(UserCompany.user)).filter(
UserCompany.user_id.in_(user_ids)
).all() if user_ids else []
email_by_user = {uc.user_id: uc.user.email for uc in ucs if uc.user}
for m in all_members:
members_by_group.setdefault(m.group_id, []).append(m)
items = []
for g in groups:
members = members_by_group.get(g.id, [])
user_emails = [email_by_user[m.user_id] for m in members if m.user_id in email_by_user]
items.append({
'id': g.id,
'name': g.name,
'description': g.description,
'company_id': g.company_id,
'member_count': len(members),
'created_at': g.created_at.isoformat() if g.created_at else None,
})
return jsonify({'items': items, 'nextCursor': None})
@admin_api_bp.route('/api/super-admin/partners', methods=['GET'])
@require_super_admin()
@limiter.limit("20/hour")
def api_super_admin_partners():
"""Return partner organizations (companies with partner status)."""
from app.models import db, Company, UserCompany, Invite
partners = Company.query.filter(
Company.is_partner == True, # noqa: E712 — audit M1: dedicated partner flag
Company.is_deleted == False, # noqa: E712 — audit L4: exclude soft-deleted
).order_by(Company.created_at.desc()).all()
items = []
for p in partners:
member_count = UserCompany.query.filter_by(company_id=p.id).count()
invite_count = Invite.query.filter_by(company_id=p.id, status='pending').count()
owner = UserCompany.query.filter_by(company_id=p.id, role='owner').first()
owner_name = 'Unknown'
if owner and owner.user:
owner_name = f"{owner.user.first_name} {owner.user.last_name}"
items.append({
'id': p.id,
'name': p.name,
'domain': p.domain,
'member_count': member_count,
'pending_invites': invite_count,
'owner': owner_name,
'created_at': p.created_at.isoformat() if p.created_at else None,
})
return jsonify({'items': items, 'nextCursor': None})
# ──────────────────────────────────────────────────────────────────────
# Commission / Partnership management
# ──────────────────────────────────────────────────────────────────────
@admin_api_bp.route('/api/super-admin/commissions/partnerships', methods=['GET'])
@require_super_admin()
def api_admin_commissions_list():
"""List all active commission partnerships."""
from app.models import PartnerCommission, User, Company
partnerships = PartnerCommission.query.filter_by(
status='active'
).order_by(PartnerCommission.created_at.desc()).all()
items = []
for p in partnerships:
partner_name = 'Unknown'
partner_email = ''
if p.partner:
partner_name = f"{p.partner.first_name} {p.partner.last_name}"
partner_email = p.partner.email or ''
items.append({
'id': p.id,
'partner_id': p.partner_id,
'partner_name': partner_name,
'partner_email': partner_email,
'company_id': p.company_id,
'company_name': p.company.name if p.company else 'Unknown',
'commission_rate': p.commission_rate,
'revenue_source': p.revenue_source,
'effective_date': p.effective_date.isoformat() if p.effective_date else None,
'created_at': p.created_at.isoformat() if p.created_at else None,
})
return jsonify({'items': items})
@admin_api_bp.route('/api/super-admin/commissions/partnerships', methods=['POST'])
@require_super_admin()
@require_csrf
def api_admin_commissions_create():
"""Create a new commission partnership agreement."""
from app.models import PartnerCommission, User, Company
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
partner_email = data.get('partner_email')
company_id = data.get('company_id')
commission_rate = data.get('commission_rate', 0.15)
if not partner_email or not company_id:
return jsonify({'error': 'partner_email and company_id required'}), 400
# Look up partner by email
partner = User.query.filter_by(email=partner_email).first()
if not partner:
return jsonify({'error': f'No user found with email {partner_email}'}), 404
# Look up company
company = Company.query.get(company_id)
if not company:
return jsonify({'error': f'No company found with id {company_id}'}), 404
# Check for existing partnership
existing = PartnerCommission.query.filter_by(
partner_id=partner.id,
company_id=company_id
).first()
if existing and existing.status == 'active':
return jsonify({'error': 'Active partnership already exists for this pair'}), 409
# Create partnership
partnership = PartnerCommission(
partner_id=partner.id,
company_id=company_id,
commission_rate=commission_rate,
revenue_source=data.get('revenue_source', 'self_reported'),
status='active',
metadata_json=data.get('metadata', {}),
)
db.session.add(partnership)
db.session.commit()
return jsonify({
'id': partnership.id,
'message': f'Partnership created: {partner_email} @ {company.name} @ {commission_rate*100}%'
}), 201
@admin_api_bp.route('/api/super-admin/commissions/partnerships/<partnership_id>', methods=['PUT'])
@require_super_admin()
@require_csrf
def api_admin_commissions_update(partnership_id):
"""Update or terminate a commission partnership."""
from app.models import PartnerCommission
partnership = PartnerCommission.query.get(partnership_id)
if not partnership:
return jsonify({'error': 'Partnership not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
# Handle termination
if data.get('terminate'):
partnership.status = 'terminated'
partnership.terminated_date = datetime.now(timezone.utc)
db.session.commit()
return jsonify({'message': 'Partnership terminated'})
# Update rate
if 'commission_rate' in data:
partnership.commission_rate = data['commission_rate']
if 'revenue_source' in data:
partnership.revenue_source = data['revenue_source']
if 'metadata' in data:
partnership.metadata_json = data['metadata']
db.session.commit()
return jsonify({
'id': partnership.id,
'commission_rate': partnership.commission_rate,
'status': partnership.status,
'message': 'Partnership updated'
})
@admin_api_bp.route('/api/super-admin/commissions/records', methods=['GET'])
@require_super_admin()
def api_admin_commissions_records():
"""List commission records with optional filtering."""
from app.models import CommissionRecord, PartnerCommission
query = CommissionRecord.query.join(PartnerCommission).order_by(
CommissionRecord.period.desc(),
CommissionRecord.created_at.desc()
)
# Optional filters
partner_id = request.args.get('partner_id')
if partner_id:
query = query.filter(PartnerCommission.partner_id == partner_id)
status = request.args.get('status')
if status:
query = query.filter(CommissionRecord.status == status)
period = request.args.get('period')
if period:
query = query.filter(CommissionRecord.period == period)
records = query.limit(200).all()
items = []
for r in records:
items.append({
'id': r.id,
'partner_commission_id': r.partner_commission_id,
'partner_name': f"{r.partner_commission.partner.first_name} {r.partner_commission.partner.last_name}" if r.partner_commission and r.partner_commission.partner else 'Unknown',
'company_name': r.partner_commission.company.name if r.partner_commission and r.partner_commission.company else 'Unknown',
'period': r.period,
'revenue_amount': r.revenue_amount,
'commission_rate': r.commission_rate,
'commission_amount': r.commission_amount,
'status': r.status,
'revenue_confidence': r.revenue_confidence,
'calculated_at': r.calculated_at.isoformat() if r.calculated_at else None,
})
return jsonify({'items': items})
@admin_api_bp.route('/api/super-admin/commissions/records/<record_id>/approve', methods=['POST'])
@require_super_admin()
@require_csrf
def api_admin_commissions_approve_record(record_id):
"""Approve a commission record for payout."""
from app.models import CommissionRecord
record = CommissionRecord.query.get(record_id)
if not record:
return jsonify({'error': 'Record not found'}), 404
data = request.get_json() or {}
record.status = 'approved'
if data.get('notes'):
record.notes = data['notes']
db.session.commit()
return jsonify({'message': 'Commission record approved'})
@admin_api_bp.route('/api/super-admin/commissions/payouts', methods=['GET'])
@require_super_admin()
def api_admin_commissions_payouts():
"""List all payouts."""
from app.models import Payout, User
payouts = Payout.query.order_by(Payout.scheduled_date.desc()).limit(100).all()
items = []
for p in payouts:
partner_name = 'Unknown'
if p.partner:
partner_name = f"{p.partner.first_name} {p.partner.last_name}"
# Count linked records
record_count = 0
total_commission = 0
for pr in p.payout_records:
if pr.commission_record:
record_count += 1
total_commission += pr.commission_record.commission_amount
items.append({
'id': p.id,
'partner_id': p.partner_id,
'partner_name': partner_name,
'amount': p.amount,
'currency': p.currency,
'status': p.status,
'scheduled_date': p.scheduled_date.isoformat() if p.scheduled_date else None,
'paid_date': p.paid_date.isoformat() if p.paid_date else None,
'payment_method': p.payment_method,
'transaction_id': p.transaction_id,
'record_count': record_count,
'created_at': p.created_at.isoformat() if p.created_at else None,
})
return jsonify({'items': items})
@admin_api_bp.route('/api/super-admin/commissions/payouts', methods=['POST'])
@require_super_admin()
@require_csrf
def api_admin_commissions_create_payout():
"""Create a payout from approved commission records."""
from app.models import Payout, PayoutRecord, CommissionRecord, PartnerCommission
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
partner_id = data.get('partner_id')
if not partner_id:
return jsonify({'error': 'partner_id required'}), 400
# Find approved, unpaid records for this partner
records = CommissionRecord.query.join(PartnerCommission).filter(
PartnerCommission.partner_id == partner_id,
CommissionRecord.status == 'approved',
CommissionRecord.payout_id == None
).all()
if not records:
return jsonify({'error': 'No approved, unpaid records found for this partner'}), 404
# If specific record IDs provided, filter to those
if data.get('record_ids'):
records = [r for r in records if r.id in data['record_ids']]
if not records:
return jsonify({'error': 'No matching records found'}), 404
total = sum(r.commission_amount for r in records)
# Create payout
payout = Payout(
partner_id=partner_id,
amount=total,
currency=data.get('currency', 'USD'),
status='scheduled',
scheduled_date=data.get('scheduled_date') or datetime.now(timezone.utc).replace(day=15),
payment_method=data.get('payment_method', 'bank_transfer'),
notes=data.get('notes', ''),
)
db.session.add(payout)
db.session.flush() # Get payout.id
# Link records
for r in records:
payout_record = PayoutRecord(
payout_id=payout.id,
commission_record_id=r.id
)
db.session.add(payout_record)
r.status = 'paid'
r.payout_id = payout.id
r.updated_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
'payout_id': payout.id,
'amount': total,
'record_count': len(records),
'scheduled_date': payout.scheduled_date.isoformat(),
'message': f'Payout of ${total:.2f} created with {len(records)} records'
}), 201
@admin_api_bp.route('/api/super-admin/commissions/payouts/<payout_id>', methods=['PUT'])
@require_super_admin()
@require_csrf
def api_admin_commissions_update_payout(payout_id):
"""Mark a payout as paid/failed and record transaction details."""
from app.models import Payout
payout = Payout.query.get(payout_id)
if not payout:
return jsonify({'error': 'Payout not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
if 'status' in data:
payout.status = data['status']
if data['status'] == 'completed':
payout.paid_date = datetime.now(timezone.utc)
if 'transaction_id' in data:
payout.transaction_id = data['transaction_id']
if 'notes' in data:
payout.notes = data['notes']
db.session.commit()
return jsonify({
'id': payout.id,
'status': payout.status,
'paid_date': payout.paid_date.isoformat() if payout.paid_date else None,
'message': 'Payout updated'
})
@admin_api_bp.route('/api/super-admin/commissions/calculate', methods=['POST'])
@require_super_admin()
@require_csrf
def api_admin_commissions_calculate():
"""Trigger commission calculation for a specific period.
Pulls revenue from whichever source is available:
1. Connector-synced KPI data
2. Forecast values
3. Company.annual_revenue fallback
"""
from app.models import PartnerCommission, CommissionRecord, Company, KPIValue, Forecast
from datetime import datetime, timezone
data = request.get_json() or {}
period = data.get('period') # YYYY-MM format
if not period:
# Default to previous month
now = datetime.now(timezone.utc)
if now.month == 1:
period = f"{now.year - 1}-12"
else:
period = f"{now.year}-{now.month - 1:02d}"
# Parse period dates
year, month = map(int, period.split('-'))
from calendar import monthrange
period_start = datetime(year, month, 1, tzinfo=timezone.utc)
_, last_day = monthrange(year, month)
period_end = datetime(year, month, last_day, 23, 59, 59, tzinfo=timezone.utc)
# Get all active partnerships
partnerships = PartnerCommission.query.filter_by(status='active').all()
created_count = 0
updated_count = 0
for partnership in partnerships:
company = partnership.company
if not company:
continue
# Check if record already exists for this period
existing = CommissionRecord.query.filter_by(
partner_commission_id=partnership.id,
period=period
).first()
# Determine revenue amount
revenue_amount = 0.0
revenue_confidence = 'estimated'
# Try: self-reported value first (partner fills in revenue)
if partnership.revenue_source == 'self_reported':
# Allow manual override in request body
if 'revenue_overrides' in data and company.id in data['revenue_overrides']:
revenue_amount = float(data['revenue_overrides'][company.id])
revenue_confidence = 'self_reported'
else:
# Fallback to annual_revenue / 12
revenue_amount = (company.annual_revenue or 0) / 12
revenue_confidence = 'estimated'
elif partnership.revenue_source == 'forecast':
# Pull from Forecast table
forecast = Forecast.query.filter_by(company_id=company.id).first()
if forecast:
revenue_amount = forecast.forecasted_revenue or 0
revenue_confidence = 'verified'
else:
revenue_amount = (company.annual_revenue or 0) / 12
revenue_confidence = 'estimated'
elif partnership.revenue_source == 'connector':
# Pull from KPIValue (connector-synced revenue)
# Look for a monthly revenue KPI
kpi = KPIValue.query.filter_by(
company_id=company.id
).order_by(KPIValue.value_date.desc()).first()
if kpi:
revenue_amount = kpi.current_value or 0
revenue_confidence = 'verified'
else:
revenue_amount = (company.annual_revenue or 0) / 12
revenue_confidence = 'estimated'
commission_amount = revenue_amount * partnership.commission_rate
if existing:
# Update existing record
existing.revenue_amount = revenue_amount
existing.revenue_confidence = revenue_confidence
existing.commission_rate = partnership.commission_rate
existing.commission_amount = commission_amount
existing.calculated_at = datetime.now(timezone.utc)
updated_count += 1
else:
# Create new record
record = CommissionRecord(
partner_commission_id=partnership.id,
period=period,
period_start=period_start,
period_end=period_end,
revenue_amount=revenue_amount,
revenue_confidence=revenue_confidence,
commission_rate=partnership.commission_rate,
commission_amount=commission_amount,
status='calculated',
calculated_at=datetime.now(timezone.utc),
)
db.session.add(record)
created_count += 1
db.session.commit()
return jsonify({
'period': period,
'created': created_count,
'updated': updated_count,
'partnerships_processed': len(partnerships),
})
@admin_api_bp.route('/api/super-admin/organizations/<org_id>', methods=['DELETE'])
@require_super_admin()
@require_csrf
def api_super_admin_delete_organization(org_id):
"""Delete an organization (company) and all related data — super-admin only.
Cascades to user_company memberships, projects, goals, KPI values,
forecasts, revenue_leaks, optimization_moves, coaching_assignments,
coaching_scorecards, connectors, connector_logs, activity_logs,
notifications, settings, templates, custom_fields, roi_calculations.
"""
from app.models import db, Company
company = db.session.get(Company, org_id)
if not company:
return jsonify({'error': 'Organization not found'}), 404
deleted_name = company.name
db.session.delete(company)
db.session.commit()
return jsonify({
'success': True,
'message': f'Organization "{deleted_name}" deleted',
})
@admin_api_bp.route('/api/super-admin/organizations', methods=['GET'])
@require_super_admin()
@limiter.limit("20/hour")
def api_super_admin_organizations():
"""Return list of organizations (companies) for super-admin.
SPA calls /api/super-admin/organizations — super-admin gets all companies.
"""
from app.models import Company, UserCompany, Project, KPIValue
# Super-admin: return all companies
pagination, meta = paginate_query(Company.query.order_by(Company.id))
companies = pagination.items
organizations = []
for company in companies:
# Get active projects count
active_projects = Project.query.filter_by(
company_id=company.id,
status='active'
).count()
# Build slug from company name
slug = company.name.lower().replace(' ', '-').replace('&', 'and').replace(',', '')
# Dynamic tier from settings_json
tier = (company.settings_json or {}).get('tier', 'starter')
organizations.append({
'id': company.id,
'name': company.name,
'slug': slug,
'industry': company.industry or 'Home Services',
'status': 'active',
'subStatus': 'active',
'tier': tier,
'groupId': None,
'referrerPartnerId': None,
'annual_revenue': company.annual_revenue or 0,
'target_revenue': company.target_revenue or 0,
'revenue_growth': ((company.annual_revenue or 0) - (company.target_revenue or 0)) / max(company.target_revenue or 1, 1) * 100,
'active_projects': active_projects,
'size': company.size or 'Small',
'location': f"{company.city or ''}, {company.state or ''}".strip(', '),
'website': company.website or '',
'commission_rate': 0.15,
'last_updated': company.updated_at.isoformat() if company.updated_at else None
})
return jsonify({
'items': organizations,
'total': meta['total'],
'nextCursor': None
})
@admin_api_bp.route('/api/super-admin/organizations', methods=['POST'])
@require_super_admin()
@require_csrf
@limiter.limit("10/hour")
def api_super_admin_create_organization():
"""Create a new organization (company) — super-admin only.
Accepts name (required) plus optional fields: industry, annual_revenue,
address, city, state, postal_code, country, phone, website, owner_id,
partner_id, stripe_customer_id.
If owner_id is provided, a UserCompany membership is created linking
that user as 'owner'.
"""
from app.models import db, Company, UserCompany, User
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
name = data.get('name', '').strip()
if not name:
return jsonify({'error': 'Organization name is required'}), 400
if len(name) > 200:
return jsonify({'error': 'Organization name must be 200 characters or less'}), 400
# Check for duplicate name
existing = Company.query.filter_by(name=name).first()
if existing:
return jsonify({'error': f'An organization with the name "{name}" already exists'}), 409
# Validate owner_id if provided
owner_id = data.get('owner_id')
if owner_id:
owner = User.query.get(owner_id)
if not owner:
return jsonify({'error': f'User with id "{owner_id}" not found'}), 404
# Optional numeric fields
annual_revenue = data.get('annual_revenue')
if annual_revenue is not None:
try:
annual_revenue = float(annual_revenue)
except (ValueError, TypeError):
return jsonify({'error': 'annual_revenue must be a number'}), 400
new_company = Company(
name=name,
industry=data.get('industry', '').strip() or '',
annual_revenue=annual_revenue,
address=data.get('address', '').strip() or '',
city=data.get('city', '').strip() or '',
state=data.get('state', '').strip() or '',
zip_code=data.get('postal_code', '').strip() or '',
website=data.get('website', '').strip() or '',
stripe_customer_id=data.get('stripe_customer_id', '') or '',
)
db.session.add(new_company)
db.session.flush()
# If owner_id provided, create a UserCompany membership
if owner_id:
# Check if user is already a member
existing_membership = UserCompany.query.filter_by(
user_id=owner_id,
company_id=new_company.id
).first()
if not existing_membership:
membership = UserCompany(
user_id=owner_id,
company_id=new_company.id,
role='owner'
)
db.session.add(membership)
db.session.commit()
return jsonify({
'success': True,
'organization': {
'id': new_company.id,
'name': new_company.name,
'industry': new_company.industry,
'annual_revenue': new_company.annual_revenue,
'address': new_company.address,
'city': new_company.city,
'state': new_company.state,
'zip_code': new_company.zip_code,
'website': new_company.website,
'stripe_customer_id': new_company.stripe_customer_id,
}
}), 201
@admin_api_bp.route('/api/partner/organizations', methods=['GET'])
@require_auth_json()
def api_partner_organizations():
"""Return list of organizations (companies) for the partner.
Partner users see only companies they have access to.
Each org includes the user's role within that company.
"""
from app.models import Company, UserCompany, Project, KPIValue
pagination, meta = paginate_query(
UserCompany.query.filter_by(user_id=current_user.id).order_by(UserCompany.company_id)
)
user_companies = pagination.items
organizations = []
for uc in user_companies:
company = uc.company
if not company:
continue
# Get active projects count
active_projects = Project.query.filter_by(
company_id=company.id,
status='active'
).count()
# Build slug from company name
slug = company.name.lower().replace(' ', '-').replace('&', 'and').replace(',', '')
# Dynamic tier from settings_json
tier = (company.settings_json or {}).get('tier', 'starter')
organizations.append({
'id': company.id,
'name': company.name,
'slug': slug,
'industry': company.industry or 'Home Services',
'status': 'active',
'subStatus': 'active',
'tier': tier,
'groupId': None,
'referrerPartnerId': None,
'annual_revenue': company.annual_revenue or 0,
'target_revenue': company.target_revenue or 0,
'revenue_growth': ((company.annual_revenue or 0) - (company.target_revenue or 0)) / max(company.target_revenue or 1, 1) * 100,
'active_projects': active_projects,
'size': company.size or 'Small',
'location': f"{company.city or ''}, {company.state or ''}".strip(', '),
'website': company.website or '',
'commission_rate': 0.15,
'last_updated': company.updated_at.isoformat() if company.updated_at else None,
'companyRole': uc.role, # NEW: user's role in this company
})
return jsonify({
'items': organizations,
'total': meta['total'],
'nextCursor': None
})
@admin_api_bp.route('/api/partner/organizations', methods=['POST'])
@require_auth_json()
@require_csrf
def api_partner_create_organization():
"""Create a new organization — owner role required."""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
# Verify owner-level permission for creating organizations
from app.models import UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='create_org')
if error:
return error
name = data.get('name', '').strip()
if not name:
return jsonify({'error': 'Organization name is required'}), 400
new_company = Company(
name=name,
industry=data.get('industry', ''),
size=data.get('size', ''),
)
db.session.add(new_company)
db.session.flush()
# Add the creator as owner
new_uc = UserCompany(
user_id=current_user.id,
company_id=new_company.id,
role='owner',
)
db.session.add(new_uc)
db.session.commit()
return jsonify({
'success': True,
'organization': {
'id': new_company.id,
'name': new_company.name,
'companyRole': 'owner',
}
}), 201
@admin_api_bp.route('/api/partner/organizations/<org_id>/activity', methods=['GET'])
@require_auth_json()
def api_partner_organization_activity(org_id):
"""Return recent activity for a specific organization.
Requires at least viewer role in the organization.
"""
from app.models import ActivityLog
# Verify user has access to this organization (at least viewer)
err, membership = _check_company_access(org_id, min_permission='view_reports')
if err:
return err
# Get recent activity logs
activities = ActivityLog.query.filter_by(
company_id=org_id
).order_by(ActivityLog.created_at.desc()).limit(20).all()
activity_list = []
for a in activities:
activity_list.append({
'id': str(a.id),
'type': a.activity_type or 'activity',
'description': a.description or '',
'timestamp': a.created_at.isoformat() if a.created_at else None,
'user_id': str(a.user_id) if a.user_id else None
})
return jsonify({
'organization_id': org_id,
'organization_name': membership.company.name,
'activities': activity_list,
'total': len(activity_list)
})
"""Super-admin stub endpoints for SPA sections."""
@admin_api_bp.route('/api/super-admin', methods=['GET'])
@require_super_admin()
def api_super_admin_root():
"""Super-admin overview — SPA calls /api/super-admin for dashboard data."""
from app.models import Company, UserCompany
user_companies = UserCompany.query.filter_by(user_id=current_user.id).all()
company_ids = [uc.company_id for uc in user_companies]
companies = Company.query.filter(Company.id.in_(company_ids)).all() if company_ids else []
total_revenue = sum(c.annual_revenue or 0 for c in companies)
avg_revenue = total_revenue / len(companies) if companies else 0
active = len([c for c in companies if getattr(c, 'status', None) != 'suspended'])
return jsonify({
'totalOrganizations': len(companies),
'totalActive': active,
'totalSuspended': len(companies) - active,
'totalUsers': 1,
'totalRevenue': total_revenue,
'avgRevenue': round(avg_revenue, 2),
'recentActivity': [],
'systemHealth': {
'emailFailures': 0,
'integrationErrors': 0,
'activeAlerts': 0,
'lastSync': None
}
})
@admin_api_bp.route('/api/super-admin/overview', methods=['GET'])
@require_super_admin()
def api_super_admin_overview():
"""Overview dashboard stats — super admin sees ALL companies, not just their own."""
from app.models import Company, UserCompany, User
companies, meta = paginate_query(Company.query.order_by(Company.id), per_page=200)
total_users = User.query.count()
return jsonify({
'totalOrganizations': meta['total'],
'totalUsers': total_users,
'totalActive': len(companies),
'totalSuspended': 0,
'totalRevenue': sum(c.annual_revenue or 0 for c in companies),
'avgRevenue': (sum(c.annual_revenue or 0 for c in companies) / max(len(companies), 1)),
'recentActivity': []
})
@admin_api_bp.route('/api/super-admin/audit', methods=['GET'])
@require_super_admin()
def api_super_admin_audit():
"""Audit log entries."""
limit = int(request.args.get('limit', 50))
entity_type = request.args.get('entityType')
actor_email = request.args.get('actorEmail')
from app.models import ActivityLog
query = ActivityLog.query
if entity_type:
query = query.filter_by(activity_type=entity_type)
if actor_email:
from app.models import User
user = User.query.filter(User.email.ilike(f'%{actor_email}%')).first()
if user:
query = query.filter_by(user_id=user.id)
logs = query.order_by(ActivityLog.created_at.desc()).limit(limit).all()
items = [{
'id': str(l.id),
'action': l.activity_type or 'unknown',
'entityType': l.activity_type or 'unknown',
'entityId': str(l.company_id) if l.company_id else None,
'actorEmail': current_user.email,
'timestamp': l.created_at.isoformat() if l.created_at else None,
'description': l.description or ''
} for l in logs]
return jsonify({'items': items, 'total': len(items), 'nextCursor': None})
"""Public demo request submission endpoint."""
def validate_email(email):
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return bool(re.match(pattern, email))
def validate_phone(phone):
digits = re.sub(r'[^0-9]', '', phone)
return 7 <= len(digits) <= 15
@admin_api_bp.route('/api/leads/demo-request', methods=['POST'])
@limiter.limit("5 per hour")
def api_demo_request():
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
name = data.get('name', '').strip()
email = data.get('email', '').strip()
company = data.get('company', '').strip()
phone = data.get('phone', '').strip()
message = data.get('message', '').strip()
if not name or not email:
return jsonify({'error': 'Name and email are required'}), 400
if not validate_email(email):
return jsonify({'error': 'Invalid email format'}), 400
if phone and not validate_phone(phone):
return jsonify({'error': 'Invalid phone format'}), 400
if len(name) > 100:
name = name[:100]
if len(company) > 200:
company = company[:200]
if len(message) > 2000:
message = message[:2000]
from app.models import db, DemoRequest
from app.utils.email_service import send_demo_notification
demo_request = DemoRequest(
name=name,
email=email,
company=company or None,
phone=phone or None,
message=message or None,
status='new',
)
db.session.add(demo_request)
db.session.commit()
send_demo_notification(
name=name,
email=email,
company=company,
phone=phone,
message=message,
)
return jsonify({
'success': True,
'message': 'Demo request submitted successfully',
'id': demo_request.id
}), 201
@admin_api_bp.route('/api/csrf-token', methods=['GET'])
def get_csrf_token():
"""Return CSRF token for SPA forms."""
from app.utils.csrf import generate_csrf_token
token = generate_csrf_token()
from flask import session
session['csrf_token'] = token
return jsonify({'token': token})
@admin_api_bp.route('/api/super-admin/leads', methods=['GET'])
@require_super_admin()
def api_super_admin_leads():
"""Leads list."""
from app.models import DemoRequest, db
pagination, meta = paginate_query(
db.session.query(DemoRequest).order_by(DemoRequest.created_at.desc())
)
requests = pagination.items
items = [{
'id': r.id,
'name': r.name,
'email': r.email,
'company': r.company or '',
'phone': r.phone or '',
'message': r.message or '',
'status': r.status,
'createdAt': r.created_at.isoformat() if r.created_at else None,
'assignedTo': None
} for r in requests]
return jsonify({'items': items, 'total': meta['total'], 'nextCursor': None})
@admin_api_bp.route('/api/super-admin/lead-recipients', methods=['GET'])
@require_super_admin()
def api_super_admin_lead_recipients():
"""Lead recipients (partners/companies that can receive leads)."""
from app.models import Company, UserCompany
companies = Company.query.join(UserCompany, Company.id == UserCompany.company_id).filter(
UserCompany.user_id == current_user.id
).all()
recipients = [{
'id': c.id,
'name': c.name,
'email': f'leads@{c.name.lower().replace(" ", "")}.com',
'type': 'organization'
} for c in companies]
return jsonify({'items': recipients})
@admin_api_bp.route('/api/super-admin/email-failures', methods=['GET'])
@require_super_admin()
def api_super_admin_email_failures():
"""Email delivery failures."""
status = request.args.get('status', 'open')
return jsonify({'items': [], 'total': 0, 'filter': status, 'nextCursor': None})
@admin_api_bp.route('/api/super-admin/users', methods=['POST'])
@require_super_admin()
@require_csrf
@limiter.limit("10/hour")
def api_super_admin_create_user():
"""Create a new user. Only super admins can call this.
Body: { email, full_name?, password, role?, company_name?, company_role? }
"""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
email = str(data.get('email', '')).strip()
full_name = str(data.get('full_name', '')).strip()
password = data.get('password', '')
role = str(data.get('role', 'user')).strip()
company_name = str(data.get('company_name', '')).strip()
company_role = str(data.get('company_role', 'member')).strip().lower()
if not email or not password:
return jsonify({'error': 'Email and password are required'}), 400
if len(password) < 8:
return jsonify({'error': 'Password must be at least 8 characters'}), 400
valid_roles = ['super_admin', 'admin', 'partner', 'user']
if role not in valid_roles:
return jsonify({'error': f'Invalid role. Must be one of: {", ".join(valid_roles)}'}), 400
valid_company_roles = ['owner', 'admin', 'manager', 'rep', 'viewer', 'member']
if company_role not in valid_company_roles:
return jsonify({'error': f'Invalid company role. Must be one of: {", ".join(valid_company_roles)}'}), 400
from app.models import User, Company, UserCompany, db
# Check if user already exists
existing = User.query.filter_by(email=email).first()
if existing:
return jsonify({'error': 'Email already registered'}), 409
new_user = User(
email=email,
full_name=full_name or email.split('@')[0],
role=role,
)
new_user.set_password(password)
db.session.add(new_user)
db.session.flush() # Get the ID
# If a company name was provided, create/find it and add membership
company_id = None
if company_name:
company = Company.query.filter(Company.name.ilike(company_name)).first()
if not company:
company = Company(name=company_name)
db.session.add(company)
db.session.flush()
company_id = company.id
# Check seat limit before adding membership
from app.utils.seats import check_seat_limit
seat_error, seat_usage = check_seat_limit(company_id)
if seat_error:
db.session.rollback()
return seat_error
membership = UserCompany(
user_id=new_user.id,
company_id=company_id,
role=company_role,
)
db.session.add(membership)
db.session.commit()
return jsonify({
'success': True,
'user': {
'id': new_user.id,
'email': new_user.email,
'full_name': new_user.full_name,
'role': new_user.role,
'is_active': new_user.is_active,
}
}), 201
@admin_api_bp.route('/api/super-admin/users/<user_id>', methods=['DELETE'])
@require_super_admin()
@require_csrf
@limiter.limit("10/hour")
def api_super_admin_delete_user(user_id):
"""Delete a user. Only super admins can call this.
Prevents deleting yourself. Cascades to user_company memberships.
"""
if int(user_id) == current_user.id:
return jsonify({'error': 'You cannot delete your own account'}), 400
from app.models import User, db
user = db.session.get(User, int(user_id))
if not user:
return jsonify({'error': 'User not found'}), 404
deleted_email = user.email
db.session.delete(user)
db.session.commit()
return jsonify({
'success': True,
'message': f'User {deleted_email} deleted',
})
@admin_api_bp.route('/api/super-admin/users', methods=['GET'])
@require_super_admin()
def api_super_admin_users():
"""All users list with optional role filter, including primary company role."""
limit = int(request.args.get('limit', 25))
role = request.args.get('role')
from app.models import User, UserCompany
query = User.query
if role:
query = query.filter_by(role=role)
users = query.limit(limit).all()
items = []
for u in users:
uc = UserCompany.query.filter_by(user_id=u.id).first()
items.append({
'id': u.id,
'email': u.email,
'fullName': u.full_name or u.email.split('@')[0],
'role': u.role or 'user',
'companyRole': uc.role if uc else None,
'status': 'active',
'lastLogin': u.last_login.isoformat() if u.last_login else None
})
return jsonify({'items': items, 'total': len(items), 'nextCursor': None})
@admin_api_bp.route('/api/super-admin/users/<user_id>/role', methods=['PUT'])
@require_super_admin()
@require_csrf
@limiter.limit("20/hour")
def api_super_admin_update_user_role(user_id):
"""Update a user's role. Only super admins can call this.
Body: { role: 'super_admin'|'partner'|'user' }
"""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
new_role = data.get('role', '').strip()
valid_roles = ['super_admin', 'admin', 'partner', 'user']
if new_role not in valid_roles:
return jsonify({'error': f'Invalid role. Must be one of: {", ".join(valid_roles)}'}), 400
from app.models import User, db
user = db.session.get(User, user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
old_role = user.role
user.role = new_role
db.session.commit()
return jsonify({
'success': True,
'user': {
'id': user.id,
'email': user.email,
'full_name': user.full_name,
'role': user.role,
'is_active': user.is_active,
'lastLogin': user.last_login.isoformat() if user.last_login else None
},
'previousRole': old_role
})
@admin_api_bp.route('/api/super-admin/users/<user_id>/company-role', methods=['PUT'])
@require_super_admin()
@require_csrf
@limiter.limit("20/hour")
def api_super_admin_update_user_company_role(user_id):
"""Update a user's company role (owner, admin, manager, rep, viewer). Only super admins can call this.
Body: { role: 'owner'|'admin'|'manager'|'rep'|'viewer', company_id?: int }
If company_id is provided and no UserCompany exists, creates the membership.
If company_id is provided and UserCompany exists with different company, updates to new company.
If no company_id, updates the first/primary UserCompany record for the user.
"""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
new_role = data.get('role', '').strip().lower()
valid_company_roles = ['owner', 'admin', 'manager', 'rep', 'viewer']
if new_role not in valid_company_roles:
return jsonify({'error': f'Invalid company role. Must be one of: {", ".join(valid_company_roles)}'}), 400
company_id = data.get('company_id')
from app.models import User, UserCompany, db
user = db.session.get(User, user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
uc = UserCompany.query.filter_by(user_id=user.id).first()
# If company_id provided and no existing membership, create it
if not uc and company_id:
uc = UserCompany(user_id=user.id, company_id=company_id, role=new_role)
db.session.add(uc)
db.session.commit()
return jsonify({
'success': True,
'created': True,
'user': {
'id': user.id,
'email': user.email,
'full_name': user.full_name,
'role': user.role,
'companyRole': uc.role,
'companyId': uc.company_id,
},
})
# If company_id provided and existing membership is for different company, update
if company_id and uc and uc.company_id != company_id:
uc.company_id = company_id
uc.role = new_role
db.session.commit()
return jsonify({
'success': True,
'user': {
'id': user.id,
'email': user.email,
'full_name': user.full_name,
'role': user.role,
'companyRole': uc.role,
'companyId': uc.company_id,
},
})
# Existing membership, update role only
if uc:
old_role = uc.role
uc.role = new_role
db.session.commit()
return jsonify({
'success': True,
'user': {
'id': user.id,
'email': user.email,
'full_name': user.full_name,
'role': user.role,
'companyRole': uc.role,
},
'previousCompanyRole': old_role
})
# No membership and no company_id provided
return jsonify({'error': 'User has no company membership. Provide company_id to create one.'}), 404
@admin_api_bp.route('/api/super-admin/users/<user_id>', methods=['PUT'])
@require_super_admin()
@require_csrf
@limiter.limit("20/hour")
def api_super_admin_update_user(user_id):
"""Update any user field. Only super admins can call this.
Body can include: full_name, email, role, is_active
"""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
from app.models import User, db
user = db.session.get(User, user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
if 'full_name' in data:
user.full_name = str(data['full_name']).strip()
if 'email' in data:
new_email = str(data['email']).strip()
existing = User.query.filter(User.email == new_email, User.id != user_id).first()
if existing:
return jsonify({'error': 'Email already in use'}), 409
user.email = new_email
if 'role' in data:
valid_roles = ['super_admin', 'admin', 'partner', 'user']
if data['role'] not in valid_roles:
return jsonify({'error': f'Invalid role. Must be one of: {", ".join(valid_roles)}'}), 400
user.role = data['role']
if 'is_active' in data:
user.is_active = bool(data['is_active'])
db.session.commit()
return jsonify({
'success': True,
'user': {
'id': user.id,
'email': user.email,
'full_name': user.full_name,
'role': user.role,
'is_active': user.is_active,
'lastLogin': user.last_login.isoformat() if user.last_login else None
}
})
@admin_api_bp.route('/api/super-admin/analytics', methods=['GET'])
@require_super_admin()
def api_super_admin_analytics():
"""Analytics overview."""
from app.models import Company, UserCompany
companies = Company.query.join(UserCompany, Company.id == UserCompany.company_id).filter(
UserCompany.user_id == current_user.id
).all()
return jsonify({
'totalOrganizations': len(companies),
'totalRevenue': sum(c.annual_revenue or 0 for c in companies),
'avgRevenue': (sum(c.annual_revenue or 0 for c in companies) / max(len(companies), 1)),
'revenueByIndustry': {},
'revenueTrend': [],
'userGrowth': [],
'topOrganizations': [{
'id': c.id,
'name': c.name,
'revenue': c.annual_revenue or 0
} for c in sorted(companies, key=lambda x: x.annual_revenue or 0, reverse=True)[:10]]
})
@admin_api_bp.route('/api/partner/commissions', methods=['GET'])
@require_auth_json()
def api_partner_commissions():
"""Return commission summary for the partner.
Real data from PartnerCommission, CommissionRecord, and Payout models.
Falls back to legacy Company.annual_revenue if no partnerships exist yet.
"""
from app.models import Company, UserCompany, PartnerCommission, CommissionRecord, Payout
# Check user has at least viewer-level company role
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='view_reports')
if error:
return error
# Find all active partnerships for this partner
partnerships = PartnerCommission.query.filter_by(
partner_id=current_user.id,
status='active'
).all()
if partnerships:
# --- Real commission tracking ---
partnership_ids = [p.id for p in partnerships]
records = CommissionRecord.query.filter(
CommissionRecord.partner_commission_id.in_(partnership_ids)
).all()
# Build a lookup: partnership_id -> company
company_map = {p.id: p.company for p in partnerships}
rate_map = {p.id: p.commission_rate for p in partnerships}
# Aggregate by status
total_commission = sum(r.commission_amount for r in records)
paid_commission = sum(r.commission_amount for r in records if r.status == 'paid')
pending_commission = total_commission - paid_commission
# Monthly breakdown grouped by company
# Group records by (partner_commission_id, period)
from collections import defaultdict
by_company = defaultdict(list)
for r in records:
by_company[r.partner_commission_id].append(r)
monthly_breakdown = []
for pc_id, pc_records in by_company.items():
company = company_map.get(pc_id)
if not company:
continue
# Use the most recent record per company for the summary row
latest = max(pc_records, key=lambda r: r.period)
monthly_breakdown.append({
'organization_id': company.id,
'organization_name': company.name,
'monthly_revenue': latest.revenue_amount,
'commission_amount': latest.commission_amount,
'status': latest.status,
'period': latest.period,
'revenue_confidence': latest.revenue_confidence,
})
# Use the rate from the first partnership as the "default" rate
default_rate = partnerships[0].commission_rate if partnerships else 0.15
# Next scheduled payout
next_payout = Payout.query.filter_by(
partner_id=current_user.id,
status='scheduled'
).order_by(Payout.scheduled_date).first()
next_payout_date = next_payout.scheduled_date.strftime('%Y-%m-%d') if next_payout else None
return jsonify({
'total_commission': round(total_commission, 2),
'commission_rate': default_rate,
'pending': round(pending_commission, 2),
'paid': round(paid_commission, 2),
'organizations': len(partnerships),
'monthly_breakdown': monthly_breakdown,
'next_payout_date': next_payout_date,
'currency': 'USD',
})
# --- Fallback: legacy mode (no partnerships configured yet) ---
companies = Company.query.join(
UserCompany, Company.id == UserCompany.company_id
).filter(
UserCompany.user_id == current_user.id
).all()
total_revenue = sum(c.annual_revenue or 0 for c in companies)
commission_rate = 0.15
total_commission = total_revenue * commission_rate
monthly_commissions = []
for c in companies:
monthly_commissions.append({
'organization_id': c.id,
'organization_name': c.name,
'monthly_revenue': (c.annual_revenue or 0) / 12,
'commission_amount': (c.annual_revenue or 0) / 12 * commission_rate,
'status': 'pending',
'period': None,
'revenue_confidence': 'estimated',
})
return jsonify({
'total_commission': round(total_commission, 2),
'commission_rate': commission_rate,
'pending': round(total_commission, 2),
'paid': 0.0,
'organizations': len(companies),
'monthly_breakdown': monthly_commissions,
'next_payout_date': None,
'currency': 'USD',
})
@admin_api_bp.route('/api/partner/team', methods=['GET'])
@require_auth_json()
def api_partner_team_list():
"""List team members of the partner's primary company.
Requires at least manager role (manage_team permission).
"""
from app.models import User, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='manage_team')
if error:
return error
company_id = uc.company_id
# Get all members of this company
pagination, meta = paginate_query(
UserCompany.query.filter_by(company_id=company_id).order_by(UserCompany.user_id)
)
members = pagination.items
team = []
for m in members:
if m.user:
team.append({
'id': m.user.id,
'name': m.user.full_name or m.user.email,
'email': m.user.email,
'role': m.role,
'status': 'active',
'lastLogin': m.user.last_login.isoformat() if m.user.last_login else None,
})
return jsonify({'items': team, 'total': meta['total']})
@admin_api_bp.route('/api/partner/team', methods=['POST'])
@require_auth_json()
@require_csrf
def api_partner_team_add():
"""Add a new team member to the partner's primary company.
Requires at least admin role (manage_team permission).
"""
from app.models import User, UserCompany, db
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='manage_team')
if error:
return error
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
email = data.get('email', '').strip()
if not email:
return jsonify({'error': 'Email is required'}), 400
new_role = data.get('role', 'viewer')
if new_role not in COMPANY_ROLE_LEVELS:
return jsonify({'error': f'Invalid role. Must be one of: {list(COMPANY_ROLE_LEVELS.keys())}'}), 400
# Check if user already exists
user = User.query.filter_by(email=email).first()
if not user:
# Create new user
user = User(
email=email,
password='', # Would be set via invitation in production
full_name=data.get('name', email.split('@')[0]),
is_active=True,
)
db.session.add(user)
db.session.flush()
# Add to company
existing = UserCompany.query.filter_by(
user_id=user.id, company_id=uc.company_id
).first()
if existing:
return jsonify({'error': 'User is already a member of this company'}), 409
new_uc = UserCompany(
user_id=user.id,
company_id=uc.company_id,
role=new_role,
)
db.session.add(new_uc)
db.session.commit()
return jsonify({
'success': True,
'member': {
'id': user.id,
'name': user.full_name or user.email,
'email': user.email,
'role': new_role,
}
}), 201
@admin_api_bp.route('/api/partner/team/<int:user_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def api_partner_team_update(user_id):
"""Update a team member's role.
Requires at least admin role (manage_team permission).
"""
from app.models import User, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='manage_team')
if error:
return error
data = request.get_json()
new_role = data.get('role')
if not new_role or new_role not in COMPANY_ROLE_LEVELS:
return jsonify({'error': f'Invalid role. Must be one of: {list(COMPANY_ROLE_LEVELS.keys())}'}), 400
member = UserCompany.query.filter_by(
user_id=user_id, company_id=uc.company_id
).first()
if not member:
return jsonify({'error': 'Team member not found'}), 404
# Owners cannot demote themselves or other owners
if member.role == 'owner' and current_user.id != member.user_id:
return jsonify({'error': 'Cannot change an owner\'s role'}), 403
member.role = new_role
db.session.commit()
return jsonify({
'success': True,
'member': {
'id': member.user_id,
'role': new_role,
}
})
@admin_api_bp.route('/api/partner/team/<int:user_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def api_partner_team_remove(user_id):
"""Remove a team member from the company.
Requires at least admin role (manage_team permission).
"""
from app.models import User, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='manage_team')
if error:
return error
member = UserCompany.query.filter_by(
user_id=user_id, company_id=uc.company_id
).first()
if not member:
return jsonify({'error': 'Team member not found'}), 404
# Cannot remove yourself if you're the only owner
if member.user_id == current_user.id:
owners = UserCompany.query.filter_by(company_id=uc.company_id, role='owner').count()
if member.role == 'owner' and owners <= 1:
return jsonify({'error': 'Cannot remove yourself — at least one owner is required'}), 400
db.session.delete(member)
db.session.commit()
return jsonify({'success': True, 'message': 'Team member removed'})
# ============================================================================
# Revenue Leak Detectors — settings API (Phase 0 auto-detection foundation)
# ============================================================================
_VALID_LEAK_SEVERITIES = ('low', 'medium', 'high', 'critical')
@admin_api_bp.route('/api/company/<company_id>/revenue-leaks/detectors', methods=['GET'])
@require_auth_json()
def list_leak_detectors(company_id):
"""List available leak detectors with per-company settings merged in."""
from app.services.leak_detectors import registry
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
company = db.session.get(Company, company_id)
if not company:
return jsonify({'error': 'Company not found'}), 404
return jsonify({'detectors': registry.describe(company)})
@admin_api_bp.route('/api/company/<company_id>/revenue-leaks/detectors', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_leak_detectors(company_id):
"""Update per-company leak detector settings.
Body: { "detectors": { "<detector_id>": { "enabled": bool,
"severity_override": str|null, "threshold_override": obj|null } } }
"""
from app.services.leak_detectors import registry, SETTINGS_KEY
from sqlalchemy.orm.attributes import flag_modified
error_response, membership = _check_company_access(company_id, min_role='admin')
if error_response:
return error_response
company = db.session.get(Company, company_id)
if not company:
return jsonify({'error': 'Company not found'}), 404
data = request.get_json(silent=True) or {}
updates = data.get('detectors')
if not isinstance(updates, dict):
return jsonify({'error': 'detectors must be an object keyed by detector id'}), 400
settings = dict(company.settings_json or {})
detector_settings = dict(settings.get(SETTINGS_KEY) or {})
for detector_id, conf in updates.items():
if registry.get(detector_id) is None:
return jsonify({'error': f'Unknown detector: {detector_id}'}), 400
if not isinstance(conf, dict):
return jsonify({'error': f'Settings for {detector_id} must be an object'}), 400
entry = dict(detector_settings.get(detector_id) or {})
if 'enabled' in conf:
if not isinstance(conf['enabled'], bool):
return jsonify({'error': f'{detector_id}: enabled must be a boolean'}), 400
entry['enabled'] = conf['enabled']
if 'severity_override' in conf:
sev = conf['severity_override']
if sev is not None and sev not in _VALID_LEAK_SEVERITIES:
return jsonify({'error': f'{detector_id}: severity_override must be one of {", ".join(_VALID_LEAK_SEVERITIES)} or null'}), 400
entry['severity_override'] = sev
if 'threshold_override' in conf:
thr = conf['threshold_override']
if thr is not None and not isinstance(thr, dict):
return jsonify({'error': f'{detector_id}: threshold_override must be an object or null'}), 400
entry['threshold_override'] = thr
detector_settings[detector_id] = entry
settings[SETTINGS_KEY] = detector_settings
company.settings_json = settings
flag_modified(company, 'settings_json')
db.session.commit()
return jsonify({'success': True, 'detectors': registry.describe(company)})