Files
sub2api/backend/internal/service/openai_gateway_usage.go
T
shaw d4952154ff fix: bill Grok video per second and harden video usage logging
Follow-up fixes for the #3775 audit findings:

- Bill Grok video generation per second of output, matching the xAI rate
  card: parse the request duration (1-15s, upstream default 8s) and compute
  cost as per-second price x duration x count. The built-in rate card values
  were already xAI per-second prices but were previously charged per video,
  undercharging up to 15x with a user-controlled duration.
- Group video_price_* fields are now documented and surfaced as per-second
  rates (USD/s); admin UI labels, placeholders and hints updated accordingly.
- Persist video_count/video_resolution/video_duration_seconds on usage_logs
  (migration 172) so video billing is auditable, and exempt any row with
  video_count > 0 from the image_size check constraint: a video billed via a
  token-mode channel price produces billing_mode='token' with image_count=1
  and no image_size, which the previous constraint rejected, dropping the
  whole billing transaction.
- Only refetch the group in apiKeyWithFreshGroupMediaPricing when the group
  object actually looks like it is missing media pricing fields (both media
  multipliers zero and all prices nil, impossible for a normally loaded
  group), removing a per-usage DB query for groups without overrides.
- Frontend: drop the unused admin.groups.mediaPricing locale block, map
  cleared price inputs to null (create) / -1 (update, cleared via backend
  normalizePrice) instead of sending "" that failed *float64 unmarshalling,
  and align video price placeholders with the text-to-video default model
  (grok-imagine-video 0.05/0.07, 1080p only on 1.5 at 0.25).
2026-07-09 15:38:59 +08:00

