Skip to content

A
aisbf
Commit 7b4a5b2d authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

fix: refine analytics cost and provider quota insights

parent 18da1eed
Hide whitespace changes
Inline Side-by-side
Showing with 1360 additions and 232 deletions
aisbf/analytics.py
View file @ 7b4a5b2d
......@@ -57,6 +57,37 @@ class Analytics:
'kilo': {'prompt': 0.0, 'completion': 0.0}, # Kilo providers are free/subscription
'codex': {'prompt': 2.5, 'completion': 10.0}, # ChatGPT pricing (input cheaper, output more expensive)
}
PROVIDER_FREE_TIER_DEFAULTS = {
'codex': {
'period': 'week',
'limit_type': 'requests',
'limit': 1,
'premium_monthly_cost': 20.0,
'description': 'Codex/ChatGPT free tier equivalent'
},
'kiro': {
'period': 'month',
'limit_type': 'requests',
'limit': 50,
'premium_monthly_cost': 19.0,
'description': 'Kiro free tier equivalent'
},
'claude': {
'period': 'month',
'limit_type': 'requests',
'limit': 1,
'premium_monthly_cost': 20.0,
'description': 'Claude subscription/free-tier equivalent'
},
'kilo': {
'period': 'month',
'limit_type': 'requests',
'limit': 1,
'premium_monthly_cost': 0.0,
'description': 'Kilo subscription/free-tier equivalent'
},
}
def __init__(self, db_manager, pricing: Optional[Dict] = None):
"""
......@@ -118,6 +149,123 @@ class Analytics:
# Ultimate fallback
return self.pricing.get('openrouter', {'prompt': 5.0, 'completion': 15.0})
def _get_provider_model_pricing(self, provider_id: str, model_name: Optional[str] = None) -> Dict[str, float]:
pricing = self._get_provider_pricing(provider_id)
if not model_name:
return pricing
try:
from .config import config
provider_config = config.get_provider(provider_id, warn=False)
if provider_config:
for model in getattr(provider_config, 'models', []) or []:
if getattr(model, 'name', None) != model_name:
continue
model_pricing = getattr(model, 'pricing', None) or {}
prompt_price = model_pricing.get('prompt')
completion_price = model_pricing.get('completion')
if prompt_price is not None or completion_price is not None:
return {
'prompt': float(prompt_price) if prompt_price is not None else pricing.get('prompt', 0.0),
'completion': float(completion_price) if completion_price is not None else pricing.get('completion', 0.0)
}
except Exception:
pass
return pricing
def _estimate_completion_ratio(self, prompt_tokens: float, completion_tokens: float, total_tokens: float) -> float:
if completion_tokens > 0 and total_tokens > 0:
return min(max(completion_tokens / total_tokens, 0.05), 0.95)
if prompt_tokens > 0 and total_tokens > prompt_tokens:
return min(max((total_tokens - prompt_tokens) / total_tokens, 0.05), 0.95)
if total_tokens >= 32000:
return 0.2
if total_tokens >= 8000:
return 0.3
if total_tokens >= 2000:
return 0.4
return 0.5
def _get_provider_free_tier_info(self, provider_id: str) -> Optional[Dict[str, Any]]:
try:
from .config import config
provider_config = config.get_provider(provider_id, warn=False)
if provider_config and getattr(provider_config, 'free_tier_limit', None):
return {
'period': getattr(provider_config, 'free_tier_period', None) or 'month',
'limit_type': getattr(provider_config, 'free_tier_limit_type', None) or 'requests',
'limit': int(getattr(provider_config, 'free_tier_limit', 0) or 0),
'premium_monthly_cost': float(getattr(provider_config, 'premium_reference_monthly_cost', 0) or 0),
'description': getattr(provider_config, 'free_tier_description', None) or f'{provider_id} free tier',
'provider_family': provider_id,
'source': 'config'
}
except Exception:
pass
provider_lower = (provider_id or '').lower()
for key, info in self.PROVIDER_FREE_TIER_DEFAULTS.items():
if key in provider_lower:
return {**info, 'provider_family': key, 'source': 'default'}
return None
def _derive_quota_from_usage(self, provider_id: str, usage_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
if not usage_data or not isinstance(usage_data, dict):
return None
free_tier = usage_data.get('free_tier')
if isinstance(free_tier, dict) and free_tier.get('limit'):
return {
'limit_type': free_tier.get('limit_type', 'requests'),
'limit': int(free_tier.get('limit') or 0),
'period': free_tier.get('period') or 'month',
'used': free_tier.get('used') or 0,
'remaining': free_tier.get('remaining'),
'source': free_tier.get('source', 'provider')
}
provider_lower = (provider_id or '').lower()
if 'codex' in provider_lower:
for key in ('weekly_requests_limit', 'weekly_limit', 'free_requests_per_week'):
if usage_data.get(key):
return {
'limit_type': 'requests',
'limit': int(usage_data.get(key) or 0),
'period': 'week',
'used': int(usage_data.get('weekly_requests_used') or usage_data.get('used_requests') or 0),
'source': 'derived'
}
if 'kiro' in provider_lower:
if usage_data.get('monthly_requests_limit') or usage_data.get('free_requests_per_month'):
return {
'limit_type': 'requests',
'limit': int(usage_data.get('monthly_requests_limit') or usage_data.get('free_requests_per_month') or 0),
'period': 'month',
'used': int(usage_data.get('monthly_requests_used') or usage_data.get('used_requests') or 0),
'source': 'derived'
}
if 'claude' in provider_lower:
if usage_data.get('quota_7d_utilization'):
try:
utilization = float(str(usage_data.get('quota_7d_utilization')).rstrip('%'))
except (TypeError, ValueError):
utilization = None
if utilization is not None:
return {
'limit_type': 'requests',
'limit': 1,
'period': 'week',
'used': max(1, int(utilization // 100)),
'source': 'derived'
}
return None
def record_request(
......@@ -134,7 +282,8 @@ class Analytics:
token_id: Optional[int] = None,
prompt_tokens: Optional[int] = None,
completion_tokens: Optional[int] = None,
actual_cost: Optional[float] = None
actual_cost: Optional[float] = None,
analytics_kind: str = 'execution'
):
"""
Record a request for analytics.
......@@ -153,6 +302,7 @@ class Analytics:
prompt_tokens: Optional number of input/prompt tokens
completion_tokens: Optional number of output/completion tokens
actual_cost: Optional actual cost returned by provider (in USD)
analytics_kind: Whether this row represents actual execution or selector metadata
"""
logger.info(f"🎯 Analytics.record_request ENTERED: provider={provider_id}, model={model_name}, tokens={tokens_used}, user_id={user_id}, success={success}")
......@@ -181,7 +331,7 @@ class Analytics:
if tokens_used > 0:
logger.info(f"Analytics.record_request: Recording to DB - provider={provider_id}, latency_ms={latency_ms}, success={success}, prompt={prompt_tokens}, completion={completion_tokens}, cost={actual_cost}")
try:
self.db.record_token_usage(provider_id, model_name, tokens_used, user_id, success, latency_ms, error_type, token_id, prompt_tokens, completion_tokens, actual_cost, rotation_id, autoselect_id)
self.db.record_token_usage(provider_id, model_name, tokens_used, user_id, success, latency_ms, error_type, token_id, prompt_tokens, completion_tokens, actual_cost, rotation_id, autoselect_id, analytics_kind)
logger.info(f"✅ Analytics recording completed successfully for {provider_id}")
except Exception as e:
logger.error(f"❌ Analytics recording FAILED for {provider_id}: {e}")
......@@ -359,12 +509,68 @@ class Analytics:
'actual_cost': total_actual_cost # Actual cost from provider responses
}
def _build_analytics_filters(
self,
placeholder: str,
user_filter: Optional[int] = None,
provider_filter: Optional[str] = None,
model_filter: Optional[str] = None,
rotation_filter: Optional[str] = None,
autoselect_filter: Optional[str] = None,
analytics_kind: Optional[str] = 'execution'
):
conditions = []
params = []
if user_filter == -1:
conditions.append("user_id IS NULL")
elif user_filter is not None:
conditions.append(f"user_id = {placeholder}")
params.append(user_filter)
if provider_filter:
conditions.append(f"provider_id = {placeholder}")
params.append(provider_filter)
if model_filter:
conditions.append(f"model_name = {placeholder}")
params.append(model_filter)
if rotation_filter:
conditions.append(f"rotation_id = {placeholder}")
params.append(rotation_filter)
if autoselect_filter:
conditions.append(f"autoselect_id = {placeholder}")
params.append(autoselect_filter)
if analytics_kind:
conditions.append(f"analytics_kind = {placeholder}")
params.append(analytics_kind)
return conditions, params
def _estimate_aggregate_cost(self, provider_totals: Dict[str, Dict[str, Any]]) -> float:
total_cost = 0.0
for provider_id, usage in provider_totals.items():
actual_cost = usage.get('actual_cost')
if actual_cost is not None and actual_cost > 0:
total_cost += actual_cost
continue
total_cost += self.estimate_cost(
provider_id,
usage.get('tokens_used', 0),
usage.get('prompt_tokens', 0),
usage.get('completion_tokens', 0),
model_name=usage.get('model_name')
)
return total_cost
def get_token_usage_by_date_range(
self,
provider_id: Optional[str] = None,
from_datetime: Optional[datetime] = None,
to_datetime: Optional[datetime] = None,
user_filter: Optional[int] = None
user_filter: Optional[int] = None,
model_filter: Optional[str] = None,
rotation_filter: Optional[str] = None,
autoselect_filter: Optional[str] = None
) -> Dict[str, Any]:
"""
Get token usage for a specific date range.
......@@ -388,45 +594,70 @@ class Analytics:
formatted_start = self._format_timestamp(start)
formatted_end = self._format_timestamp(end)
# Build user condition
if user_filter == -1:
user_condition = " AND user_id IS NULL"
params_suffix = []
elif user_filter is not None:
user_condition = f" AND user_id = {placeholder}"
params_suffix = [user_filter]
else:
user_condition = ""
params_suffix = []
filter_conditions, filter_params = self._build_analytics_filters(
placeholder,
user_filter=user_filter,
provider_filter=provider_id,
model_filter=model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter,
analytics_kind='execution'
)
extra_where = ''.join(f" AND {condition}" for condition in filter_conditions)
provider_tokens = {}
provider_prompt_tokens = {}
provider_completion_tokens = {}
if provider_id:
params = [provider_id, formatted_start, formatted_end] + params_suffix
params = [formatted_start, formatted_end] + filter_params
cursor.execute(f'''
SELECT SUM(tokens_used) as total_tokens
SELECT SUM(tokens_used) as total_tokens,
SUM(COALESCE(prompt_tokens, 0)) as total_prompt_tokens,
SUM(COALESCE(completion_tokens, 0)) as total_completion_tokens
FROM token_usage
WHERE provider_id = {placeholder} AND timestamp >= {placeholder} AND timestamp <= {placeholder}
{user_condition}
WHERE timestamp >= {placeholder} AND timestamp <= {placeholder}
{extra_where}
''', params)
row = cursor.fetchone()
total_tokens = row[0] if row and row[0] else 0
total_prompt_tokens = row[1] if row and row[1] else 0
total_completion_tokens = row[2] if row and row[2] else 0
else:
params = [formatted_start, formatted_end] + params_suffix
params = [formatted_start, formatted_end] + filter_params
cursor.execute(f'''
SELECT provider_id, SUM(tokens_used) as total_tokens
SELECT provider_id,
SUM(tokens_used) as total_tokens,
SUM(COALESCE(prompt_tokens, 0)) as total_prompt_tokens,
SUM(COALESCE(completion_tokens, 0)) as total_completion_tokens
FROM token_usage
WHERE timestamp >= {placeholder} AND timestamp <= {placeholder}
{user_condition}
{extra_where}
GROUP BY provider_id
''', params)
provider_tokens = {}
total_tokens = 0
total_prompt_tokens = 0
total_completion_tokens = 0
for row in cursor.fetchall():
provider_tokens[row[0]] = row[1]
provider_prompt_tokens[row[0]] = row[2] or 0
provider_completion_tokens[row[0]] = row[3] or 0
total_tokens += row[1]
# Calculate cost
cost = self.estimate_cost(provider_id or 'all', total_tokens)
total_prompt_tokens += row[2] or 0
total_completion_tokens += row[3] or 0
if provider_id:
cost = self.estimate_cost(provider_id, total_tokens, total_prompt_tokens, total_completion_tokens)
else:
cost = self._estimate_aggregate_cost({
pid: {
'tokens_used': tokens,
'prompt_tokens': provider_prompt_tokens.get(pid, 0),
'completion_tokens': provider_completion_tokens.get(pid, 0)
}
for pid, tokens in provider_tokens.items()
})
# Calculate duration in days for display
duration_days = (end - start).total_seconds() / 86400
......@@ -437,7 +668,9 @@ class Analytics:
'start': start.isoformat(),
'end': end.isoformat(),
'duration_days': duration_days,
'provider_tokens': provider_tokens if not provider_id else None
'provider_tokens': provider_tokens if not provider_id else None,
'prompt_tokens': total_prompt_tokens,
'completion_tokens': total_completion_tokens
}
def get_all_providers_stats(
......@@ -462,29 +695,17 @@ class Analytics:
cursor = conn.cursor()
placeholder = '?' if self.db.db_type == 'sqlite' else '%s'
if user_filter == -1:
user_condition = " AND user_id IS NULL"
params = [self._format_timestamp(start), self._format_timestamp(end)]
elif user_filter is not None:
user_condition = f" AND user_id = {placeholder}"
params = [self._format_timestamp(start), self._format_timestamp(end), user_filter]
else:
user_condition = ""
params = [self._format_timestamp(start), self._format_timestamp(end)]
extra_conditions = ""
if provider_filter:
extra_conditions += f" AND provider_id = {placeholder}"
params.append(provider_filter)
if model_filter:
extra_conditions += f" AND model_name = {placeholder}"
params.append(model_filter)
if rotation_filter:
extra_conditions += f" AND rotation_id = {placeholder}"
params.append(rotation_filter)
if autoselect_filter:
extra_conditions += f" AND autoselect_id = {placeholder}"
params.append(autoselect_filter)
filter_conditions, filter_params = self._build_analytics_filters(
placeholder,
user_filter=user_filter,
provider_filter=provider_filter,
model_filter=model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter,
analytics_kind='execution'
)
params = [self._format_timestamp(start), self._format_timestamp(end)] + filter_params
extra_conditions = ''.join(f" AND {condition}" for condition in filter_conditions)
cursor.execute(f'''
SELECT
......@@ -502,7 +723,6 @@ class Analytics:
FROM token_usage
WHERE timestamp >= {placeholder}
AND timestamp <= {placeholder}
{user_condition}
{extra_conditions}
GROUP BY provider_id, model_name, rotation_id, autoselect_id
ORDER BY provider_id, model_name, rotation_id, autoselect_id
......@@ -675,24 +895,17 @@ class Analytics:
conditions = [f"timestamp >= {placeholder}", f"timestamp <= {placeholder}"]
params = [formatted_cutoff, formatted_end_time]
if user_filter == -1:
conditions.append("user_id IS NULL")
elif user_filter is not None:
conditions.append(f"user_id = {placeholder}")
params.append(user_filter)
if provider_id:
conditions.append(f"provider_id = {placeholder}")
params.append(provider_id)
if model_filter:
conditions.append(f"model_name = {placeholder}")
params.append(model_filter)
if rotation_filter:
conditions.append(f"rotation_id = {placeholder}")
params.append(rotation_filter)
if autoselect_filter:
conditions.append(f"autoselect_id = {placeholder}")
params.append(autoselect_filter)
filter_conditions, filter_params = self._build_analytics_filters(
placeholder,
user_filter=user_filter,
provider_filter=provider_id,
model_filter=model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter,
analytics_kind='execution'
)
conditions.extend(filter_conditions)
params.extend(filter_params)
where_clause = " AND ".join(conditions)
......@@ -767,6 +980,20 @@ class Analytics:
elif user_filter is not None:
query += f' AND user_id = {placeholder}'
params.append(user_filter)
query += f' AND analytics_kind = {placeholder}'
params.append('execution')
if provider_filter:
query += f' AND provider_id = {placeholder}'
params.append(provider_filter)
if model_filter:
query += f' AND model_name = {placeholder}'
params.append(model_filter)
if rotation_filter:
query += f' AND rotation_id = {placeholder}'
params.append(rotation_filter)
if autoselect_filter:
query += f' AND autoselect_id = {placeholder}'
params.append(autoselect_filter)
query += ' ORDER BY provider_id, model_name'
cursor.execute(query, params)
......@@ -819,11 +1046,17 @@ class Analytics:
if not is_autoselect or (autoselect_id and autoselect_id != autoselect_filter):
continue
# Get provider request stats with date range
provider_stats = self.get_provider_stats(provider_id, from_datetime, to_datetime, user_filter)
# Skip if no data for this provider in the selected time range
if provider_stats['requests']['total'] == 0:
combo_stats = next((item for item in self.get_all_providers_stats(
from_datetime,
to_datetime,
user_filter=user_filter,
provider_filter=provider_id,
model_filter=model_name,
rotation_filter=rotation_id if rotation_id else rotation_filter,
autoselect_filter=autoselect_id if autoselect_id else autoselect_filter
) if item['provider_id'] == provider_id and item['model_name'] == model_name and item['rotation_id'] == rotation_id and item['autoselect_id'] == autoselect_id), None)
if not combo_stats or combo_stats['requests']['total'] == 0:
continue
# Get provider type from config or user providers
......@@ -855,11 +1088,11 @@ class Analytics:
'effective_context': dim.get('effective_context'),
'condense_context': dim.get('condense_context'),
'condense_method': dim.get('condense_method'),
'tokens_per_minute': provider_stats['tokens']['TPM'],
'tokens_per_hour': provider_stats['tokens']['TPH'],
'tokens_per_day': provider_stats['tokens']['TPD'],
'error_rate': provider_stats.get('error_rate', 0),
'avg_latency_ms': provider_stats.get('avg_latency_ms', 0),
'tokens_per_minute': combo_stats['tokens']['TPM'],
'tokens_per_hour': combo_stats['tokens']['TPH'],
'tokens_per_day': combo_stats['tokens']['TPD'],
'error_rate': combo_stats.get('error_rate', 0),
'avg_latency_ms': combo_stats.get('avg_latency_ms', 0),
'is_rotation': is_rotation,
'is_autoselect': is_autoselect,
'rotation_id': rotation_id,
......@@ -931,7 +1164,8 @@ class Analytics:
provider_id: str,
tokens_used: int,
prompt_tokens: Optional[int] = None,
completion_tokens: Optional[int] = None
completion_tokens: Optional[int] = None,
model_name: Optional[str] = None
) -> float:
"""
Estimate cost for token usage.
......@@ -945,14 +1179,10 @@ class Analytics:
Returns:
Estimated cost in USD
"""
# Handle special case for 'all' providers (aggregate stats)
if provider_id == 'all':
# For aggregate stats across all providers, skip cost estimation
# as it would require complex per-provider calculation
return 0.0
# Get provider-specific pricing (checks subscription status and custom pricing)
provider_pricing = self._get_provider_pricing(provider_id)
provider_pricing = self._get_provider_model_pricing(provider_id, model_name)
logger.info(f"💰 Cost calculation for {provider_id}:")
logger.info(f" tokens_used: {tokens_used}, prompt: {prompt_tokens}, completion: {completion_tokens}")
......@@ -979,14 +1209,127 @@ class Analytics:
logger.info(f" Calculated (estimated completion): ${prompt_cost:.8f} + ${completion_cost:.8f} = ${total_cost:.8f}")
return total_cost
else:
# No reliable breakdown — use estimation (25% input, 75% output is typical for chat)
prompt_tokens_est = tokens_used * 0.25
completion_tokens_est = tokens_used * 0.75
completion_ratio = self._estimate_completion_ratio(prompt_tokens, completion_tokens, tokens_used)
prompt_tokens_est = tokens_used * (1 - completion_ratio)
completion_tokens_est = tokens_used * completion_ratio
prompt_cost = (prompt_tokens_est / 1_000_000) * provider_pricing.get('prompt', 0)
completion_cost = (completion_tokens_est / 1_000_000) * provider_pricing.get('completion', 0)
total_cost = prompt_cost + completion_cost
logger.info(f" Calculated (estimated breakdown 25/75): ${prompt_cost:.8f} + ${completion_cost:.8f} = ${total_cost:.8f}")
logger.info(f" Calculated (estimated breakdown {1 - completion_ratio:.2f}/{completion_ratio:.2f}): ${prompt_cost:.8f} + ${completion_cost:.8f} = ${total_cost:.8f}")
return total_cost
def get_savings_overview(
self,
from_datetime: Optional[datetime] = None,
to_datetime: Optional[datetime] = None,
user_filter: Optional[int] = None,
provider_filter: Optional[str] = None,
model_filter: Optional[str] = None,
rotation_filter: Optional[str] = None,
autoselect_filter: Optional[str] = None
) -> Dict[str, Any]:
start = from_datetime or (datetime.now() - timedelta(days=1))
end = to_datetime or datetime.now()
provider_stats = self.get_all_providers_stats(
from_datetime,
to_datetime,
user_filter=user_filter,
provider_filter=provider_filter,
model_filter=model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter
)
provider_totals: Dict[str, Dict[str, Any]] = {}
for row in provider_stats:
bucket = provider_totals.setdefault(row['provider_id'], {
'tokens_used': 0,
'prompt_tokens': 0,
'completion_tokens': 0,
'actual_cost': 0.0,
'request_count': 0
})
bucket['tokens_used'] += row['tokens'].get('total', 0) or 0
bucket['prompt_tokens'] += row['tokens'].get('prompt', 0) or 0
bucket['completion_tokens'] += row['tokens'].get('completion', 0) or 0
bucket['actual_cost'] += row.get('actual_cost', 0) or 0
bucket['request_count'] += row['requests'].get('total', 0) or 0
equivalent_token_savings = 0
equivalent_cost_savings = 0.0
provider_equivalents = []
for provider_id, usage in provider_totals.items():
if usage['tokens_used'] <= 0:
continue
tier_info = self._get_provider_free_tier_info(provider_id)
if not tier_info:
continue
cached_usage = None
try:
cached_row = self.db.get_provider_usage(None if user_filter == -1 else user_filter, provider_id)
if cached_row:
cached_usage = cached_row.get('usage_data')
except Exception:
cached_usage = None
runtime_quota = self._derive_quota_from_usage(provider_id, cached_usage)
free_limit = max(int((runtime_quota or {}).get('limit') or tier_info.get('limit', 0) or 0), 1)
limit_type = (runtime_quota or {}).get('limit_type') or tier_info.get('limit_type') or 'requests'
period = (runtime_quota or {}).get('period') or tier_info.get('period') or 'month'
if limit_type == 'tokens':
usage_amount = usage.get('tokens_used', 0) or 0
else:
usage_amount = (runtime_quota or {}).get('used') or usage.get('request_count', 0) or 0
if usage_amount <= free_limit:
continue
extra_free_tiers = usage_amount // free_limit
if extra_free_tiers <= 0:
continue
saved_tokens = usage['tokens_used'] * extra_free_tiers
equivalent_token_savings += saved_tokens
premium_price = float(tier_info.get('premium_monthly_cost', 0) or 0)
premium_equivalent_cost = premium_price * extra_free_tiers
provider_equivalents.append({
'provider_id': provider_id,
'tokens_used': usage['tokens_used'],
'request_count': usage.get('request_count', 0) or 0,
'usage_amount': usage_amount,
'free_tier_limit': free_limit,
'free_tier_period': period,
'free_tier_limit_type': limit_type,
'extra_free_tiers': extra_free_tiers,
'equivalent_saved_tokens': saved_tokens,
'premium_reference_name': tier_info.get('description'),
'premium_reference_monthly_cost': premium_price,
'equivalent_saved_cost': premium_equivalent_cost,
'quota_source': 'runtime' if runtime_quota else tier_info.get('source', 'default')
})
equivalent_cost_savings += premium_equivalent_cost
return {
'total_tokens_saved': equivalent_token_savings,
'total_cost_saved': equivalent_cost_savings,
'date_range': {
'start': start.isoformat(),
'end': end.isoformat()
},
'provider_equivalents': provider_equivalents,
'savings_by_type': {
'free_tier_equivalent': {
'count': len(provider_equivalents),
'tokens_saved': equivalent_token_savings,
'cost_saved': equivalent_cost_savings,
'avg_tokens_saved': int(equivalent_token_savings / len(provider_equivalents)) if provider_equivalents else 0,
'max_tokens_saved': max((item['equivalent_saved_tokens'] for item in provider_equivalents), default=0)
}
} if provider_equivalents else {}
}
def get_cost_overview(
self,
......@@ -1013,9 +1356,6 @@ class Analytics:
start = from_datetime or (datetime.now() - timedelta(days=1))
end = to_datetime or datetime.now()
# Get token usage by date range
range_usage = self.get_token_usage_by_date_range(None, start, end, user_filter)
# Get providers that have data
providers = self.get_all_providers_stats(from_datetime, to_datetime, user_filter,
provider_filter, model_filter, rotation_filter, autoselect_filter)
......@@ -1025,6 +1365,7 @@ class Analytics:
for provider in providers:
provider_id = provider['provider_id']
model_name = provider.get('model_name')
tokens = provider['tokens']
# Get token usage for this provider in the date range
......@@ -1038,12 +1379,13 @@ class Analytics:
if actual_cost > 0:
cost = actual_cost
else:
cost = self.estimate_cost(provider_id, total_tokens, prompt_tokens, completion_tokens)
cost = self.estimate_cost(provider_id, total_tokens, prompt_tokens, completion_tokens, model_name=model_name)
total_cost += cost
provider_costs.append({
'provider_id': provider_id,
'model_name': model_name,
'tokens_today': total_tokens,
'estimated_cost': cost,
'is_actual': actual_cost > 0
......@@ -1208,6 +1550,9 @@ class Analytics:
to_datetime: Optional[datetime] = None,
user_filter: Optional[int] = None,
rotation_filter: Optional[str] = None,
provider_filter: Optional[str] = None,
model_filter: Optional[str] = None,
autoselect_filter: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""
Per-rotation breakdown: which provider/model received what share of hits and tokens.
......@@ -1221,20 +1566,17 @@ class Analytics:
cursor = conn.cursor()
ph = '?' if self.db.db_type == 'sqlite' else '%s'
if user_filter == -1:
user_cond = " AND user_id IS NULL"
params = [self._format_timestamp(start), self._format_timestamp(end)]
elif user_filter is not None:
user_cond = f" AND user_id = {ph}"
params = [self._format_timestamp(start), self._format_timestamp(end), user_filter]
else:
user_cond = ""
params = [self._format_timestamp(start), self._format_timestamp(end)]
rot_cond = ""
if rotation_filter:
rot_cond = f" AND rotation_id = {ph}"
params.append(rotation_filter)
filter_conditions, filter_params = self._build_analytics_filters(
ph,
user_filter=user_filter,
provider_filter=provider_filter,
model_filter=model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter,
analytics_kind='execution'
)
params = [self._format_timestamp(start), self._format_timestamp(end)] + filter_params
filter_clause = ''.join(f" AND {condition}" for condition in filter_conditions)
cursor.execute(f'''
SELECT rotation_id, provider_id, model_name,
......@@ -1244,7 +1586,7 @@ class Analytics:
FROM token_usage
WHERE rotation_id IS NOT NULL
AND timestamp >= {ph} AND timestamp <= {ph}
{user_cond} {rot_cond}
{filter_clause}
GROUP BY rotation_id, provider_id, model_name
ORDER BY rotation_id, requests DESC
''', params)
......@@ -1277,6 +1619,9 @@ class Analytics:
to_datetime: Optional[datetime] = None,
user_filter: Optional[int] = None,
autoselect_filter: Optional[str] = None,
provider_filter: Optional[str] = None,
model_filter: Optional[str] = None,
rotation_filter: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""
Per-autoselect breakdown: which rotation/model was selected, hit %, tokens, and selection latency.
......@@ -1291,20 +1636,17 @@ class Analytics:
cursor = conn.cursor()
ph = '?' if self.db.db_type == 'sqlite' else '%s'
if user_filter == -1:
user_cond = " AND user_id IS NULL"
params = [self._format_timestamp(start), self._format_timestamp(end)]
elif user_filter is not None:
user_cond = f" AND user_id = {ph}"
params = [self._format_timestamp(start), self._format_timestamp(end), user_filter]
else:
user_cond = ""
params = [self._format_timestamp(start), self._format_timestamp(end)]
as_cond = ""
if autoselect_filter:
as_cond = f" AND autoselect_id = {ph}"
params.append(autoselect_filter)
filter_conditions, filter_params = self._build_analytics_filters(
ph,
user_filter=user_filter,
provider_filter=provider_filter,
model_filter=model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter,
analytics_kind='execution'
)
params = [self._format_timestamp(start), self._format_timestamp(end)] + filter_params
filter_clause = ''.join(f" AND {condition}" for condition in filter_conditions)
cursor.execute(f'''
SELECT autoselect_id, model_name,
......@@ -1314,7 +1656,7 @@ class Analytics:
FROM token_usage
WHERE autoselect_id IS NOT NULL
AND timestamp >= {ph} AND timestamp <= {ph}
{user_cond} {as_cond}
{filter_clause}
GROUP BY autoselect_id, model_name
ORDER BY autoselect_id, requests DESC
''', params)
......@@ -1411,9 +1753,9 @@ class Analytics:
'tokens_1d': tokens_1d,
'tokens_7d': tokens_7d,
'provider_totals': provider_totals,
'estimated_cost_1h': self.estimate_cost('global', tokens_1h),
'estimated_cost_1d': self.estimate_cost('global', tokens_1d),
'estimated_cost_7d': self.estimate_cost('global', tokens_7d)
'estimated_cost_1h': self._estimate_aggregate_cost({pid: {'tokens_used': total} for pid, total in provider_totals.items()}),
'estimated_cost_1d': self._estimate_aggregate_cost({pid: {'tokens_used': total} for pid, total in provider_totals.items()}),
'estimated_cost_7d': self._estimate_aggregate_cost({pid: {'tokens_used': total} for pid, total in provider_totals.items()})
}
def get_user_token_usage_over_time(
......@@ -1537,4 +1879,4 @@ def initialize_analytics(db_manager):
"""
global _analytics
_analytics = Analytics(db_manager)
logger.info("Analytics module initialized")
\ No newline at end of file
logger.info("Analytics module initialized")
aisbf/app/model_cache.py
View file @ 7b4a5b2d
......@@ -272,6 +272,7 @@ async def _refresh_provider_usage_if_stale(provider_id: str, user_id):
return
usage_data = await handler.get_usage()
if usage_data:
usage_data = handler.normalize_usage_data(usage_data)
db.save_provider_usage(user_id, provider_id, usage_data)
_apply_usage_disable(db, user_id, provider_id, usage_data)
except Exception as e:
......
aisbf/config.py
View file @ 7b4a5b2d
......@@ -94,6 +94,11 @@ class ProviderConfig(BaseModel):
cache_ttl: Optional[int] = None # Cache TTL in seconds for Google Context Caching API
min_cacheable_tokens: Optional[int] = 1024 # Minimum token count for content to be cacheable (default matches OpenAI)
prompt_cache_key: Optional[str] = None # Optional cache key for OpenAI's load balancer routing optimization
free_tier_limit: Optional[int] = None # Upstream provider free-tier quota amount
free_tier_period: Optional[str] = None # Upstream free-tier period: day, week, month
free_tier_limit_type: Optional[str] = None # requests or tokens
premium_reference_monthly_cost: Optional[float] = None # Equivalent premium value for one extra upstream free tier
free_tier_description: Optional[str] = None # Human description for upstream free tier
# Response caching control
enable_response_cache: Optional[bool] = None # Enable/disable response caching for this provider (None = use global default)
......@@ -421,7 +426,7 @@ class Config:
print(f"Created default config file: {dst}")
# Copy markdown prompt files if they don't exist
for prompt_file in ['condensation_conversational.md', 'condensation_semantic.md', 'autoselect.md']:
for prompt_file in ['condensation_conversational.md', 'condensation_semantic.md', 'autoselect.md', 'STUDIO_SYSTEM.md']:
src = source_dir / prompt_file
dst = config_dir / prompt_file
......
aisbf/cost_extractor.py
View file @ 7b4a5b2d
......@@ -25,6 +25,50 @@ from typing import Dict, Optional, Any
logger = logging.getLogger(__name__)
def _coerce_cost_value(value: Any) -> Optional[float]:
if value is None or value == '':
return None
try:
if isinstance(value, dict):
for key in ('usd', 'amount', 'value', 'total'):
if key in value:
coerced = _coerce_cost_value(value[key])
if coerced is not None:
return coerced
return None
if isinstance(value, str):
cleaned = value.strip().replace('$', '')
if cleaned.lower().startswith('usd '):
cleaned = cleaned[4:]
return float(cleaned)
return float(value)
except (TypeError, ValueError):
return None
def _find_nested_cost(obj: Any) -> Optional[float]:
if isinstance(obj, dict):
prioritized_keys = (
'cost', 'total_cost', 'total_price', 'price', 'amount', 'usd_cost',
'billed_cost', 'request_cost', 'charge', 'credits_cost'
)
for key in prioritized_keys:
if key in obj:
coerced = _coerce_cost_value(obj.get(key))
if coerced is not None:
return coerced
for value in obj.values():
nested = _find_nested_cost(value)
if nested is not None:
return nested
elif isinstance(obj, list):
for item in obj:
nested = _find_nested_cost(item)
if nested is not None:
return nested
return None
def extract_cost_from_response(response: Dict[str, Any], provider_id: str) -> Optional[float]:
"""
Extract actual cost from provider response if available.
......@@ -40,49 +84,18 @@ def extract_cost_from_response(response: Dict[str, Any], provider_id: str) -> Op
return None
try:
# AWS Bedrock - may include cost in usage
if provider_id in ['amazon', 'bedrock', 'aws']:
usage = response.get('usage', {})
if isinstance(usage, dict):
cost = usage.get('cost')
if cost is not None:
return float(cost)
# Cohere - has billed_units but not direct cost
# Would need pricing config to convert
if provider_id == 'cohere':
meta = response.get('meta', {})
if isinstance(meta, dict):
billed_units = meta.get('billed_units', {})
if billed_units:
# Return None - we'll calculate from tokens
# Could enhance this to calculate from billed_units
pass
# Replicate - has prediction time
if provider_id == 'replicate':
metrics = response.get('metrics', {})
if isinstance(metrics, dict):
predict_time = metrics.get('predict_time')
if predict_time:
# Would need pricing per second to calculate
# Return None for now - calculate from tokens
pass
# Check for generic cost fields that some providers might use
for cost_field in ['cost', 'price', 'amount', 'total_cost']:
if cost_field in response:
cost = response[cost_field]
if cost is not None:
return float(cost)
# Check in usage object
usage = response.get('usage', {})
if isinstance(usage, dict) and cost_field in usage:
cost = usage[cost_field]
if cost is not None:
return float(cost)
provider_id_lower = (provider_id or '').lower()
if 'x_openai_metadata' in response:
cost = _find_nested_cost(response['x_openai_metadata'])
if cost is not None:
return cost
if provider_id_lower in ['amazon', 'bedrock', 'aws', 'openrouter', 'openai', 'codex', 'claude', 'anthropic', 'google', 'qwen', 'kiro', 'kilo', 'replicate', 'cohere']:
cost = _find_nested_cost(response)
if cost is not None:
return cost
return None
except Exception as e:
......@@ -108,20 +121,7 @@ def extract_cost_from_streaming_chunk(chunk: Dict[str, Any], provider_id: str) -
return None
try:
# Check if this is a final chunk with usage/cost info
usage = chunk.get('usage', {})
if isinstance(usage, dict):
# Try to extract cost from usage
cost = usage.get('cost')
if cost is not None:
return float(cost)
# Some providers might include cost at top level in final chunk
cost = chunk.get('cost')
if cost is not None:
return float(cost)
return None
return _find_nested_cost(chunk)
except Exception as e:
logger.debug(f"Error extracting cost from {provider_id} streaming chunk: {e}")
......
aisbf/database.py
View file @ 7b4a5b2d
......@@ -24,6 +24,7 @@ Database module for persistent tracking of context dimensions and rate limiting.
import sqlite3
import json
import hashlib
import time
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Any
from datetime import datetime, timedelta
......@@ -398,7 +399,8 @@ class DatabaseManager:
completion_tokens: Optional[int] = None,
actual_cost: Optional[float] = None,
rotation_id: Optional[str] = None,
autoselect_id: Optional[str] = None
autoselect_id: Optional[str] = None,
analytics_kind: str = 'execution'
):
"""
Record token usage for rate limiting and analytics.
......@@ -434,10 +436,10 @@ class DatabaseManager:
try:
# Try to insert with all columns
sql = f'''
INSERT INTO token_usage (user_id, provider_id, model_name, tokens_used, prompt_tokens, completion_tokens, actual_cost, success, latency_ms, error_type, token_id, rotation_id, autoselect_id, timestamp)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, CURRENT_TIMESTAMP)
INSERT INTO token_usage (user_id, provider_id, model_name, tokens_used, prompt_tokens, completion_tokens, actual_cost, success, latency_ms, error_type, token_id, rotation_id, autoselect_id, analytics_kind, timestamp)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, CURRENT_TIMESTAMP)
'''
params = (user_id, provider_id, model_name, tokens_used, prompt_tokens, completion_tokens, actual_cost, success, latency_int, error_type, token_id, rotation_id, autoselect_id)
params = (user_id, provider_id, model_name, tokens_used, prompt_tokens, completion_tokens, actual_cost, success, latency_int, error_type, token_id, rotation_id, autoselect_id, analytics_kind)
logger.debug(f"Trying full INSERT with {len(params)} parameters")
cursor.execute(sql, params)
logger.debug(f"Inserted with full column set, rows affected: {cursor.rowcount}")
......@@ -954,6 +956,212 @@ class DatabaseManager:
}
return None
def list_studio_assets(self, user_id: int, asset_type: str) -> List[Dict]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
cursor.execute(f'''
SELECT name, description, metadata_json, files_json, quote_text, created_at, updated_at
FROM studio_assets
WHERE user_id = {placeholder} AND asset_type = {placeholder}
ORDER BY updated_at DESC
''', (user_id, asset_type))
rows = cursor.fetchall()
items = []
for row in rows:
meta = _studio_json_loads(row[2], {})
files = _studio_json_loads(row[3], [])
item = {
'name': row[0],
'description': row[1] or '',
'kind': asset_type,
'created_at': self._normalize_db_timestamp(row[5]),
'updated_at': self._normalize_db_timestamp(row[6]),
}
item.update(meta)
if asset_type in ('character', 'environment'):
item['ref_images'] = files
item['image_count'] = len(files)
elif asset_type == 'voice':
item['sample_files'] = files
item['quote'] = row[4] or ''
item['transcript'] = row[4] or ''
items.append(item)
return items
def get_studio_asset(self, user_id: int, asset_type: str, name: str) -> Optional[Dict]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
cursor.execute(f'''
SELECT name, description, metadata_json, files_json, quote_text, created_at, updated_at
FROM studio_assets
WHERE user_id = {placeholder} AND asset_type = {placeholder} AND name = {placeholder}
''', (user_id, asset_type, name))
row = cursor.fetchone()
if not row:
return None
meta = _studio_json_loads(row[2], {})
files = _studio_json_loads(row[3], [])
item = {
'name': row[0],
'description': row[1] or '',
'kind': asset_type,
'created_at': self._normalize_db_timestamp(row[5]),
'updated_at': self._normalize_db_timestamp(row[6]),
}
item.update(meta)
if asset_type in ('character', 'environment'):
item['ref_images'] = files
item['image_count'] = len(files)
elif asset_type == 'voice':
item['sample_files'] = files
item['quote'] = row[4] or ''
item['transcript'] = row[4] or ''
return item
def upsert_studio_asset(self, user_id: int, asset_type: str, name: str, description: str, metadata: Dict, files: List[str], quote_text: str = '') -> Dict:
existing = self.get_studio_asset(user_id, asset_type, name)
created_at = existing.get('created_at') if existing else int(time.time())
now = int(time.time())
payload_meta = dict(metadata or {})
payload_meta.setdefault('name', name)
payload_meta.setdefault('kind', asset_type)
payload_meta['created_at'] = created_at
payload_meta['updated_at'] = now
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
if self.db_type == 'sqlite':
cursor.execute(f'''
INSERT INTO studio_assets (user_id, asset_type, name, description, metadata_json, files_json, quote_text, created_at, updated_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, datetime({placeholder}, 'unixepoch'), datetime({placeholder}, 'unixepoch'))
ON CONFLICT(user_id, asset_type, name) DO UPDATE SET
description=excluded.description,
metadata_json=excluded.metadata_json,
files_json=excluded.files_json,
quote_text=excluded.quote_text,
updated_at=excluded.updated_at
''', (user_id, asset_type, name, description, _studio_json_dumps(payload_meta), _studio_json_dumps(files), quote_text, created_at, now))
else:
cursor.execute(f'''
INSERT INTO studio_assets (user_id, asset_type, name, description, metadata_json, files_json, quote_text, created_at, updated_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, FROM_UNIXTIME({placeholder}), FROM_UNIXTIME({placeholder}))
ON DUPLICATE KEY UPDATE
description=VALUES(description),
metadata_json=VALUES(metadata_json),
files_json=VALUES(files_json),
quote_text=VALUES(quote_text),
updated_at=VALUES(updated_at)
''', (user_id, asset_type, name, description, _studio_json_dumps(payload_meta), _studio_json_dumps(files), quote_text, created_at, now))
conn.commit()
return self.get_studio_asset(user_id, asset_type, name)
def delete_studio_asset(self, user_id: int, asset_type: str, name: str) -> bool:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
cursor.execute(f'DELETE FROM studio_assets WHERE user_id = {placeholder} AND asset_type = {placeholder} AND name = {placeholder}', (user_id, asset_type, name))
conn.commit()
return cursor.rowcount > 0
def list_studio_pipelines(self, user_id: int) -> List[Dict]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
cursor.execute(f'''
SELECT pipeline_id, name, description, steps_json, created_at, updated_at
FROM studio_pipelines
WHERE user_id = {placeholder}
ORDER BY updated_at DESC
''', (user_id,))
rows = cursor.fetchall()
return [{
'id': row[0],
'name': row[1],
'description': row[2] or '',
'steps': _studio_json_loads(row[3], []),
'created_at': self._normalize_db_timestamp(row[4]),
'updated_at': self._normalize_db_timestamp(row[5]),
} for row in rows]
def get_studio_pipeline(self, user_id: int, pipeline_id: str) -> Optional[Dict]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
cursor.execute(f'''
SELECT pipeline_id, name, description, steps_json, created_at, updated_at
FROM studio_pipelines
WHERE user_id = {placeholder} AND pipeline_id = {placeholder}
''', (user_id, pipeline_id))
row = cursor.fetchone()
if not row:
return None
return {
'id': row[0],
'name': row[1],
'description': row[2] or '',
'steps': _studio_json_loads(row[3], []),
'created_at': self._normalize_db_timestamp(row[4]),
'updated_at': self._normalize_db_timestamp(row[5]),
}
def upsert_studio_pipeline(self, user_id: int, pipeline_id: str, name: str, description: str, steps: List[Dict]) -> Dict:
existing = self.get_studio_pipeline(user_id, pipeline_id)
created_at = existing.get('created_at') if existing else int(time.time())
now = int(time.time())
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
if self.db_type == 'sqlite':
cursor.execute(f'''
INSERT INTO studio_pipelines (user_id, pipeline_id, name, description, steps_json, created_at, updated_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, datetime({placeholder}, 'unixepoch'), datetime({placeholder}, 'unixepoch'))
ON CONFLICT(user_id, pipeline_id) DO UPDATE SET
name=excluded.name,
description=excluded.description,
steps_json=excluded.steps_json,
updated_at=excluded.updated_at
''', (user_id, pipeline_id, name, description, _studio_json_dumps(steps), created_at, now))
else:
cursor.execute(f'''
INSERT INTO studio_pipelines (user_id, pipeline_id, name, description, steps_json, created_at, updated_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, FROM_UNIXTIME({placeholder}), FROM_UNIXTIME({placeholder}))
ON DUPLICATE KEY UPDATE
name=VALUES(name),
description=VALUES(description),
steps_json=VALUES(steps_json),
updated_at=VALUES(updated_at)
''', (user_id, pipeline_id, name, description, _studio_json_dumps(steps), created_at, now))
conn.commit()
return self.get_studio_pipeline(user_id, pipeline_id)
def delete_studio_pipeline(self, user_id: int, pipeline_id: str) -> bool:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
cursor.execute(f'DELETE FROM studio_pipelines WHERE user_id = {placeholder} AND pipeline_id = {placeholder}', (user_id, pipeline_id))
conn.commit()
return cursor.rowcount > 0
def _normalize_db_timestamp(self, value) -> int:
if value is None:
return int(time.time())
if isinstance(value, datetime):
return int(value.timestamp())
if isinstance(value, str):
try:
return int(datetime.fromisoformat(value).timestamp())
except Exception:
try:
return int(datetime.strptime(value, '%Y-%m-%d %H:%M:%S').timestamp())
except Exception:
return int(time.time())
try:
return int(value)
except Exception:
return int(time.time())
def set_verification_token(self, user_id: int, token: str, expires_at: datetime):
"""
Set email verification token for a user.
......@@ -3723,6 +3931,7 @@ def DatabaseManager__initialize_database(self):
token_id INTEGER,
rotation_id VARCHAR(255),
autoselect_id VARCHAR(255),
analytics_kind VARCHAR(32) DEFAULT 'execution',
timestamp TIMESTAMP DEFAULT {timestamp_default}
)
''')
......@@ -3742,6 +3951,7 @@ def DatabaseManager__initialize_database(self):
('token_id', 'INTEGER'),
('rotation_id', 'VARCHAR(255)'),
('autoselect_id', 'VARCHAR(255)'),
('analytics_kind', "VARCHAR(32) DEFAULT 'execution'"),
]:
if col not in columns:
cursor.execute(f'ALTER TABLE token_usage ADD COLUMN {col} {defn}')
......@@ -3762,6 +3972,7 @@ def DatabaseManager__initialize_database(self):
('token_id', 'INTEGER'),
('rotation_id', 'VARCHAR(255)'),
('autoselect_id', 'VARCHAR(255)'),
('analytics_kind', "VARCHAR(32) DEFAULT 'execution'"),
]:
if col not in existing:
cursor.execute(f'ALTER TABLE token_usage ADD COLUMN {col} {defn}')
......@@ -3806,6 +4017,19 @@ def DatabaseManager__initialize_database(self):
logger.info(f"Database tables initialized successfully for {self.database_type} database")
def _studio_json_dumps(value: Any) -> str:
return json.dumps(value)
def _studio_json_loads(value: Any, default):
if not value:
return default
try:
return json.loads(value)
except Exception:
return default
def DatabaseManager__create_cache_tables(self, cursor, auto_increment, timestamp_default, boolean_type):
"""Create only temporary cache tables (CACHE DB ONLY)"""
......@@ -3824,6 +4048,7 @@ def DatabaseManager__create_cache_tables(self, cursor, auto_increment, timestamp
latency_ms INTEGER,
error_type VARCHAR(255),
token_id INTEGER,
analytics_kind VARCHAR(32) DEFAULT 'execution',
timestamp TIMESTAMP DEFAULT {timestamp_default}
)
''')
......@@ -3998,6 +4223,44 @@ def DatabaseManager__run_config_migrations(self, cursor, auto_increment, timesta
except Exception as e:
logger.warning(f"Migration check for account_tiers table: {e}")
try:
cursor.execute(f'''
CREATE TABLE IF NOT EXISTS studio_assets (
id INTEGER PRIMARY KEY {auto_increment},
user_id INTEGER NOT NULL,
asset_type VARCHAR(32) NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT,
metadata_json TEXT,
files_json TEXT,
quote_text TEXT,
created_at TIMESTAMP DEFAULT {timestamp_default},
updated_at TIMESTAMP DEFAULT {timestamp_default},
FOREIGN KEY (user_id) REFERENCES users(id),
UNIQUE(user_id, asset_type, name)
)
''')
except Exception as e:
logger.warning(f"Migration check for studio_assets table: {e}")
try:
cursor.execute(f'''
CREATE TABLE IF NOT EXISTS studio_pipelines (
id INTEGER PRIMARY KEY {auto_increment},
user_id INTEGER NOT NULL,
pipeline_id VARCHAR(255) NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT,
steps_json TEXT,
created_at TIMESTAMP DEFAULT {timestamp_default},
updated_at TIMESTAMP DEFAULT {timestamp_default},
FOREIGN KEY (user_id) REFERENCES users(id),
UNIQUE(user_id, pipeline_id)
)
''')
except Exception as e:
logger.warning(f"Migration check for studio_pipelines table: {e}")
# Migration: Add missing columns to account_tiers
try:
tier_columns = [
......
aisbf/geolocation.py
View file @ 7b4a5b2d
......@@ -14,6 +14,8 @@ _IPNet = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
# failures are not cached so they are retried on the next request.
_CACHE_TTL = 30 * 24 * 3600 # 30 days
_subnet_cache: Dict[str, Tuple[str, float]] = {}
_failure_cache: Dict[str, float] = {}
_FAILURE_TTL = 15 * 60
def _fallback_prefix(addr: _IPAddr) -> _IPNet:
......@@ -69,12 +71,25 @@ async def get_ip_country(ip: str) -> Optional[str]:
if cached is not None:
return cached
failure_expires = _failure_cache.get(ip)
if failure_expires and time.time() < failure_expires:
logger.debug("geo lookup: skipping %s due to recent failure backoff", ip)
return None
logger.debug("geo lookup: querying ipapi.co for %s", ip)
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(f"https://ipapi.co/{ip}/json/")
if response.status_code != 200:
logger.debug("geo lookup: %s returned HTTP %d", ip, response.status_code)
retry_after = response.headers.get('Retry-After') or response.headers.get('retry-after')
wait_seconds = _FAILURE_TTL
if retry_after:
try:
wait_seconds = max(int(retry_after), _FAILURE_TTL)
except (TypeError, ValueError):
pass
_failure_cache[ip] = time.time() + wait_seconds
return None
data = response.json()
......@@ -99,6 +114,7 @@ async def get_ip_country(ip: str) -> Optional[str]:
except Exception as exc:
logger.debug("geo lookup: %s failed with %s: %s", ip, type(exc).__name__, exc)
_failure_cache[ip] = time.time() + _FAILURE_TTL
return None
......
aisbf/handlers.py
View file @ 7b4a5b2d
......@@ -38,6 +38,7 @@ from .models import ChatCompletionRequest, ChatCompletionResponse
from .providers import get_provider_handler, RateLimitError
from .config import config
from .studio import infer_model_capabilities
from .studio_adapters import effective_studio_adapter, infer_studio_adapter_profile, adapt_studio_payload_with_profile
from .utils import (
count_messages_tokens,
split_messages_into_chunks,
......@@ -121,6 +122,66 @@ class RequestHandler:
self.user_rotations = {}
self.user_autoselects = {}
def _get_provider_config(self, provider_id: str):
if self.user_id and provider_id in getattr(self, 'user_providers', {}):
return self.user_providers[provider_id]
return self.config.get_provider(provider_id)
def _get_provider_type(self, provider_id: str) -> str:
provider_config = self._get_provider_config(provider_id)
if isinstance(provider_config, dict):
return provider_config.get('type', 'openai') or 'openai'
return getattr(provider_config, 'type', 'openai') or 'openai'
def _adapt_studio_workflow_payload(self, provider_id: str, endpoint_path: str, body: Dict) -> Dict:
payload = dict(body or {})
provider_type = (self._get_provider_type(provider_id) or '').lower()
model_metadata = payload.get('_studio_model_metadata') if isinstance(payload.get('_studio_model_metadata'), dict) else {}
if provider_id and '_studio_provider_id' not in payload:
payload['_studio_provider_id'] = provider_id
if model_metadata.get('provider_endpoint') and '_studio_provider_endpoint' not in payload:
payload['_studio_provider_endpoint'] = model_metadata.get('provider_endpoint')
adapter = effective_studio_adapter(provider_type, model_metadata)
profile = infer_studio_adapter_profile(provider_id, provider_type, model_metadata)
def first_value(*keys):
for key in keys:
value = payload.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
return None
if endpoint_path == 'v1/video/dub':
selected = first_value('video_model', 'stt_model', 'tts_model', 'model')
if selected:
payload['model'] = selected
elif endpoint_path == 'v1/audio/clone':
selected = first_value('model', 'tts_model')
if selected:
payload['model'] = selected
elif endpoint_path == 'v1/audio/convert':
selected = first_value('model', 'audio_model', 'tts_model', 'stt_model')
if selected:
payload['model'] = selected
elif endpoint_path in {'v1/audio/split', 'v1/audio/denoise'}:
selected = first_value('model', 'audio_model')
if selected:
payload['model'] = selected
elif endpoint_path in {'v1/images/faceswap', 'v1/images/outfit'}:
selected = first_value('model', 'image_model', 'video_model')
if selected:
payload['model'] = selected
elif endpoint_path in {'v1/images/to3d', 'v1/images/from3d', 'v1/video/to3d', 'v1/video/from3d', 'v1/3d/generate'}:
selected = first_value('model', 'render_model', 'image_model', 'video_model')
if selected:
payload['model'] = selected
payload = adapt_studio_payload_with_profile(adapter, profile, endpoint_path, payload)
payload.pop('_studio_model_metadata', None)
payload.pop('_studio_provider_id', None)
payload.pop('_studio_provider_endpoint', None)
return payload
def _load_user_configs(self):
"""Load user-specific configurations from database"""
self.reload_user_config()
......@@ -662,7 +723,8 @@ class RequestHandler:
token_id=getattr(request.state, 'token_id', None),
prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
completion_tokens=completion_tokens if completion_tokens > 0 else None,
actual_cost=actual_cost
actual_cost=actual_cost,
analytics_kind='execution'
)
# Record context dimensions for model performance tracking
......@@ -715,7 +777,8 @@ class RequestHandler:
token_id=getattr(request.state, 'token_id', None),
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
actual_cost=None
actual_cost=None,
analytics_kind='execution'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording for failed request failed: {analytics_error}")
......@@ -1355,7 +1418,8 @@ class RequestHandler:
token_id=getattr(request.state, 'token_id', None),
prompt_tokens=effective_context,
completion_tokens=completion_tokens,
actual_cost=None # Streaming responses typically don't include cost
actual_cost=None,
analytics_kind='execution'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording for streaming request failed: {analytics_error}")
......@@ -1393,7 +1457,8 @@ class RequestHandler:
token_id=getattr(request.state, 'token_id', None),
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
actual_cost=None
actual_cost=None,
analytics_kind='execution'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording for failed streaming request failed: {analytics_error}")
......@@ -1678,6 +1743,7 @@ class RequestHandler:
import httpx
import logging
logger = logging.getLogger(__name__)
body = self._adapt_studio_workflow_payload(provider_id, endpoint_path, body)
# Support user-defined providers (dict format) and global providers (object format)
if self.user_id and provider_id in self.user_providers:
......@@ -2990,7 +3056,8 @@ class RotationHandler:
token_id=token_id,
prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
completion_tokens=completion_tokens if completion_tokens > 0 else None,
actual_cost=actual_cost
actual_cost=actual_cost,
analytics_kind='execution'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording failed: {analytics_error}")
......@@ -3065,7 +3132,8 @@ class RotationHandler:
token_id=token_id,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
actual_cost=None
actual_cost=None,
analytics_kind='selector'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording for failed rotation failed: {analytics_error}")
......@@ -3735,7 +3803,8 @@ class RotationHandler:
token_id=token_id,
prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
completion_tokens=completion_tokens_count if completion_tokens_count > 0 else None,
actual_cost=None
actual_cost=None,
analytics_kind='execution'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording for streaming rotation failed: {analytics_error}")
......@@ -4852,7 +4921,8 @@ class AutoselectHandler:
token_id=token_id,
prompt_tokens=prompt_tokens if prompt_tokens > 0 else None,
completion_tokens=completion_tokens if completion_tokens > 0 else None,
actual_cost=None
actual_cost=None,
analytics_kind='selector'
)
except Exception as analytics_error:
logger.warning(f"Analytics recording for autoselect failed: {analytics_error}")
......
aisbf/models.py
View file @ 7b4a5b2d
......@@ -89,6 +89,11 @@ class Provider(BaseModel):
is_subscription: bool = False # If True, pricing is 0 (subscription-based provider)
price_per_million_prompt: Optional[float] = None # Price per million prompt tokens (USD)
price_per_million_completion: Optional[float] = None # Price per million completion tokens (USD)
free_tier_limit: Optional[int] = None
free_tier_period: Optional[str] = None
free_tier_limit_type: Optional[str] = None
premium_reference_monthly_cost: Optional[float] = None
free_tier_description: Optional[str] = None
class ErrorTracking(BaseModel):
failures: int
......
aisbf/providers/base.py
View file @ 7b4a5b2d
......@@ -1541,6 +1541,24 @@ class BaseProviderHandler:
"""Fetch current usage/quota data from the provider. Returns None if unsupported."""
return None
def normalize_usage_data(self, usage_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Normalize provider usage/quota data into a shared schema when possible."""
if not usage_data or not isinstance(usage_data, dict):
return usage_data
normalized = dict(usage_data)
free_tier = normalized.get('free_tier')
if isinstance(free_tier, dict):
normalized['free_tier'] = {
'limit': free_tier.get('limit'),
'used': free_tier.get('used'),
'remaining': free_tier.get('remaining'),
'period': free_tier.get('period'),
'limit_type': free_tier.get('limit_type', 'requests'),
'source': free_tier.get('source', 'provider')
}
return normalized
async def handle_request_with_batching(self, model: str, messages: List[Dict], max_tokens: Optional[int] = None,
temperature: Optional[float] = 1.0, stream: Optional[bool] = False,
tools: Optional[List[Dict]] = None, tool_choice: Optional[Union[str, Dict]] = None) -> Union[Dict, object]:
......
aisbf/providers/claude.py
View file @ 7b4a5b2d
......@@ -957,6 +957,28 @@ class ClaudeProviderHandler(BaseProviderHandler):
return True
return False
def normalize_usage_data(self, usage_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
normalized = super().normalize_usage_data(usage_data)
if not normalized or not isinstance(normalized, dict):
return normalized
utilization = normalized.get('quota_7d_utilization')
if utilization is not None:
try:
util_float = float(str(utilization).rstrip('%'))
normalized['free_tier'] = {
'limit': 100,
'used': max(util_float, 0.0),
'remaining': max(100.0 - util_float, 0.0),
'period': 'week',
'limit_type': 'percent',
'source': 'provider'
}
except (TypeError, ValueError):
pass
return normalized
async def _ensure_session(self):
"""Ensure session is initialized and valid before making requests."""
......
aisbf/providers/codex.py
View file @ 7b4a5b2d
......@@ -996,7 +996,24 @@ class CodexProviderHandler(BaseProviderHandler):
data = response.json()
import json as _json
logger.debug(f"CodexProviderHandler: Usage raw body for {self.provider_id}:\n{_json.dumps(data, indent=2)}")
return data
normalized = dict(data)
weekly_limit = data.get('weekly_requests_limit') or data.get('weekly_limit') or data.get('free_requests_per_week')
weekly_used = data.get('weekly_requests_used') or data.get('used_requests') or data.get('requests_used')
if weekly_limit is not None:
try:
weekly_limit_int = int(weekly_limit)
weekly_used_int = int(weekly_used or 0)
normalized['free_tier'] = {
'limit': weekly_limit_int,
'used': weekly_used_int,
'remaining': max(weekly_limit_int - weekly_used_int, 0),
'period': 'week',
'limit_type': 'requests',
'source': 'provider'
}
except (TypeError, ValueError):
pass
return self.normalize_usage_data(normalized)
except Exception as e:
logger.warning(f"CodexProviderHandler: Failed to fetch usage: {e}")
return None
aisbf/providers/kiro/handler.py
View file @ 7b4a5b2d
......@@ -29,7 +29,7 @@ import json
import uuid
import logging
from pathlib import Path
from typing import Dict, List, Optional, Union
from typing import Any, Dict, List, Optional, Union
from ...config import config
from ...models import Model
......@@ -366,6 +366,30 @@ class KiroProviderHandler(BaseProviderHandler):
return openai_response
def normalize_usage_data(self, usage_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
normalized = super().normalize_usage_data(usage_data)
if not normalized or not isinstance(normalized, dict):
return normalized
monthly_limit = normalized.get('monthly_requests_limit') or normalized.get('free_requests_per_month')
monthly_used = normalized.get('monthly_requests_used') or normalized.get('used_requests')
if monthly_limit is not None:
try:
monthly_limit_int = int(monthly_limit)
monthly_used_int = int(monthly_used or 0)
normalized['free_tier'] = {
'limit': monthly_limit_int,
'used': monthly_used_int,
'remaining': max(monthly_limit_int - monthly_used_int, 0),
'period': 'month',
'limit_type': 'requests',
'source': 'provider'
}
except (TypeError, ValueError):
pass
return normalized
async def _handle_streaming_request(self, kiro_api_url: str, payload: dict, headers: dict, model: str):
"""Handle streaming request to Kiro API."""
logger = logging.getLogger(__name__)
......
aisbf/routes/dashboard/providers.py
View file @ 7b4a5b2d
......@@ -7,7 +7,9 @@ from datetime import datetime, timedelta
from aisbf.database import DatabaseRegistry
from aisbf.database import _hash_password as _db_hash_password
from aisbf import __version__
from aisbf.studio import build_studio_catalog, stamp_inferred_capabilities
from aisbf.studio import build_studio_catalog, stamp_inferred_capabilities, serialize_studio_capability_choices, derive_aggregate_capabilities, normalize_capabilities
from aisbf.studio_adapters import serialize_studio_adapter_choices, serialize_studio_adapter_profile_choices, effective_studio_adapter, infer_studio_adapter_profile
from aisbf.studio_services import studio_service
from aisbf.app.templates import url_for, get_base_url
from aisbf.app.startup import _reload_global_config, _apply_condense_defaults_provider, _apply_condense_defaults_rotation, _providers_json_path, _rotations_json_path, _autoselect_json_path, _claude_cli_mode
from aisbf.app.middleware import _is_local_client
......@@ -23,16 +25,80 @@ _server_config = None
logger = logging.getLogger(__name__)
def _serialize_provider_usage_snapshot(snapshot):
if not snapshot:
return None
last_updated = snapshot.get('last_updated')
if isinstance(last_updated, datetime):
last_updated = last_updated.isoformat()
return {
'usage_data': snapshot.get('usage_data'),
'last_updated': last_updated,
}
def _provider_model_capability_lookup(providers: dict) -> dict[str, list[str]]:
lookup: dict[str, list[str]] = {}
for provider_id, provider_config in (providers or {}).items():
models = provider_config.get('models') if isinstance(provider_config, dict) else getattr(provider_config, 'models', None)
provider_type = provider_config.get('type', 'openai') if isinstance(provider_config, dict) else getattr(provider_config, 'type', 'openai')
for model in models or []:
stamped = stamp_inferred_capabilities(model if isinstance(model, dict) else model.model_dump(), provider_type)
model_name = stamped.get('name') or stamped.get('id')
if model_name:
lookup[f"{provider_id}/{model_name}"] = normalize_capabilities(stamped.get('studio_capabilities'))
return lookup
def _rotation_inherited_capabilities(rotation_config: dict, provider_lookup: dict[str, list[str]]) -> dict:
capability_sets = []
for provider in rotation_config.get('providers') or []:
provider_id = provider.get('provider_id')
models = provider.get('models') or []
for model in models:
model_name = model.get('name')
if provider_id and model_name:
capability_sets.append(provider_lookup.get(f"{provider_id}/{model_name}", []))
derived = derive_aggregate_capabilities(capability_sets)
rotation_config['capabilities'] = derived.capabilities
rotation_config['partial_capabilities'] = derived.partial_capabilities
return rotation_config
def _autoselect_inherited_capabilities(autoselect_config: dict, provider_lookup: dict[str, list[str]], rotations: dict | None = None) -> dict:
capability_sets = []
rotations = rotations or {}
for model in autoselect_config.get('available_models') or []:
model_id = (model.get('model_id') or '').strip()
if not model_id:
continue
if model_id in rotations:
rot_caps = normalize_capabilities(rotations[model_id].get('capabilities'))
capability_sets.append(rot_caps)
else:
capability_sets.append(provider_lookup.get(model_id, []))
derived = derive_aggregate_capabilities(capability_sets)
autoselect_config['capabilities'] = derived.capabilities
autoselect_config['partial_capabilities'] = derived.partial_capabilities
return autoselect_config
def _stamp_provider_models(provider_config: dict) -> dict:
stamped = dict(provider_config)
provider_id = stamped.get("provider_id") or ""
provider_type = stamped.get("type", "openai")
models = stamped.get("models")
if isinstance(models, list):
stamped["models"] = [
stamp_inferred_capabilities(model, provider_type)
if isinstance(model, dict) else model
for model in models
]
stamped_models = []
for model in models:
if not isinstance(model, dict):
stamped_models.append(model)
continue
stamped_model = stamp_inferred_capabilities(model, provider_type)
stamped_model["studio_adapter"] = effective_studio_adapter(provider_type, stamped_model)
stamped_model["studio_adapter_profile"] = infer_studio_adapter_profile(provider_id, provider_type, {**stamped_model, "provider_endpoint": stamped.get("endpoint")})
stamped_models.append(stamped_model)
stamped["models"] = stamped_models
return stamped
def init(config, templates, server_config=None):
......@@ -226,6 +292,9 @@ async def dashboard_providers(request: Request):
"session": request.session,
"__version__": __version__,
"providers_json": json.dumps(providers_data),
"studio_capability_choices_json": json.dumps(serialize_studio_capability_choices()),
"studio_adapter_choices_json": json.dumps(serialize_studio_adapter_choices()),
"studio_adapter_profile_choices_json": json.dumps(serialize_studio_adapter_profile_choices()),
"claude_cli_mode": _claude_cli_mode,
"is_local_client": _is_local_client(request),
"success": "Configuration saved successfully!" if success else None
......@@ -241,6 +310,9 @@ async def dashboard_providers(request: Request):
"session": request.session,
"__version__": __version__,
"user_providers_json": json.dumps(providers_data),
"studio_capability_choices_json": json.dumps(serialize_studio_capability_choices()),
"studio_adapter_choices_json": json.dumps(serialize_studio_adapter_choices()),
"studio_adapter_profile_choices_json": json.dumps(serialize_studio_adapter_profile_choices()),
"user_id": current_user_id,
"claude_cli_mode": _claude_cli_mode,
"is_local_client": _is_local_client(request),
......@@ -257,8 +329,9 @@ async def dashboard_studio(request: Request):
return auth_check
current_user_id = request.session.get("user_id")
scope = "admin" if request.session.get("role") == "admin" else "user"
db = None if scope == "admin" else DatabaseRegistry.get_config_database()
is_config_admin = request.session.get("role") == "admin" and current_user_id is None
scope = "admin" if is_config_admin else "user"
db = None if is_config_admin else DatabaseRegistry.get_config_database()
catalog = build_studio_catalog(
scope=scope,
owner_id=current_user_id,
......@@ -274,6 +347,10 @@ async def dashboard_studio(request: Request):
"session": request.session,
"__version__": __version__,
"studio_bootstrap_json": json.dumps(catalog),
"studio_root_path_json": json.dumps("/api/v1") if is_config_admin else json.dumps(f"/api/u/{request.session.get('username', '')}"),
"studio_username_json": json.dumps(request.session.get("username", "")),
"studio_is_global_admin_json": json.dumps(is_config_admin),
"studio_system_prompt_json": json.dumps(studio_service.load_studio_system_prompt(scope, current_user_id)),
"studio_body_mode": "wide",
},
)
......@@ -507,6 +584,9 @@ async def dashboard_providers_save(request: Request, config: str = Form(...)):
"session": request.session,
"__version__": __version__,
"providers_json": json.dumps(providers_data),
"studio_capability_choices_json": json.dumps(serialize_studio_capability_choices()),
"studio_adapter_choices_json": json.dumps(serialize_studio_adapter_choices()),
"studio_adapter_profile_choices_json": json.dumps(serialize_studio_adapter_profile_choices()),
"claude_cli_mode": _claude_cli_mode,
"is_local_client": _is_local_client(request),
"success": success_msg
......@@ -523,6 +603,9 @@ async def dashboard_providers_save(request: Request, config: str = Form(...)):
"session": request.session,
"__version__": __version__,
"user_providers_json": json.dumps(providers_data),
"studio_capability_choices_json": json.dumps(serialize_studio_capability_choices()),
"studio_adapter_choices_json": json.dumps(serialize_studio_adapter_choices()),
"studio_adapter_profile_choices_json": json.dumps(serialize_studio_adapter_profile_choices()),
"user_id": current_user_id,
"claude_cli_mode": _claude_cli_mode,
"is_local_client": _is_local_client(request),
......@@ -555,6 +638,9 @@ async def dashboard_providers_save(request: Request, config: str = Form(...)):
"session": request.session,
"__version__": __version__,
"providers_json": json.dumps(providers_data),
"studio_capability_choices_json": json.dumps(serialize_studio_capability_choices()),
"studio_adapter_choices_json": json.dumps(serialize_studio_adapter_choices()),
"studio_adapter_profile_choices_json": json.dumps(serialize_studio_adapter_profile_choices()),
"claude_cli_mode": _claude_cli_mode,
"is_local_client": _is_local_client(request),
"error": f"Invalid JSON: {str(e)}"
......@@ -572,6 +658,9 @@ async def dashboard_providers_save(request: Request, config: str = Form(...)):
"session": request.session,
"__version__": __version__,
"user_providers_json": json.dumps(user_providers),
"studio_capability_choices_json": json.dumps(serialize_studio_capability_choices()),
"studio_adapter_choices_json": json.dumps(serialize_studio_adapter_choices()),
"studio_adapter_profile_choices_json": json.dumps(serialize_studio_adapter_profile_choices()),
"user_id": current_user_id,
"claude_cli_mode": _claude_cli_mode,
"is_local_client": _is_local_client(request),
......@@ -934,6 +1023,8 @@ async def dashboard_rotations_save(request: Request, config: str = Form(...)):
try:
rotations_data = json.loads(config)
provider_source = getattr(_config, 'providers', {}) if is_config_admin else {row['provider_id']: row['config'] for row in DatabaseRegistry.get_config_database().get_user_providers(current_user_id)}
provider_lookup = _provider_model_capability_lookup(provider_source)
# Apply defaults: if condense_method is set but condense_context is not, default to 80
if 'rotations' in rotations_data:
......@@ -945,6 +1036,7 @@ async def dashboard_rotations_save(request: Request, config: str = Form(...)):
if 'condense_method' in model and model.get('condense_method'):
if 'condense_context' not in model or model.get('condense_context') is None:
model['condense_context'] = 80
rotations_data['rotations'][rotation_key] = _rotation_inherited_capabilities(rotation, provider_lookup)
if is_config_admin:
# Config admin: save to JSON files
......@@ -1213,6 +1305,9 @@ async def dashboard_autoselect_save(request: Request, config: str = Form(...)):
try:
autoselect_data = json.loads(config)
provider_source = getattr(_config, 'providers', {}) if is_config_admin else {row['provider_id']: row['config'] for row in DatabaseRegistry.get_config_database().get_user_providers(current_user_id)}
provider_lookup = _provider_model_capability_lookup(provider_source)
rotation_source = getattr(_config, 'rotations', {}) if is_config_admin else {row['rotation_id']: row['config'] for row in DatabaseRegistry.get_config_database().get_user_rotations(current_user_id)}
# Sanitize every autoselect entry before saving
for key, cfg in autoselect_data.items():
......@@ -1225,6 +1320,7 @@ async def dashboard_autoselect_save(request: Request, config: str = Form(...)):
# Default selection_model to "internal" when blank
if not (cfg.get('selection_model') or '').strip():
cfg['selection_model'] = 'internal'
autoselect_data[key] = _autoselect_inherited_capabilities(cfg, provider_lookup, rotation_source)
if is_config_admin:
# Config admin: save to JSON files
......@@ -1580,6 +1676,7 @@ async def api_provider_usage(request: Request, provider_id: str):
if cached:
return JSONResponse({"success": True, "supported": True, "usage": cached['usage_data'], "stale": True})
return JSONResponse({"success": True, "supported": True, "usage": None})
usage_data = handler.normalize_usage_data(usage_data)
db.save_provider_usage(current_user_id, provider_id, usage_data)
_apply_usage_disable(db, current_user_id, provider_id, usage_data)
return JSONResponse({"success": True, "supported": True, "usage": usage_data})
......@@ -1871,18 +1968,48 @@ async def dashboard_analytics(
provider_filter=effective_provider_filter, model_filter=effective_model_filter,
rotation_filter=rotation_filter, autoselect_filter=autoselect_filter
)
optimization_savings = analytics.get_savings_overview(
from_datetime,
to_datetime,
user_filter=user_filter_int,
provider_filter=effective_provider_filter,
model_filter=effective_model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter
)
date_range_usage = None
if from_datetime or to_datetime:
start = from_datetime or (datetime.now() - timedelta(days=1))
end = to_datetime or datetime.now()
date_range_usage = analytics.get_token_usage_by_date_range(
effective_provider_filter, start, end, user_filter=user_filter_int)
effective_provider_filter,
start,
end,
user_filter=user_filter_int,
model_filter=effective_model_filter,
rotation_filter=rotation_filter,
autoselect_filter=autoselect_filter
)
rotation_breakdown = analytics.get_rotation_breakdown(
from_datetime, to_datetime, user_filter=user_filter_int, rotation_filter=rotation_filter)
from_datetime,
to_datetime,
user_filter=user_filter_int,
rotation_filter=rotation_filter,
provider_filter=effective_provider_filter,
model_filter=effective_model_filter,
autoselect_filter=autoselect_filter
)
autoselect_breakdown = analytics.get_autoselect_breakdown(
from_datetime, to_datetime, user_filter=user_filter_int, autoselect_filter=autoselect_filter)
from_datetime,
to_datetime,
user_filter=user_filter_int,
autoselect_filter=autoselect_filter,
provider_filter=effective_provider_filter,
model_filter=effective_model_filter,
rotation_filter=rotation_filter
)
is_config_admin = is_admin and current_user_id is None
......@@ -1904,7 +2031,7 @@ async def dashboard_analytics(
"model_performance": model_performance,
"cost_overview": cost_overview,
"recommendations": [],
"optimization_savings": 0,
"optimization_savings": optimization_savings,
"selected_time_range": time_range,
"from_date": from_date,
"to_date": to_date,
......@@ -1925,3 +2052,68 @@ async def dashboard_analytics(
"autoselect_breakdown": autoselect_breakdown,
}
)
@router.get("/dashboard/analytics/provider-quotas", response_class=HTMLResponse)
async def dashboard_analytics_provider_quotas(request: Request):
auth_check = require_dashboard_auth(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
current_user_id = request.session.get('user_id')
is_admin = request.session.get('role') == 'admin'
provider_ids = []
if _config and getattr(_config, 'providers', None):
provider_ids.extend(list(_config.providers.keys()))
if current_user_id is not None:
try:
for provider in db.get_user_providers(current_user_id):
provider_id = provider.get('provider_id')
if provider_id and provider_id not in provider_ids:
provider_ids.append(provider_id)
except Exception:
pass
provider_rows = []
analytics = None
try:
from aisbf.analytics import get_analytics
analytics = get_analytics(db)
except Exception:
analytics = None
for provider_id in sorted(set(provider_ids)):
global_snapshot = _serialize_provider_usage_snapshot(db.get_provider_usage(None, provider_id)) if is_admin else None
user_snapshot = _serialize_provider_usage_snapshot(db.get_provider_usage(current_user_id, provider_id)) if current_user_id is not None else None
free_tier_info = analytics._get_provider_free_tier_info(provider_id) if analytics else None
normalized_quota = None
if analytics:
source_usage = None
if user_snapshot and user_snapshot.get('usage_data'):
source_usage = user_snapshot['usage_data']
elif global_snapshot and global_snapshot.get('usage_data'):
source_usage = global_snapshot['usage_data']
normalized_quota = analytics._derive_quota_from_usage(provider_id, source_usage)
provider_rows.append({
'provider_id': provider_id,
'free_tier_info': free_tier_info,
'normalized_quota': normalized_quota,
'global_snapshot': global_snapshot,
'user_snapshot': user_snapshot,
})
return _templates.TemplateResponse(
request=request,
name="dashboard/provider_quotas.html",
context={
'request': request,
'session': request.session,
'is_admin': is_admin,
'provider_rows': provider_rows,
}
)
config/providers.json
View file @ 7b4a5b2d
{
"providers": {
"codex": {
"id": "codex",
"name": "Codex / ChatGPT OAuth2",
"endpoint": "https://chatgpt.com/backend-api",
"type": "codex",
"api_key_required": false,
"is_subscription": true,
"free_tier_limit": 1,
"free_tier_period": "week",
"free_tier_limit_type": "requests",
"premium_reference_monthly_cost": 20,
"free_tier_description": "Codex upstream weekly free-tier equivalent",
"rate_limit": 0,
"models": []
},
"kiro": {
"id": "kiro",
"name": "Kiro CLI",
"endpoint": "https://api.kiro.dev",
"type": "kiro",
"api_key_required": false,
"is_subscription": true,
"free_tier_limit": 50,
"free_tier_period": "month",
"free_tier_limit_type": "requests",
"premium_reference_monthly_cost": 19,
"free_tier_description": "Kiro upstream 50 requests/month free tier",
"rate_limit": 0,
"models": []
},
"claude": {
"id": "claude",
"name": "Claude OAuth2",
"endpoint": "https://api.anthropic.com",
"type": "claude",
"api_key_required": false,
"is_subscription": true,
"free_tier_limit": 1,
"free_tier_period": "month",
"free_tier_limit_type": "requests",
"premium_reference_monthly_cost": 20,
"free_tier_description": "Claude upstream subscription/free-tier equivalent",
"rate_limit": 0,
"models": []
},
"kilo": {
"id": "kilo",
"name": "Kilocode (OAuth2)",
......@@ -7,6 +52,11 @@
"type": "kilo",
"api_key_required": false,
"is_subscription": true,
"free_tier_limit": 1,
"free_tier_period": "month",
"free_tier_limit_type": "requests",
"premium_reference_monthly_cost": 0,
"free_tier_description": "Kilo upstream free-tier/subscription equivalent",
"rate_limit": 0,
"kilo_config": {
"credentials_file": "~/.kilo_credentials.json"
......
setup.py
View file @ 7b4a5b2d
......@@ -155,6 +155,7 @@ setup(
'config/autoselect.md',
'config/condensation_conversational.md',
'config/condensation_semantic.md',
'config/STUDIO_SYSTEM.md',
'config/aisbf.json',
]),
# Install aisbf package to share directory for venv installation
......@@ -307,6 +308,7 @@ setup(
'templates/dashboard/providers.html',
'templates/dashboard/pricing.html',
'templates/dashboard/paypal_connect.html',
'templates/dashboard/provider_quotas.html',
'templates/dashboard/error.html',
'templates/dashboard/forgot_password.html',
'templates/dashboard/rate_limits.html',
......
templates/dashboard/analytics.html
View file @ 7b4a5b2d
......@@ -603,8 +603,14 @@ window.onclick = function(event) {
<div style="background: #3498db; color: white; padding: 20px; border-radius: 8px;">
<h4 style="font-size: 14px; margin-bottom: 10px;">💰 Estimated Savings</h4>
<p style="font-size: 28px; font-weight: bold;" id="savings-amount">{{ currency_symbol }}0.00</p>
<small style="color: rgba(255,255,255,0.8);">From cache hits & optimization</small>
<p style="font-size: 28px; font-weight: bold;" id="savings-amount">{{ currency_symbol }}{{ "%.2f"|format((optimization_savings.total_cost_saved if optimization_savings else 0) or 0) }}</p>
<small style="color: rgba(255,255,255,0.8);">
{% if optimization_savings and optimization_savings.provider_equivalents %}
Provider free-tier equivalents + optimization
{% else %}
Cache hits & optimization
{% endif %}
</small>
</div>
{% for pc in cost_overview.providers %}
......@@ -635,6 +641,48 @@ window.onclick = function(event) {
</div>
</div>
{% if optimization_savings.provider_equivalents %}
<div style="background: var(--bg-page); padding: 16px; border-radius: 8px; margin-bottom: 20px;">
<h4 style="margin: 0 0 10px 0;">Free-Tier Equivalent Savings</h4>
<p style="margin: 0 0 10px 0; color: var(--color-muted);">
The table below estimates provider-side free-tier equivalents using known upstream free limits such as weekly or monthly request allowances.
When the filtered usage exceeds a provider's base free allowance, the overflow is counted as additional free-tier-equivalent capacity and translated into token and premium-value equivalents.
</p>
<table>
<tr>
<th>Provider</th>
<th>Used Tokens</th>
<th>Requests</th>
<th>Measured Usage</th>
<th>Free Limit</th>
<th>Extra Free Tiers</th>
<th>Equivalent Saved Tokens</th>
<th>Premium Reference</th>
<th>Equivalent Saved Cost</th>
</tr>
{% for item in optimization_savings.provider_equivalents %}
<tr>
<td><strong>{{ item.provider_id }}</strong></td>
<td>{{ format_tokens(item.tokens_used) }}</td>
<td>{{ item.request_count }}</td>
<td>{{ item.usage_amount }} {{ item.free_tier_limit_type }}</td>
<td>{{ item.free_tier_limit }}/{{ item.free_tier_period }} {{ item.free_tier_limit_type }} <small style="color: var(--color-muted);">({{ item.quota_source }})</small></td>
<td>{{ item.extra_free_tiers }}</td>
<td>{{ format_tokens(item.equivalent_saved_tokens) }}</td>
<td>
{% if item.premium_reference_name %}
{{ item.premium_reference_name }} ({{ currency_symbol }}{{ "%.2f"|format(item.premium_reference_monthly_cost) }}/mo)
{% else %}
N/A
{% endif %}
</td>
<td>{{ currency_symbol }}{{ "%.2f"|format(item.equivalent_saved_cost) }}</td>
</tr>
{% endfor %}
</table>
</div>
{% endif %}
{% if optimization_savings.savings_by_type %}
<table>
<tr>
......@@ -666,26 +714,9 @@ window.onclick = function(event) {
</table>
{% endif %}
{% else %}
<p style="color: var(--color-muted);">No optimization savings recorded yet. Enable deduplication, condensation, batching, or caching to see savings.</p>
<p style="color: var(--color-muted);">No optimization savings recorded yet. Increase usage on providers with known upstream free tiers, or enable deduplication, condensation, batching, or caching to see savings.</p>
{% endif %}
<script>
// Fetch and display savings estimation
fetch('{{ url_for(request, "/dashboard/response-cache/stats") }}')
.then(response => response.json())
.then(data => {
if (data.enabled && data.hits > 0) {
// Estimate savings: assume average cost of $0.002 per cached request
const avgCostPerRequest = 0.002;
const estimatedSavings = data.hits * avgCostPerRequest;
document.getElementById('savings-amount').textContent = '{{ currency_symbol }}' + estimatedSavings.toFixed(2);
}
})
.catch(error => {
console.error('Error fetching cache stats:', error);
});
</script>
<h3 style="margin-top: 30px; margin-bottom: 15px;">Model Performance</h3>
{% if model_performance %}
<table>
......@@ -1015,9 +1046,10 @@ function confirmDeleteAnalytics(scope) {
<div style="margin-top: 30px; display: flex; gap: 10px; flex-wrap: wrap;">
<a href="/dashboard" class="btn btn-secondary" data-i18n="analytics_page.back">Back to Dashboard</a>
<a href="{{ url_for(request, '/dashboard/analytics/provider-quotas') }}" class="btn btn-secondary">Provider Quota Debug</a>
{% if is_admin %}
<a href="{{ url_for(request, '/dashboard/rate-limits') }}" class="btn">Rate Limits</a>
<a href="{{ url_for(request, '/dashboard/response-cache') }}" class="btn">Response Cache</a>
{% endif %}
</div>
{% endblock %}
\ No newline at end of file
{% endblock %}
templates/dashboard/provider_quotas.html 0 → 100644
View file @ 7b4a5b2d
{% extends "base.html" %}
{% block title %}Provider Quotas - AISBF Dashboard{% endblock %}
{% block content %}
<h2 style="margin-bottom: 20px;">Provider Quota Debug</h2>
<p style="color: var(--color-muted); margin-bottom: 20px;">
Debug view for normalized provider quota/free-tier data persisted from upstream usage endpoints. This page is intentionally not linked from the main navigation.
</p>
{% if provider_rows %}
<table>
<tr>
<th>Provider</th>
<th>Configured Free Tier</th>
<th>Normalized Runtime Quota</th>
<th>User Snapshot</th>
{% if is_admin %}<th>Global Snapshot</th>{% endif %}
</tr>
{% for row in provider_rows %}
<tr>
<td><strong>{{ row.provider_id }}</strong></td>
<td>
{% if row.free_tier_info %}
<div><strong>{{ row.free_tier_info.description or row.provider_id }}</strong></div>
<div style="color: var(--color-muted); font-size: 13px;">
{{ row.free_tier_info.limit }}/{{ row.free_tier_info.period }} {{ row.free_tier_info.limit_type }}
· {{ row.free_tier_info.source }}
</div>
{% else %}
<span style="color: var(--color-muted);">N/A</span>
{% endif %}
</td>
<td>
{% if row.normalized_quota %}
<div class="code">{{ row.normalized_quota | tojson(indent=2) }}</div>
{% else %}
<span style="color: var(--color-muted);">N/A</span>
{% endif %}
</td>
<td>
{% if row.user_snapshot %}
<div style="margin-bottom: 8px; color: var(--color-muted); font-size: 13px;">{{ row.user_snapshot.last_updated or 'unknown time' }}</div>
<div class="code">{{ row.user_snapshot.usage_data | tojson(indent=2) }}</div>
{% else %}
<span style="color: var(--color-muted);">N/A</span>
{% endif %}
</td>
{% if is_admin %}
<td>
{% if row.global_snapshot %}
<div style="margin-bottom: 8px; color: var(--color-muted); font-size: 13px;">{{ row.global_snapshot.last_updated or 'unknown time' }}</div>
<div class="code">{{ row.global_snapshot.usage_data | tojson(indent=2) }}</div>
{% else %}
<span style="color: var(--color-muted);">N/A</span>
{% endif %}
</td>
{% endif %}
</tr>
{% endfor %}
</table>
{% else %}
<p style="color: var(--color-muted);">No provider quota data available.</p>
{% endif %}
<div style="margin-top: 24px; display: flex; gap: 10px; flex-wrap: wrap;">
<a href="{{ url_for(request, '/dashboard/analytics') }}" class="btn btn-secondary">Back to Analytics</a>
</div>
{% endblock %}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment