chore: remove build/ from tracking
This commit is contained in:
parent
0a8c946f31
commit
cab2aaaf6d
1
.gitignore
vendored
1
.gitignore
vendored
@ -2,3 +2,4 @@ __pycache__/
|
||||
*.pyc
|
||||
*.egg-info/
|
||||
models/mysql.ddl.sql
|
||||
build/
|
||||
|
||||
@ -1,2 +0,0 @@
|
||||
"""sage_datamart — Sage 数据集市模块"""
|
||||
from .init import load_sage_datamart
|
||||
@ -1,561 +0,0 @@
|
||||
"""
|
||||
统一 Dashboard API — 从集市表 / llmusage 查询运营指标 + 性能指标
|
||||
支持客户(customer)和分销商(distributor)两种视图
|
||||
实时指标直接查 llmusage，非实时从集市表取
|
||||
"""
|
||||
from sqlor.dbpools import get_sor_context
|
||||
from datetime import datetime, timedelta
|
||||
import redis.asyncio as redis
|
||||
|
||||
# ── 角色/范围辅助 ──
|
||||
|
||||
CUSTOMER = 'customer'
|
||||
DISTRIBUTOR = 'distributor'
|
||||
|
||||
|
||||
def _fact_where(scope, org_filter):
|
||||
"""根据角色返回 dm_model_call_fact 的 WHERE 条件"""
|
||||
if scope == DISTRIBUTOR and org_filter:
|
||||
return "distributor_orgid = ${org}$"
|
||||
elif scope == CUSTOMER and org_filter:
|
||||
return "userorgid = ${org}$"
|
||||
return "1=1"
|
||||
|
||||
|
||||
def _usage_where(scope, org_filter):
|
||||
"""根据角色返回 llmusage 的 WHERE 条件"""
|
||||
if scope == DISTRIBUTOR and org_filter:
|
||||
return "distributor_orgid = ${org}$"
|
||||
elif scope == CUSTOMER and org_filter:
|
||||
return "userorgid = ${org}$"
|
||||
return "1=1"
|
||||
|
||||
|
||||
# ── 实时指标卡片（直接查 llmusage，无 ETL 延迟）──
|
||||
|
||||
async def get_realtime_calls(sor, today, scope=None, org_filter=None):
|
||||
"""实时今日调用笔数"""
|
||||
w = _usage_where(scope, org_filter)
|
||||
sql = ("SELECT COUNT(*) as cnt FROM llmusage "
|
||||
"WHERE use_date = ${today}$ AND " + w)
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_realtime_amount(sor, today, scope=None, org_filter=None):
|
||||
"""实时今日交易金额（原币汇总）"""
|
||||
w = _usage_where(scope, org_filter)
|
||||
sql = ("SELECT COALESCE(SUM(amount), 0) as total FROM llmusage "
|
||||
"WHERE use_date = ${today}$ AND " + w)
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return round(float(recs[0].total), 2) if recs else 0
|
||||
|
||||
|
||||
async def get_total_users_count(sor, scope=None, org_filter=None):
|
||||
"""用户总数"""
|
||||
if scope == CUSTOMER and org_filter:
|
||||
sql = "SELECT COUNT(*) as cnt FROM users WHERE orgid = ${org}$"
|
||||
ns = {'org': org_filter}
|
||||
elif scope == DISTRIBUTOR and org_filter:
|
||||
# 分销商下所有客户机构的用户
|
||||
sql = ("SELECT COUNT(DISTINCT u.id) as cnt FROM users u "
|
||||
"JOIN organization o ON o.id = u.orgid "
|
||||
"WHERE o.resellerid = ${org}$")
|
||||
ns = {'org': org_filter}
|
||||
else:
|
||||
sql = "SELECT COUNT(*) as cnt FROM users"
|
||||
ns = {}
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_realtime_active_users(sor, today, scope=None, org_filter=None):
|
||||
"""实时今日活跃用户数"""
|
||||
w = _usage_where(scope, org_filter)
|
||||
sql = ("SELECT COUNT(DISTINCT userid) as cnt FROM llmusage "
|
||||
"WHERE use_date = ${today}$ AND " + w)
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_realtime_stats(sor, today, scope=None, org_filter=None):
|
||||
"""一次查llmusage返回全部实时指标，确保数据一致"""
|
||||
w = _usage_where(scope, org_filter)
|
||||
sql = ("SELECT "
|
||||
"COUNT(*) as calls, "
|
||||
"COALESCE(SUM(amount), 0) as amount, "
|
||||
"COUNT(DISTINCT userid) as active_users, "
|
||||
"SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success, "
|
||||
"SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as fail "
|
||||
"FROM llmusage "
|
||||
"WHERE use_date = ${today}$ AND " + w)
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
if recs:
|
||||
r = recs[0]
|
||||
total = int(r.calls or 0)
|
||||
success = int(r.success or 0)
|
||||
fail = int(r.fail or 0)
|
||||
return {
|
||||
'calls': total,
|
||||
'amount': round(float(r.amount or 0), 2),
|
||||
'active_users': int(r.active_users or 0),
|
||||
'success_rate': round(success * 100.0 / total, 1) if total else 0,
|
||||
'fail_count': fail,
|
||||
}
|
||||
return {'calls': 0, 'amount': 0, 'active_users': 0, 'success_rate': 0, 'fail_count': 0}
|
||||
|
||||
|
||||
async def get_online_users(env, scope=None, org_filter=None):
|
||||
"""在线用户数 — 从 Redis session 中统计"""
|
||||
try:
|
||||
config = env.get_config()
|
||||
redis_url = getattr(config, 'redis_url', 'redis://127.0.0.1:6379')
|
||||
r = redis.from_url(redis_url, decode_responses=True)
|
||||
count = 0
|
||||
cursor = 0
|
||||
while True:
|
||||
cursor, keys = await r.scan(cursor, match='AIOHTTP_SESSION_*', count=100)
|
||||
for k in keys:
|
||||
data = await r.get(k)
|
||||
if data:
|
||||
count += 1
|
||||
if cursor == 0:
|
||||
break
|
||||
await r.close()
|
||||
return count
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
|
||||
async def get_online_users_by_org(env, org_filter, scope=None):
|
||||
"""按机构统计在线用户数"""
|
||||
try:
|
||||
config = env.get_config()
|
||||
redis_url = getattr(config, 'redis_url', 'redis://127.0.0.1:6379')
|
||||
r = redis.from_url(redis_url, decode_responses=True)
|
||||
count = 0
|
||||
cursor = 0
|
||||
while True:
|
||||
cursor, keys = await r.scan(cursor, match='AIOHTTP_SESSION_*', count=100)
|
||||
for k in keys:
|
||||
data = await r.get(k)
|
||||
if data and org_filter in data:
|
||||
count += 1
|
||||
if cursor == 0:
|
||||
break
|
||||
await r.close()
|
||||
return count
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
|
||||
# ── 统计卡片（集市表，ETL 更新）──
|
||||
|
||||
async def get_today_usage(env, sor):
|
||||
"""当天调用总数"""
|
||||
sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$"
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_today_amount(env, sor):
|
||||
"""当天总金额(CNY)"""
|
||||
sql = "SELECT COALESCE(SUM(amount_cny), 0) as total FROM dm_model_call_fact WHERE call_date = ${today}$"
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||||
return round(float(recs[0].total), 2) if recs else 0
|
||||
|
||||
|
||||
async def get_success_rate(env, sor):
|
||||
"""当天成功率"""
|
||||
sql = """SELECT
|
||||
COUNT(*) as total,
|
||||
SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success
|
||||
FROM dm_model_call_fact WHERE call_date = ${today}$"""
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||||
if recs and recs[0].total:
|
||||
return round(recs[0].success * 100.0 / recs[0].total, 1)
|
||||
return 0
|
||||
|
||||
|
||||
async def get_fail_count(env, sor):
|
||||
"""当天失败数"""
|
||||
sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$ AND status = 'FAILED'"
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_active_users(env, sor):
|
||||
"""当天活跃用户数"""
|
||||
sql = "SELECT COUNT(DISTINCT userid) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$"
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_avg_ttft(env, sor):
|
||||
"""当天平均 TTFT"""
|
||||
sql = "SELECT AVG(avg_ttft_ms) as v FROM dm_model_perf_daily WHERE stat_date = ${today}$"
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||||
if recs and recs[0].v:
|
||||
return round(float(recs[0].v), 0)
|
||||
return 0
|
||||
|
||||
|
||||
async def get_currency_breakdown(env, sor, org_filter=None):
|
||||
"""各币种分布（全量+可选org过滤）"""
|
||||
w = ""
|
||||
ns = {}
|
||||
if org_filter:
|
||||
w = " AND userorgid = ${org}$"
|
||||
ns['org'] = org_filter
|
||||
sql = """SELECT
|
||||
COALESCE(currency, 'CNY') as currency,
|
||||
COUNT(*) as call_count,
|
||||
COALESCE(SUM(amount), 0) as amount,
|
||||
COALESCE(SUM(amount_cny), 0) as amount_cny
|
||||
FROM dm_model_call_fact
|
||||
WHERE 1=1""" + w + """
|
||||
GROUP BY dm_model_call_fact.currency
|
||||
ORDER BY call_count DESC"""
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return [{
|
||||
'currency': r.currency or 'CNY',
|
||||
'call_count': int(r.call_count),
|
||||
'amount': round(float(r.amount), 2),
|
||||
'amount_cny': round(float(r.amount_cny), 2)
|
||||
} for r in recs]
|
||||
|
||||
|
||||
# ── 排行榜 ──
|
||||
|
||||
async def get_top_models(env, sor, limit=5, org_filter=None):
|
||||
"""Top N 模型（按调用次数，全量+可选org过滤）"""
|
||||
w = ""
|
||||
ns = {'limit': limit}
|
||||
if org_filter:
|
||||
w = " AND userorgid = ${org}$"
|
||||
ns['org'] = org_filter
|
||||
sql = ("SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount "
|
||||
"FROM dm_model_call_fact WHERE 1=1" + w +
|
||||
" GROUP BY model ORDER BY cnt DESC LIMIT ${limit}$")
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return [{'model_name': r.model or 'Unknown', 'cnt': int(r.cnt),
|
||||
'total_amount': round(float(r.total_amount), 2)} for r in recs]
|
||||
|
||||
|
||||
async def get_top_providers(env, sor, limit=5, org_filter=None):
|
||||
"""Top N 供应商（按金额，全量+可选org过滤）"""
|
||||
w = ""
|
||||
ns = {'limit': limit}
|
||||
if org_filter:
|
||||
w = " AND f.userorgid = ${org}$"
|
||||
ns['org'] = org_filter
|
||||
sql = """SELECT f.providerid, COUNT(*) as cnt, COALESCE(SUM(f.amount_cny), 0) as total_amount,
|
||||
COALESCE(o.orgname, f.providerid) as provider_name
|
||||
FROM dm_model_call_fact f
|
||||
LEFT JOIN organization o ON o.id = f.providerid
|
||||
WHERE f.providerid IS NOT NULL""" + w + """
|
||||
GROUP BY f.providerid ORDER BY total_amount DESC LIMIT ${limit}$"""
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return [{'provider_name': r.provider_name or r.providerid or 'Unknown',
|
||||
'cnt': int(r.cnt), 'total_amount': round(float(r.total_amount), 2)} for r in recs]
|
||||
|
||||
|
||||
async def get_top_users(env, sor, limit=5, org_filter=None):
|
||||
"""全量 Top N 用户（按金额+可选org过滤）"""
|
||||
w = ""
|
||||
ns = {'limit': limit}
|
||||
if org_filter:
|
||||
w = " AND f.userorgid = ${org}$"
|
||||
ns['org'] = org_filter
|
||||
sql = """SELECT f.userid, COUNT(*) as cnt, COALESCE(SUM(f.amount_cny), 0) as total_amount,
|
||||
COALESCE(u.username, f.userid) as user_name
|
||||
FROM dm_model_call_fact f
|
||||
LEFT JOIN users u ON u.id = f.userid
|
||||
WHERE 1=1""" + w + """
|
||||
GROUP BY f.userid ORDER BY total_amount DESC LIMIT ${limit}$"""
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return [{'user_name': r.user_name or r.userid, 'cnt': int(r.cnt),
|
||||
'total_amount': round(float(r.total_amount), 2)} for r in recs]
|
||||
|
||||
|
||||
# ── 客户维度 ──
|
||||
|
||||
async def get_customer_daily_models(env, sor, userorgid):
|
||||
"""客户当天各模型用量"""
|
||||
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||||
FROM dm_model_call_fact WHERE call_date = ${today}$ AND userorgid = ${orgid}$
|
||||
GROUP BY model ORDER BY cnt DESC"""
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'orgid': userorgid})
|
||||
return [{'model_name': r.model or 'Unknown', 'cnt': int(r.cnt),
|
||||
'total_amount': round(float(r.total_amount), 4)} for r in recs]
|
||||
|
||||
|
||||
async def get_customer_monthly_models(env, sor, userorgid):
|
||||
"""客户当月各模型用量"""
|
||||
now = datetime.now()
|
||||
month_start = now.strftime('%Y-%m-01')
|
||||
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||||
FROM dm_model_call_fact WHERE call_date >= ${start}$ AND userorgid = ${orgid}$
|
||||
GROUP BY model ORDER BY cnt DESC"""
|
||||
recs = await sor.sqlExe(sql, {'start': month_start, 'orgid': userorgid})
|
||||
return [{'model_name': r.model or 'Unknown', 'cnt': int(r.cnt),
|
||||
'total_amount': round(float(r.total_amount), 4)} for r in recs]
|
||||
|
||||
|
||||
async def get_customer_user_today(env, sor, userorgid):
|
||||
"""客户当天各用户用量"""
|
||||
sql = """SELECT userid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||||
FROM dm_model_call_fact WHERE call_date = ${today}$ AND userorgid = ${orgid}$
|
||||
GROUP BY userid ORDER BY cnt DESC"""
|
||||
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'orgid': userorgid})
|
||||
return [{'user_name': r.userid, 'cnt': int(r.cnt),
|
||||
'total_amount': round(float(r.total_amount), 4)} for r in recs]
|
||||
|
||||
|
||||
# ── 趋势 ──
|
||||
|
||||
async def get_daily_trend(env, sor, days=7):
|
||||
"""最近 N 天调用量和金额趋势"""
|
||||
sql = """SELECT call_date, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||||
FROM dm_model_call_fact
|
||||
WHERE call_date >= ${start}$
|
||||
GROUP BY call_date ORDER BY call_date"""
|
||||
start = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
||||
recs = await sor.sqlExe(sql, {'start': start})
|
||||
return [{'date': r.call_date, 'cnt': int(r.cnt),
|
||||
'amount': round(float(r.total_amount), 2)} for r in recs]
|
||||
|
||||
|
||||
async def get_hourly_concurrency(env, sor):
|
||||
"""每小时并发（全量）"""
|
||||
sql = """SELECT call_hour, COUNT(DISTINCT userid) as users, COUNT(*) as calls
|
||||
FROM dm_model_call_fact
|
||||
GROUP BY call_hour ORDER BY call_hour"""
|
||||
recs = await sor.sqlExe(sql, {})
|
||||
return [{'hour': r.call_hour, 'users': int(r.users), 'calls': int(r.calls)} for r in recs]
|
||||
|
||||
|
||||
# ── 分销商专用 ──
|
||||
|
||||
async def get_distributor_customers_count(sor, distributor_orgid):
|
||||
"""分销商旗下客户数"""
|
||||
sql = """SELECT COUNT(DISTINCT userorgid) as cnt FROM dm_model_call_fact
|
||||
WHERE distributor_orgid = ${org}$"""
|
||||
recs = await sor.sqlExe(sql, {'org': distributor_orgid})
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_distributor_active_customers(sor, today, distributor_orgid):
|
||||
"""分销商今日活跃客户数"""
|
||||
sql = """SELECT COUNT(DISTINCT userorgid) as cnt FROM dm_model_call_fact
|
||||
WHERE call_date = ${today}$ AND distributor_orgid = ${org}$"""
|
||||
recs = await sor.sqlExe(sql, {'today': today, 'org': distributor_orgid})
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_distributor_top_customers(sor, distributor_orgid, limit=5):
|
||||
"""分销商旗下客户排行（按金额）"""
|
||||
sql = """SELECT userorgid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||||
FROM dm_model_call_fact WHERE distributor_orgid = ${org}$
|
||||
GROUP BY userorgid ORDER BY total_amount DESC LIMIT ${limit}$"""
|
||||
recs = await sor.sqlExe(sql, {'org': distributor_orgid, 'limit': limit})
|
||||
return [{'org_id': r.userorgid, 'cnt': int(r.cnt),
|
||||
'total_amount': round(float(r.total_amount), 2)} for r in recs]
|
||||
|
||||
|
||||
# ── scope 过滤版（在全局函数基础上加 org_filter）──
|
||||
|
||||
async def get_scoped_today_usage(sor, today, scope=None, org_filter=None):
|
||||
w = _fact_where(scope, org_filter)
|
||||
sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$ AND " + w
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
async def get_scoped_today_amount(sor, today, scope=None, org_filter=None):
|
||||
w = _fact_where(scope, org_filter)
|
||||
sql = "SELECT COALESCE(SUM(amount_cny), 0) as total FROM dm_model_call_fact WHERE call_date = ${today}$ AND " + w
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return round(float(recs[0].total), 2) if recs else 0
|
||||
|
||||
|
||||
async def get_scoped_success_rate(sor, today, scope=None, org_filter=None):
|
||||
w = _fact_where(scope, org_filter)
|
||||
sql = ("SELECT COUNT(*) as total, SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success "
|
||||
"FROM dm_model_call_fact WHERE call_date = ${today}$ AND " + w)
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
if recs and recs[0].total:
|
||||
return round(recs[0].success * 100.0 / recs[0].total, 1)
|
||||
return 0
|
||||
|
||||
|
||||
async def get_scoped_fail_count(sor, today, scope=None, org_filter=None):
|
||||
w = _fact_where(scope, org_filter)
|
||||
sql = "SELECT COUNT(*) as cnt FROM dm_model_call_fact WHERE call_date = ${today}$ AND status = 'FAILED' AND " + w
|
||||
ns = {'today': today}
|
||||
if org_filter:
|
||||
ns['org'] = org_filter
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
return int(recs[0].cnt) if recs else 0
|
||||
|
||||
|
||||
# ── 分销商财务统计 ──
|
||||
|
||||
from datetime import date as _date_module_date
|
||||
|
||||
async def get_distributor_financial(sor, org_filter, period='today'):
|
||||
"""分销商财务概览：按币种统计收入/成本/毛利
|
||||
period: 'today' | 'month' | 'year' | 'all'
|
||||
收入含直接客户 + 二级分销商
|
||||
"""
|
||||
today_str = _date_module_date.today().isoformat()
|
||||
|
||||
date_sql = ""
|
||||
ns = {'org': org_filter, 'today': today_str}
|
||||
if period == 'month':
|
||||
date_sql = "AND f.call_date BETWEEN ${month_start}$ AND ${today}$"
|
||||
ns['month_start'] = today_str[:7] + '-01'
|
||||
elif period == 'year':
|
||||
date_sql = "AND f.call_date BETWEEN ${year_start}$ AND ${today}$"
|
||||
ns['year_start'] = today_str[:4] + '-01-01'
|
||||
elif period == 'today':
|
||||
date_sql = "AND f.call_date = ${today}$"
|
||||
|
||||
# 收入按币种（含直接客户 + 二级分销商）
|
||||
sql_rev = f"""SELECT
|
||||
COALESCE(f.currency, 'CNY') as currency,
|
||||
COUNT(*) as call_count,
|
||||
COALESCE(SUM(f.amount), 0) as revenue,
|
||||
COALESCE(SUM(f.amount_cny), 0) as revenue_cny
|
||||
FROM dm_model_call_fact f
|
||||
WHERE f.distributor_orgid = ${{org}}$ {date_sql}
|
||||
GROUP BY f.currency ORDER BY revenue_cny DESC"""
|
||||
|
||||
recs = await sor.sqlExe(sql_rev, ns)
|
||||
|
||||
revenue_by_currency = []
|
||||
total_revenue_cny = 0.0
|
||||
total_calls = 0
|
||||
for r in recs:
|
||||
total_revenue_cny += float(r.revenue_cny or 0)
|
||||
total_calls += int(r.call_count or 0)
|
||||
revenue_by_currency.append({
|
||||
'currency': r.currency,
|
||||
'call_count': int(r.call_count or 0),
|
||||
'revenue': round(float(r.revenue or 0), 2),
|
||||
'revenue_cny': round(float(r.revenue_cny or 0), 2),
|
||||
})
|
||||
|
||||
# 成本（尝试从provider_cost归集，关联distributor_orgid）
|
||||
total_cost_cny = 0.0
|
||||
try:
|
||||
sql_cost = f"""SELECT COALESCE(SUM(c.total_amount_cny), 0) as total_cost_cny
|
||||
FROM dm_provider_cost_daily c
|
||||
JOIN dm_model_call_fact f ON f.call_date = c.stat_date
|
||||
AND f.providerid = c.providerid AND f.catelogid = c.catelogid
|
||||
WHERE f.distributor_orgid = ${{org}}$ {date_sql}"""
|
||||
cost_recs = await sor.sqlExe(sql_cost, ns)
|
||||
if cost_recs:
|
||||
total_cost_cny = round(float(cost_recs[0].total_cost_cny or 0), 2)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
gross_profit_cny = round(total_revenue_cny - total_cost_cny, 2)
|
||||
|
||||
return {
|
||||
'revenue_by_currency': revenue_by_currency,
|
||||
'total_revenue_cny': total_revenue_cny,
|
||||
'total_cost_cny': total_cost_cny,
|
||||
'gross_profit_cny': gross_profit_cny,
|
||||
'total_calls': total_calls,
|
||||
'period': period,
|
||||
}
|
||||
|
||||
|
||||
async def get_customer_currency_stats(sor, org_filter, period='today'):
|
||||
"""客户费用统计：按币种+时间维度"""
|
||||
today_str = _date_module_date.today().isoformat()
|
||||
date_sql = ""
|
||||
ns = {'org': org_filter, 'today': today_str}
|
||||
if period == 'month':
|
||||
date_sql = "AND f.call_date BETWEEN ${month_start}$ AND ${today}$"
|
||||
ns['month_start'] = today_str[:7] + '-01'
|
||||
elif period == 'year':
|
||||
date_sql = "AND f.call_date BETWEEN ${year_start}$ AND ${today}$"
|
||||
ns['year_start'] = today_str[:4] + '-01-01'
|
||||
elif period == 'today':
|
||||
date_sql = "AND f.call_date = ${today}$"
|
||||
|
||||
sql = f"""SELECT COALESCE(f.currency, 'CNY') as currency,
|
||||
COUNT(*) as call_count,
|
||||
COALESCE(SUM(f.amount), 0) as revenue,
|
||||
COALESCE(SUM(f.amount_cny), 0) as revenue_cny
|
||||
FROM dm_model_call_fact f
|
||||
WHERE f.userorgid = ${{org}}$ {date_sql}
|
||||
GROUP BY f.currency ORDER BY revenue_cny DESC"""
|
||||
async def get_financial_summary(sor, period='all'):
|
||||
"""平台级财务汇总：收入/成本/毛利（不分机构）
|
||||
period: 'today' | 'month' | 'year' | 'all'
|
||||
"""
|
||||
today_str = _date_module_date.today().isoformat()
|
||||
ns = {'today': today_str}
|
||||
date_sql = ""
|
||||
if period == 'month':
|
||||
date_sql = "AND call_date BETWEEN ${month_start}$ AND ${today}$"
|
||||
ns['month_start'] = today_str[:7] + '-01'
|
||||
elif period == 'year':
|
||||
date_sql = "AND call_date BETWEEN ${year_start}$ AND ${today}$"
|
||||
ns['year_start'] = today_str[:4] + '-01-01'
|
||||
elif period == 'today':
|
||||
date_sql = "AND call_date = ${today}$"
|
||||
|
||||
sql_rev = f"""SELECT
|
||||
COALESCE(SUM(amount_cny), 0) as revenue_cny,
|
||||
COUNT(*) as call_count
|
||||
FROM dm_model_call_fact
|
||||
WHERE 1=1 {date_sql}"""
|
||||
recs = await sor.sqlExe(sql_rev, ns)
|
||||
total_revenue_cny = round(float(recs[0].revenue_cny or 0), 2) if recs else 0
|
||||
total_calls = int(recs[0].call_count or 0) if recs else 0
|
||||
|
||||
total_cost_cny = 0.0
|
||||
try:
|
||||
cost_sql = f"""SELECT COALESCE(SUM(c.total_amount_cny), 0) as total_cost_cny
|
||||
FROM dm_provider_cost_daily c
|
||||
WHERE 1=1 {date_sql.replace('call_date', 'c.stat_date')}"""
|
||||
cost_recs = await sor.sqlExe(cost_sql, ns)
|
||||
if cost_recs:
|
||||
total_cost_cny = round(float(cost_recs[0].total_cost_cny or 0), 2)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
gross_profit_cny = round(total_revenue_cny - total_cost_cny, 2)
|
||||
|
||||
return {
|
||||
'total_revenue_cny': total_revenue_cny,
|
||||
'total_cost_cny': total_cost_cny,
|
||||
'gross_profit_cny': gross_profit_cny,
|
||||
'total_calls': total_calls,
|
||||
'period': period,
|
||||
}
|
||||
@ -1,283 +0,0 @@
|
||||
"""
|
||||
数据集市 ETL — 增量同步 llmusage → 集市表
|
||||
建表由 models/*.json + json2ddl + build.sh 部署时完成，ETL 只负责数据读写。
|
||||
"""
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from appPublic.uniqueID import getID
|
||||
from appPublic.dictObject import DictObject
|
||||
from appPublic.log import info
|
||||
|
||||
MODULE_NAME = "sage_datamart"
|
||||
|
||||
|
||||
def parse_usages(usages_str):
|
||||
"""从 llmusage.usages JSON 中提取 token 数"""
|
||||
try:
|
||||
u = json.loads(usages_str) if isinstance(usages_str, str) else usages_str
|
||||
if isinstance(u, dict):
|
||||
return (
|
||||
u.get('prompt_tokens', 0) or 0,
|
||||
u.get('completion_tokens', 0) or 0,
|
||||
u.get('total_tokens', 0) or 0
|
||||
)
|
||||
except:
|
||||
pass
|
||||
return 0, 0, 0
|
||||
|
||||
|
||||
async def sync_call_fact(sor, last_sync=None):
|
||||
"""增量同步 llmusage → dm_model_call_fact（含币种折算）
|
||||
|
||||
金额优先从 product_usage_log.sell_price 取实际计费金额，
|
||||
取不到时 fallback 到 llmusage.amount。
|
||||
币种从 llmusage.amount_currency 取，按 exchange_rate 折算 CNY。
|
||||
"""
|
||||
if last_sync is None:
|
||||
last_sync = (datetime.now() - timedelta(minutes=10)).strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
# 预取汇率
|
||||
stat_date = datetime.now().strftime('%Y-%m-%d')
|
||||
rates = await _get_exchange_rates(sor, stat_date)
|
||||
|
||||
# 预取 product_usage_log 实际计费金额（按 source_ref_table='llmusage' 关联）
|
||||
usage_log_map = await _load_usage_log_bills(sor, last_sync)
|
||||
|
||||
sql = """
|
||||
SELECT lu.*, l.model as llm_model, l.providerid,
|
||||
(SELECT llmcatelogid FROM llm_api_map
|
||||
WHERE llmid = lu.llmid AND isdefaultcatelog = '1' LIMIT 1) as catelogid
|
||||
FROM llmusage lu
|
||||
JOIN llm l ON l.id = lu.llmid
|
||||
WHERE lu.use_time >= ${last_sync}$
|
||||
ORDER BY lu.use_time
|
||||
"""
|
||||
rows = await sor.sqlExe(sql, {'last_sync': last_sync})
|
||||
info('sage_datamart ETL: syncing ' + str(len(rows)) + ' rows from ' + last_sync)
|
||||
|
||||
count = 0
|
||||
for r in rows:
|
||||
r = DictObject(r)
|
||||
# Skip existing
|
||||
existing = await sor.sqlExe(
|
||||
'SELECT id FROM dm_model_call_fact WHERE luid=${luid}$',
|
||||
{'luid': r.id})
|
||||
if existing:
|
||||
continue
|
||||
|
||||
prompt_tokens, completion_tokens, _ = parse_usages(r.usages)
|
||||
call_time = datetime.strptime(r.use_time[:19], '%Y-%m-%d %H:%M:%S')
|
||||
currency = getattr(r, 'amount_currency', None) or 'CNY'
|
||||
|
||||
# 优先取 product_usage_log.sell_price（实际计费金额），fallback 到 llmusage.amount
|
||||
bill_data = usage_log_map.get(r.id)
|
||||
if bill_data:
|
||||
amount = bill_data.get('sell_price', 0) or 0
|
||||
# product_usage_log 的币种来自 product_subscription.purchase_currency
|
||||
bill_currency = bill_data.get('purchase_currency')
|
||||
if bill_currency:
|
||||
currency = bill_currency
|
||||
else:
|
||||
amount = r.amount or 0
|
||||
|
||||
rate = rates.get(currency, 1.0)
|
||||
amount_cny = round(float(amount) * rate, 4)
|
||||
|
||||
ns = {
|
||||
'id': getID(),
|
||||
'luid': r.id,
|
||||
'llmid': r.llmid,
|
||||
'model': r.llm_model or r.model,
|
||||
'catelogid': r.catelogid,
|
||||
'userid': r.userid,
|
||||
'userorgid': r.userorgid,
|
||||
'ownerid': r.ownerid,
|
||||
'providerid': r.providerid,
|
||||
'call_date': r.use_date or call_time.strftime('%Y-%m-%d'),
|
||||
'call_hour': call_time.hour,
|
||||
'call_time': r.use_time,
|
||||
'ttft_ms': int(r.responsed_seconds * 1000) if r.responsed_seconds else None,
|
||||
'ttot_ms': int(r.finish_seconds * 1000) if r.finish_seconds else None,
|
||||
'prompt_tokens': prompt_tokens,
|
||||
'completion_tokens': completion_tokens,
|
||||
'status': r.status or 'SUCCEEDED',
|
||||
'amount': amount,
|
||||
'currency': currency,
|
||||
'amount_cny': amount_cny,
|
||||
}
|
||||
await sor.C('dm_model_call_fact', ns)
|
||||
count += 1
|
||||
|
||||
info('sage_datamart ETL: inserted ' + str(count) + ' new rows')
|
||||
return count
|
||||
|
||||
|
||||
async def _load_usage_log_bills(sor, last_sync):
|
||||
"""从 product_usage_log 预取实际计费金额"""
|
||||
usage_log_map = {}
|
||||
try:
|
||||
sql = """
|
||||
SELECT pul.source_ref_id, pul.sell_price, ps.purchase_currency
|
||||
FROM product_usage_log pul
|
||||
LEFT JOIN product_subscription ps ON ps.id = pul.subscription_id
|
||||
WHERE pul.source_ref_table = 'llmusage'
|
||||
AND pul.use_time >= ${last_sync}$
|
||||
"""
|
||||
rows = await sor.sqlExe(sql, {'last_sync': last_sync})
|
||||
for r in rows:
|
||||
ref_id = r.source_ref_id if hasattr(r, 'source_ref_id') else r['source_ref_id']
|
||||
sell_price = r.sell_price if hasattr(r, 'sell_price') else r['sell_price']
|
||||
purchase_currency = r.purchase_currency if hasattr(r, 'purchase_currency') else r.get('purchase_currency')
|
||||
if ref_id:
|
||||
usage_log_map[ref_id] = {
|
||||
'sell_price': float(sell_price) if sell_price else 0.0,
|
||||
'purchase_currency': purchase_currency,
|
||||
}
|
||||
except Exception as e:
|
||||
info('sage_datamart: product_usage_log lookup failed: ' + str(e))
|
||||
return usage_log_map
|
||||
|
||||
|
||||
async def _get_exchange_rates(sor, stat_date):
|
||||
"""查询汇率表"""
|
||||
rates = {'CNY': 1.0}
|
||||
try:
|
||||
sql = """SELECT from_currency, mid_rate FROM exchange_rate
|
||||
WHERE to_currency = 'CNY' AND effective_date <= ${d}$
|
||||
ORDER BY effective_date DESC"""
|
||||
rows = await sor.sqlExe(sql, {'d': stat_date})
|
||||
for r in rows:
|
||||
cur = r.from_currency if hasattr(r, 'from_currency') else r['from_currency']
|
||||
rate = r.mid_rate if hasattr(r, 'mid_rate') else r['mid_rate']
|
||||
if cur not in rates and rate:
|
||||
rates[cur] = float(rate)
|
||||
except Exception as e:
|
||||
info('sage_datamart: exchange_rate lookup failed: ' + str(e))
|
||||
return rates
|
||||
|
||||
|
||||
async def aggregate_daily_perf(sor, stat_date=None):
|
||||
"""聚合当天 dm_model_call_fact → dm_model_perf_daily（按币种分组）"""
|
||||
if stat_date is None:
|
||||
stat_date = datetime.now().strftime('%Y-%m-%d')
|
||||
|
||||
rates = await _get_exchange_rates(sor, stat_date)
|
||||
|
||||
sql = """
|
||||
SELECT
|
||||
call_date, llmid, model, catelogid, providerid, ownerid,
|
||||
COALESCE(userorgid, '') as userorgid,
|
||||
COALESCE(currency, 'CNY') as currency,
|
||||
COUNT(*) as total_calls,
|
||||
SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success_calls,
|
||||
SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as fail_calls,
|
||||
AVG(ttft_ms) as avg_ttft,
|
||||
AVG(ttot_ms) as avg_ttot,
|
||||
SUM(prompt_tokens) as total_prompt,
|
||||
SUM(completion_tokens) as total_completion,
|
||||
SUM(amount) as total_amount
|
||||
FROM dm_model_call_fact
|
||||
WHERE call_date = ${stat_date}$
|
||||
GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, dm_model_call_fact.userorgid, dm_model_call_fact.currency
|
||||
"""
|
||||
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
|
||||
info('sage_datamart AGG: aggregating ' + str(len(rows)) + ' groups for ' + stat_date)
|
||||
|
||||
for r in rows:
|
||||
r = DictObject(r)
|
||||
cur = r.currency or 'CNY'
|
||||
rate = rates.get(cur, 1.0)
|
||||
amount_cny = round(float(r.total_amount or 0) * rate, 4)
|
||||
|
||||
await sor.sqlExe(
|
||||
'DELETE FROM dm_model_perf_daily WHERE stat_date=${d}$ AND llmid=${l}$ AND COALESCE(currency,"CNY")=${c}$ AND COALESCE(userorgid,"")=${u}$',
|
||||
{'d': stat_date, 'l': r.llmid, 'c': cur, 'u': r.userorgid or ''})
|
||||
|
||||
ns = {
|
||||
'id': getID(),
|
||||
'stat_date': stat_date,
|
||||
'llmid': r.llmid,
|
||||
'model': r.model,
|
||||
'catelogid': r.catelogid,
|
||||
'providerid': r.providerid,
|
||||
'ownerid': r.ownerid,
|
||||
'total_calls': r.total_calls,
|
||||
'success_calls': r.success_calls,
|
||||
'fail_calls': r.fail_calls,
|
||||
'success_rate': round(r.success_calls * 100.0 / r.total_calls, 2) if r.total_calls else 0,
|
||||
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
|
||||
'avg_ttot_ms': round(r.avg_ttot, 1) if r.avg_ttot else None,
|
||||
'total_prompt_tokens': r.total_prompt or 0,
|
||||
'total_completion_tokens': r.total_completion or 0,
|
||||
'total_amount': r.total_amount or 0,
|
||||
'currency': cur,
|
||||
'total_amount_cny': amount_cny,
|
||||
'userorgid': r.userorgid or None,
|
||||
}
|
||||
await sor.C('dm_model_perf_daily', ns)
|
||||
|
||||
return len(rows)
|
||||
|
||||
|
||||
async def aggregate_provider_cost(sor, stat_date=None):
|
||||
"""聚合同模型供应商性价比 — dm_provider_cost_daily（按币种分组）"""
|
||||
if stat_date is None:
|
||||
stat_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
||||
|
||||
rates = await _get_exchange_rates(sor, stat_date)
|
||||
|
||||
sql = """
|
||||
SELECT
|
||||
model, providerid,
|
||||
COALESCE(currency, 'CNY') as currency,
|
||||
COUNT(*) as total_calls,
|
||||
SUM(prompt_tokens + completion_tokens) as total_tokens,
|
||||
SUM(amount) as total_amount,
|
||||
AVG(ttft_ms) as avg_ttft
|
||||
FROM dm_model_call_fact
|
||||
WHERE call_date = ${stat_date}$
|
||||
GROUP BY model, providerid, dm_model_call_fact.currency
|
||||
"""
|
||||
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
|
||||
|
||||
for r in rows:
|
||||
r = DictObject(r)
|
||||
cur = r.currency or 'CNY'
|
||||
rate = rates.get(cur, 1.0)
|
||||
amount_cny = round(float(r.total_amount or 0) * rate, 4)
|
||||
|
||||
await sor.sqlExe(
|
||||
'DELETE FROM dm_provider_cost_daily WHERE stat_date=${d}$ AND model=${m}$ AND providerid=${p}$ AND COALESCE(currency,"CNY")=${c}$',
|
||||
{'d': stat_date, 'm': r.model, 'p': r.providerid, 'c': cur})
|
||||
|
||||
ns = {
|
||||
'id': getID(),
|
||||
'stat_date': stat_date,
|
||||
'model': r.model,
|
||||
'providerid': r.providerid,
|
||||
'total_calls': r.total_calls,
|
||||
'total_tokens': r.total_tokens or 0,
|
||||
'total_amount': r.total_amount or 0,
|
||||
'currency': cur,
|
||||
'total_amount_cny': amount_cny,
|
||||
'unit_price': round(float(r.total_amount) / r.total_tokens, 8) if r.total_tokens else None,
|
||||
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
|
||||
}
|
||||
await sor.C('dm_provider_cost_daily', ns)
|
||||
|
||||
return len(rows)
|
||||
|
||||
|
||||
async def run_etl_sync(sor):
|
||||
"""每5分钟运行: 增量同步"""
|
||||
await sync_call_fact(sor)
|
||||
|
||||
|
||||
async def run_etl_aggregate(sor):
|
||||
"""每小时运行: 聚合天级性能 + 并发统计"""
|
||||
await aggregate_daily_perf(sor)
|
||||
|
||||
|
||||
async def run_etl_provider_cost(sor):
|
||||
"""每天运行: 供应商性价比"""
|
||||
await aggregate_provider_cost(sor)
|
||||
@ -1,26 +0,0 @@
|
||||
{
|
||||
"API Key排行": "API Key Ranking",
|
||||
"供应商性价比": "Provider Cost-Performance",
|
||||
"供应商排行": "Provider Ranking",
|
||||
"币种分布": "Currency Distribution",
|
||||
"我的模型(按金额Top5)": "My Models (Top 5 by Amount)",
|
||||
"模型性能排行": "Model Performance Ranking",
|
||||
"热门模型": "Popular Models",
|
||||
"用户排行": "User Ranking",
|
||||
"财务概览": "Financial Overview",
|
||||
"今天调用": "Today",
|
||||
"用户": "Users",
|
||||
"今日调用笔数": "Today Calls",
|
||||
"今日交易金额": "Today Amount",
|
||||
"用户总数": "Total Users",
|
||||
"今日活跃": "Today Active",
|
||||
"在线用户": "Online Users",
|
||||
"成功率": "Success Rate",
|
||||
"失败数": "Failed",
|
||||
"币种成本收入": "Currency Cost & Revenue",
|
||||
"币种费用": "Currency Expenses",
|
||||
"今天": "Today",
|
||||
"本月": "This Month",
|
||||
"本年": "This Year",
|
||||
"累计": "Cumulative"
|
||||
}
|
||||
@ -1,26 +0,0 @@
|
||||
{
|
||||
"API Key排行": "APIキーランキング",
|
||||
"供应商性价比": "プロバイダーコストパフォーマンス",
|
||||
"供应商排行": "プロバイダーランキング",
|
||||
"币种分布": "通貨分布",
|
||||
"我的模型(按金额Top5)": "マイモデル（金額Top5）",
|
||||
"模型性能排行": "モデルパフォーマンスランキング",
|
||||
"热门模型": "人気モデル",
|
||||
"用户排行": "ユーザーランキング",
|
||||
"财务概览": "財務概要",
|
||||
"今天调用": "本日",
|
||||
"用户": "ユーザー",
|
||||
"今日调用笔数": "本日コール数",
|
||||
"今日交易金额": "本日取引額",
|
||||
"用户总数": "総ユーザー数",
|
||||
"今日活跃": "本日アクティブ",
|
||||
"在线用户": "オンラインユーザー",
|
||||
"成功率": "成功率",
|
||||
"失败数": "失敗数",
|
||||
"币种成本收入": "通貨コスト収入",
|
||||
"币种费用": "通貨費用",
|
||||
"今天": "今日",
|
||||
"本月": "今月",
|
||||
"本年": "今年",
|
||||
"累计": "累計"
|
||||
}
|
||||
@ -1,26 +0,0 @@
|
||||
{
|
||||
"API Key排行": "API 키 랭킹",
|
||||
"供应商性价比": "공급업체 가성비",
|
||||
"供应商排行": "공급업체 랭킹",
|
||||
"币种分布": "통화 분포",
|
||||
"我的模型(按金额Top5)": "내 모델 (금액 Top5)",
|
||||
"模型性能排行": "모델 성능 랭킹",
|
||||
"热门模型": "인기 모델",
|
||||
"用户排行": "사용자 랭킹",
|
||||
"财务概览": "재무 개요",
|
||||
"今天调用": "오늘",
|
||||
"用户": "사용자",
|
||||
"今日调用笔数": "오늘 호출",
|
||||
"今日交易金额": "오늘 거래액",
|
||||
"用户总数": "전체 사용자",
|
||||
"今日活跃": "오늘 활성",
|
||||
"在线用户": "온라인 사용자",
|
||||
"成功率": "성공률",
|
||||
"失败数": "실패수",
|
||||
"币种成本收入": "통화 비용 수익",
|
||||
"币种费用": "통화 비용",
|
||||
"今天": "오늘",
|
||||
"本月": "이번 달",
|
||||
"本年": "올해",
|
||||
"累计": "누계"
|
||||
}
|
||||
@ -1,26 +0,0 @@
|
||||
{
|
||||
"API Key排行": "API Key排行",
|
||||
"供应商性价比": "供应商性价比",
|
||||
"供应商排行": "供应商排行",
|
||||
"币种分布": "币种分布",
|
||||
"我的模型(按金额Top5)": "我的模型(按金额Top5)",
|
||||
"模型性能排行": "模型性能排行",
|
||||
"热门模型": "热门模型",
|
||||
"用户排行": "用户排行",
|
||||
"财务概览": "财务概览",
|
||||
"今天调用": "今天调用",
|
||||
"用户": "用户",
|
||||
"今日调用笔数": "今日调用笔数",
|
||||
"今日交易金额": "今日交易金额",
|
||||
"用户总数": "用户总数",
|
||||
"今日活跃": "今日活跃",
|
||||
"在线用户": "在线用户",
|
||||
"成功率": "成功率",
|
||||
"失败数": "失败数",
|
||||
"币种成本收入": "币种成本收入",
|
||||
"币种费用": "币种费用",
|
||||
"今天": "今天",
|
||||
"本月": "本月",
|
||||
"本年": "本年",
|
||||
"累计": "累计"
|
||||
}
|
||||
@ -1,621 +0,0 @@
|
||||
"""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,
|
||||
get_distributor_financial, get_customer_currency_stats,
|
||||
get_financial_summary,
|
||||
)
|
||||
|
||||
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
|
||||
|
||||
|
||||
async def j2_customer_currency(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_customer_currency_stats(sor, org_filter, period)
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
# ── 监控组件 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 []
|
||||
|
||||
|
||||
async def api_financial_summary(request):
|
||||
"""平台级财务汇总 API"""
|
||||
try:
|
||||
env = request._run_ns
|
||||
period = (request._run_ns.params_kw or {}).get('period', 'all')
|
||||
async with get_sor_context(env, MODULE_NAME) as sor:
|
||||
return await get_financial_summary(sor, period)
|
||||
except Exception as e:
|
||||
debug("sage_datamart API error: " + str(e))
|
||||
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_customer_currency = j2_customer_currency
|
||||
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.api_financial_summary = api_financial_summary
|
||||
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")
|
||||
Loading…
x
Reference in New Issue
Block a user