Files
sub2api-overdraft/backend/internal/service/ratelimit_cn_providers.go
Randark 901a0439f1 feat: 国产供应商一等支持(Kimi/Zhipu/DeepSeek 多协议 + 配额/余额监控)
后端:
- 协议凭证维度 credentials[api_protocol] ∈ chat_completions(默认)/anthropic/responses(deepseek)
- /v1/messages 零转换直通原生 Anthropic 端点(kimi/zhipu/deepseek),CC/Responses
  入站交叉组合走 apicompat 双向转换链(responses/chat_completions anthropic-native 转发器)
- count_tokens:anthropic 协议透传原生端点;其余 CN 协议本地 tiktoken 估算
- Coding Plan 额度探测(5h/weekly 滚动窗口)+ payg 余额探测(kimi/deepseek),
  deepseek 双币种 CNY+USD 明细,任一币种达标不停调
- 周期任务 [CNBalance] 并发探测 + 预算随工作量放大;响应式 429 冷却到最早窗口
  重置点;余额不足可恢复临时停调;智谱 CREDIT_LIMIT 不污染窗口解析
- CC→anthropic 流式客户端断开后继续排水上游保住 usage 计量

前端:
- 创建/编辑弹窗 account_mode + api_protocol + base_url 联动预设(含 watcher 竞态防护)
- 用量单元格:kimi/zhipu coding 显示 5h/weekly 窗口,kimi/deepseek payg 显示余额,
  多币种并列展示;探测失败保留快照;挂载自动探测 5min 去抖
- 调度阈值设置面板补 kimi/zhipu 平台(对齐后端 AllowedSchedulingThresholdPlatforms)
2026-08-15 10:37:51 +00:00

