595 lines
20 KiB
Python
595 lines
20 KiB
Python
"""sage_datamart 模块初始化 — 统一 Dashboard API，支持客户/分销商角色"""
|
||
from ahserver.serverenv import ServerEnv
|
||
from appPublic.log import debug
|
||
from sqlor.dbpools import get_sor_context
|
||
from .etl import sync_call_fact, aggregate_daily_perf, aggregate_provider_cost
|
||
|
||
from .dashboards import (
|
||
CUSTOMER, DISTRIBUTOR,
|
||
get_today_usage, get_today_amount, get_success_rate, get_fail_count,
|
||
get_active_users, get_top_models, get_top_providers, get_top_users,
|
||
get_customer_daily_models, get_customer_monthly_models,
|
||
get_customer_user_today, get_daily_trend, get_hourly_concurrency,
|
||
get_avg_ttft, get_currency_breakdown,
|
||
get_realtime_calls, get_realtime_amount, get_total_users_count,
|
||
get_realtime_active_users, get_online_users, get_realtime_stats,
|
||
get_distributor_customers_count, get_distributor_active_customers,
|
||
get_distributor_top_customers,
|
||
get_scoped_today_usage, get_scoped_today_amount,
|
||
get_scoped_success_rate, get_scoped_fail_count,
|
||
)
|
||
|
||
MODULE_NAME = "sage_datamart"
|
||
|
||
|
||
async def _resolve_view(request):
|
||
"""解析当前用户角色和机构过滤条件
|
||
返回 (scope, org_filter) — scope: 'customer'|'distributor'|None, org_filter: org_id|None
|
||
org_filter=None 表示不按机构过滤（平台管理员看全部数据）
|
||
"""
|
||
env = request._run_ns
|
||
userorgid = await env.get_userorgid()
|
||
scope = CUSTOMER
|
||
org_filter = userorgid
|
||
# 平台根机构('0')无过滤，看全部数据
|
||
if not userorgid or userorgid == '0':
|
||
return None, None
|
||
try:
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT org_type FROM organization WHERE id = ${id}$",
|
||
{'id': userorgid})
|
||
if recs:
|
||
org_type = (recs[0].org_type if hasattr(recs[0], 'org_type')
|
||
else recs[0].get('org_type', ''))
|
||
if org_type == 'reseller':
|
||
scope = DISTRIBUTOR
|
||
except Exception as e:
|
||
debug("sage_datamart _resolve_view error: " + str(e))
|
||
|
||
pass
|
||
return scope, org_filter
|
||
|
||
|
||
# ── 性能指标 API ──
|
||
|
||
async def api_model_perf(sor, params_kw=None):
|
||
try:
|
||
stat_date = (params_kw or {}).get('date')
|
||
sql = """SELECT COALESCE(l.name, f.model) as model,
|
||
CASE WHEN SUM(total_calls)>0 THEN ROUND(SUM(total_calls*COALESCE(avg_ttft_ms,avg_ttot_ms))/SUM(total_calls),1) ELSE NULL END as avg_latency_ms,
|
||
CASE WHEN SUM(total_calls)>0 THEN ROUND(SUM(total_calls*avg_ttot_ms)/SUM(total_calls),1) ELSE NULL END as avg_ttot_ms,
|
||
ROUND(SUM(success_calls)*100.0/GREATEST(SUM(total_calls),1),1) as success_rate
|
||
FROM dm_model_perf_daily f
|
||
LEFT JOIN llm l ON l.id = f.llmid
|
||
WHERE 1=1"""
|
||
ns = {}
|
||
if stat_date:
|
||
sql += " AND f.stat_date=${stat_date}$"
|
||
ns['stat_date'] = stat_date
|
||
sql += " GROUP BY f.model, f.catelogid, f.providerid"
|
||
sql += " ORDER BY avg_latency_ms ASC LIMIT 5"
|
||
rows = await sor.sqlExe(sql, ns)
|
||
return [dict(r) for r in rows]
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
async def api_provider_roi(sor, params_kw=None):
|
||
try:
|
||
model = (params_kw or {}).get('model')
|
||
stat_date = (params_kw or {}).get('date')
|
||
sql = """SELECT p.*, COALESCE(o.orgname, p.providerid) as provider_name
|
||
FROM dm_provider_cost_daily p
|
||
LEFT JOIN organization o ON o.id = p.providerid
|
||
WHERE 1=1"""
|
||
ns = {}
|
||
if stat_date:
|
||
sql += " AND stat_date=${stat_date}$"
|
||
ns['stat_date'] = stat_date
|
||
if model:
|
||
sql += " AND model=${model}$"
|
||
ns['model'] = model
|
||
sql += " ORDER BY total_amount DESC LIMIT 50"
|
||
rows = await sor.sqlExe(sql, ns)
|
||
return [dict(r) for r in rows]
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
# ── 运营指标 API ──
|
||
|
||
async def api_stats(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return {
|
||
'today_usage': await get_today_usage(env, sor),
|
||
'today_amount': await get_today_amount(env, sor),
|
||
'success_rate': await get_success_rate(env, sor),
|
||
'fail_count': await get_fail_count(env, sor),
|
||
'active_users': await get_active_users(env, sor),
|
||
}
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return {}
|
||
|
||
|
||
async def api_top_models(request):
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_top_models(env, sor, org_filter=org_filter)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
async def api_top_providers(request):
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_top_providers(env, sor, org_filter=org_filter)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
async def api_top_users(request):
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_top_users(env, sor, org_filter=org_filter)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
async def api_customer_models(request):
|
||
try:
|
||
env = request._run_ns
|
||
userorgid = await env.get_userorgid()
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return {
|
||
'daily': await get_customer_daily_models(env, sor, userorgid),
|
||
'monthly': await get_customer_monthly_models(env, sor, userorgid),
|
||
}
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return {'daily': [], 'monthly': []}
|
||
|
||
|
||
async def api_daily_trend(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_daily_trend(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
async def api_hourly_concurrency(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_hourly_concurrency(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
async def api_currency_stats(request):
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_currency_breakdown(env, sor, org_filter=org_filter)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
# ── 实时指标 API（角色感知）──
|
||
|
||
async def api_realtime_cards(request):
|
||
"""实时统计卡片 — /sage_datamart/api/realtime_cards.dspy"""
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
today = env.curDateString()
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
if scope == DISTRIBUTOR:
|
||
return {
|
||
'scope': scope,
|
||
'calls': await get_realtime_calls(sor, today, scope, org_filter),
|
||
'amount': await get_realtime_amount(sor, today, scope, org_filter),
|
||
'total_users': await get_distributor_customers_count(sor, org_filter),
|
||
'active_customers': await get_distributor_active_customers(sor, today, org_filter),
|
||
'online_users': await get_online_users(env),
|
||
'success_rate': await get_scoped_success_rate(sor, today, scope, org_filter),
|
||
'fail_count': await get_scoped_fail_count(sor, today, scope, org_filter),
|
||
}
|
||
else:
|
||
return {
|
||
'scope': scope,
|
||
'calls': await get_realtime_calls(sor, today, scope, org_filter),
|
||
'amount': await get_realtime_amount(sor, today, scope, org_filter),
|
||
'total_users': await get_total_users_count(sor, scope, org_filter),
|
||
'active_users': await get_realtime_active_users(sor, today, scope, org_filter),
|
||
'online_users': await get_online_users(env),
|
||
'success_rate': await get_scoped_success_rate(sor, today, scope, org_filter),
|
||
'fail_count': await get_scoped_fail_count(sor, today, scope, org_filter),
|
||
}
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return {}
|
||
|
||
|
||
# ── 分销商专用 API ──
|
||
|
||
async def api_distributor_customers(request):
|
||
"""分销商客户排行 — /sage_datamart/api/distributor_customers.dspy"""
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_distributor_top_customers(sor, org_filter)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return []
|
||
|
||
|
||
# ── Jinja2 模板函数（实时，角色感知）──
|
||
|
||
async def _get_realtime_snapshot(request):
|
||
"""一次查询返回全部实时指标(含角色)，同一请求内缓存复用"""
|
||
cache_key = '_sage_datamart_rt_cache'
|
||
if cache_key in request:
|
||
return request[cache_key]
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
today = env.curDateString()
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
result = await get_realtime_stats(sor, today, scope, org_filter)
|
||
request[cache_key] = result
|
||
return result
|
||
|
||
|
||
async def j2_realtime_calls(request):
|
||
try:
|
||
s = await _get_realtime_snapshot(request)
|
||
return s.get('calls', 0)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
async def j2_realtime_amount(request):
|
||
try:
|
||
s = await _get_realtime_snapshot(request)
|
||
return s.get('amount', 0)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
async def j2_total_users(request):
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
if scope == DISTRIBUTOR:
|
||
return await get_distributor_customers_count(sor, org_filter)
|
||
return await get_total_users_count(sor, scope, org_filter)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
async def j2_active_today(request):
|
||
try:
|
||
s = await _get_realtime_snapshot(request)
|
||
return s.get('active_users', 0)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
async def j2_online_users(request):
|
||
try:
|
||
env = request._run_ns
|
||
return await get_online_users(env)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
async def j2_scoped_success_rate(request):
|
||
try:
|
||
s = await _get_realtime_snapshot(request)
|
||
return s.get('success_rate', 0)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
async def j2_scoped_fail_count(request):
|
||
try:
|
||
s = await _get_realtime_snapshot(request)
|
||
return s.get('fail_count', 0)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return 0
|
||
|
||
|
||
# ── 向后兼容 Jinja2 函数 ──
|
||
|
||
async def j2_today_usage(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_today_usage(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return 0
|
||
|
||
|
||
async def j2_today_amount(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_today_amount(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return 0
|
||
|
||
|
||
async def j2_active_users(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_active_users(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return 0
|
||
|
||
|
||
async def j2_fail_count(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_fail_count(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return 0
|
||
|
||
|
||
async def j2_success_rate(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_success_rate(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return 0
|
||
|
||
|
||
async def j2_avg_ttft(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_avg_ttft(env, sor)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
|
||
return 0
|
||
|
||
|
||
|
||
async def api_user_apikeys(request):
|
||
"""当前用户每个 API Key 的使用排行"""
|
||
try:
|
||
env = request._run_ns
|
||
userid = await env.get_userid()
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
sql = """SELECT uak.name as apikey_name, COUNT(*) as cnt, COALESCE(SUM(lu.amount), 0) as total_amount
|
||
FROM llmusage lu
|
||
JOIN userapikey uak ON uak.apikey = lu.apikey
|
||
WHERE lu.userid = ${userid}$
|
||
GROUP BY uak.name ORDER BY cnt DESC LIMIT 10"""
|
||
recs = await sor.sqlExe(sql, {'userid': userid})
|
||
return [{'apikey_name': r.apikey_name or 'Unknown',
|
||
'cnt': int(r.cnt or 0),
|
||
'total_amount': round(float(r.total_amount or 0), 2)} for r in recs]
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return []
|
||
|
||
|
||
async def api_distributor_financial(request):
|
||
"""分销商财务统计 API"""
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
period = (request._run_ns.params_kw or {}).get('period', 'today')
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_distributor_financial(sor, org_filter, period)
|
||
except Exception as e:
|
||
debug("sage_datamart API error: " + str(e))
|
||
return {}
|
||
|
||
|
||
async def j2_distributor_financial(request, period='today'):
|
||
"""Jinja2: 分销商财务数据"""
|
||
try:
|
||
env = request._run_ns
|
||
scope, org_filter = await _resolve_view(request)
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_distributor_financial(sor, org_filter, period)
|
||
except Exception:
|
||
return {'total_revenue_cny': 0, 'total_cost_cny': 0, 'gross_profit_cny': 0,
|
||
'revenue_by_currency': [], 'total_calls': 0, 'period': period}
|
||
|
||
|
||
async def j2_is_distributor(request):
|
||
"""Jinja2: 是否分销商角色"""
|
||
try:
|
||
scope, _ = await _resolve_view(request)
|
||
return scope == DISTRIBUTOR
|
||
except Exception:
|
||
return False
|
||
|
||
# ── 监控组件 j2 函数，每段独立 .ui 通过 RefreshWidget 加载 ──
|
||
|
||
async def j2_model_perf_data(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await api_model_perf(sor)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
async def j2_currency_data(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_currency_breakdown(env, sor)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
async def j2_provider_roi_data(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await api_provider_roi(sor)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
async def j2_top_models_data(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_top_models(env, sor)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
async def j2_top_users_data(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_top_users(env, sor)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
async def j2_top_providers_data(request):
|
||
try:
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
return await get_top_providers(env, sor)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
# ── Cron ETL ──
|
||
|
||
async def cron_etl_sync(request):
|
||
ip = request['client_ip']
|
||
if ip not in ['127.0.0.1']:
|
||
return {'error': f'IP {ip} not allowed'}
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
count = await sync_call_fact(sor)
|
||
return {'status': 'ok', 'synced': count}
|
||
|
||
|
||
async def cron_etl_aggregate(request):
|
||
ip = request['client_ip']
|
||
if ip not in ['127.0.0.1']:
|
||
return {'error': f'IP {ip} not allowed'}
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
count = await aggregate_daily_perf(sor)
|
||
return {'status': 'ok', 'aggregated': count}
|
||
|
||
|
||
async def cron_etl_provider_cost(request):
|
||
ip = request['client_ip']
|
||
if ip not in ['127.0.0.1']:
|
||
return {'error': f'IP {ip} not allowed'}
|
||
env = request._run_ns
|
||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||
count = await aggregate_provider_cost(sor)
|
||
return {'status': 'ok', 'aggregated': count}
|
||
|
||
|
||
def load_sage_datamart():
|
||
env = ServerEnv()
|
||
env.api_model_perf = api_model_perf
|
||
env.api_provider_roi = api_provider_roi
|
||
env.api_stats = api_stats
|
||
env.api_top_models = api_top_models
|
||
env.api_top_providers = api_top_providers
|
||
env.api_top_users = api_top_users
|
||
env.api_customer_models = api_customer_models
|
||
env.api_daily_trend = api_daily_trend
|
||
env.api_hourly_concurrency = api_hourly_concurrency
|
||
env.api_currency_stats = api_currency_stats
|
||
env.api_realtime_cards = api_realtime_cards
|
||
env.api_distributor_customers = api_distributor_customers
|
||
env.j2_today_usage = j2_today_usage
|
||
env.j2_today_amount = j2_today_amount
|
||
env.j2_active_users = j2_active_users
|
||
env.j2_fail_count = j2_fail_count
|
||
env.j2_success_rate = j2_success_rate
|
||
env.j2_avg_ttft = j2_avg_ttft
|
||
env.j2_realtime_calls = j2_realtime_calls
|
||
env.j2_realtime_amount = j2_realtime_amount
|
||
env.j2_total_users = j2_total_users
|
||
env.j2_active_today = j2_active_today
|
||
env.j2_online_users = j2_online_users
|
||
env.j2_scoped_success_rate = j2_scoped_success_rate
|
||
env.j2_scoped_fail_count = j2_scoped_fail_count
|
||
env.api_user_apikeys = api_user_apikeys
|
||
env.api_distributor_financial = api_distributor_financial
|
||
env.j2_distributor_financial = j2_distributor_financial
|
||
env.j2_is_distributor = j2_is_distributor
|
||
env.j2_model_perf_data = j2_model_perf_data
|
||
env.j2_currency_data = j2_currency_data
|
||
env.j2_provider_roi_data = j2_provider_roi_data
|
||
env.j2_top_models_data = j2_top_models_data
|
||
env.j2_top_users_data = j2_top_users_data
|
||
env.j2_top_providers_data = j2_top_providers_data
|
||
env.cron_etl_sync = cron_etl_sync
|
||
env.cron_etl_aggregate = cron_etl_aggregate
|
||
env.cron_etl_provider_cost = cron_etl_provider_cost
|
||
|
||
debug("[sage_datamart] registered all API + realtime + role-aware endpoints")
|