package sql import ( "database/sql" "fmt" "time" "ccLoad/internal/model" ) func scanAggregatedMetricsRows(rows *sql.Rows) (map[int64]*model.MetricPoint, map[int64]*metricAggregationHelper, map[int64]bool, error) { mapp := make(map[int64]*model.MetricPoint) channelIDsToFetch := make(map[int64]bool) helperMap := make(map[int64]*metricAggregationHelper) for rows.Next() { var bucketTsFloat float64 var channelID sql.NullInt64 var success, errorCount int var avgFirstByteTime sql.NullFloat64 var avgDuration sql.NullFloat64 var streamSuccessFirstByteCount int var durationSuccessCount int var totalCost, effectiveCost float64 var inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens int64 if err := rows.Scan(&bucketTsFloat, &channelID, &success, &errorCount, &avgFirstByteTime, &avgDuration, &streamSuccessFirstByteCount, &durationSuccessCount, &totalCost, &effectiveCost, &inputTokens, &outputTokens, &cacheReadTokens, &cacheCreationTokens); err != nil { return nil, nil, nil, err } bucketTs := int64(bucketTsFloat) mp, ok := mapp[bucketTs] if !ok { mp = &model.MetricPoint{ Ts: time.Unix(bucketTs, 0), Channels: make(map[string]model.ChannelMetric), } mapp[bucketTs] = mp } helper, ok := helperMap[bucketTs] if !ok { helper = &metricAggregationHelper{} helperMap[bucketTs] = helper } mp.Success += success mp.Error += errorCount if mp.TotalCost == nil { mp.TotalCost = new(float64) } *mp.TotalCost += totalCost if mp.EffectiveCost == nil { mp.EffectiveCost = new(float64) } *mp.EffectiveCost += effectiveCost // 累加 token 统计 mp.InputTokens += inputTokens mp.OutputTokens += outputTokens mp.CacheReadTokens += cacheReadTokens mp.CacheCreationTokens += cacheCreationTokens if avgFirstByteTime.Valid { helper.totalFirstByteTime += avgFirstByteTime.Float64 * float64(streamSuccessFirstByteCount) helper.firstByteCount += streamSuccessFirstByteCount } if avgDuration.Valid { helper.totalDuration += avgDuration.Float64 * float64(durationSuccessCount) helper.durationCount += durationSuccessCount } channelKey := "未知渠道" if channelID.Valid { channelKey = fmt.Sprintf("ch_%d", channelID.Int64) channelIDsToFetch[channelID.Int64] = true } var avgFBT *float64 if avgFirstByteTime.Valid { avgFBT = new(float64) *avgFBT = avgFirstByteTime.Float64 } var avgDur *float64 if avgDuration.Valid { avgDur = new(float64) *avgDur = avgDuration.Float64 } var chCost *float64 if totalCost > 0 { chCost = new(float64) *chCost = totalCost } var chEffective *float64 if effectiveCost > 0 || totalCost > 0 { chEffective = new(float64) *chEffective = effectiveCost } mp.Channels[channelKey] = model.ChannelMetric{ Success: success, Error: errorCount, AvgFirstByteTimeSeconds: avgFBT, AvgDurationSeconds: avgDur, TotalCost: chCost, EffectiveCost: chEffective, InputTokens: inputTokens, OutputTokens: outputTokens, CacheReadTokens: cacheReadTokens, CacheCreationTokens: cacheCreationTokens, } } if err := rows.Err(); err != nil { return nil, nil, nil, err } return mapp, helperMap, channelIDsToFetch, nil }