Files
new-api-analytics/lib/queries.ts

672 lines
23 KiB
TypeScript

import { query } from "./db";
// ── 查询缓存 ─────────────────────────────────────────────────
const queryCache = new Map<string, { data: unknown; ts: number }>();
const CACHE_TTL = 120_000; // 2 分钟
function cached<T>(key: string, fn: () => Promise<T>): Promise<T> {
const hit = queryCache.get(key);
if (hit && Date.now() - hit.ts < CACHE_TTL) return Promise.resolve(hit.data as T);
return fn().then((data) => {
queryCache.set(key, { data, ts: Date.now() });
return data;
});
}
function cacheKey(...parts: unknown[]): string {
return parts.map((p) => String(p ?? "")).join(":");
}
// 真实模型名表达式:优先取 other 里的 upstream_model_name
const REAL_MODEL = `COALESCE(
CASE WHEN other IS NOT NULL AND other != '' AND other::jsonb ? 'upstream_model_name'
THEN other::jsonb->>'upstream_model_name' END,
model_name)`;
const CACHE_CREATION = `COALESCE(
CASE WHEN other IS NOT NULL AND other != '' AND other::jsonb ? 'cache_creation_tokens'
THEN (other::jsonb->>'cache_creation_tokens')::bigint END,
0)`;
const CACHE_READ = `COALESCE(
CASE WHEN other IS NOT NULL AND other != '' AND other::jsonb ? 'cache_tokens'
THEN (other::jsonb->>'cache_tokens')::bigint END,
0)`;
// ── 数据时间边界 ────────────────────────────────────────────────
export async function getDateRange(): Promise<{ minDate: string; maxDate: string }> {
const rows = await query(
`SELECT
((MIN(to_timestamp(created_at)) AT TIME ZONE 'Asia/Shanghai')::date)::text as min_date,
((MAX(to_timestamp(created_at)) AT TIME ZONE 'Asia/Shanghai')::date)::text as max_date
FROM logs WHERE type = 2`
);
return { minDate: rows[0]?.min_date ?? "", maxDate: rows[0]?.max_date ?? "" };
}
// 时间条件构建
function timeWhere(
params: (string | number | boolean | null)[],
startTs?: number,
endTs?: number
): string {
let where = "type = 2";
if (startTs) {
params.push(startTs);
where += ` AND created_at >= $${params.length}`;
}
if (endTs) {
params.push(endTs);
where += ` AND created_at <= $${params.length}`;
}
return where;
}
// ── 总览 ──────────────────────────────────────────────────────
export interface OverviewData {
total_calls: number;
total_tokens: number;
total_prompt: number;
total_completion: number;
total_cache_creation: number;
total_cache_read: number;
total_quota: number;
active_users: number;
active_models: number;
active_channels: number;
}
export function getOverview(
startTs?: number,
endTs?: number
): Promise<OverviewData> {
return cached(cacheKey("overview", startTs, endTs), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
const rows = await query(
`SELECT
COUNT(*)::int as total_calls,
COALESCE(SUM(prompt_tokens + completion_tokens), 0)::bigint as total_tokens,
COALESCE(SUM(prompt_tokens), 0)::bigint as total_prompt,
COALESCE(SUM(completion_tokens), 0)::bigint as total_completion,
COALESCE(SUM(${CACHE_CREATION}), 0)::bigint as total_cache_creation,
COALESCE(SUM(${CACHE_READ}), 0)::bigint as total_cache_read,
COALESCE(SUM(quota), 0)::bigint as total_quota,
COUNT(DISTINCT user_id)::int as active_users,
COUNT(DISTINCT ${REAL_MODEL})::int as active_models,
COUNT(DISTINCT channel_id)::int as active_channels
FROM logs WHERE ${where}`,
params
);
const r = rows[0];
return {
total_calls: Number(r.total_calls),
total_tokens: Number(r.total_tokens) + Number(r.total_cache_creation) + Number(r.total_cache_read),
total_prompt: Number(r.total_prompt),
total_completion: Number(r.total_completion),
total_cache_creation: Number(r.total_cache_creation),
total_cache_read: Number(r.total_cache_read),
total_quota: Number(r.total_quota),
active_users: Number(r.active_users),
active_models: Number(r.active_models),
active_channels: Number(r.active_channels),
};
});
}
// ── 趋势 ──────────────────────────────────────────────────────
export interface TrendPoint {
date: string;
calls: number;
prompt_tokens: number;
completion_tokens: number;
cache_creation_tokens: number;
cache_read_tokens: number;
total_tokens: number;
quota: number;
}
export type TrendGranularity = "hour" | "day" | "week" | "month";
export interface TrendFilters {
username?: string;
model?: string;
channelId?: number;
}
function appendTrendFilters(
where: string,
params: (string | number | boolean | null)[],
filters: TrendFilters = {}
): string {
if (filters.username) {
params.push(filters.username);
where += ` AND username = $${params.length}`;
}
if (filters.model) {
params.push(filters.model);
where += ` AND ${REAL_MODEL} = $${params.length}`;
}
if (filters.channelId !== undefined) {
params.push(filters.channelId);
where += ` AND channel_id = $${params.length}`;
}
return where;
}
export function getTrends(
granularity: TrendGranularity = "day",
startTs?: number,
endTs?: number,
filters: TrendFilters = {}
): Promise<TrendPoint[]> {
return cached(cacheKey("trends", granularity, startTs, endTs, filters.username, filters.model, filters.channelId), async () => {
const params: (string | number | boolean | null)[] = [];
const where = appendTrendFilters(timeWhere(params, startTs, endTs), params, filters);
const truncExpr =
granularity === "hour"
? `to_char(date_trunc('hour', to_timestamp(created_at) AT TIME ZONE 'Asia/Shanghai'), 'YYYY-MM-DD HH24:00')`
: granularity === "day"
? `((to_timestamp(created_at) AT TIME ZONE 'Asia/Shanghai')::date)::text`
: `(date_trunc('${granularity}', to_timestamp(created_at) AT TIME ZONE 'Asia/Shanghai')::date)::text`;
const rows = await query(
`SELECT
${truncExpr} as date,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens), 0)::bigint as prompt_tokens,
COALESCE(SUM(completion_tokens), 0)::bigint as completion_tokens,
COALESCE(SUM(${CACHE_CREATION}), 0)::bigint as cache_creation_tokens,
COALESCE(SUM(${CACHE_READ}), 0)::bigint as cache_read_tokens,
COALESCE(SUM(quota), 0)::bigint as quota
FROM logs WHERE ${where}
GROUP BY date ORDER BY date`,
params
);
return rows.map((r) => ({
date: String(r.date).slice(0, granularity === "hour" ? 16 : 10),
calls: Number(r.calls),
prompt_tokens: Number(r.prompt_tokens),
completion_tokens: Number(r.completion_tokens),
cache_creation_tokens: Number(r.cache_creation_tokens),
cache_read_tokens: Number(r.cache_read_tokens),
total_tokens: Number(r.prompt_tokens) + Number(r.completion_tokens) + Number(r.cache_creation_tokens) + Number(r.cache_read_tokens),
quota: Number(r.quota),
}));
});
}// ── 排名 ──────────────────────────────────────────────────────
export interface RankingItem {
rank: number;
name: string;
id?: number;
calls: number;
prompt_tokens: number;
completion_tokens: number;
cache_creation_tokens: number;
cache_read_tokens: number;
total_tokens: number;
quota: number;
quota_usd: number;
}
// 用户显示名称映射
async function getDisplayNames(): Promise<Record<number, string>> {
const rows = await query(
"SELECT id, display_name FROM users WHERE display_name IS NOT NULL AND display_name != ''"
);
return Object.fromEntries(rows.map((r) => [r.id, r.display_name]));
}
// 渠道名称映射
async function getChannelNames(): Promise<Record<number, string>> {
const rows = await query("SELECT id, name FROM channels");
return Object.fromEntries(rows.map((r) => [r.id, r.name]));
}
export function getUserRanking(
startTs?: number,
endTs?: number,
limit = 50
): Promise<RankingItem[]> {
return cached(cacheKey("userRank", startTs, endTs, limit), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
params.push(limit);
const displayNames = await getDisplayNames();
const rows = await query(
`SELECT user_id, username,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens), 0)::bigint as prompt,
COALESCE(SUM(completion_tokens), 0)::bigint as completion,
COALESCE(SUM(${CACHE_CREATION}), 0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}), 0)::bigint as cache_read,
COALESCE(SUM(quota), 0)::bigint as quota
FROM logs WHERE ${where}
GROUP BY user_id, username
ORDER BY COALESCE(SUM(prompt_tokens),0) + COALESCE(SUM(completion_tokens),0) + COALESCE(SUM(${CACHE_CREATION}),0) + COALESCE(SUM(${CACHE_READ}),0) DESC
LIMIT $${params.length}`,
params
);
return rows.map((r, i) => ({
rank: i + 1,
name: displayNames[r.user_id] || r.username,
username: r.username as string,
id: Number(r.user_id),
calls: Number(r.calls),
prompt_tokens: Number(r.prompt),
completion_tokens: Number(r.completion),
cache_creation_tokens: Number(r.cache_creation),
cache_read_tokens: Number(r.cache_read),
total_tokens: Number(r.prompt) + Number(r.completion) + Number(r.cache_creation) + Number(r.cache_read),
quota: Number(r.quota),
quota_usd: Number(r.quota) / 500000,
}));
});
}
export function getModelRanking(
startTs?: number,
endTs?: number,
limit = 50
): Promise<RankingItem[]> {
return cached(cacheKey("modelRank", startTs, endTs, limit), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
params.push(limit);
const rows = await query(
`SELECT ${REAL_MODEL} as model,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens), 0)::bigint as prompt,
COALESCE(SUM(completion_tokens), 0)::bigint as completion,
COALESCE(SUM(${CACHE_CREATION}), 0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}), 0)::bigint as cache_read,
COALESCE(SUM(quota), 0)::bigint as quota
FROM logs WHERE ${where}
GROUP BY model
ORDER BY COALESCE(SUM(prompt_tokens),0) + COALESCE(SUM(completion_tokens),0) + COALESCE(SUM(${CACHE_CREATION}),0) + COALESCE(SUM(${CACHE_READ}),0) DESC
LIMIT $${params.length}`,
params
);
return rows.map((r, i) => ({
rank: i + 1,
name: r.model || "(unknown)",
calls: Number(r.calls),
prompt_tokens: Number(r.prompt),
completion_tokens: Number(r.completion),
cache_creation_tokens: Number(r.cache_creation),
cache_read_tokens: Number(r.cache_read),
total_tokens: Number(r.prompt) + Number(r.completion) + Number(r.cache_creation) + Number(r.cache_read),
quota: Number(r.quota),
quota_usd: Number(r.quota) / 500000,
}));
});
}
export function getChannelRanking(
startTs?: number,
endTs?: number,
limit = 50
): Promise<RankingItem[]> {
return cached(cacheKey("channelRank", startTs, endTs, limit), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
params.push(limit);
const channelNames = await getChannelNames();
const rows = await query(
`SELECT channel_id,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens), 0)::bigint as prompt,
COALESCE(SUM(completion_tokens), 0)::bigint as completion,
COALESCE(SUM(${CACHE_CREATION}), 0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}), 0)::bigint as cache_read,
COALESCE(SUM(quota), 0)::bigint as quota
FROM logs WHERE ${where}
GROUP BY channel_id
ORDER BY COALESCE(SUM(prompt_tokens),0) + COALESCE(SUM(completion_tokens),0) + COALESCE(SUM(${CACHE_CREATION}),0) + COALESCE(SUM(${CACHE_READ}),0) DESC
LIMIT $${params.length}`,
params
);
return rows.map((r, i) => ({
rank: i + 1,
name: channelNames[r.channel_id] || `已删除(${r.channel_id})`,
id: Number(r.channel_id),
calls: Number(r.calls),
prompt_tokens: Number(r.prompt),
completion_tokens: Number(r.completion),
cache_creation_tokens: Number(r.cache_creation),
cache_read_tokens: Number(r.cache_read),
total_tokens: Number(r.prompt) + Number(r.completion) + Number(r.cache_creation) + Number(r.cache_read),
quota: Number(r.quota),
quota_usd: Number(r.quota) / 500000,
}));
});
}
// ── 下钻详情 ──────────────────────────────────────────────────
export interface DetailBreakdown {
name: string;
calls: number;
total_tokens: number;
quota: number;
}
export function getUserDetail(
username: string,
startTs?: number,
endTs?: number
) {
return cached(cacheKey("userDetail", username, startTs, endTs), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
params.push(username);
// 用户总览
const overview = await query(
`SELECT COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens),0)::bigint as prompt,
COALESCE(SUM(completion_tokens),0)::bigint as completion,
COALESCE(SUM(${CACHE_CREATION}),0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}),0)::bigint as cache_read,
COALESCE(SUM(quota),0)::bigint as quota
FROM logs WHERE ${where} AND username = $${params.length}`,
params
);
// 用户的模型分布
const params2: (string | number | boolean | null)[] = [];
const where2 = timeWhere(params2, startTs, endTs);
params2.push(username);
const models = await query(
`SELECT ${REAL_MODEL} as model,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens + completion_tokens),0)::bigint as tokens,
COALESCE(SUM(${CACHE_CREATION}),0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}),0)::bigint as cache_read,
COALESCE(SUM(quota),0)::bigint as quota
FROM logs WHERE ${where2} AND username = $${params2.length}
GROUP BY model
ORDER BY tokens DESC LIMIT 20`,
params2
);
const displayNames = await getDisplayNames();
const userRow = await query(
"SELECT id FROM users WHERE username = $1 LIMIT 1",
[username]
);
const displayName =
userRow.length > 0 ? displayNames[userRow[0].id] || username : username;
const o = overview[0];
return {
display_name: displayName,
calls: Number(o.calls),
prompt_tokens: Number(o.prompt),
completion_tokens: Number(o.completion),
cache_creation_tokens: Number(o.cache_creation),
cache_read_tokens: Number(o.cache_read),
total_tokens: Number(o.prompt) + Number(o.completion) + Number(o.cache_creation) + Number(o.cache_read),
quota: Number(o.quota),
models: models.map((m) => ({
name: m.model,
calls: Number(m.calls),
total_tokens: Number(m.tokens) + Number(m.cache_creation) + Number(m.cache_read),
quota: Number(m.quota),
})),
};
});
}
export function getModelDetail(
model: string,
startTs?: number,
endTs?: number
) {
return cached(cacheKey("modelDetail", model, startTs, endTs), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
params.push(model);
const overview = await query(
`SELECT COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens),0)::bigint as prompt,
COALESCE(SUM(completion_tokens),0)::bigint as completion,
COALESCE(SUM(${CACHE_CREATION}),0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}),0)::bigint as cache_read,
COALESCE(SUM(quota),0)::bigint as quota
FROM logs WHERE ${where} AND ${REAL_MODEL} = $${params.length}`,
params
);
const params2: (string | number | boolean | null)[] = [];
const where2 = timeWhere(params2, startTs, endTs);
params2.push(model);
const displayNames = await getDisplayNames();
const users = await query(
`SELECT user_id, username,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens + completion_tokens),0)::bigint as tokens,
COALESCE(SUM(${CACHE_CREATION}),0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}),0)::bigint as cache_read,
COALESCE(SUM(quota),0)::bigint as quota
FROM logs WHERE ${where2} AND ${REAL_MODEL} = $${params2.length}
GROUP BY user_id, username
ORDER BY tokens DESC LIMIT 20`,
params2
);
const o = overview[0];
return {
calls: Number(o.calls),
prompt_tokens: Number(o.prompt),
completion_tokens: Number(o.completion),
cache_creation_tokens: Number(o.cache_creation),
cache_read_tokens: Number(o.cache_read),
total_tokens: Number(o.prompt) + Number(o.completion) + Number(o.cache_creation) + Number(o.cache_read),
quota: Number(o.quota),
users: users.map((u) => ({
name: displayNames[u.user_id] || u.username,
calls: Number(u.calls),
total_tokens: Number(u.tokens) + Number(u.cache_creation) + Number(u.cache_read),
quota: Number(u.quota),
})),
};
});
}
export function getChannelDetail(
channelId: number,
startTs?: number,
endTs?: number
) {
return cached(cacheKey("channelDetail", channelId, startTs, endTs), async () => {
const params: (string | number | boolean | null)[] = [];
const where = timeWhere(params, startTs, endTs);
params.push(channelId);
const channelNames = await getChannelNames();
const overview = await query(
`SELECT COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens),0)::bigint as prompt,
COALESCE(SUM(completion_tokens),0)::bigint as completion,
COALESCE(SUM(${CACHE_CREATION}),0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}),0)::bigint as cache_read,
COALESCE(SUM(quota),0)::bigint as quota
FROM logs WHERE ${where} AND channel_id = $${params.length}`,
params
);
const params2: (string | number | boolean | null)[] = [];
const where2 = timeWhere(params2, startTs, endTs);
params2.push(channelId);
const models = await query(
`SELECT ${REAL_MODEL} as model,
COUNT(*)::int as calls,
COALESCE(SUM(prompt_tokens + completion_tokens),0)::bigint as tokens,
COALESCE(SUM(${CACHE_CREATION}),0)::bigint as cache_creation,
COALESCE(SUM(${CACHE_READ}),0)::bigint as cache_read,
COALESCE(SUM(quota),0)::bigint as quota
FROM logs WHERE ${where2} AND channel_id = $${params2.length}
GROUP BY model
ORDER BY tokens DESC LIMIT 20`,
params2
);
const o = overview[0];
return {
channel_name: channelNames[channelId] || `已删除(${channelId})`,
calls: Number(o.calls),
prompt_tokens: Number(o.prompt),
completion_tokens: Number(o.completion),
cache_creation_tokens: Number(o.cache_creation),
cache_read_tokens: Number(o.cache_read),
total_tokens: Number(o.prompt) + Number(o.completion) + Number(o.cache_creation) + Number(o.cache_read),
quota: Number(o.quota),
models: models.map((m) => ({
name: m.model,
calls: Number(m.calls),
total_tokens: Number(m.tokens) + Number(m.cache_creation) + Number(m.cache_read),
quota: Number(m.quota),
})),
};
});
}
// ── 明细日志 ──────────────────────────────────────────────────
export interface LogEntry {
id: number;
created_at: string;
username: string;
display_name: string;
real_model: string;
request_model: string;
channel_name: string;
channel_id: number;
prompt_tokens: number;
completion_tokens: number;
cache_creation_tokens: number;
cache_read_tokens: number;
total_tokens: number;
quota: number;
quota_usd: number;
use_time: number;
is_stream: boolean;
token_name: string;
}
export interface LogsResult {
logs: LogEntry[];
total: number;
page: number;
page_size: number;
}
export async function getLogs(options: {
page?: number;
pageSize?: number;
startTs?: number;
endTs?: number;
username?: string;
model?: string;
channelId?: number;
tokenName?: string;
}): Promise<LogsResult> {
const ck = cacheKey("logs", options.page, options.pageSize, options.startTs, options.endTs, options.username, options.model, options.channelId, options.tokenName);
return cached(ck, async () => {
const { page = 1, pageSize = 100 } = options;
const params: (string | number | boolean | null)[] = [];
let where = timeWhere(params, options.startTs, options.endTs);
if (options.username) {
params.push(options.username);
where += ` AND username = $${params.length}`;
}
if (options.model) {
params.push(options.model);
where += ` AND ${REAL_MODEL} = $${params.length}`;
}
if (options.channelId) {
params.push(options.channelId);
where += ` AND channel_id = $${params.length}`;
}
if (options.tokenName) {
params.push(`%${options.tokenName}%`);
where += ` AND token_name ILIKE $${params.length}`;
}
// Count
const countRows = await query(
`SELECT COUNT(*)::int as total FROM logs WHERE ${where}`,
params
);
const total = Number(countRows[0].total);
// Data
const offset = (page - 1) * pageSize;
const dataParams = [...params, pageSize, offset];
const displayNames = await getDisplayNames();
const channelNames = await getChannelNames();
const rows = await query(
`SELECT id, created_at, user_id, username, model_name,
${REAL_MODEL} as real_model,
channel_id, prompt_tokens, completion_tokens, quota,
${CACHE_CREATION} as cache_creation,
${CACHE_READ} as cache_read,
use_time, is_stream, token_name
FROM logs WHERE ${where}
ORDER BY id DESC
LIMIT $${dataParams.length - 1} OFFSET $${dataParams.length}`,
dataParams
);
return {
total,
page,
page_size: pageSize,
logs: rows.map((r) => ({
id: Number(r.id),
created_at: new Date(Number(r.created_at) * 1000).toISOString(),
username: r.username,
display_name: displayNames[r.user_id] || r.username,
real_model: r.real_model || r.model_name,
request_model: r.model_name,
channel_name: channelNames[r.channel_id] || `已删除(${r.channel_id})`,
channel_id: Number(r.channel_id),
prompt_tokens: Number(r.prompt_tokens),
completion_tokens: Number(r.completion_tokens),
cache_creation_tokens: Number(r.cache_creation),
cache_read_tokens: Number(r.cache_read),
total_tokens: Number(r.prompt_tokens) + Number(r.completion_tokens) + Number(r.cache_creation) + Number(r.cache_read),
quota: Number(r.quota),
quota_usd: Number(r.quota) / 500000,
use_time: Number(r.use_time),
is_stream: Boolean(r.is_stream),
token_name: r.token_name || "",
})),
};
});
}