155 lines
6.1 KiB
Go
Raw Permalink 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
import (
"context"
"log/slog"
"net/http"
"strings"
"time"
)
// 国产供应商(kimi/zhipu/deepseek)的响应式冷却辅助。
//
// 与 openai/anthropic 不同:
// - 余额不足是「可恢复」状态(充值/检测恢复后自动重新调度),不能走 handleAuthError
// 永久置 status=error。这里改为 SetTempUnschedulable,由 CN 余额检测周期任务
// (cn_provider_balance_check_service.go)在余额恢复后 ClearTempUnschedulable。
// - Coding Plan 滚动窗口耗尽(429)的冷却终点应是真实的窗口重置时间(已由
// CNProviderQuotaService 落入 account.Extra 快照),而非默认的秒级兜底。
// cnBalanceExtraSuffixLow 标记账号响应过「余额不足」,供余额检测任务区分
// 「确属余额不足」与「尚未探测」。
const cnBalanceExtraSuffixLow = "balance_low"
// cnBalanceLowReasonPrefix 是余额不足临时停调 reason 的稳定前缀。
// 周期余额检测任务据此识别「是我们停调的」并在余额恢复后安全清除——不会误清
// 其他子系统(阈值/限流/401)写入的临时停调。
const cnBalanceLowReasonPrefix = "cn_balance_low"
// cnBalanceLowReason 构造余额不足临时停调的 reason(带稳定前缀)。
func cnBalanceLowReason(upstreamMsg string) string {
if upstreamMsg = strings.TrimSpace(upstreamMsg); upstreamMsg != "" {
return cnBalanceLowReasonPrefix + ": " + upstreamMsg
}
return cnBalanceLowReasonPrefix + ": 余额不足,账号临时停调"
}
// cnProviderResponseIndicatesInsufficientBalance 通过响应体文案识别余额不足
// (智谱 payg 无独立余额端点,仅能靠响应文案识别)。
func cnProviderResponseIndicatesInsufficientBalance(body []byte) bool {
if len(body) == 0 {
return false
}
s := strings.ToLower(string(body))
return strings.Contains(s, "余额不足") ||
strings.Contains(s, "insufficient balance") ||
strings.Contains(s, "insufficient_credit") ||
strings.Contains(s, "balance is not enough") ||
strings.Contains(s, "no enough balance")
}
// handleCNProviderInsufficientBalance 把余额不足标记为可恢复的临时停调:
// 写入 balance_low 快照 + SetTempUnschedulable 一个余额检测周期,
// 由周期任务在余额恢复后清除。返回前已通知调度阻塞。
func (s *RateLimitService) handleCNProviderInsufficientBalance(
ctx context.Context,
account *Account,
upstreamMsg string,
) {
msg := cnBalanceLowReason(upstreamMsg)
if err := s.accountRepo.UpdateExtra(ctx, account.ID, map[string]any{
cnExtraKey(account.Platform, cnBalanceExtraSuffixLow): true,
}); err != nil {
slog.Warn("cn_balance_low_mark_failed", "account_id", account.ID, "error", err)
}
until := time.Now().Add(s.cnBalanceCooldownDuration())
s.notifyAccountSchedulingBlocked(account, until, "cn_insufficient_balance")
if err := s.accountRepo.SetTempUnschedulable(ctx, account.ID, until, msg); err != nil {
slog.Warn("cn_balance_set_temp_unschedulable_failed", "account_id", account.ID, "error", err)
return
}
slog.Info("cn_provider_insufficient_balance",
"account_id", account.ID,
"platform", account.Platform,
"until", until.UTC(),
)
}
// cnBalanceCooldownDuration 返回余额不足临时停调的持续时长(= 2× 余额检测周期,
// 默认 20 分钟)。周期任务会在余额恢复后提前清除,故此处只需保证冷却覆盖到下一次
// 周期检测即可。
func (s *RateLimitService) cnBalanceCooldownDuration() time.Duration {
minutes := 10
if s != nil && s.cfg != nil {
if cfgMin := s.cfg.Gateway.CNProviders.BalanceCheckIntervalMinutes; cfgMin > 0 {
minutes = cfgMin
}
}
cooldown := time.Duration(minutes) * time.Minute * 2
if cooldown < time.Minute {
cooldown = 10 * time.Minute
}
return cooldown
}
// cnProviderQuotaSnapshotReset 读取 Coding Plan 账号快照中最早一个仍在未来的窗口
// 重置时间(5h / weekly)。429 多数由 5h 滚动窗口触发,取较早的重置点可避免
// 把账号冷却到 weekly 重置(可达数天)的过度停调;如果确是 weekly 窗口耗尽,
// 周期额度探测刷新快照后阈值评估会再次停调到正确的时间点。
// 无快照或均已过期返回 nil。
func cnProviderQuotaSnapshotReset(account *Account, now time.Time) *time.Time {
if account == nil || !account.IsCNProvider() || !account.IsCodingPlan() || len(account.Extra) == 0 {
return nil
}
provider := account.Platform
var earliest *time.Time
for _, suffix := range []string{cnExtraSuffix5hReset, cnExtraSuffixWeeklyReset} {
t := parseSchedulingResetAt(account.Extra[cnExtraKey(provider, suffix)])
if t == nil || !t.After(now) {
continue
}
if earliest == nil || t.Before(*earliest) {
earliest = t
}
}
return earliest
}
// applyCNProviderReactive429 处理国产供应商的 429 响应。
// 返回 true 表示已处理(调用方应 return),false 表示未命中、继续走默认 429 逻辑。
func (s *RateLimitService) applyCNProviderReactive429(
ctx context.Context,
account *Account,
headers http.Header,
responseBody []byte,
) bool {
if !account.IsCNProvider() {
return false
}
// 1) 余额不足文案:可恢复临时停调(含智谱 payg 这类无余额端点的场景)。
if cnProviderResponseIndicatesInsufficientBalance(responseBody) {
s.handleCNProviderInsufficientBalance(ctx, account, extractUpstreamErrorMessage(responseBody))
return true
}
// 2) Coding Plan 窗口耗尽:冷却到快照中最早的窗口重置点(见
// cnProviderQuotaSnapshotReset:429 多由 5h 窗口触发,取较早点避免过度停调)。
if account.IsCodingPlan() {
if until := cnProviderQuotaSnapshotReset(account, time.Now()); until != nil {
s.notifyAccountSchedulingBlocked(account, *until, "429")
if err := s.accountRepo.SetRateLimited(ctx, account.ID, *until); err != nil {
slog.Warn("rate_limit_set_failed", "account_id", account.ID, "error", err)
return true
}
slog.Info("cn_coding_plan_rate_limited",
"account_id", account.ID,
"platform", account.Platform,
"reset_at", *until,
)
return true
}
}
return false
}