787 lines
28 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package service
// 本文件由 openai_gateway_service.go 纯移动拆分而来:用量记录、计费成本计算与
// Codex 用量快照。仅做代码搬迁,无任何行为变更。
import (
"context"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
"go.uber.org/zap"
)
// OpenAIRecordUsageInput input for recording usage
type OpenAIRecordUsageInput struct {
Result *OpenAIForwardResult
APIKey *APIKey
User *User
Account *Account
Subscription *UserSubscription
InboundEndpoint string
UpstreamEndpoint string
UserAgent string // 请求的 User-Agent
IPAddress string // 请求的客户端 IP 地址
RequestPayloadHash string
APIKeyService APIKeyQuotaUpdater
QuotaPlatform string // user×platform quota platform resolved by the handler before async billing.
// CyberBlocked 为 true 时把该用量行标记为 cyber(request_type=cyber),计费逻辑不变。
CyberBlocked bool
ChannelUsageFields
}
// CyberPolicyUsageInput 是 cyber 拒绝、未走正常 RecordUsage 的请求记录用量的入参。
// 用量按上游真实 token 计费,与 WS cyber 及正常请求口径一致(InputTokens/OutputTokens
// 取自上游 response.failed 报告的 usage,即 mark.UpstreamInTok/OutTok)。
type CyberPolicyUsageInput struct {
APIKey *APIKey
Account *Account
Subscription *UserSubscription
RequestID string
Model string
Stream bool
InputTokens int
OutputTokens int
// 渠道归因与请求级 meta,使 cyber 计费行与正常 RecordUsage 行口径一致
// (否则 cyber 行 channel_id 等为空,渠道维度统计会遗漏 cyber 命中)。
InboundEndpoint string
UpstreamEndpoint string
UserAgent string
IPAddress string
RequestPayloadHash string
APIKeyService APIKeyQuotaUpdater
ChannelUsageFields
}
// RecordCyberPolicyUsageLog 为被上游 cyber_policy 拒绝、未走正常 RecordUsage 的请求
// (HTTP forward 返回错误路径)记录用量并按上游真实 token 计费,使其与 WS cyber 路径、
// 与正常请求的计费口径统一(不再是 tokens=0 免费行)。token 取自上游 response.failed
// 报告的 usage(非流式直接拒通常为 0,cost 随之为 0)。复用 RecordUsage 完成成本计算、
// 扣费与用量行写入(request_type=cyber 由 CyberBlocked 置位)。仅 forward 返回错误的
// 路径由 handler 调用,避免与成功路径的正常 RecordUsage 重复。
func (s *OpenAIGatewayService) RecordCyberPolicyUsageLog(ctx context.Context, in CyberPolicyUsageInput) {
if s == nil || in.APIKey == nil || in.APIKey.User == nil || in.Account == nil || strings.TrimSpace(in.Model) == "" {
return
}
result := &OpenAIForwardResult{
RequestID: in.RequestID,
Model: in.Model,
Stream: in.Stream,
Usage: OpenAIUsage{
InputTokens: in.InputTokens,
OutputTokens: in.OutputTokens,
},
}
if err := s.RecordUsage(ctx, &OpenAIRecordUsageInput{
Result: result,
APIKey: in.APIKey,
User: in.APIKey.User,
Account: in.Account,
Subscription: in.Subscription,
InboundEndpoint: in.InboundEndpoint,
UpstreamEndpoint: in.UpstreamEndpoint,
UserAgent: in.UserAgent,
IPAddress: in.IPAddress,
RequestPayloadHash: in.RequestPayloadHash,
APIKeyService: in.APIKeyService,
ChannelUsageFields: in.ChannelUsageFields,
CyberBlocked: true,
}); err != nil {
logger.LegacyPrintf("service.openai_gateway", "cyber usage record failed: request_id=%s err=%v", in.RequestID, err)
}
}
// RecordUsage records usage and deducts balance
func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRecordUsageInput) error {
if input == nil {
return errors.New("openai usage input is nil")
}
result := input.Result
if result == nil {
return errors.New("openai usage result is nil")
}
if s.rateLimitService != nil && input.Account != nil && input.Account.Platform == PlatformOpenAI {
s.rateLimitService.ResetOpenAI403Counter(ctx, input.Account.ID)
}
apiKey := input.APIKey
user := input.User
account := input.Account
subscription := input.Subscription
if !isGrokVideoUsageResult(result, nil) {
ApplyOpenAIImageBillingResolution(result)
}
// 计算实际的新输入token(减去缓存读取的token)
// 因为 input_tokens 包含了 cache_read_tokens,而缓存读取的token不应按输入价格计费
actualInputTokens := result.Usage.InputTokens - result.Usage.CacheReadInputTokens
if actualInputTokens < 0 {
actualInputTokens = 0
}
// Calculate cost
tokens := UsageTokens{
InputTokens: actualInputTokens,
ImageInputTokens: result.Usage.ImageInputTokens,
OutputTokens: result.Usage.OutputTokens,
CacheCreationTokens: result.Usage.CacheCreationInputTokens,
CacheReadTokens: result.Usage.CacheReadInputTokens,
ImageOutputTokens: result.Usage.ImageOutputTokens,
}
// Get rate multiplier
multiplier := 1.0
if s.cfg != nil {
multiplier = s.cfg.Default.RateMultiplier
}
if apiKey.GroupID != nil && apiKey.Group != nil {
resolver := s.userGroupRateResolver
if resolver == nil {
resolver = newUserGroupRateResolver(nil, nil, resolveUserGroupRateCacheTTL(s.cfg), nil, "service.openai_gateway")
}
multiplier = resolver.Resolve(ctx, user.ID, *apiKey.GroupID, apiKey.Group.RateMultiplier)
}
// token 倍率叠加高峰因子(token 计费含图片 token,图片按次倍率不受影响)。高峰因子按请求时刻现算,
// 不并入上面的 Resolve,以免污染 user:group 倍率缓存。
baseMultiplier := multiplier
multiplier, imageMultiplier := computePeakAwareMultipliers(apiKey, baseMultiplier, timezone.Now())
videoMultiplier := resolveVideoRateMultiplier(apiKey, baseMultiplier)
var cost *CostBreakdown
var err error
billingModel := forwardResultBillingModel(result.Model, result.UpstreamModel)
if result.BillingModel != "" {
billingModel = strings.TrimSpace(result.BillingModel)
}
if input.BillingModelSource == BillingModelSourceChannelMapped && input.ChannelMappedModel != "" && input.ChannelMappedModel != input.OriginalModel {
billingModel = input.ChannelMappedModel
}
if input.BillingModelSource == BillingModelSourceRequested && input.OriginalModel != "" {
billingModel = input.OriginalModel
}
billingModels := usageBillingModelCandidates(
billingModel,
result.BillingModel,
input.ChannelMappedModel,
input.OriginalModel,
result.UpstreamModel,
result.Model,
)
serviceTier := ""
if result.ServiceTier != nil {
serviceTier = strings.TrimSpace(*result.ServiceTier)
}
cost, err = s.calculateOpenAIRecordUsageCost(ctx, result, apiKey, billingModels, multiplier, imageMultiplier, videoMultiplier, tokens, serviceTier)
if err != nil {
if !isUsagePricingUnavailableError(err) {
return err
}
logger.L().With(
zap.String("component", "service.openai_gateway"),
zap.Strings("billing_models", billingModels),
zap.String("requested_model", input.OriginalModel),
zap.String("mapped_model", input.ChannelMappedModel),
zap.String("upstream_model", result.UpstreamModel),
zap.Int64("api_key_id", apiKey.ID),
zap.Int64("account_id", account.ID),
).Warn("openai_usage.pricing_missing_record_zero_cost", zap.Error(err))
cost = &CostBreakdown{BillingMode: string(BillingModeToken)}
}
// Determine billing type
isSubscriptionBilling := subscription != nil && apiKey.Group != nil && apiKey.Group.IsSubscriptionType()
billingType := BillingTypeBalance
if isSubscriptionBilling {
billingType = BillingTypeSubscription
}
// Create usage log
durationMs := int(result.Duration.Milliseconds())
accountRateMultiplier := account.BillingRateMultiplier()
requestID := resolveUsageBillingRequestID(ctx, result.RequestID)
if result.OpenAIWSMode {
if upstreamRequestID := strings.TrimSpace(result.RequestID); upstreamRequestID != "" {
requestID = upstreamRequestID
}
}
// 确定 RequestedModel(渠道映射前的原始模型)
requestedModel := result.Model
if input.OriginalModel != "" {
requestedModel = input.OriginalModel
}
usageLog := &UsageLog{
UserID: user.ID,
APIKeyID: apiKey.ID,
AccountID: account.ID,
RequestID: requestID,
Model: result.Model,
RequestedModel: requestedModel,
UpstreamModel: optionalNonEqualStringPtr(result.UpstreamModel, result.Model),
ServiceTier: result.ServiceTier,
ReasoningEffort: result.ReasoningEffort,
InboundEndpoint: optionalTrimmedStringPtr(input.InboundEndpoint),
UpstreamEndpoint: optionalTrimmedStringPtr(input.UpstreamEndpoint),
InputTokens: actualInputTokens,
OutputTokens: result.Usage.OutputTokens,
CacheCreationTokens: result.Usage.CacheCreationInputTokens,
CacheReadTokens: result.Usage.CacheReadInputTokens,
ImageOutputTokens: result.Usage.ImageOutputTokens,
ImageCount: result.ImageCount,
ImageSize: optionalTrimmedStringPtr(result.ImageSize),
ImageInputSize: optionalTrimmedStringPtr(result.ImageInputSize),
ImageOutputSize: optionalTrimmedStringPtr(result.ImageOutputSize),
ImageSizeSource: optionalTrimmedStringPtr(result.ImageSizeSource),
ImageSizeBreakdown: result.ImageSizeBreakdown,
}
isVideoUsage := isGrokVideoUsageResult(result, billingModels)
if isVideoUsage {
usageLog.VideoCount = result.VideoCount
usageLog.VideoResolution = optionalTrimmedStringPtr(NormalizeVideoBillingResolutionOrDefault(result.VideoResolution))
videoDurationSeconds := NormalizeVideoBillingDurationSecondsOrDefault(result.VideoDurationSeconds)
usageLog.VideoDurationSeconds = &videoDurationSeconds
}
if cost != nil {
usageLog.InputCost = cost.InputCost
usageLog.OutputCost = cost.OutputCost
usageLog.ImageOutputCost = cost.ImageOutputCost
usageLog.CacheCreationCost = cost.CacheCreationCost
usageLog.CacheReadCost = cost.CacheReadCost
usageLog.TotalCost = cost.TotalCost
usageLog.ActualCost = cost.ActualCost
}
if isVideoUsage && (cost == nil || cost.BillingMode != string(BillingModeToken)) {
usageLog.RateMultiplier = videoMultiplier
} else if result.ImageCount > 0 && (cost == nil || cost.BillingMode != string(BillingModeToken)) {
usageLog.RateMultiplier = imageMultiplier
} else {
usageLog.RateMultiplier = multiplier
}
usageLog.AccountRateMultiplier = &accountRateMultiplier
usageLog.BillingType = billingType
usageLog.Stream = result.Stream
if input.CyberBlocked {
usageLog.RequestType = RequestTypeCyberBlocked
}
usageLog.OpenAIWSMode = result.OpenAIWSMode
usageLog.DurationMs = &durationMs
usageLog.FirstTokenMs = result.FirstTokenMs
usageLog.CreatedAt = time.Now()
// 设置渠道信息
usageLog.ChannelID = optionalInt64Ptr(input.ChannelID)
usageLog.ModelMappingChain = optionalTrimmedStringPtr(input.ModelMappingChain)
// 设置计费模式
if cost != nil && cost.BillingMode != "" {
billingMode := cost.BillingMode
usageLog.BillingMode = &billingMode
} else if isVideoUsage {
billingMode := string(BillingModeVideo)
usageLog.BillingMode = &billingMode
} else if result.ImageCount > 0 {
billingMode := string(BillingModeImage)
usageLog.BillingMode = &billingMode
} else {
billingMode := string(BillingModeToken)
usageLog.BillingMode = &billingMode
}
// 添加 UserAgent
if input.UserAgent != "" {
usageLog.UserAgent = &input.UserAgent
}
// 添加 IPAddress
if input.IPAddress != "" {
usageLog.IPAddress = &input.IPAddress
}
if apiKey.GroupID != nil {
usageLog.GroupID = apiKey.GroupID
}
if subscription != nil {
usageLog.SubscriptionID = &subscription.ID
}
// 计算账号统计定价费用(使用最终上游模型匹配自定义规则)
if apiKey.GroupID != nil {
applyAccountStatsCost(ctx, usageLog, s.channelService, s.billingService,
account.ID, *apiKey.GroupID, result.UpstreamModel, result.Model,
tokens, cost.TotalCost,
)
}
if s.cfg != nil && s.cfg.RunMode == config.RunModeSimple {
writeUsageLogBestEffort(ctx, s.usageLogRepo, usageLog, "service.openai_gateway")
logger.LegacyPrintf("service.openai_gateway", "[SIMPLE MODE] Usage recorded (not billed): user=%d, tokens=%d", usageLog.UserID, usageLog.TotalTokens())
s.deferredService.ScheduleLastUsedUpdate(account.ID)
return nil
}
// Async usage billing runs outside the original request context, so it
// cannot recover ForcePlatform there. Fall back for internal/test callers.
quotaPlatform := input.QuotaPlatform
if quotaPlatform == "" {
quotaPlatform = PlatformFromAPIKey(apiKey)
}
billingErr := func() error {
_, err := applyUsageBilling(ctx, requestID, usageLog, &postUsageBillingParams{
Cost: cost,
User: user,
APIKey: apiKey,
Account: account,
Subscription: subscription,
RequestPayloadHash: resolveUsageBillingPayloadFingerprint(ctx, input.RequestPayloadHash),
IsSubscriptionBill: isSubscriptionBilling,
AccountRateMultiplier: accountRateMultiplier,
APIKeyService: input.APIKeyService,
Platform: quotaPlatform,
}, s.billingDeps(), s.usageBillingRepo)
return err
}()
if billingErr != nil {
return billingErr
}
writeUsageLogBestEffort(ctx, s.usageLogRepo, usageLog, "service.openai_gateway")
return nil
}
func (s *OpenAIGatewayService) calculateOpenAIRecordUsageCost(
ctx context.Context,
result *OpenAIForwardResult,
apiKey *APIKey,
billingModels []string,
multiplier float64,
imageMultiplier float64,
videoMultiplier float64,
tokens UsageTokens,
serviceTier string,
) (*CostBreakdown, error) {
billingModel := firstUsageBillingModel(billingModels)
if isGrokVideoUsageResult(result, billingModels) {
if resolved := s.resolveOpenAIChannelPricing(ctx, billingModel, apiKey); resolved == nil || resolved.Mode != BillingModeToken {
return s.calculateOpenAIVideoCost(ctx, billingModel, apiKey, result, videoMultiplier), nil
}
}
if result != nil && result.ImageCount > 0 {
// 渠道定价为 token 计费时走 token 路径,否则走图片计费
if resolved := s.resolveOpenAIChannelPricing(ctx, billingModel, apiKey); resolved == nil || resolved.Mode != BillingModeToken {
return s.calculateOpenAIImageCost(ctx, billingModel, apiKey, result, imageMultiplier), nil
}
}
if len(billingModels) == 0 || billingModel == "" {
return nil, errors.New("openai usage billing model is empty")
}
var lastErr error
for _, candidate := range billingModels {
candidate = strings.TrimSpace(candidate)
if candidate == "" {
continue
}
cost, err := s.calculateOpenAIRecordUsageTokenCost(ctx, apiKey, candidate, multiplier, tokens, serviceTier)
if err == nil {
return cost, nil
}
lastErr = err
}
if lastErr == nil {
lastErr = errors.New("no non-empty billing model candidates")
}
return nil, fmt.Errorf("calculate OpenAI usage cost failed for billing models %s: %w", strings.Join(billingModels, ","), lastErr)
}
func isGrokVideoBillingModel(model string) bool {
return strings.HasPrefix(strings.ToLower(strings.TrimSpace(model)), "grok-imagine-video")
}
func isGrokVideoUsageResult(result *OpenAIForwardResult, billingModels []string) bool {
if result == nil || result.VideoCount <= 0 {
return false
}
candidates := append([]string{}, billingModels...)
candidates = append(candidates, result.BillingModel, result.Model, result.UpstreamModel)
for _, candidate := range candidates {
if isGrokVideoBillingModel(candidate) {
return true
}
}
return false
}
func isUsagePricingUnavailableError(err error) bool {
if err == nil {
return false
}
if errors.Is(err, ErrModelPricingUnavailable) {
return true
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "no pricing available") || strings.Contains(msg, "pricing not found")
}
func (s *OpenAIGatewayService) calculateOpenAIRecordUsageTokenCost(
ctx context.Context,
apiKey *APIKey,
billingModel string,
multiplier float64,
tokens UsageTokens,
serviceTier string,
) (*CostBreakdown, error) {
if s.resolver != nil && apiKey.Group != nil {
gid := apiKey.Group.ID
return s.billingService.CalculateCostUnified(CostInput{
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
Tokens: tokens,
RequestCount: 1,
RateMultiplier: multiplier,
ServiceTier: serviceTier,
Resolver: s.resolver,
})
}
return s.billingService.CalculateCostWithServiceTier(billingModel, tokens, multiplier, serviceTier)
}
func (s *OpenAIGatewayService) calculateOpenAIImageCost(
ctx context.Context,
billingModel string,
apiKey *APIKey,
result *OpenAIForwardResult,
multiplier float64,
) *CostBreakdown {
sizeTier := NormalizeImageBillingTierOrDefault(result.ImageSize)
groupConfig := imagePriceConfigFromAPIKey(apiKey)
if apiKeyHasConfiguredImagePrice(apiKey, sizeTier) {
return s.billingService.CalculateImageCost(billingModel, sizeTier, result.ImageCount, groupConfig, multiplier)
}
if refreshed := s.apiKeyWithFreshGroupMediaPricing(ctx, apiKey); refreshed != apiKey {
apiKey = refreshed
groupConfig = imagePriceConfigFromAPIKey(apiKey)
if apiKeyHasConfiguredImagePrice(apiKey, sizeTier) {
return s.billingService.CalculateImageCost(billingModel, sizeTier, result.ImageCount, groupConfig, multiplier)
}
}
if resolved := s.resolveOpenAIChannelPricing(ctx, billingModel, apiKey); resolved != nil &&
(resolved.Mode == BillingModePerRequest || resolved.Mode == BillingModeImage) {
gid := apiKey.Group.ID
cost, err := s.billingService.CalculateCostUnified(CostInput{
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
RequestCount: result.ImageCount,
SizeTier: sizeTier,
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
})
if err == nil {
return cost
}
logger.LegacyPrintf("service.openai_gateway", "Calculate image channel cost failed: %v", err)
}
return s.billingService.CalculateImageCost(billingModel, sizeTier, result.ImageCount, groupConfig, multiplier)
}
func (s *OpenAIGatewayService) calculateOpenAIVideoCost(
ctx context.Context,
billingModel string,
apiKey *APIKey,
result *OpenAIForwardResult,
multiplier float64,
) *CostBreakdown {
videoCount := result.VideoCount
if videoCount <= 0 {
videoCount = 1
}
resolution := NormalizeVideoBillingResolutionOrDefault(result.VideoResolution)
durationSeconds := NormalizeVideoBillingDurationSecondsOrDefault(result.VideoDurationSeconds)
groupConfig := videoPriceConfigFromAPIKey(apiKey)
if apiKeyHasConfiguredVideoPrice(apiKey, resolution) {
return s.billingService.CalculateVideoCost(billingModel, resolution, videoCount, durationSeconds, groupConfig, multiplier)
}
if refreshed := s.apiKeyWithFreshGroupMediaPricing(ctx, apiKey); refreshed != apiKey {
apiKey = refreshed
groupConfig = videoPriceConfigFromAPIKey(apiKey)
if apiKeyHasConfiguredVideoPrice(apiKey, resolution) {
return s.billingService.CalculateVideoCost(billingModel, resolution, videoCount, durationSeconds, groupConfig, multiplier)
}
}
if resolved := s.resolveOpenAIChannelPricing(ctx, billingModel, apiKey); resolved != nil &&
(resolved.Mode == BillingModePerRequest || resolved.Mode == BillingModeImage) {
// 渠道 per_request/image 定价保持"按请求次数"口径(价格由管理员按次配置),不乘视频时长。
gid := apiKey.Group.ID
cost, err := s.billingService.CalculateCostUnified(CostInput{
Ctx: ctx,
Model: billingModel,
GroupID: &gid,
RequestCount: videoCount,
SizeTier: resolution,
RateMultiplier: multiplier,
Resolver: s.resolver,
Resolved: resolved,
})
if err == nil {
cost.BillingMode = string(BillingModeVideo)
return cost
}
logger.LegacyPrintf("service.openai_gateway", "Calculate video channel cost failed: %v", err)
}
return s.billingService.CalculateVideoCost(billingModel, resolution, videoCount, durationSeconds, groupConfig, multiplier)
}
func (s *OpenAIGatewayService) apiKeyWithFreshGroupMediaPricing(ctx context.Context, apiKey *APIKey) *APIKey {
if apiKey == nil || apiKey.GroupID == nil || *apiKey.GroupID <= 0 {
return apiKey
}
if !groupMediaPricingLooksIncomplete(apiKey.Group) {
return apiKey
}
if s == nil || s.channelService == nil || s.channelService.groupRepo == nil {
return apiKey
}
group, err := s.channelService.groupRepo.GetByIDLite(ctx, *apiKey.GroupID)
if err != nil || group == nil {
return apiKey
}
clone := *apiKey
clone.Group = group
return &clone
}
// groupMediaPricingLooksIncomplete 判断分组对象是否可能缺失媒体计费字段(例如由不含
// 这些字段的旧快照或手工构造的上下文对象生成)。image/video 独立倍率在数据库中的
// 默认值均为 1.0,正常加载的分组不可能两个倍率同时为 0 且未开启独立倍率、全部媒体
// 价为 nil——只有这种情况才回源查库,避免对未配置覆盖价的分组每条媒体用量都多打一次 DB 查询。
func groupMediaPricingLooksIncomplete(group *Group) bool {
if group == nil {
return true
}
if group.ImageRateIndependent || group.VideoRateIndependent {
return false
}
if group.ImageRateMultiplier != 0 || group.VideoRateMultiplier != 0 {
return false
}
return group.ImagePrice1K == nil && group.ImagePrice2K == nil && group.ImagePrice4K == nil &&
group.VideoPrice480P == nil && group.VideoPrice720P == nil && group.VideoPrice1080P == nil
}
func (s *OpenAIGatewayService) resolveOpenAIChannelPricing(ctx context.Context, billingModel string, apiKey *APIKey) *ResolvedPricing {
if s.resolver == nil || apiKey == nil || apiKey.Group == nil {
return nil
}
gid := apiKey.Group.ID
resolved := s.resolver.Resolve(ctx, PricingInput{Model: billingModel, GroupID: &gid})
if resolved.Source == PricingSourceChannel {
return resolved
}
return nil
}
// ParseCodexRateLimitHeaders extracts Codex usage limits from response headers.
// Exported for use in ratelimit_service when handling OpenAI 429 responses.
func ParseCodexRateLimitHeaders(headers http.Header) *OpenAICodexUsageSnapshot {
snapshot := &OpenAICodexUsageSnapshot{}
hasData := false
// Helper to parse float64 from header
parseFloat := func(key string) *float64 {
if v := headers.Get(key); v != "" {
if f, err := strconv.ParseFloat(v, 64); err == nil {
return &f
}
}
return nil
}
// Helper to parse int from header
parseInt := func(key string) *int {
if v := headers.Get(key); v != "" {
if i, err := strconv.Atoi(v); err == nil {
return &i
}
}
return nil
}
// Primary (weekly) limits
if v := parseFloat("x-codex-primary-used-percent"); v != nil {
snapshot.PrimaryUsedPercent = v
hasData = true
}
if v := parseInt("x-codex-primary-reset-after-seconds"); v != nil {
snapshot.PrimaryResetAfterSeconds = v
hasData = true
}
if v := parseInt("x-codex-primary-window-minutes"); v != nil {
snapshot.PrimaryWindowMinutes = v
hasData = true
}
// Secondary (5h) limits
if v := parseFloat("x-codex-secondary-used-percent"); v != nil {
snapshot.SecondaryUsedPercent = v
hasData = true
}
if v := parseInt("x-codex-secondary-reset-after-seconds"); v != nil {
snapshot.SecondaryResetAfterSeconds = v
hasData = true
}
if v := parseInt("x-codex-secondary-window-minutes"); v != nil {
snapshot.SecondaryWindowMinutes = v
hasData = true
}
// Overflow ratio
if v := parseFloat("x-codex-primary-over-secondary-limit-percent"); v != nil {
snapshot.PrimaryOverSecondaryPercent = v
hasData = true
}
if !hasData {
return nil
}
snapshot.UpdatedAt = time.Now().Format(time.RFC3339)
return snapshot
}
func codexSnapshotBaseTime(snapshot *OpenAICodexUsageSnapshot, fallback time.Time) time.Time {
if snapshot == nil {
return fallback
}
if snapshot.UpdatedAt == "" {
return fallback
}
base, err := time.Parse(time.RFC3339, snapshot.UpdatedAt)
if err != nil {
return fallback
}
return base
}
func codexResetAtRFC3339(base time.Time, resetAfterSeconds *int) *string {
if resetAfterSeconds == nil {
return nil
}
sec := *resetAfterSeconds
if sec < 0 {
sec = 0
}
resetAt := base.Add(time.Duration(sec) * time.Second).Format(time.RFC3339)
return &resetAt
}
func buildCodexUsageExtraUpdates(snapshot *OpenAICodexUsageSnapshot, fallbackNow time.Time) map[string]any {
if snapshot == nil {
return nil
}
baseTime := codexSnapshotBaseTime(snapshot, fallbackNow)
updates := make(map[string]any)
// 保存原始 primary/secondary 字段,便于排查问题
if snapshot.PrimaryUsedPercent != nil {
updates["codex_primary_used_percent"] = *snapshot.PrimaryUsedPercent
}
if snapshot.PrimaryResetAfterSeconds != nil {
updates["codex_primary_reset_after_seconds"] = *snapshot.PrimaryResetAfterSeconds
}
if snapshot.PrimaryWindowMinutes != nil {
updates["codex_primary_window_minutes"] = *snapshot.PrimaryWindowMinutes
}
if snapshot.SecondaryUsedPercent != nil {
updates["codex_secondary_used_percent"] = *snapshot.SecondaryUsedPercent
}
if snapshot.SecondaryResetAfterSeconds != nil {
updates["codex_secondary_reset_after_seconds"] = *snapshot.SecondaryResetAfterSeconds
}
if snapshot.SecondaryWindowMinutes != nil {
updates["codex_secondary_window_minutes"] = *snapshot.SecondaryWindowMinutes
}
if snapshot.PrimaryOverSecondaryPercent != nil {
updates["codex_primary_over_secondary_percent"] = *snapshot.PrimaryOverSecondaryPercent
}
updates["codex_usage_updated_at"] = baseTime.Format(time.RFC3339)
// 归一化到 5h/7d 规范字段
if normalized := snapshot.Normalize(); normalized != nil {
if normalized.Used5hPercent != nil {
updates["codex_5h_used_percent"] = *normalized.Used5hPercent
}
if normalized.Reset5hSeconds != nil {
updates["codex_5h_reset_after_seconds"] = *normalized.Reset5hSeconds
}
if normalized.Window5hMinutes != nil {
updates["codex_5h_window_minutes"] = *normalized.Window5hMinutes
}
if normalized.Used7dPercent != nil {
updates["codex_7d_used_percent"] = *normalized.Used7dPercent
}
if normalized.Reset7dSeconds != nil {
updates["codex_7d_reset_after_seconds"] = *normalized.Reset7dSeconds
}
if normalized.Window7dMinutes != nil {
updates["codex_7d_window_minutes"] = *normalized.Window7dMinutes
}
if reset5hAt := codexResetAtRFC3339(baseTime, normalized.Reset5hSeconds); reset5hAt != nil {
updates["codex_5h_reset_at"] = *reset5hAt
}
if reset7dAt := codexResetAtRFC3339(baseTime, normalized.Reset7dSeconds); reset7dAt != nil {
updates["codex_7d_reset_at"] = *reset7dAt
}
}
return updates
}
// updateCodexUsageSnapshot saves the Codex usage snapshot to account's Extra field
// updateCodexUsageSnapshot 把 /responses 的 x-codex-* 全局头快照写入账号 codex_* Extra。
// ⚠️ 调用方必须排除 spark 影子账号(account.IsShadow()):影子的 codex_* 仅由 QueryUsage
// (/wham/usage bengalfox 道)更新,不能被全局头口径污染(外审第7轮 P1)。本函数仅持 accountID,
// 无法在此自检影子,故守卫前置到各调用点。
func (s *OpenAIGatewayService) updateCodexUsageSnapshot(ctx context.Context, accountID int64, snapshot *OpenAICodexUsageSnapshot) {
if snapshot == nil {
return
}
if s == nil || s.accountRepo == nil {
return
}
now := time.Now()
updates := buildCodexUsageExtraUpdates(snapshot, now)
if len(updates) == 0 {
return
}
if !s.getCodexSnapshotThrottle().Allow(accountID, now) {
return
}
go func() {
updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = s.accountRepo.UpdateExtra(updateCtx, accountID, updates)
}()
}
func (s *OpenAIGatewayService) UpdateCodexUsageSnapshotFromHeaders(ctx context.Context, accountID int64, headers http.Header) {
if accountID <= 0 || headers == nil {
return
}
if snapshot := ParseCodexRateLimitHeaders(headers); snapshot != nil {
s.updateCodexUsageSnapshot(ctx, accountID, snapshot)
}
}