package model import ( "context" "errors" "fmt" "strings" "time" "github.com/QuantumNous/new-api/common" "github.com/QuantumNous/new-api/logger" "github.com/QuantumNous/new-api/types" "github.com/gin-gonic/gin" "gorm.io/gorm" ) func applyExplicitLogTextFilter(tx *gorm.DB, column string, value string) (*gorm.DB, error) { if value == "" { return tx, nil } if strings.Contains(value, "%") { condition, pattern, err := buildLogLikeCondition(column, value) if err != nil { return nil, err } return tx.Where(condition, pattern), nil } return tx.Where(column+" = ?", value), nil } func buildLogLikeCondition(column string, value string) (string, string, error) { if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { pattern, err := sanitizeClickHouseLikePattern(value) if err != nil { return "", "", err } return column + " LIKE ?", pattern, nil } pattern, err := sanitizeLikePattern(value) if err != nil { return "", "", err } return column + " LIKE ? ESCAPE '!'", pattern, nil } func sanitizeClickHouseLikePattern(input string) (string, error) { input = strings.ReplaceAll(input, `\`, `\\`) input = strings.ReplaceAll(input, `_`, `\_`) if err := validateLikePattern(input); err != nil { return "", err } return input, nil } type Log struct { Id int `json:"id" gorm:"index:idx_created_at_id,priority:2;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:1;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:''"` UpstreamRequestId string `json:"upstream_request_id,omitempty" gorm:"type:varchar(128);index:idx_logs_upstream_request_id;default:''"` Other string `json:"other"` } // don't use iota, avoid change log type value const ( LogTypeUnknown = 0 LogTypeTopup = 1 LogTypeConsume = 2 LogTypeManage = 3 LogTypeSystem = 4 LogTypeError = 5 LogTypeRefund = 6 LogTypeLogin = 7 ) func ensureLogRequestId(log *Log) { if log != nil && log.RequestId == "" { log.RequestId = common.NewRequestId() } } func createLog(log *Log) error { ensureLogRequestId(log) return LOG_DB.Create(log).Error } func clickHouseLogOrder(prefix string) string { return prefix + "created_at desc, " + prefix + "request_id desc" } func assignDisplayLogIds(logs []*Log, startIdx int) { for i := range logs { logs[i].Id = startIdx + i + 1 } } 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 { // Remove admin-only debug fields. delete(otherMap, "admin_info") // Remove operation-audit details (operator/route info), admin-only. delete(otherMap, "audit_info") // delete(otherMap, "reject_reason") delete(otherMap, "stream_status") } logs[i].Other = common.MapToJsonStr(otherMap) } assignDisplayLogIds(logs, startIdx) } func GetLogByTokenId(tokenId int) (logs []*Log, err error) { order := "id desc" if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { order = clickHouseLogOrder("") } err = LOG_DB.Model(&Log{}).Where("token_id = ?", tokenId).Order(order).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 := createLog(log) if err != nil { common.SysLog("failed to record log: " + err.Error()) } } // RecordLogWithAdminInfo 记录操作日志,并将管理员相关信息存入 Other.admin_info, func RecordLogWithAdminInfo(userId int, logType int, content string, adminInfo map[string]interface{}) { if logType == LogTypeConsume && !common.LogConsumeEnabled { return } username, _ := GetUsernameById(userId, false) log := &Log{ UserId: userId, Username: username, CreatedAt: common.GetTimestamp(), Type: logType, Content: content, } if len(adminInfo) > 0 { other := map[string]interface{}{ "admin_info": adminInfo, } log.Other = common.MapToJsonStr(other) } if err := createLog(log); err != nil { common.SysLog("failed to record log: " + err.Error()) } } // buildOpField 构建语言无关的操作描述(写入 Other.op)。 // 前端依据 action(稳定操作标识) + params(结构化参数) 在渲染期用 i18n 本地化展示, // 因此不在数据库中存储自然语言句子。 func buildOpField(action string, params map[string]interface{}) map[string]interface{} { op := map[string]interface{}{ "action": action, } if len(params) > 0 { op["params"] = params } return op } // RecordLoginLog 记录用户登录成功的审计日志(type=LogTypeLogin)。 // username 由调用方传入(登录流程已持有用户对象),避免额外的数据库查询。 // content 为英文兜底文本(用于导出);action+params 供前端本地化渲染。 // extra 可携带 login_method、user_agent 等附加信息(普通用户可见)。 func RecordLoginLog(userId int, username string, content string, ip string, action string, params map[string]interface{}, extra map[string]interface{}) { other := map[string]interface{}{} for k, v := range extra { other[k] = v } other["op"] = buildOpField(action, params) log := &Log{ UserId: userId, Username: username, CreatedAt: common.GetTimestamp(), Type: LogTypeLogin, Content: content, Ip: ip, Other: common.MapToJsonStr(other), } if err := createLog(log); err != nil { common.SysLog("failed to record login log: " + err.Error()) } } // RecordOperationAuditLog 记录管理/高危操作审计日志(type=LogTypeManage)。 // logUserId 为日志归属者,管理审计日志应归属实际操作者;目标资源/用户放入 // action params。username 内部按 logUserId 查询。content 为英文兜底文本(供导出使用)。 // action+params 写入 Other.op,供前端本地化渲染(普通用户可见,不含敏感信息)。 // adminInfo 存放操作者身份(写入 Other.admin_info,普通用户查询时剥离); // auditInfo 存放路由/方法/结果等中间件兜底信息(写入 Other.audit_info,普通用户查询时剥离)。 func RecordOperationAuditLog(logUserId int, content string, ip string, action string, params map[string]interface{}, adminInfo map[string]interface{}, auditInfo map[string]interface{}) { username, _ := GetUsernameById(logUserId, false) other := map[string]interface{}{ "op": buildOpField(action, params), } if len(adminInfo) > 0 { other["admin_info"] = adminInfo } if len(auditInfo) > 0 { other["audit_info"] = auditInfo } log := &Log{ UserId: logUserId, Username: username, CreatedAt: common.GetTimestamp(), Type: LogTypeManage, Content: content, Ip: ip, Other: common.MapToJsonStr(other), } if err := createLog(log); err != nil { common.SysLog("failed to record operation audit 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(), "node_name": common.NodeName, "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 := createLog(log) 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, common.LocalLogPreview(content))) username := c.GetString("username") requestId := c.GetString(common.RequestIdKey) upstreamRequestId := c.GetString(common.UpstreamRequestIdKey) otherStr := common.MapToJsonStr(other) // 判断是否需要记录 IP 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, UpstreamRequestId: upstreamRequestId, Other: otherStr, } err := createLog(log) 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) upstreamRequestId := c.GetString(common.UpstreamRequestIdKey) createdAt := common.GetTimestamp() otherStr := common.MapToJsonStr(params.Other) // 判断是否需要记录 IP needRecordIp := false if settingMap, err := GetUserSetting(userId, false); err == nil { if settingMap.RecordIpLog { needRecordIp = true } } log := &Log{ UserId: userId, Username: username, CreatedAt: createdAt, 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, UpstreamRequestId: upstreamRequestId, Other: otherStr, } err := createLog(log) if err != nil { logger.LogError(c, "failed to record log: "+err.Error()) } if common.DataExportEnabled { LogQuotaData(QuotaDataLogParams{ UserID: userId, Username: username, ModelName: params.ModelName, Quota: params.Quota, CreatedAt: createdAt, TokenUsed: params.PromptTokens + params.CompletionTokens, UseGroup: params.Group, TokenID: params.TokenId, ChannelID: params.ChannelId, NodeName: common.NodeName, }) } } type RecordTaskBillingLogParams struct { UserId int LogType int Content string ChannelId int ModelName string Quota int TokenId int Group string Other map[string]interface{} NodeName string // 任务发起节点;为空时回退当前节点 } 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 } } createdAt := common.GetTimestamp() log := &Log{ UserId: params.UserId, Username: username, CreatedAt: createdAt, 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 := createLog(log) if err != nil { common.SysLog("failed to record task billing log: " + err.Error()) } if params.LogType == LogTypeConsume && common.DataExportEnabled { nodeName := params.NodeName if nodeName == "" { nodeName = common.NodeName } LogQuotaData(QuotaDataLogParams{ UserID: params.UserId, Username: username, ModelName: params.ModelName, Quota: params.Quota, CreatedAt: createdAt, UseGroup: params.Group, TokenID: params.TokenId, ChannelID: params.ChannelId, NodeName: nodeName, }) } } func GetAllLogs(logType int, startTimestamp int64, endTimestamp int64, modelName string, username string, tokenName string, startIdx int, num int, channel int, group string, requestId string, upstreamRequestId 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 tx, err = applyExplicitLogTextFilter(tx, "logs.model_name", modelName); err != nil { return nil, 0, err } if tx, err = applyExplicitLogTextFilter(tx, "logs.username", username); err != nil { return nil, 0, err } if tokenName != "" { tx = tx.Where("logs.token_name = ?", tokenName) } if requestId != "" { tx = tx.Where("logs.request_id = ?", requestId) } if upstreamRequestId != "" { tx = tx.Where("logs.upstream_request_id = ?", upstreamRequestId) } 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 } order := "logs.created_at desc, logs.id desc" if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { order = clickHouseLogOrder("logs.") } err = tx.Order(order).Limit(num).Offset(startIdx).Find(&logs).Error if err != nil { return nil, 0, err } if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { assignDisplayLogIds(logs, startIdx) } 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 { // Cache get channel 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 { // Bulk query channels from DB 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, upstreamRequestId 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 tx, err = applyExplicitLogTextFilter(tx, "logs.model_name", modelName); err != nil { return nil, 0, err } if tokenName != "" { tx = tx.Where("logs.token_name = ?", tokenName) } if requestId != "" { tx = tx.Where("logs.request_id = ?", requestId) } if upstreamRequestId != "" { tx = tx.Where("logs.upstream_request_id = ?", upstreamRequestId) } 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("查询日志失败") } order := "logs.id desc" if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { order = clickHouseLogOrder("logs.") } err = tx.Order(order).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("COALESCE(sum(quota), 0) quota") // 为rpm和tpm创建单独的查询 rpmTpmQuery := LOG_DB.Table("logs").Select("count(*) rpm, COALESCE(sum(prompt_tokens), 0) + COALESCE(sum(completion_tokens), 0) tpm") if tx, err = applyExplicitLogTextFilter(tx, "username", username); err != nil { return stat, err } if rpmTpmQuery, err = applyExplicitLogTextFilter(rpmTpmQuery, "username", username); err != nil { return stat, err } 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 tx, err = applyExplicitLogTextFilter(tx, "model_name", modelName); err != nil { return stat, err } if rpmTpmQuery, err = applyExplicitLogTextFilter(rpmTpmQuery, "model_name", modelName); err != nil { return stat, err } 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) // 只统计最近60秒的rpm和tpm 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("查询统计数据失败") } 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("COALESCE(sum(prompt_tokens), 0) + COALESCE(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 CountOldLog(ctx context.Context, targetTimestamp int64) (int64, error) { var total int64 if err := LOG_DB.WithContext(ctx).Model(&Log{}).Where("created_at < ?", targetTimestamp).Count(&total).Error; err != nil { return 0, err } return total, nil } func DeleteOldLogBatch(ctx context.Context, targetTimestamp int64, limit int) (int64, error) { if limit <= 0 { limit = 100 } if nil != ctx.Err() { return 0, ctx.Err() } if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { // ClickHouse DELETE is a heavy mutation that rewrites data parts, so // per-batch mutations would be pathologically slow. Remove all matching // rows in a single synchronous mutation regardless of limit; the reported // count lets the caller's progress loop complete in one pass. total, err := CountOldLog(ctx, targetTimestamp) if err != nil { return 0, err } if total == 0 { return 0, nil } if err := LOG_DB.WithContext(ctx).Exec( "ALTER TABLE logs DELETE WHERE created_at < ? SETTINGS mutations_sync = 1", targetTimestamp, ).Error; err != nil { return 0, err } return total, nil } result := LOG_DB.WithContext(ctx).Where("created_at < ?", targetTimestamp).Limit(limit).Delete(&Log{}) if nil != result.Error { return 0, result.Error } return result.RowsAffected, nil }