"""Business blueprint — revenue leaks, optimization moves, forecasts, coaching assignments/scorecards.
Split from api_proxy.py monolith for maintainability (July 2026).
These are company-scoped CRUD routes requiring editor+ role for mutations.
Imports shared utilities from api_proxy.
"""
from flask import Blueprint, request, jsonify
from flask_login import current_user
from datetime import datetime, timezone
from app.routes.api_proxy import (
require_auth_json,
require_csrf,
_check_company_access,
)
from app.utils.pagination import paginate_query
from app.models import db
business_api_bp = Blueprint('business_api', __name__)
# ============================================================================
# Revenue Leaks API
# ============================================================================
@business_api_bp.route('/api/company/<company_id>/revenue-leaks', methods=['GET'])
@require_auth_json()
def list_revenue_leaks(company_id):
"""List revenue leaks for a company."""
from app.models import RevenueLeak
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
severity_filter = request.args.get('severity')
resolved_filter = request.args.get('resolved')
query = RevenueLeak.query.filter_by(company_id=company_id)
if severity_filter:
query = query.filter_by(severity=severity_filter)
if resolved_filter is not None:
query = query.filter_by(resolved=resolved_filter.lower() == 'true')
pagination, meta = paginate_query(query.order_by(RevenueLeak.created_at.desc()))
leaks = pagination.items
return jsonify({
**meta,
'revenue_leaks': [{
'id': leak.id,
'source': leak.source,
'description': leak.description,
'estimated_loss': leak.estimated_loss,
'severity': leak.severity,
'resolved': leak.resolved,
'resolved_at': leak.resolved_at.isoformat() if leak.resolved_at else None,
'resolution_notes': leak.resolution_notes,
'detected_at': leak.detected_at.isoformat() if leak.detected_at else None,
'metadata_json': leak.metadata_json,
'detection_type': leak.detection_type,
'detector_id': leak.detector_id,
'source_connector': leak.source_connector,
'created_at': leak.created_at.isoformat() if leak.created_at else None,
} for leak in leaks]
})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks', methods=['POST'])
@require_auth_json()
@require_csrf
def create_revenue_leak(company_id):
"""Create a new revenue leak."""
from app.models import db, RevenueLeak
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
source = data.get('source', '').strip()
if not source:
return jsonify({'error': 'Source is required'}), 400
severity = data.get('severity', 'medium')
if severity not in ('low', 'medium', 'high', 'critical'):
return jsonify({'error': 'Severity must be low, medium, high, or critical'}), 400
estimated_loss = data.get('estimated_loss')
if estimated_loss is not None:
try:
estimated_loss = float(estimated_loss)
except (ValueError, TypeError):
return jsonify({'error': 'estimated_loss must be a number'}), 400
leak = RevenueLeak(
company_id=company_id,
source=source,
description=data.get('description', '').strip(),
estimated_loss=estimated_loss,
severity=severity,
metadata_json=data.get('metadata_json') or {},
)
db.session.add(leak)
db.session.commit()
return jsonify({
'success': True,
'revenue_leak': {
'id': leak.id,
'source': leak.source,
'description': leak.description,
'estimated_loss': leak.estimated_loss,
'severity': leak.severity,
'resolved': leak.resolved,
'detected_at': leak.detected_at.isoformat() if leak.detected_at else None,
'created_at': leak.created_at.isoformat() if leak.created_at else None,
}
}), 201
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/<leak_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_revenue_leak(company_id, leak_id):
"""Update a revenue leak."""
from app.models import db, RevenueLeak
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
leak = db.session.get(RevenueLeak, leak_id)
if not leak or leak.company_id != company_id:
return jsonify({'error': 'Revenue leak not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
if 'source' in data:
leak.source = data['source'].strip()
if 'description' in data:
leak.description = data['description'].strip()
if 'estimated_loss' in data:
try:
leak.estimated_loss = float(data['estimated_loss'])
except (ValueError, TypeError):
return jsonify({'error': 'estimated_loss must be a number'}), 400
if 'severity' in data:
if data['severity'] not in ('low', 'medium', 'high', 'critical'):
return jsonify({'error': 'Severity must be low, medium, high, or critical'}), 400
leak.severity = data['severity']
if 'resolution_notes' in data:
leak.resolution_notes = data['resolution_notes']
if 'metadata_json' in data:
leak.metadata_json = data['metadata_json']
db.session.commit()
return jsonify({
'success': True,
'revenue_leak': {
'id': leak.id,
'source': leak.source,
'description': leak.description,
'estimated_loss': leak.estimated_loss,
'severity': leak.severity,
'resolved': leak.resolved,
'resolved_at': leak.resolved_at.isoformat() if leak.resolved_at else None,
'resolution_notes': leak.resolution_notes,
'created_at': leak.created_at.isoformat() if leak.created_at else None,
}
})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/<leak_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def delete_revenue_leak(company_id, leak_id):
"""Delete a revenue leak."""
from app.models import db, RevenueLeak
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
leak = db.session.get(RevenueLeak, leak_id)
if not leak or leak.company_id != company_id:
return jsonify({'error': 'Revenue leak not found'}), 404
db.session.delete(leak)
db.session.commit()
return jsonify({'success': True, 'message': 'Revenue leak deleted'})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/<leak_id>/resolve', methods=['PATCH'])
@require_auth_json()
@require_csrf
def resolve_revenue_leak(company_id, leak_id):
"""Mark a revenue leak as resolved."""
from app.models import db, RevenueLeak
from datetime import datetime, timezone
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
leak = db.session.get(RevenueLeak, leak_id)
if not leak or leak.company_id != company_id:
return jsonify({'error': 'Revenue leak not found'}), 404
data = request.get_json() or {}
leak.resolved = True
leak.resolved_at = datetime.now(timezone.utc)
if 'resolution_notes' in data:
leak.resolution_notes = data['resolution_notes']
db.session.commit()
return jsonify({
'success': True,
'revenue_leak': {
'id': leak.id,
'resolved': leak.resolved,
'resolved_at': leak.resolved_at.isoformat(),
'resolution_notes': leak.resolution_notes,
}
})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/scan', methods=['POST'])
@require_auth_json()
@require_csrf
def scan_revenue_leaks(company_id):
"""Trigger an immediate auto leak-detection scan for this company.
Optional JSON body: { "detectors": ["qb_overdue_30d", ...] } to scan a
subset; omit to run all detectors enabled for the company.
"""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json(silent=True) or {}
detector_ids = data.get('detectors')
if detector_ids is not None:
if not isinstance(detector_ids, list) or not all(isinstance(d, str) for d in detector_ids):
return jsonify({'error': 'detectors must be a list of detector id strings'}), 400
from app.services.leak_detectors import registry
unknown = [d for d in detector_ids if registry.get(d) is None]
if unknown:
return jsonify({'error': f"Unknown detector(s): {', '.join(unknown)}"}), 400
from app.services.leak_detectors import run_all_detectors
result = run_all_detectors(company_id, detector_ids=detector_ids)
# Phase 4: send alerts for new leaks
alerts = {}
try:
from app.services.leak_alerts import send_new_leak_alerts
alerts = send_new_leak_alerts(
company_id,
result.get('leaks', []),
result,
)
except Exception as exc:
alerts = {'error': str(exc)}
status_code = 200
payload = {
'success': not result.get('errors'),
'scanned': result.get('scanned', 0),
'new_leaks': result.get('new_leaks', 0),
'updated_leaks': result.get('updated_leaks', 0),
'already_resolved': result.get('already_resolved', 0),
'leaks': result.get('leaks', []),
'alerts': alerts,
}
if result.get('errors'):
payload['errors'] = result['errors']
return jsonify(payload), status_code
# ============================================================================
# Phase 4 — Remediation & Proactive Ops API
# ============================================================================
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/<leak_id>/remediation', methods=['GET'])
@require_auth_json()
def get_leak_remediation(company_id, leak_id):
"""Get remediation suggestions for a specific revenue leak."""
from app.models import db, RevenueLeak
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
leak = db.session.get(RevenueLeak, leak_id)
if not leak or leak.company_id != company_id:
return jsonify({'error': 'Revenue leak not found'}), 404
from app.services.leak_remediation import get_remediation_suggestions
leak_dict = {
'detector_id': leak.detector_id,
'severity': leak.severity,
'estimated_loss': leak.estimated_loss,
'source': leak.source,
'description': leak.description,
}
suggestions = get_remediation_suggestions(leak_dict)
return jsonify({
'leak_id': leak_id,
'detector_id': leak.detector_id,
'suggestions': suggestions,
})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/<leak_id>/remediation', methods=['POST'])
@require_auth_json()
@require_csrf
def apply_leak_remediation(company_id, leak_id):
"""Apply a remediation suggestion by creating an OptimizationMove.
Body: { "suggestion_index": 0, "title": "...", "description": "..." }
or { "suggestion": { full suggestion dict } }
"""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
from app.models import RevenueLeak
leak = db.session.get(RevenueLeak, leak_id)
if not leak or leak.company_id != company_id:
return jsonify({'error': 'Revenue leak not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
from app.services.leak_remediation import get_remediation_suggestions, apply_suggestion
leak_dict = {
'detector_id': leak.detector_id,
'severity': leak.severity,
'estimated_loss': leak.estimated_loss,
'source': leak.source,
'description': leak.description,
}
suggestions = get_remediation_suggestions(leak_dict)
# Get the suggestion to apply
suggestion = None
if 'suggestion_index' in data:
idx = data['suggestion_index']
if 0 <= idx < len(suggestions):
suggestion = suggestions[idx]
else:
return jsonify({'error': f'Invalid suggestion_index: {idx}'}), 400
elif 'suggestion' in data:
suggestion = data['suggestion']
else:
return jsonify({
'error': 'Provide suggestion_index or suggestion object',
'available_suggestions': suggestions,
}), 400
if suggestion is None:
return jsonify({'error': 'No suggestion to apply'}), 400
move_id = apply_suggestion(
company_id,
leak_id,
suggestion,
user_id=current_user.id if current_user.is_authenticated else None,
)
if move_id:
# Record resolution attempt
try:
from app.services.leak_resolution import record_resolution
record_resolution(
leak_id,
current_user.id if current_user.is_authenticated else 'system',
f"Applied remediation: {suggestion.get('title', 'Unknown')}",
)
except Exception:
pass
return jsonify({
'success': True,
'optimization_move_id': move_id,
'suggestion': suggestion,
})
else:
return jsonify({'error': 'Failed to apply suggestion'}), 500
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/<leak_id>/resolution-history', methods=['GET'])
@require_auth_json()
def get_resolution_history(company_id, leak_id):
"""Get the resolution history for a specific leak."""
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
from app.models import RevenueLeak
from app.services.leak_resolution import get_resolution_history
leak = db.session.get(RevenueLeak, leak_id)
if not leak or leak.company_id != company_id:
return jsonify({'error': 'Revenue leak not found'}), 404
history = get_resolution_history(leak_id)
return jsonify({
'leak_id': leak_id,
'resolution_history': history,
})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/recurring', methods=['GET'])
@require_auth_json()
def get_recurring_leaks_endpoint(company_id):
"""Get recurring leaks (same detector+source appearing multiple times)."""
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
days = request.args.get('days', 90, type=int)
from app.services.leak_resolution import get_recurring_leaks
recurring = get_recurring_leaks(company_id, days=days)
return jsonify({
'days': days,
'recurring_leaks': recurring,
})
@business_api_bp.route('/api/company/<company_id>/revenue-leaks/stats', methods=['GET'])
@require_auth_json()
def get_resolution_stats_endpoint(company_id):
"""Get resolution statistics for a company."""
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
days = request.args.get('days', 30, type=int)
from app.services.leak_resolution import get_resolution_stats
stats = get_resolution_stats(company_id, days=days)
return jsonify({
'days': days,
'stats': stats,
})
@business_api_bp.route('/api/company/<company_id>/leak-settings/severity', methods=['GET'])
@require_auth_json()
def get_severity_rules_endpoint(company_id):
"""Get severity tuning rules for a company."""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
from app.services.leak_severity import get_severity_rules
rules = get_severity_rules(company_id)
return jsonify({
'severity_rules': rules,
})
@business_api_bp.route('/api/company/<company_id>/leak-settings/severity', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_severity_rules_endpoint(company_id):
"""Update severity tuning rules for a company."""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
from app.services.leak_severity import update_severity_rules
result = update_severity_rules(company_id, data)
if 'error' in result:
return jsonify(result), 404
return jsonify(result)
@business_api_bp.route('/api/company/<company_id>/leak-settings/alerts', methods=['GET'])
@require_auth_json()
def get_alert_settings(company_id):
"""Get alert delivery settings for a company."""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
from app.models import Company, Connector
company = db.session.get(Company, company_id)
if not company:
return jsonify({'error': 'Company not found'}), 404
settings = company.settings_json or {}
alert_settings = settings.get('leak_alerts') or {}
# Check connected Slack connectors
slack_connectors = Connector.query.filter_by(
company_id=company_id,
service='slack',
status='connected',
).all()
return jsonify({
'alert_settings': alert_settings,
'slack_connected': len(slack_connectors) > 0,
'slack_workspaces': [
{
'id': c.id,
'workspace': (c.config or {}).get('team', {}).get('name', 'Unknown'),
}
for c in slack_connectors
],
})
@business_api_bp.route('/api/company/<company_id>/leak-settings/alerts', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_alert_settings(company_id):
"""Update alert delivery settings for a company.
Body: {
"slack_channel": "#revenue-alerts",
"slack_connector_id": "conn_abc123",
"severity_threshold": "medium", # only alert for this severity and above
}
"""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
from app.models import Company, Connector
company = db.session.get(Company, company_id)
if not company:
return jsonify({'error': 'Company not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
# Validate slack_connector_id if provided
if 'slack_connector_id' in data and data['slack_connector_id']:
connector = db.session.get(Connector, data['slack_connector_id'])
if not connector or connector.company_id != company_id:
return jsonify({'error': 'Invalid Slack connector ID'}), 400
settings = company.settings_json or {}
current_alerts = settings.get('leak_alerts') or {}
# Merge new settings
current_alerts.update({
k: v for k, v in data.items()
if v is not None
})
settings['leak_alerts'] = current_alerts
company.settings_json = settings
db.session.commit()
return jsonify({
'success': True,
'alert_settings': current_alerts,
})
# ============================================================================
# Optimization Moves API
# ============================================================================
@business_api_bp.route('/api/company/<company_id>/optimization-moves', methods=['GET'])
@require_auth_json()
def list_optimization_moves(company_id):
"""List optimization moves for a company."""
from app.models import OptimizationMove
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
status_filter = request.args.get('status')
query = OptimizationMove.query.filter_by(company_id=company_id)
if status_filter:
query = query.filter_by(status=status_filter)
pagination, meta = paginate_query(query.order_by(OptimizationMove.created_at.desc()))
moves = pagination.items
return jsonify({
**meta,
'optimization_moves': [{
'id': move.id,
'move_type': move.move_type,
'description': move.description,
'source_channel': move.source_channel,
'target_channel': move.target_channel,
'current_spend': move.current_spend,
'recommended_spend': move.recommended_spend,
'expected_impact': move.expected_impact,
'confidence': move.confidence,
'status': move.status,
'implemented_at': move.implemented_at.isoformat() if move.implemented_at else None,
'actual_result': move.actual_result,
'created_at': move.created_at.isoformat() if move.created_at else None,
'updated_at': move.updated_at.isoformat() if move.updated_at else None,
} for move in moves]
})
@business_api_bp.route('/api/company/<company_id>/optimization-moves', methods=['POST'])
@require_auth_json()
@require_csrf
def create_optimization_move(company_id):
"""Create a new optimization move."""
from app.models import db, OptimizationMove
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
move_type = data.get('move_type', 'budget_shift')
if move_type not in ('budget_shift', 'channel_change', 'target_audience'):
return jsonify({'error': 'move_type must be budget_shift, channel_change, or target_audience'}), 400
confidence = data.get('confidence')
if confidence is not None:
try:
confidence = float(confidence)
except (ValueError, TypeError):
return jsonify({'error': 'confidence must be a number between 0 and 1'}), 400
current_spend = data.get('current_spend')
recommended_spend = data.get('recommended_spend')
expected_impact = data.get('expected_impact')
for val in [current_spend, recommended_spend, expected_impact]:
if val is not None:
try:
float(val)
except (ValueError, TypeError):
return jsonify({'error': 'Spend and impact values must be numbers'}), 400
move = OptimizationMove(
company_id=company_id,
move_type=move_type,
description=data.get('description', '').strip(),
source_channel=data.get('source_channel', '').strip(),
target_channel=data.get('target_channel', '').strip(),
current_spend=float(current_spend) if current_spend is not None else None,
recommended_spend=float(recommended_spend) if recommended_spend is not None else None,
expected_impact=float(expected_impact) if expected_impact is not None else None,
confidence=confidence,
)
db.session.add(move)
db.session.commit()
return jsonify({
'success': True,
'optimization_move': {
'id': move.id,
'move_type': move.move_type,
'description': move.description,
'source_channel': move.source_channel,
'target_channel': move.target_channel,
'current_spend': move.current_spend,
'recommended_spend': move.recommended_spend,
'expected_impact': move.expected_impact,
'confidence': move.confidence,
'status': move.status,
'created_at': move.created_at.isoformat() if move.created_at else None,
}
}), 201
@business_api_bp.route('/api/company/<company_id>/optimization-moves/<move_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_optimization_move(company_id, move_id):
"""Update an optimization move."""
from app.models import db, OptimizationMove
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
move = db.session.get(OptimizationMove, move_id)
if not move or move.company_id != company_id:
return jsonify({'error': 'Optimization move not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
if 'move_type' in data:
if data['move_type'] not in ('budget_shift', 'channel_change', 'target_audience'):
return jsonify({'error': 'move_type must be budget_shift, channel_change, or target_audience'}), 400
move.move_type = data['move_type']
if 'description' in data:
move.description = data['description'].strip()
if 'source_channel' in data:
move.source_channel = data['source_channel'].strip()
if 'target_channel' in data:
move.target_channel = data['target_channel'].strip()
if 'current_spend' in data:
move.current_spend = float(data['current_spend'])
if 'recommended_spend' in data:
move.recommended_spend = float(data['recommended_spend'])
if 'expected_impact' in data:
move.expected_impact = float(data['expected_impact'])
if 'confidence' in data:
move.confidence = float(data['confidence'])
if 'status' in data:
if data['status'] not in ('recommended', 'accepted', 'rejected', 'implemented'):
return jsonify({'error': 'status must be recommended, accepted, rejected, or implemented'}), 400
move.status = data['status']
if 'actual_result' in data:
move.actual_result = float(data['actual_result']) if data['actual_result'] is not None else None
db.session.commit()
return jsonify({
'success': True,
'optimization_move': {
'id': move.id,
'move_type': move.move_type,
'description': move.description,
'source_channel': move.source_channel,
'target_channel': move.target_channel,
'current_spend': move.current_spend,
'recommended_spend': move.recommended_spend,
'expected_impact': move.expected_impact,
'confidence': move.confidence,
'status': move.status,
'created_at': move.created_at.isoformat() if move.created_at else None,
'updated_at': move.updated_at.isoformat() if move.updated_at else None,
}
})
@business_api_bp.route('/api/company/<company_id>/optimization-moves/<move_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def delete_optimization_move(company_id, move_id):
"""Delete an optimization move."""
from app.models import db, OptimizationMove
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
move = db.session.get(OptimizationMove, move_id)
if not move or move.company_id != company_id:
return jsonify({'error': 'Optimization move not found'}), 404
db.session.delete(move)
db.session.commit()
return jsonify({'success': True, 'message': 'Optimization move deleted'})
@business_api_bp.route('/api/company/<company_id>/optimization-moves/<move_id>/accept', methods=['PATCH'])
@require_auth_json()
@require_csrf
def accept_optimization_move(company_id, move_id):
"""Accept an optimization move."""
from app.models import db, OptimizationMove
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
move = db.session.get(OptimizationMove, move_id)
if not move or move.company_id != company_id:
return jsonify({'error': 'Optimization move not found'}), 404
move.status = 'accepted'
db.session.commit()
return jsonify({
'success': True,
'optimization_move': {
'id': move.id,
'status': move.status,
}
})
@business_api_bp.route('/api/company/<company_id>/optimization-moves/<move_id>/reject', methods=['PATCH'])
@require_auth_json()
@require_csrf
def reject_optimization_move(company_id, move_id):
"""Reject an optimization move."""
from app.models import db, OptimizationMove
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
move = db.session.get(OptimizationMove, move_id)
if not move or move.company_id != company_id:
return jsonify({'error': 'Optimization move not found'}), 404
move.status = 'rejected'
db.session.commit()
return jsonify({
'success': True,
'optimization_move': {
'id': move.id,
'status': move.status,
}
})
# ============================================================================
# Forecasts API
# ============================================================================
@business_api_bp.route('/api/company/<company_id>/forecasts', methods=['GET'])
@require_auth_json()
def list_forecasts(company_id):
"""List forecasts for a company."""
from app.models import Forecast
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
type_filter = request.args.get('type')
period_filter = request.args.get('period')
query = Forecast.query.filter_by(company_id=company_id)
if type_filter:
query = query.filter_by(forecast_type=type_filter)
if period_filter:
query = query.filter_by(period=period_filter)
pagination, meta = paginate_query(query.order_by(Forecast.created_at.desc()))
forecasts = pagination.items
return jsonify({
**meta,
'forecasts': [{
'id': f.id,
'forecast_type': f.forecast_type,
'period': f.period,
'period_start': f.period_start.isoformat() if f.period_start else None,
'period_end': f.period_end.isoformat() if f.period_end else None,
'projected_value': f.projected_value,
'actual_value': f.actual_value,
'confidence': f.confidence,
'methodology': f.methodology,
'assumptions_json': f.assumptions_json,
'variance': f.variance,
'variance_pct': round(f.variance_pct, 2) if f.variance_pct is not None else None,
'created_at': f.created_at.isoformat() if f.created_at else None,
'updated_at': f.updated_at.isoformat() if f.updated_at else None,
} for f in forecasts]
})
@business_api_bp.route('/api/company/<company_id>/forecasts', methods=['POST'])
@require_auth_json()
@require_csrf
def create_forecast(company_id):
"""Create a new forecast."""
from app.models import db, Forecast
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
forecast_type = data.get('forecast_type', 'revenue')
if forecast_type not in ('revenue', 'pipeline', 'cost'):
return jsonify({'error': 'forecast_type must be revenue, pipeline, or cost'}), 400
projected_value = data.get('projected_value')
if projected_value is None:
return jsonify({'error': 'projected_value is required'}), 400
try:
projected_value = float(projected_value)
except (ValueError, TypeError):
return jsonify({'error': 'projected_value must be a number'}), 400
actual_value = data.get('actual_value')
if actual_value is not None:
try:
actual_value = float(actual_value)
except (ValueError, TypeError):
return jsonify({'error': 'actual_value must be a number'}), 400
confidence = data.get('confidence')
if confidence is not None:
try:
confidence = float(confidence)
except (ValueError, TypeError):
return jsonify({'error': 'confidence must be a number'}), 400
forecast = Forecast(
company_id=company_id,
forecast_type=forecast_type,
period=data.get('period', 'monthly'),
projected_value=projected_value,
actual_value=actual_value,
confidence=confidence,
methodology=data.get('methodology', ''),
assumptions_json=data.get('assumptions_json') or {},
)
if data.get('period_start'):
forecast.period_start = data['period_start']
if data.get('period_end'):
forecast.period_end = data['period_end']
db.session.add(forecast)
db.session.commit()
return jsonify({
'success': True,
'forecast': {
'id': forecast.id,
'forecast_type': forecast.forecast_type,
'period': forecast.period,
'projected_value': forecast.projected_value,
'actual_value': forecast.actual_value,
'confidence': forecast.confidence,
'methodology': forecast.methodology,
'created_at': forecast.created_at.isoformat() if forecast.created_at else None,
}
}), 201
@business_api_bp.route('/api/company/<company_id>/forecasts/<forecast_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_forecast(company_id, forecast_id):
"""Update a forecast."""
from app.models import db, Forecast
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
forecast = db.session.get(Forecast, forecast_id)
if not forecast or forecast.company_id != company_id:
return jsonify({'error': 'Forecast not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
if 'forecast_type' in data:
if data['forecast_type'] not in ('revenue', 'pipeline', 'cost'):
return jsonify({'error': 'forecast_type must be revenue, pipeline, or cost'}), 400
forecast.forecast_type = data['forecast_type']
if 'period' in data:
forecast.period = data['period']
if 'projected_value' in data:
forecast.projected_value = float(data['projected_value'])
if 'actual_value' in data:
forecast.actual_value = float(data['actual_value']) if data['actual_value'] is not None else None
if 'confidence' in data:
forecast.confidence = float(data['confidence']) if data['confidence'] is not None else None
if 'methodology' in data:
forecast.methodology = data['methodology']
if 'assumptions_json' in data:
forecast.assumptions_json = data['assumptions_json']
if 'period_start' in data:
forecast.period_start = data['period_start']
if 'period_end' in data:
forecast.period_end = data['period_end']
db.session.commit()
return jsonify({
'success': True,
'forecast': {
'id': forecast.id,
'forecast_type': forecast.forecast_type,
'period': forecast.period,
'period_start': forecast.period_start.isoformat() if forecast.period_start else None,
'period_end': forecast.period_end.isoformat() if forecast.period_end else None,
'projected_value': forecast.projected_value,
'actual_value': forecast.actual_value,
'confidence': forecast.confidence,
'variance': forecast.variance,
'variance_pct': round(forecast.variance_pct, 2) if forecast.variance_pct is not None else None,
'updated_at': forecast.updated_at.isoformat() if forecast.updated_at else None,
}
})
@business_api_bp.route('/api/company/<company_id>/forecasts/<forecast_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def delete_forecast(company_id, forecast_id):
"""Delete a forecast."""
from app.models import db, Forecast
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
forecast = db.session.get(Forecast, forecast_id)
if not forecast or forecast.company_id != company_id:
return jsonify({'error': 'Forecast not found'}), 404
db.session.delete(forecast)
db.session.commit()
return jsonify({'success': True, 'message': 'Forecast deleted'})
@business_api_bp.route('/api/company/<company_id>/forecasts/sync', methods=['POST'])
@require_auth_json()
@require_csrf
def sync_forecasts(company_id):
"""Trigger an immediate forecast sync for this company.
Optional JSON body: { "types": ["revenue", "pipeline", "cost"] } to sync
a subset; omit to sync all available types.
"""
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
from app.services.forecast_updater import sync_all_forecasts
result = sync_all_forecasts(company_id)
return jsonify({
'success': True,
'revenue_updated': result.get('revenue_updated', 0),
'pipeline_updated': result.get('pipeline_updated', 0),
'cost_updated': result.get('cost_updated', 0),
'future_created': result.get('future_created', 0),
'duration': result.get('duration_seconds', 0),
})
@business_api_bp.route('/api/company/<company_id>/forecasts/sync-status', methods=['GET'])
@require_auth_json()
def get_forecast_sync_status(company_id):
"""Get the last forecast sync status for a company."""
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
from app.models import Forecast
from datetime import datetime, timezone
# Find the most recently updated forecast for this company
latest = (
Forecast.query
.filter_by(company_id=company_id)
.order_by(Forecast.updated_at.desc())
.first()
)
last_sync_at = None
if latest and latest.updated_at:
last_sync_at = latest.updated_at.isoformat()
return jsonify({
'company_id': company_id,
'last_sync_at': last_sync_at,
'auto_sync': 'daily',
'next_sync': 'within 24 hours',
})
# ============================================================================
# Coaching Assignments API
# ============================================================================
@business_api_bp.route('/api/company/<company_id>/coaching-assignments', methods=['GET'])
@require_auth_json()
def list_coaching_assignments(company_id):
"""List coaching assignments for a company."""
from app.models import CoachingAssignment
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
status_filter = request.args.get('status')
query = CoachingAssignment.query.filter_by(company_id=company_id)
if status_filter:
query = query.filter_by(status=status_filter)
pagination, meta = paginate_query(query.order_by(CoachingAssignment.created_at.desc()))
assignments = pagination.items
return jsonify({
**meta,
'coaching_assignments': [{
'id': a.id,
'coach_id': a.coach_id,
'coach_name': a.coach.full_name if a.coach else '',
'coach_email': a.coach.email if a.coach else '',
'rep_id': a.rep_id,
'rep_name': a.rep.full_name if a.rep else '',
'rep_email': a.rep.email if a.rep else '',
'focus_area': a.focus_area,
'description': a.description,
'start_date': a.start_date.isoformat() if a.start_date else None,
'end_date': a.end_date.isoformat() if a.end_date else None,
'status': a.status,
'created_at': a.created_at.isoformat() if a.created_at else None,
} for a in assignments]
})
@business_api_bp.route('/api/company/<company_id>/coaching-assignments', methods=['POST'])
@require_auth_json()
@require_csrf
def create_coaching_assignment(company_id):
"""Create a new coaching assignment."""
from app.models import db, CoachingAssignment
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
coach_id = data.get('coach_id')
rep_id = data.get('rep_id')
if not coach_id or not rep_id:
return jsonify({'error': 'coach_id and rep_id are required'}), 400
status = data.get('status', 'active')
if status not in ('active', 'completed', 'paused'):
return jsonify({'error': 'status must be active, completed, or paused'}), 400
assignment = CoachingAssignment(
company_id=company_id,
coach_id=coach_id,
rep_id=rep_id,
focus_area=data.get('focus_area', '').strip(),
description=data.get('description', '').strip(),
status=status,
)
if data.get('start_date'):
assignment.start_date = data['start_date']
if data.get('end_date'):
assignment.end_date = data['end_date']
db.session.add(assignment)
db.session.commit()
return jsonify({
'success': True,
'coaching_assignment': {
'id': assignment.id,
'coach_id': assignment.coach_id,
'coach_name': assignment.coach.full_name if assignment.coach else '',
'rep_id': assignment.rep_id,
'rep_name': assignment.rep.full_name if assignment.rep else '',
'focus_area': assignment.focus_area,
'description': assignment.description,
'status': assignment.status,
'start_date': assignment.start_date.isoformat() if assignment.start_date else None,
'end_date': assignment.end_date.isoformat() if assignment.end_date else None,
'created_at': assignment.created_at.isoformat() if assignment.created_at else None,
}
}), 201
@business_api_bp.route('/api/company/<company_id>/coaching-assignments/<assignment_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_coaching_assignment(company_id, assignment_id):
"""Update a coaching assignment."""
from app.models import db, CoachingAssignment
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
assignment = db.session.get(CoachingAssignment, assignment_id)
if not assignment or assignment.company_id != company_id:
return jsonify({'error': 'Coaching assignment not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
if 'coach_id' in data:
assignment.coach_id = data['coach_id']
if 'rep_id' in data:
assignment.rep_id = data['rep_id']
if 'focus_area' in data:
assignment.focus_area = data['focus_area'].strip()
if 'description' in data:
assignment.description = data['description'].strip()
if 'status' in data:
if data['status'] not in ('active', 'completed', 'paused'):
return jsonify({'error': 'status must be active, completed, or paused'}), 400
assignment.status = data['status']
if 'start_date' in data:
assignment.start_date = data['start_date']
if 'end_date' in data:
assignment.end_date = data['end_date']
db.session.commit()
return jsonify({
'success': True,
'coaching_assignment': {
'id': assignment.id,
'coach_id': assignment.coach_id,
'coach_name': assignment.coach.full_name if assignment.coach else '',
'rep_id': assignment.rep_id,
'rep_name': assignment.rep.full_name if assignment.rep else '',
'focus_area': assignment.focus_area,
'status': assignment.status,
'start_date': assignment.start_date.isoformat() if assignment.start_date else None,
'end_date': assignment.end_date.isoformat() if assignment.end_date else None,
}
})
@business_api_bp.route('/api/company/<company_id>/coaching-assignments/<assignment_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def delete_coaching_assignment(company_id, assignment_id):
"""Delete a coaching assignment."""
from app.models import db, CoachingAssignment
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
assignment = db.session.get(CoachingAssignment, assignment_id)
if not assignment or assignment.company_id != company_id:
return jsonify({'error': 'Coaching assignment not found'}), 404
db.session.delete(assignment)
db.session.commit()
return jsonify({'success': True, 'message': 'Coaching assignment deleted'})
# ============================================================================
# Coaching Scorecards API
# ============================================================================
@business_api_bp.route('/api/company/<company_id>/coaching-scorecards', methods=['GET'])
@require_auth_json()
def list_coaching_scorecards(company_id):
"""List coaching scorecards for a company."""
from app.models import CoachingScorecard
error_response, membership = _check_company_access(company_id, min_role='member')
if error_response:
return error_response
pagination, meta = paginate_query(
CoachingScorecard.query.filter_by(company_id=company_id).order_by(CoachingScorecard.created_at.desc())
)
scorecards = pagination.items
return jsonify({
**meta,
'coaching_scorecards': [{
'id': s.id,
'assignment_id': s.assignment_id,
'rep_id': s.rep_id,
'rep_name': s.assignment.rep.full_name if s.assignment and s.assignment.rep else '',
'evaluation_date': s.evaluation_date.isoformat() if s.evaluation_date else None,
'overall_score': s.overall_score,
'communication_score': s.communication_score,
'technical_score': s.technical_score,
'closing_score': s.closing_score,
'follow_up_score': s.follow_up_score,
'strengths': s.strengths,
'improvement_areas': s.improvement_areas,
'notes': s.notes,
'created_at': s.created_at.isoformat() if s.created_at else None,
} for s in scorecards]
})
@business_api_bp.route('/api/company/<company_id>/coaching-scorecards', methods=['POST'])
@require_auth_json()
@require_csrf
def create_coaching_scorecard(company_id):
"""Create a new coaching scorecard."""
from app.models import db, CoachingScorecard
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
assignment_id = data.get('assignment_id')
rep_id = data.get('rep_id')
if not assignment_id or not rep_id:
return jsonify({'error': 'assignment_id and rep_id are required'}), 400
# Validate scores are 0-100
score_fields = ['overall_score', 'communication_score', 'technical_score', 'closing_score', 'follow_up_score']
for field in score_fields:
val = data.get(field)
if val is not None:
try:
val = float(val)
if val < 0 or val > 100:
return jsonify({'error': f'{field} must be between 0 and 100'}), 400
except (ValueError, TypeError):
return jsonify({'error': f'{field} must be a number'}), 400
scorecard = CoachingScorecard(
company_id=company_id,
assignment_id=assignment_id,
rep_id=rep_id,
overall_score=data.get('overall_score'),
communication_score=data.get('communication_score'),
technical_score=data.get('technical_score'),
closing_score=data.get('closing_score'),
follow_up_score=data.get('follow_up_score'),
strengths=data.get('strengths', '').strip(),
improvement_areas=data.get('improvement_areas', '').strip(),
notes=data.get('notes', '').strip(),
)
if data.get('evaluation_date'):
scorecard.evaluation_date = data['evaluation_date']
db.session.add(scorecard)
db.session.commit()
return jsonify({
'success': True,
'coaching_scorecard': {
'id': scorecard.id,
'assignment_id': scorecard.assignment_id,
'rep_id': scorecard.rep_id,
'overall_score': scorecard.overall_score,
'communication_score': scorecard.communication_score,
'technical_score': scorecard.technical_score,
'closing_score': scorecard.closing_score,
'follow_up_score': scorecard.follow_up_score,
'strengths': scorecard.strengths,
'improvement_areas': scorecard.improvement_areas,
'notes': scorecard.notes,
'evaluation_date': scorecard.evaluation_date.isoformat() if scorecard.evaluation_date else None,
'created_at': scorecard.created_at.isoformat() if scorecard.created_at else None,
}
}), 201
@business_api_bp.route('/api/company/<company_id>/coaching-scorecards/<scorecard_id>', methods=['PUT'])
@require_auth_json()
@require_csrf
def update_coaching_scorecard(company_id, scorecard_id):
"""Update a coaching scorecard."""
from app.models import db, CoachingScorecard
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
scorecard = db.session.get(CoachingScorecard, scorecard_id)
if not scorecard or scorecard.company_id != company_id:
return jsonify({'error': 'Coaching scorecard not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
score_fields = ['overall_score', 'communication_score', 'technical_score', 'closing_score', 'follow_up_score']
for field in score_fields:
if field in data:
val = data[field]
if val is not None:
try:
val = float(val)
if val < 0 or val > 100:
return jsonify({'error': f'{field} must be between 0 and 100'}), 400
setattr(scorecard, field, val)
except (ValueError, TypeError):
return jsonify({'error': f'{field} must be a number'}), 400
else:
setattr(scorecard, field, None)
if 'assignment_id' in data:
scorecard.assignment_id = data['assignment_id']
if 'rep_id' in data:
scorecard.rep_id = data['rep_id']
if 'strengths' in data:
scorecard.strengths = data['strengths'].strip()
if 'improvement_areas' in data:
scorecard.improvement_areas = data['improvement_areas'].strip()
if 'notes' in data:
scorecard.notes = data['notes'].strip()
if 'evaluation_date' in data:
scorecard.evaluation_date = data['evaluation_date']
db.session.commit()
return jsonify({
'success': True,
'coaching_scorecard': {
'id': scorecard.id,
'assignment_id': scorecard.assignment_id,
'rep_id': scorecard.rep_id,
'overall_score': scorecard.overall_score,
'communication_score': scorecard.communication_score,
'technical_score': scorecard.technical_score,
'closing_score': scorecard.closing_score,
'follow_up_score': scorecard.follow_up_score,
'strengths': scorecard.strengths,
'improvement_areas': scorecard.improvement_areas,
'notes': scorecard.notes,
'updated_at': scorecard.created_at.isoformat() if scorecard.created_at else None,
}
})
@business_api_bp.route('/api/company/<company_id>/coaching-scorecards/<scorecard_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def delete_coaching_scorecard(company_id, scorecard_id):
"""Delete a coaching scorecard."""
from app.models import db, CoachingScorecard
error_response, membership = _check_company_access(company_id, min_role='editor')
if error_response:
return error_response
scorecard = db.session.get(CoachingScorecard, scorecard_id)
if not scorecard or scorecard.company_id != company_id:
return jsonify({'error': 'Coaching scorecard not found'}), 404
db.session.delete(scorecard)
db.session.commit()
return jsonify({'success': True, 'message': 'Coaching scorecard deleted'})