364 lines
14 KiB
Python
364 lines
14 KiB
Python
"""
|
||
统一 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_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):
|
||
"""当天各币种分布"""
|
||
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 call_date = ${today}$
|
||
GROUP BY currency
|
||
ORDER BY call_count DESC"""
|
||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||
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):
|
||
"""当天 Top N 模型（按调用次数）"""
|
||
sql = """SELECT model, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||
FROM dm_model_call_fact WHERE call_date = ${today}$
|
||
GROUP BY model ORDER BY cnt DESC LIMIT ${limit}$"""
|
||
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'limit': limit})
|
||
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):
|
||
"""当天 Top N 供应商（按金额）"""
|
||
sql = """SELECT providerid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||
FROM dm_model_call_fact WHERE call_date = ${today}$ AND providerid IS NOT NULL
|
||
GROUP BY providerid ORDER BY total_amount DESC LIMIT ${limit}$"""
|
||
recs = await sor.sqlExe(sql, {'today': env.curDateString(), 'limit': limit})
|
||
return [{'provider_name': 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):
|
||
"""全量 Top N 用户（按金额）"""
|
||
sql = """SELECT userid, COUNT(*) as cnt, COALESCE(SUM(amount_cny), 0) as total_amount
|
||
FROM dm_model_call_fact GROUP BY userid ORDER BY total_amount DESC LIMIT ${limit}$"""
|
||
recs = await sor.sqlExe(sql, {'limit': limit})
|
||
return [{'user_name': 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 WHERE call_date = ${today}$
|
||
GROUP BY call_hour ORDER BY call_hour"""
|
||
recs = await sor.sqlExe(sql, {'today': env.curDateString()})
|
||
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
|