"""Usage Dashboard Routes.""" import logging from datetime import datetime, timedelta from flask import Blueprint, render_template, request, jsonify from sqlalchemy import func, desc from core.models.usage_log import UsageLog from core.models.database import db logger = logging.getLogger(__name__) usage_bp = Blueprint('usage', __name__, url_prefix='/usage', template_folder='templates') @usage_bp.route('/') @usage_bp.route('/index') def index(): """Render usage dashboard.""" return render_template('usage/dashboard.html', active_tab='usage') @usage_bp.route('/api/stats') def api_stats(): """Get usage statistics as JSON, with optional time filter.""" try: days = request.args.get('days', type=int) filters = _time_filter(days) query = UsageLog.query if filters: query = query.filter(*filters) # Total stats total_calls = query.count() totals = db.session.query( func.sum(UsageLog.tokens_used), func.sum(UsageLog.input_tokens), func.sum(UsageLog.output_tokens), func.sum(UsageLog.estimated_cost_usd), ).filter(*filters).one() total_tokens = int(totals[0] or 0) total_input_tokens = int(totals[1] or 0) total_output_tokens = int(totals[2] or 0) total_cost = float(totals[3] or 0) group_cols = [ func.count(UsageLog.id).label('calls'), func.sum(UsageLog.input_tokens).label('input_tokens'), func.sum(UsageLog.output_tokens).label('output_tokens'), func.sum(UsageLog.tokens_used).label('tokens'), func.sum(UsageLog.estimated_cost_usd).label('cost'), ] by_provider = db.session.query(UsageLog.provider, *group_cols).filter( *filters ).group_by(UsageLog.provider).all() by_model = db.session.query( UsageLog.provider, UsageLog.model, *group_cols ).filter(*filters).group_by(UsageLog.provider, UsageLog.model).all() by_module = db.session.query(UsageLog.module, *group_cols).filter( *filters ).group_by(UsageLog.module).all() by_user = db.session.query(UsageLog.user, *group_cols).filter( *filters ).group_by(UsageLog.user).all() # Daily usage (last 30 days max) daily_cutoff = datetime.utcnow() - timedelta(days=min(days or 30, 30)) daily = db.session.query( func.date(UsageLog.timestamp).label('date'), *group_cols ).filter(UsageLog.timestamp >= daily_cutoff).group_by( func.date(UsageLog.timestamp) ).order_by(func.date(UsageLog.timestamp)).all() recent = query.order_by(desc(UsageLog.timestamp)).limit(20).all() def row(base, r): return { **base, 'calls': r.calls, 'input_tokens': int(r.input_tokens or 0), 'output_tokens': int(r.output_tokens or 0), 'tokens': int(r.tokens or 0), 'cost': round(float(r.cost or 0), 4), } return jsonify({ 'summary': { 'total_calls': total_calls, 'total_tokens': total_tokens, 'total_input_tokens': total_input_tokens, 'total_output_tokens': total_output_tokens, 'total_cost': round(total_cost, 4), }, 'by_provider': [row({'provider': r.provider}, r) for r in by_provider], 'by_model': [ row({'provider': r.provider, 'model': r.model}, r) for r in by_model ], 'by_module': [row({'module': r.module or 'unknown'}, r) for r in by_module], 'by_user': [row({'user': r.user or 'anonymous'}, r) for r in by_user], 'daily': [row({'date': str(r.date)}, r) for r in daily], 'recent': [r.to_dict() for r in recent], }) except Exception as e: logger.error(f"Usage stats error: {e}", exc_info=True) return jsonify({'error': str(e)}), 500 def _time_filter(days): """Build time filter conditions.""" if days: cutoff = datetime.utcnow() - timedelta(days=days) return [UsageLog.timestamp >= cutoff] return []