629 lines
20 KiB
Go
629 lines
20 KiB
Go
package model
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/QuantumNous/new-api/common"
|
|
"github.com/QuantumNous/new-api/logger"
|
|
"github.com/QuantumNous/new-api/types"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
|
|
"github.com/bytedance/gopkg/util/gopool"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type Log struct {
|
|
Id int `json:"id" gorm:"index:idx_created_at_id,priority:1;index:idx_user_id_id,priority:2"`
|
|
UserId int `json:"user_id" gorm:"index;index:idx_user_id_id,priority:1"`
|
|
CreatedAt int64 `json:"created_at" gorm:"bigint;index:idx_created_at_id,priority:2;index:idx_created_at_type"`
|
|
Type int `json:"type" gorm:"index:idx_created_at_type"`
|
|
Content string `json:"content"`
|
|
Username string `json:"username" gorm:"index;index:index_username_model_name,priority:2;default:''"`
|
|
TokenName string `json:"token_name" gorm:"index;default:''"`
|
|
ModelName string `json:"model_name" gorm:"index;index:index_username_model_name,priority:1;default:''"`
|
|
Quota int `json:"quota" gorm:"default:0"`
|
|
PromptTokens int `json:"prompt_tokens" gorm:"default:0"`
|
|
CompletionTokens int `json:"completion_tokens" gorm:"default:0"`
|
|
UseTime int `json:"use_time" gorm:"default:0"`
|
|
IsStream bool `json:"is_stream"`
|
|
ChannelId int `json:"channel" gorm:"index"`
|
|
ChannelName string `json:"channel_name" gorm:"->"`
|
|
TokenId int `json:"token_id" gorm:"default:0;index"`
|
|
Group string `json:"group" gorm:"index"`
|
|
Ip string `json:"ip" gorm:"index;default:''"`
|
|
RequestId string `json:"request_id,omitempty" gorm:"type:varchar(64);index:idx_logs_request_id;default:''"`
|
|
Other string `json:"other"`
|
|
}
|
|
|
|
const (
|
|
LogTypeUnknown = 0
|
|
LogTypeTopup = 1
|
|
LogTypeConsume = 2
|
|
LogTypeManage = 3
|
|
LogTypeSystem = 4
|
|
LogTypeError = 5
|
|
LogTypeRefund = 6
|
|
)
|
|
|
|
func formatUserLogs(logs []*Log, startIdx int) {
|
|
for i := range logs {
|
|
logs[i].ChannelName = ""
|
|
var otherMap map[string]interface{}
|
|
otherMap, _ = common.StrToMap(logs[i].Other)
|
|
if otherMap != nil {
|
|
delete(otherMap, "admin_info")
|
|
delete(otherMap, "reject_reason")
|
|
}
|
|
logs[i].Other = common.MapToJsonStr(otherMap)
|
|
logs[i].Id = startIdx + i + 1
|
|
}
|
|
}
|
|
|
|
func GetLogByTokenId(tokenId int) (logs []*Log, err error) {
|
|
err = LOG_DB.Model(&Log{}).Where("token_id = ?", tokenId).Order("id desc").Limit(common.MaxRecentItems).Find(&logs).Error
|
|
formatUserLogs(logs, 0)
|
|
return logs, err
|
|
}
|
|
|
|
func RecordLog(userId int, logType int, content string) {
|
|
if logType == LogTypeConsume && !common.LogConsumeEnabled {
|
|
return
|
|
}
|
|
username, _ := GetUsernameById(userId, false)
|
|
log := &Log{UserId: userId, Username: username, CreatedAt: common.GetTimestamp(), Type: logType, Content: content}
|
|
err := LOG_DB.Create(log).Error
|
|
if err != nil {
|
|
common.SysLog("failed to record log: " + err.Error())
|
|
}
|
|
}
|
|
|
|
func RecordTopupLog(userId int, content string, callerIp string, paymentMethod string, callbackPaymentMethod string) {
|
|
username, _ := GetUsernameById(userId, false)
|
|
adminInfo := map[string]interface{}{
|
|
"server_ip": common.GetIp(),
|
|
"caller_ip": callerIp,
|
|
"payment_method": paymentMethod,
|
|
"callback_payment_method": callbackPaymentMethod,
|
|
"version": common.Version,
|
|
}
|
|
other := map[string]interface{}{
|
|
"admin_info": adminInfo,
|
|
}
|
|
log := &Log{
|
|
UserId: userId,
|
|
Username: username,
|
|
CreatedAt: common.GetTimestamp(),
|
|
Type: LogTypeTopup,
|
|
Content: content,
|
|
Ip: callerIp,
|
|
Other: common.MapToJsonStr(other),
|
|
}
|
|
err := LOG_DB.Create(log).Error
|
|
if err != nil {
|
|
common.SysLog("failed to record topup log: " + err.Error())
|
|
}
|
|
}
|
|
|
|
func RecordErrorLog(c *gin.Context, userId int, channelId int, modelName string, tokenName string, content string, tokenId int, useTimeSeconds int,
|
|
isStream bool, group string, other map[string]interface{}) {
|
|
logger.LogInfo(c, fmt.Sprintf("record error log: userId=%d, channelId=%d, modelName=%s, tokenName=%s, content=%s", userId, channelId, modelName, tokenName, content))
|
|
username := c.GetString("username")
|
|
requestId := c.GetString(common.RequestIdKey)
|
|
otherStr := common.MapToJsonStr(other)
|
|
needRecordIp := false
|
|
if settingMap, err := GetUserSetting(userId, false); err == nil {
|
|
if settingMap.RecordIpLog {
|
|
needRecordIp = true
|
|
}
|
|
}
|
|
log := &Log{UserId: userId, Username: username, CreatedAt: common.GetTimestamp(), Type: LogTypeError, Content: content, PromptTokens: 0, CompletionTokens: 0, TokenName: tokenName, ModelName: modelName, Quota: 0, ChannelId: channelId, TokenId: tokenId, UseTime: useTimeSeconds, IsStream: isStream, Group: group, Ip: func() string {
|
|
if needRecordIp {
|
|
return c.ClientIP()
|
|
}
|
|
return ""
|
|
}(), RequestId: requestId, Other: otherStr}
|
|
err := LOG_DB.Create(log).Error
|
|
if err != nil {
|
|
logger.LogError(c, "failed to record log: "+err.Error())
|
|
}
|
|
}
|
|
|
|
type RecordConsumeLogParams struct {
|
|
ChannelId int `json:"channel_id"`
|
|
PromptTokens int `json:"prompt_tokens"`
|
|
CompletionTokens int `json:"completion_tokens"`
|
|
ModelName string `json:"model_name"`
|
|
TokenName string `json:"token_name"`
|
|
Quota int `json:"quota"`
|
|
Content string `json:"content"`
|
|
TokenId int `json:"token_id"`
|
|
UseTimeSeconds int `json:"use_time_seconds"`
|
|
IsStream bool `json:"is_stream"`
|
|
Group string `json:"group"`
|
|
Other map[string]interface{} `json:"other"`
|
|
}
|
|
|
|
func RecordConsumeLog(c *gin.Context, userId int, params RecordConsumeLogParams) {
|
|
if !common.LogConsumeEnabled {
|
|
return
|
|
}
|
|
logger.LogInfo(c, fmt.Sprintf("record consume log: userId=%d, params=%s", userId, common.GetJsonString(params)))
|
|
username := c.GetString("username")
|
|
requestId := c.GetString(common.RequestIdKey)
|
|
otherStr := common.MapToJsonStr(params.Other)
|
|
needRecordIp := false
|
|
if settingMap, err := GetUserSetting(userId, false); err == nil {
|
|
if settingMap.RecordIpLog {
|
|
needRecordIp = true
|
|
}
|
|
}
|
|
log := &Log{UserId: userId, Username: username, CreatedAt: common.GetTimestamp(), Type: LogTypeConsume, Content: params.Content, PromptTokens: params.PromptTokens, CompletionTokens: params.CompletionTokens, TokenName: params.TokenName, ModelName: params.ModelName, Quota: params.Quota, ChannelId: params.ChannelId, TokenId: params.TokenId, UseTime: params.UseTimeSeconds, IsStream: params.IsStream, Group: params.Group, Ip: func() string {
|
|
if needRecordIp {
|
|
return c.ClientIP()
|
|
}
|
|
return ""
|
|
}(), RequestId: requestId, Other: otherStr}
|
|
err := LOG_DB.Create(log).Error
|
|
if err != nil {
|
|
logger.LogError(c, "failed to record log: "+err.Error())
|
|
}
|
|
if common.DataExportEnabled {
|
|
gopool.Go(func() {
|
|
LogQuotaData(userId, username, params.ModelName, params.Quota, common.GetTimestamp(), params.PromptTokens+params.CompletionTokens)
|
|
})
|
|
}
|
|
}
|
|
|
|
type RecordTaskBillingLogParams struct {
|
|
UserId int
|
|
LogType int
|
|
Content string
|
|
ChannelId int
|
|
ModelName string
|
|
Quota int
|
|
TokenId int
|
|
Group string
|
|
Other map[string]interface{}
|
|
}
|
|
|
|
func RecordTaskBillingLog(params RecordTaskBillingLogParams) {
|
|
if params.LogType == LogTypeConsume && !common.LogConsumeEnabled {
|
|
return
|
|
}
|
|
username, _ := GetUsernameById(params.UserId, false)
|
|
tokenName := ""
|
|
if params.TokenId > 0 {
|
|
if token, err := GetTokenById(params.TokenId); err == nil {
|
|
tokenName = token.Name
|
|
}
|
|
}
|
|
log := &Log{UserId: params.UserId, Username: username, CreatedAt: common.GetTimestamp(), Type: params.LogType, Content: params.Content, TokenName: tokenName, ModelName: params.ModelName, Quota: params.Quota, ChannelId: params.ChannelId, TokenId: params.TokenId, Group: params.Group, Other: common.MapToJsonStr(params.Other)}
|
|
err := LOG_DB.Create(log).Error
|
|
if err != nil {
|
|
common.SysLog("failed to record task billing log: " + err.Error())
|
|
}
|
|
}
|
|
|
|
type CacheDashboardMetrics struct {
|
|
CacheHitRate float64 `json:"cache_hit_rate"`
|
|
CacheReadTokens int64 `json:"cache_read_tokens"`
|
|
CacheWriteTokens int64 `json:"cache_write_tokens"`
|
|
CacheTokens int64 `json:"cache_tokens"`
|
|
Requests int64 `json:"requests"`
|
|
HitRequests int64 `json:"hit_requests"`
|
|
EligibleRequests int64 `json:"eligible_requests"`
|
|
TotalInputTokens int64 `json:"total_input_tokens"`
|
|
}
|
|
|
|
type cacheMetricRow struct {
|
|
CreatedAt int64 `gorm:"column:created_at"`
|
|
PromptTokens int `gorm:"column:prompt_tokens"`
|
|
CompletionTokens int `gorm:"column:completion_tokens"`
|
|
ModelName string `gorm:"column:model_name"`
|
|
Other string `gorm:"column:other"`
|
|
}
|
|
|
|
type DashboardConsumeStats struct {
|
|
Quota int `json:"quota"`
|
|
}
|
|
|
|
func toInt64(v interface{}) int64 {
|
|
switch t := v.(type) {
|
|
case int:
|
|
return int64(t)
|
|
case int32:
|
|
return int64(t)
|
|
case int64:
|
|
return t
|
|
case float64:
|
|
return int64(t)
|
|
case string:
|
|
if n, err := strconv.ParseInt(t, 10, 64); err == nil {
|
|
return n
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func firstPositiveInt64(values ...int64) int64 {
|
|
for _, v := range values {
|
|
if v > 0 {
|
|
return v
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func sumPositiveInt64(values ...int64) int64 {
|
|
var total int64
|
|
for _, v := range values {
|
|
if v > 0 {
|
|
total += v
|
|
}
|
|
}
|
|
return total
|
|
}
|
|
|
|
func accumulateCacheMetrics(m *CacheDashboardMetrics, row cacheMetricRow) {
|
|
m.Requests++
|
|
other, _ := common.StrToMap(row.Other)
|
|
if other == nil {
|
|
other = map[string]interface{}{}
|
|
}
|
|
|
|
read := firstPositiveInt64(
|
|
toInt64(other["cache_read_tokens"]),
|
|
toInt64(other["cache_read_input_tokens"]),
|
|
toInt64(other["cache_tokens"]),
|
|
toInt64(other["cached_tokens"]),
|
|
toInt64(other["prompt_cache_hit_tokens"]),
|
|
)
|
|
write := firstPositiveInt64(
|
|
toInt64(other["cache_write_tokens"]),
|
|
toInt64(other["cache_creation_input_tokens"]),
|
|
sumPositiveInt64(
|
|
toInt64(other["cache_creation_tokens"]),
|
|
toInt64(other["cache_creation_tokens_5m"]),
|
|
toInt64(other["cache_creation_tokens_1h"]),
|
|
),
|
|
)
|
|
baseInput := firstPositiveInt64(
|
|
toInt64(other["input_tokens_total"]),
|
|
int64(row.PromptTokens),
|
|
)
|
|
if baseInput == 0 && (read > 0 || write > 0) {
|
|
baseInput = int64(row.PromptTokens) + read + write
|
|
}
|
|
actualInput := baseInput
|
|
if actualInput < read+write {
|
|
actualInput = read + write
|
|
}
|
|
|
|
m.CacheReadTokens += read
|
|
m.CacheWriteTokens += write
|
|
if read > 0 {
|
|
m.HitRequests++
|
|
}
|
|
if read > 0 || write > 0 {
|
|
m.EligibleRequests++
|
|
m.TotalInputTokens += actualInput
|
|
}
|
|
}
|
|
|
|
func finalizeCacheMetrics(m *CacheDashboardMetrics) {
|
|
m.CacheTokens = m.CacheReadTokens + m.CacheWriteTokens
|
|
if m.TotalInputTokens > 0 {
|
|
m.CacheHitRate = float64(m.CacheReadTokens) / float64(m.TotalInputTokens)
|
|
return
|
|
}
|
|
if m.EligibleRequests > 0 {
|
|
m.CacheHitRate = float64(m.HitRequests) / float64(m.EligibleRequests)
|
|
return
|
|
}
|
|
if m.Requests > 0 {
|
|
m.CacheHitRate = float64(m.HitRequests) / float64(m.Requests)
|
|
}
|
|
}
|
|
|
|
func GetUserCacheDashboardMetrics(userId int) (CacheDashboardMetrics, CacheDashboardMetrics, error) {
|
|
now := time.Now()
|
|
start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).Unix()
|
|
overall := CacheDashboardMetrics{}
|
|
today := CacheDashboardMetrics{}
|
|
var rows []cacheMetricRow
|
|
err := LOG_DB.Model(&Log{}).
|
|
Select("created_at, prompt_tokens, completion_tokens, model_name, other").
|
|
Where("user_id = ? AND type = ?", userId, LogTypeConsume).
|
|
Order("id ASC").
|
|
FindInBatches(&rows, 1000, func(tx *gorm.DB, batch int) error {
|
|
for _, row := range rows {
|
|
accumulateCacheMetrics(&overall, row)
|
|
if row.CreatedAt >= start {
|
|
accumulateCacheMetrics(&today, row)
|
|
}
|
|
}
|
|
return nil
|
|
}).Error
|
|
if err != nil {
|
|
return CacheDashboardMetrics{}, CacheDashboardMetrics{}, err
|
|
}
|
|
finalizeCacheMetrics(&overall)
|
|
finalizeCacheMetrics(&today)
|
|
return overall, today, nil
|
|
}
|
|
|
|
func GetUserDashboardConsumeStats(userId int, startTimestamp int64, endTimestamp int64, modelName string, tokenName string, channel int, group string) (stats DashboardConsumeStats, err error) {
|
|
tx := LOG_DB.Table("logs").Select("COALESCE(sum(quota), 0) quota").Where("user_id = ? AND type = ?", userId, LogTypeConsume)
|
|
if tokenName != "" {
|
|
tx = tx.Where("token_name = ?", tokenName)
|
|
}
|
|
if startTimestamp != 0 {
|
|
tx = tx.Where("created_at >= ?", startTimestamp)
|
|
}
|
|
if endTimestamp != 0 {
|
|
tx = tx.Where("created_at <= ?", endTimestamp)
|
|
}
|
|
if modelName != "" {
|
|
modelNamePattern, err := sanitizeLikePattern(modelName)
|
|
if err != nil {
|
|
return stats, err
|
|
}
|
|
tx = tx.Where("model_name LIKE ? ESCAPE '!'", modelNamePattern)
|
|
}
|
|
if channel != 0 {
|
|
tx = tx.Where("channel_id = ?", channel)
|
|
}
|
|
if group != "" {
|
|
tx = tx.Where(logGroupCol+" = ?", group)
|
|
}
|
|
if err := tx.Scan(&stats).Error; err != nil {
|
|
common.SysError("failed to query user dashboard consume stats: " + err.Error())
|
|
return stats, errors.New("查询统计数据失败")
|
|
}
|
|
if stats.Quota < 0 {
|
|
stats.Quota = 0
|
|
}
|
|
return stats, nil
|
|
}
|
|
|
|
func GetAllLogs(logType int, startTimestamp int64, endTimestamp int64, modelName string, username string, tokenName string, startIdx int, num int, channel int, group string, requestId string) (logs []*Log, total int64, err error) {
|
|
var tx *gorm.DB
|
|
if logType == LogTypeUnknown {
|
|
tx = LOG_DB
|
|
} else {
|
|
tx = LOG_DB.Where("logs.type = ?", logType)
|
|
}
|
|
|
|
if modelName != "" {
|
|
tx = tx.Where("logs.model_name like ?", modelName)
|
|
}
|
|
if username != "" {
|
|
tx = tx.Where("logs.username = ?", username)
|
|
}
|
|
if tokenName != "" {
|
|
tx = tx.Where("logs.token_name = ?", tokenName)
|
|
}
|
|
if requestId != "" {
|
|
tx = tx.Where("logs.request_id = ?", requestId)
|
|
}
|
|
if startTimestamp != 0 {
|
|
tx = tx.Where("logs.created_at >= ?", startTimestamp)
|
|
}
|
|
if endTimestamp != 0 {
|
|
tx = tx.Where("logs.created_at <= ?", endTimestamp)
|
|
}
|
|
if channel != 0 {
|
|
tx = tx.Where("logs.channel_id = ?", channel)
|
|
}
|
|
if group != "" {
|
|
tx = tx.Where("logs."+logGroupCol+" = ?", group)
|
|
}
|
|
err = tx.Model(&Log{}).Count(&total).Error
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
err = tx.Order("logs.id desc").Limit(num).Offset(startIdx).Find(&logs).Error
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
channelIds := types.NewSet[int]()
|
|
for _, log := range logs {
|
|
if log.ChannelId != 0 {
|
|
channelIds.Add(log.ChannelId)
|
|
}
|
|
}
|
|
|
|
if channelIds.Len() > 0 {
|
|
var channels []struct {
|
|
Id int `gorm:"column:id"`
|
|
Name string `gorm:"column:name"`
|
|
}
|
|
if common.MemoryCacheEnabled {
|
|
for _, channelId := range channelIds.Items() {
|
|
if cacheChannel, err := CacheGetChannel(channelId); err == nil {
|
|
channels = append(channels, struct {
|
|
Id int `gorm:"column:id"`
|
|
Name string `gorm:"column:name"`
|
|
}{Id: channelId, Name: cacheChannel.Name})
|
|
}
|
|
}
|
|
} else {
|
|
if err = DB.Table("channels").Select("id, name").Where("id IN ?", channelIds.Items()).Find(&channels).Error; err != nil {
|
|
return logs, total, err
|
|
}
|
|
}
|
|
channelMap := make(map[int]string, len(channels))
|
|
for _, channel := range channels {
|
|
channelMap[channel.Id] = channel.Name
|
|
}
|
|
for i := range logs {
|
|
logs[i].ChannelName = channelMap[logs[i].ChannelId]
|
|
}
|
|
}
|
|
|
|
return logs, total, err
|
|
}
|
|
|
|
const logSearchCountLimit = 10000
|
|
|
|
func GetUserLogs(userId int, logType int, startTimestamp int64, endTimestamp int64, modelName string, tokenName string, startIdx int, num int, group string, requestId string) (logs []*Log, total int64, err error) {
|
|
var tx *gorm.DB
|
|
if logType == LogTypeUnknown {
|
|
tx = LOG_DB.Where("logs.user_id = ?", userId)
|
|
} else {
|
|
tx = LOG_DB.Where("logs.user_id = ? and logs.type = ?", userId, logType)
|
|
}
|
|
|
|
if modelName != "" {
|
|
modelNamePattern, err := sanitizeLikePattern(modelName)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
tx = tx.Where("logs.model_name LIKE ? ESCAPE '!'", modelNamePattern)
|
|
}
|
|
if tokenName != "" {
|
|
tx = tx.Where("logs.token_name = ?", tokenName)
|
|
}
|
|
if requestId != "" {
|
|
tx = tx.Where("logs.request_id = ?", requestId)
|
|
}
|
|
if startTimestamp != 0 {
|
|
tx = tx.Where("logs.created_at >= ?", startTimestamp)
|
|
}
|
|
if endTimestamp != 0 {
|
|
tx = tx.Where("logs.created_at <= ?", endTimestamp)
|
|
}
|
|
if group != "" {
|
|
tx = tx.Where("logs."+logGroupCol+" = ?", group)
|
|
}
|
|
err = tx.Model(&Log{}).Limit(logSearchCountLimit).Count(&total).Error
|
|
if err != nil {
|
|
common.SysError("failed to count user logs: " + err.Error())
|
|
return nil, 0, errors.New("查询日志失败")
|
|
}
|
|
err = tx.Order("logs.id desc").Limit(num).Offset(startIdx).Find(&logs).Error
|
|
if err != nil {
|
|
common.SysError("failed to search user logs: " + err.Error())
|
|
return nil, 0, errors.New("查询日志失败")
|
|
}
|
|
|
|
formatUserLogs(logs, startIdx)
|
|
return logs, total, err
|
|
}
|
|
|
|
type Stat struct {
|
|
Quota int `json:"quota"`
|
|
Rpm int `json:"rpm"`
|
|
Tpm int `json:"tpm"`
|
|
}
|
|
|
|
func SumUsedQuota(logType int, startTimestamp int64, endTimestamp int64, modelName string, username string, tokenName string, channel int, group string) (stat Stat, err error) {
|
|
tx := LOG_DB.Table("logs").Select("sum(quota) quota")
|
|
rpmTpmQuery := LOG_DB.Table("logs").Select("count(*) rpm, sum(prompt_tokens) + sum(completion_tokens) tpm")
|
|
|
|
if username != "" {
|
|
tx = tx.Where("username = ?", username)
|
|
rpmTpmQuery = rpmTpmQuery.Where("username = ?", username)
|
|
}
|
|
if tokenName != "" {
|
|
tx = tx.Where("token_name = ?", tokenName)
|
|
rpmTpmQuery = rpmTpmQuery.Where("token_name = ?", tokenName)
|
|
}
|
|
if startTimestamp != 0 {
|
|
tx = tx.Where("created_at >= ?", startTimestamp)
|
|
}
|
|
if endTimestamp != 0 {
|
|
tx = tx.Where("created_at <= ?", endTimestamp)
|
|
}
|
|
if modelName != "" {
|
|
modelNamePattern, err := sanitizeLikePattern(modelName)
|
|
if err != nil {
|
|
return stat, err
|
|
}
|
|
tx = tx.Where("model_name LIKE ? ESCAPE '!'", modelNamePattern)
|
|
rpmTpmQuery = rpmTpmQuery.Where("model_name LIKE ? ESCAPE '!'", modelNamePattern)
|
|
}
|
|
if channel != 0 {
|
|
tx = tx.Where("channel_id = ?", channel)
|
|
rpmTpmQuery = rpmTpmQuery.Where("channel_id = ?", channel)
|
|
}
|
|
if group != "" {
|
|
tx = tx.Where(logGroupCol+" = ?", group)
|
|
rpmTpmQuery = rpmTpmQuery.Where(logGroupCol+" = ?", group)
|
|
}
|
|
|
|
tx = tx.Where("type = ?", LogTypeConsume)
|
|
rpmTpmQuery = rpmTpmQuery.Where("type = ?", LogTypeConsume)
|
|
rpmTpmQuery = rpmTpmQuery.Where("created_at >= ?", time.Now().Add(-60*time.Second).Unix())
|
|
|
|
if err := tx.Scan(&stat).Error; err != nil {
|
|
common.SysError("failed to query log stat: " + err.Error())
|
|
return stat, errors.New("查询统计数据失败")
|
|
}
|
|
if err := rpmTpmQuery.Scan(&stat).Error; err != nil {
|
|
common.SysError("failed to query rpm/tpm stat: " + err.Error())
|
|
return stat, errors.New("查询统计数据失败")
|
|
}
|
|
if stat.Quota < 0 {
|
|
stat.Quota = 0
|
|
}
|
|
if stat.Rpm < 0 {
|
|
stat.Rpm = 0
|
|
}
|
|
if stat.Tpm < 0 {
|
|
stat.Tpm = 0
|
|
}
|
|
|
|
return stat, nil
|
|
}
|
|
|
|
func SumUsedToken(logType int, startTimestamp int64, endTimestamp int64, modelName string, username string, tokenName string) (token int) {
|
|
tx := LOG_DB.Table("logs").Select("ifnull(sum(prompt_tokens),0) + ifnull(sum(completion_tokens),0)")
|
|
if username != "" {
|
|
tx = tx.Where("username = ?", username)
|
|
}
|
|
if tokenName != "" {
|
|
tx = tx.Where("token_name = ?", tokenName)
|
|
}
|
|
if startTimestamp != 0 {
|
|
tx = tx.Where("created_at >= ?", startTimestamp)
|
|
}
|
|
if endTimestamp != 0 {
|
|
tx = tx.Where("created_at <= ?", endTimestamp)
|
|
}
|
|
if modelName != "" {
|
|
tx = tx.Where("model_name = ?", modelName)
|
|
}
|
|
tx.Where("type = ?", LogTypeConsume).Scan(&token)
|
|
return token
|
|
}
|
|
|
|
func DeleteOldLog(ctx context.Context, targetTimestamp int64, limit int) (int64, error) {
|
|
var total int64 = 0
|
|
|
|
for {
|
|
if nil != ctx.Err() {
|
|
return total, ctx.Err()
|
|
}
|
|
|
|
result := LOG_DB.Where("created_at < ?", targetTimestamp).Limit(limit).Delete(&Log{})
|
|
if nil != result.Error {
|
|
return total, result.Error
|
|
}
|
|
|
|
total += result.RowsAffected
|
|
|
|
if result.RowsAffected < int64(limit) {
|
|
break
|
|
}
|
|
}
|
|
|
|
return total, nil
|
|
}
|