1202 lines
43 KiB
Go
1202 lines
43 KiB
Go
package admin
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"math"
|
||
"time"
|
||
|
||
do "github.com/qianfree/team-api/internal/model/do"
|
||
|
||
"github.com/gogf/gf/v2/database/gdb"
|
||
"github.com/gogf/gf/v2/errors/gerror"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
|
||
"github.com/qianfree/team-api/api/admin/v1"
|
||
"github.com/qianfree/team-api/internal/dao"
|
||
"github.com/qianfree/team-api/internal/dispatchadapter"
|
||
"github.com/qianfree/team-api/internal/logic/common"
|
||
"github.com/qianfree/team-api/internal/logic/relay"
|
||
uc "github.com/qianfree/team-api/internal/utility/crypto"
|
||
"github.com/qianfree/team-api/internal/utility/export"
|
||
)
|
||
|
||
// providerTypeNames 供应商类型名称映射
|
||
var providerTypeNames = map[int]string{
|
||
1: "OpenAI",
|
||
2: "Claude",
|
||
3: "Gemini",
|
||
4: "Ali",
|
||
5: "Baidu",
|
||
6: "Tencent",
|
||
7: "Zhipu",
|
||
8: "DeepSeek",
|
||
9: "Moonshot",
|
||
10: "Volcengine",
|
||
11: "AWS Bedrock",
|
||
12: "Azure OpenAI",
|
||
13: "Vertex AI",
|
||
14: "Cohere",
|
||
15: "Mistral",
|
||
16: "xAI",
|
||
41: "New API",
|
||
42: "Sub2API",
|
||
}
|
||
|
||
// channelRuntimeStates 返回目录快照的渠道熔断聚合（管理后台列表/详情展示用）。
|
||
// 目录未初始化（首次 relay 请求前）时返回 nil，调用方按「正常」处理。
|
||
func channelRuntimeStates() map[int64]dispatchadapter.ChannelRuntimeState {
|
||
cat := dispatchadapter.CatalogInstance()
|
||
if cat == nil {
|
||
return nil
|
||
}
|
||
return cat.ChannelRuntimeStates()
|
||
}
|
||
|
||
// channelOptionsCache 渠道下拉选项缓存（全量渠道，单个条目）。
|
||
// 日志 / 统计页的渠道筛选每次打开都要拉一次全量渠道，而渠道量级为几十且极少变动，
|
||
// 反复对 chn_channels 全表扫描纯属浪费；整表结果缓存为一个条目，写侧主动失效。
|
||
var channelOptionsCache = common.NewCache("channel_options", 300*time.Second)
|
||
|
||
// channelOptionsCacheKey 渠道下拉是全量结果，只有一个缓存条目。
|
||
const channelOptionsCacheKey = "all"
|
||
|
||
// InvalidateChannelOptionsCache 失效渠道下拉选项缓存（渠道增 / 删 / 改 / 克隆后调用）。
|
||
func InvalidateChannelOptionsCache(ctx context.Context) {
|
||
channelOptionsCache.Delete(ctx, channelOptionsCacheKey)
|
||
}
|
||
|
||
// ListChannelOptions 渠道下拉选项（不分页，返回全部渠道）。
|
||
// 刻意不做状态过滤：日志 / 统计页的渠道筛选项要覆盖历史数据里出现过的全部渠道，
|
||
// 已停用渠道同样需要能作为筛选条件；status 原样返回，由前端决定是否标记。
|
||
// 注意与 /admin/channels 的区别：后者是渠道管理列表（分页 + 健康度联表 + 运行态），
|
||
// 下拉场景不要用它 —— 那是本接口存在的意义。
|
||
func (s *sAdmin) ListChannelOptions(ctx context.Context, _ *v1.ChannelOptionsReq) (*v1.ChannelOptionsRes, error) {
|
||
list := make([]v1.ChannelOptionItem, 0)
|
||
if channelOptionsCache.GetJSON(ctx, channelOptionsCacheKey, &list) {
|
||
return &v1.ChannelOptionsRes{List: list}, nil
|
||
}
|
||
|
||
var rows []struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Type int `json:"type"`
|
||
Status string `json:"status"`
|
||
}
|
||
err := dao.ChnChannels.Ctx(ctx).
|
||
Fields("id, name, type, status").
|
||
OrderAsc("id").
|
||
Scan(&rows)
|
||
if err = common.IgnoreScanNoRows(err); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
list = make([]v1.ChannelOptionItem, 0, len(rows))
|
||
for _, r := range rows {
|
||
list = append(list, v1.ChannelOptionItem{
|
||
ID: r.ID,
|
||
Name: r.Name,
|
||
Type: r.Type,
|
||
Status: r.Status,
|
||
})
|
||
}
|
||
|
||
channelOptionsCache.Set(ctx, channelOptionsCacheKey, list)
|
||
return &v1.ChannelOptionsRes{List: list}, nil
|
||
}
|
||
|
||
// ListChannels 获取渠道列表
|
||
func (s *sAdmin) ListChannels(ctx context.Context, req *v1.ChannelListReq) (*v1.ChannelListRes, error) {
|
||
query := dao.ChnChannels.Ctx(ctx).
|
||
LeftJoin("chn_health_scores h ON chn_channels.id = h.channel_id")
|
||
|
||
if req.Type > 0 {
|
||
query = query.Where("chn_channels.type", req.Type)
|
||
}
|
||
if req.Status != "" {
|
||
query = query.Where("chn_channels.status", req.Status)
|
||
}
|
||
if req.Search != "" {
|
||
query = query.Where("chn_channels.name LIKE ? OR chn_channels.remark LIKE ?", "%"+req.Search+"%", "%"+req.Search+"%")
|
||
}
|
||
if req.ID > 0 {
|
||
query = query.Where("chn_channels.id", req.ID)
|
||
}
|
||
if req.BaseURL != "" {
|
||
query = query.Where("chn_channels.base_url LIKE ?", "%"+req.BaseURL+"%")
|
||
}
|
||
if req.Model != "" {
|
||
// 按支持的模型名筛选：渠道需具备匹配的模型能力（chn_abilities）。
|
||
// 用 EXISTS 相关子查询而非 JOIN，避免一个渠道命中多个模型时产生重复行、破坏分页与总数。
|
||
query = query.Where("EXISTS (SELECT 1 FROM chn_abilities WHERE chn_abilities.channel_id = chn_channels.id AND (chn_abilities.model_name LIKE ? OR chn_abilities.upstream_model LIKE ?))", "%"+req.Model+"%", "%"+req.Model+"%")
|
||
}
|
||
|
||
var total int
|
||
var channels []struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Type int `json:"type"`
|
||
BaseURL string `json:"base_url"`
|
||
Status string `json:"status"`
|
||
Priority int `json:"priority"`
|
||
Weight int `json:"weight"`
|
||
MaxConcurrency int `json:"max_concurrency"`
|
||
Tier string `json:"tier"`
|
||
StrictCapacity bool `json:"strict_capacity"`
|
||
TestModel string `json:"test_model"`
|
||
Remark string `json:"remark"`
|
||
IsVIP bool `json:"is_vip"`
|
||
SharingThreshold *float64 `json:"sharing_threshold"`
|
||
PreemptionThreshold *float64 `json:"preemption_threshold"`
|
||
BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds"`
|
||
CreatedAt *gtime.Time `json:"created_at"`
|
||
HealthScore *float64 `json:"health_score"`
|
||
Settings string `json:"settings"`
|
||
}
|
||
|
||
err := query.Fields("chn_channels.id, chn_channels.name, chn_channels.type, chn_channels.base_url, chn_channels.status, chn_channels.priority, chn_channels.weight, chn_channels.max_concurrency, chn_channels.tier, chn_channels.strict_capacity, chn_channels.test_model, chn_channels.remark, chn_channels.is_vip, chn_channels.sharing_threshold, chn_channels.preemption_threshold, chn_channels.borrowing_cooldown_seconds, chn_channels.created_at, h.health_score, chn_channels.settings").
|
||
// 列表默认按 id 倒序：新建渠道排在最前，便于运营查看最新配置。
|
||
// 注意分页场景下排序必须在 DB 层完成，前端只在当前页排序会导致跨页错乱。
|
||
OrderDesc("chn_channels.id").
|
||
Page(req.Page, req.PageSize).
|
||
ScanAndCount(&channels, &total, false)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
list := make([]v1.ChannelItem, 0, len(channels))
|
||
runtimeStates := channelRuntimeStates()
|
||
for _, ch := range channels {
|
||
typeName := providerTypeNames[ch.Type]
|
||
if typeName == "" {
|
||
typeName = fmt.Sprintf("Unknown(%d)", ch.Type)
|
||
}
|
||
|
||
settings := relay.ParseChannelSettings(ch.Settings)
|
||
|
||
item := v1.ChannelItem{
|
||
ID: ch.ID,
|
||
Name: ch.Name,
|
||
Type: ch.Type,
|
||
TypeName: typeName,
|
||
BaseURL: ch.BaseURL,
|
||
Status: ch.Status,
|
||
Priority: ch.Priority,
|
||
Weight: ch.Weight,
|
||
MaxConcurrency: ch.MaxConcurrency,
|
||
Tier: ch.Tier,
|
||
StrictCapacity: ch.StrictCapacity,
|
||
TestModel: ch.TestModel,
|
||
Remark: ch.Remark,
|
||
IsVIP: ch.IsVIP,
|
||
UseProxy: settings.UseProxy,
|
||
SharingThreshold: ch.SharingThreshold,
|
||
PreemptionThreshold: ch.PreemptionThreshold,
|
||
BorrowingCooldownSeconds: ch.BorrowingCooldownSeconds,
|
||
CreatedAt: ch.CreatedAt.String(),
|
||
HealthScore: ch.HealthScore,
|
||
}
|
||
if rs, ok := runtimeStates[ch.ID]; ok {
|
||
item.BreakerState = int(rs.Breaker)
|
||
item.BreakerModels = rs.BreakerModels
|
||
item.WorstModel = rs.WorstModel
|
||
item.WorstModelScore = rs.WorstModelScore
|
||
}
|
||
list = append(list, item)
|
||
}
|
||
|
||
return &v1.ChannelListRes{
|
||
List: list,
|
||
Total: total,
|
||
Page: req.Page,
|
||
PageSize: req.PageSize,
|
||
}, nil
|
||
}
|
||
|
||
// CloneChannel 克隆渠道
|
||
func (s *sAdmin) CloneChannel(ctx context.Context, req *v1.ChannelCloneReq) (*v1.ChannelCloneRes, error) {
|
||
// 查询源渠道
|
||
var src *struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Type int `json:"type"`
|
||
BaseURL string `json:"base_url"`
|
||
Priority int `json:"priority"`
|
||
Weight int `json:"weight"`
|
||
MaxConcurrency int `json:"max_concurrency"`
|
||
TestModel string `json:"test_model"`
|
||
Remark string `json:"remark"`
|
||
Settings string `json:"settings"`
|
||
IsVIP bool `json:"is_vip"`
|
||
SharingThreshold float64 `json:"sharing_threshold"`
|
||
PreemptionThreshold float64 `json:"preemption_threshold"`
|
||
BorrowingCooldownSeconds int `json:"borrowing_cooldown_seconds"`
|
||
}
|
||
err := dao.ChnChannels.Ctx(ctx).
|
||
Fields("id, name, type, base_url, priority, weight, max_concurrency, test_model, remark, settings, is_vip, sharing_threshold, preemption_threshold, borrowing_cooldown_seconds").
|
||
Where("id", req.ID).
|
||
Scan(&src)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if src == nil {
|
||
return nil, common.NewBusinessError(404, "源渠道不存在")
|
||
}
|
||
|
||
// 确定名称
|
||
name := req.Name
|
||
if name == "" {
|
||
name = src.Name + " (副本)"
|
||
}
|
||
|
||
// 加密 Key（放在事务外，加密失败无需回滚任何写入）
|
||
encKey := relay.GetEncryptionKey()
|
||
encrypted, err := uc.EncryptString(encKey, req.ApiKey)
|
||
if err != nil {
|
||
return nil, gerror.Wrapf(err, "encrypt api key failed")
|
||
}
|
||
|
||
// 渠道 + 密钥在同一事务内创建：密钥插入失败则回滚渠道，避免留下没有任何 Key 的孤儿渠道。
|
||
var newID int64
|
||
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||
var e error
|
||
newID, e = dao.ChnChannels.Ctx(ctx).InsertAndGetId(do.ChnChannels{
|
||
Name: name,
|
||
Type: src.Type,
|
||
BaseUrl: src.BaseURL,
|
||
Status: "active",
|
||
Priority: src.Priority,
|
||
Weight: src.Weight,
|
||
MaxConcurrency: src.MaxConcurrency,
|
||
TestModel: src.TestModel,
|
||
Remark: src.Remark,
|
||
Settings: src.Settings,
|
||
IsVip: src.IsVIP,
|
||
SharingThreshold: src.SharingThreshold,
|
||
PreemptionThreshold: src.PreemptionThreshold,
|
||
BorrowingCooldownSeconds: src.BorrowingCooldownSeconds,
|
||
})
|
||
if e != nil {
|
||
return e
|
||
}
|
||
_, e = dao.ChnChannelKeys.Ctx(ctx).Insert(do.ChnChannelKeys{
|
||
ChannelId: newID,
|
||
Name: "default",
|
||
EncryptedKey: encrypted,
|
||
Status: "active",
|
||
})
|
||
return e
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 克隆 abilities
|
||
var abilities []struct {
|
||
ModelName string `json:"model_name"`
|
||
UpstreamModel string `json:"upstream_model"`
|
||
Enabled bool `json:"enabled"`
|
||
CostRatio float64 `json:"cost_ratio"`
|
||
SupportsResponses bool `json:"supports_responses"`
|
||
ChatViaResponses bool `json:"chat_via_responses"`
|
||
}
|
||
err = dao.ChnAbilities.Ctx(ctx).
|
||
Fields("model_name, upstream_model, enabled, cost_ratio, supports_responses, chat_via_responses").
|
||
Where("channel_id", req.ID).
|
||
Scan(&abilities)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for _, ab := range abilities {
|
||
_, err := dao.ChnAbilities.Ctx(ctx).Insert(do.ChnAbilities{
|
||
ChannelId: newID,
|
||
ModelName: ab.ModelName,
|
||
UpstreamModel: ab.UpstreamModel,
|
||
Enabled: ab.Enabled,
|
||
CostRatio: ab.CostRatio,
|
||
SupportsResponses: ab.SupportsResponses,
|
||
ChatViaResponses: ab.ChatViaResponses,
|
||
})
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "clone ability %s for channel %d failed: %v", ab.ModelName, newID, err)
|
||
}
|
||
}
|
||
|
||
// 初始化健康分
|
||
if err := relay.InitHealthScore(ctx, newID); err != nil {
|
||
g.Log().Warningf(ctx, "init health score for channel %d failed: %v", newID, err)
|
||
}
|
||
|
||
InvalidateChannelOptionsCache(ctx)
|
||
|
||
return &v1.ChannelCloneRes{ID: newID}, nil
|
||
}
|
||
|
||
// CreateChannel 创建渠道
|
||
func (s *sAdmin) CreateChannel(ctx context.Context, req *v1.ChannelCreateReq) (*v1.ChannelCreateRes, error) {
|
||
// 自动填充默认 Base URL
|
||
baseURL := req.BaseURL
|
||
if baseURL == "" {
|
||
baseURL = defaultProviderURL(req.Type)
|
||
}
|
||
|
||
// Build settings JSON（只写显式开启的布尔项，其余字段用 Parse 时的默认值）
|
||
settingsJSON := "{}"
|
||
if req.UseProxy || req.WebSearchToGoogleSearch {
|
||
boolFlags := map[string]bool{}
|
||
if req.UseProxy {
|
||
boolFlags["use_proxy"] = true
|
||
}
|
||
if req.WebSearchToGoogleSearch {
|
||
boolFlags["web_search_to_google_search"] = true
|
||
}
|
||
if b, err := json.Marshal(boolFlags); err == nil {
|
||
settingsJSON = string(b)
|
||
}
|
||
}
|
||
|
||
tier := req.Tier
|
||
if tier == "" {
|
||
tier = "primary"
|
||
}
|
||
id, err := dao.ChnChannels.Ctx(ctx).InsertAndGetId(do.ChnChannels{
|
||
Name: req.Name,
|
||
Type: req.Type,
|
||
BaseUrl: baseURL,
|
||
Status: "active",
|
||
Priority: req.Priority,
|
||
Weight: req.Weight,
|
||
MaxConcurrency: req.MaxConcurrency,
|
||
Tier: tier,
|
||
StrictCapacity: req.StrictCapacity,
|
||
TestModel: req.TestModel,
|
||
Remark: req.Remark,
|
||
Settings: settingsJSON,
|
||
IsVip: req.IsVIP,
|
||
SharingThreshold: req.SharingThreshold,
|
||
PreemptionThreshold: req.PreemptionThreshold,
|
||
BorrowingCooldownSeconds: req.BorrowingCooldownSeconds,
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 加密并创建 Key
|
||
encKey := relay.GetEncryptionKey()
|
||
encrypted, err := uc.EncryptString(encKey, req.ApiKey)
|
||
if err != nil {
|
||
return nil, gerror.Wrapf(err, "encrypt api key failed")
|
||
}
|
||
_, err = dao.ChnChannelKeys.Ctx(ctx).Insert(do.ChnChannelKeys{
|
||
ChannelId: id,
|
||
Name: "default",
|
||
EncryptedKey: encrypted,
|
||
Status: "active",
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 初始化健康度记录
|
||
if err := relay.InitHealthScore(ctx, id); err != nil {
|
||
g.Log().Warningf(ctx, "init health score for channel %d failed: %v", id, err)
|
||
}
|
||
|
||
InvalidateChannelOptionsCache(ctx)
|
||
|
||
return &v1.ChannelCreateRes{ID: id}, nil
|
||
}
|
||
|
||
// UpdateChannel 更新渠道
|
||
func (s *sAdmin) UpdateChannel(ctx context.Context, req *v1.ChannelUpdateReq) (*v1.ChannelUpdateRes, error) {
|
||
data := do.ChnChannels{}
|
||
if req.Name != "" {
|
||
data.Name = req.Name
|
||
}
|
||
if req.Type != nil {
|
||
data.Type = *req.Type
|
||
}
|
||
if req.BaseURL != "" {
|
||
data.BaseUrl = req.BaseURL
|
||
} else if req.Type != nil {
|
||
// 切换供应商类型且未显式传 base_url：旧地址为空或等于旧类型默认值（用户未自定义过）时，
|
||
// 自动跟随为新类型默认地址，避免残留旧供应商端点导致上游请求 404；自定义地址（如中转站）保留不动
|
||
var old struct {
|
||
Type int `json:"type"`
|
||
BaseUrl string `json:"base_url"`
|
||
}
|
||
if err := dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Fields("type, base_url").Scan(&old); err == nil {
|
||
if newDefault := defaultProviderURL(*req.Type); newDefault != "" &&
|
||
(old.BaseUrl == "" || old.BaseUrl == defaultProviderURL(old.Type)) {
|
||
data.BaseUrl = newDefault
|
||
}
|
||
}
|
||
}
|
||
// priority/weight 为指针字段，仅在显式传入时更新，避免不含这两个字段的
|
||
// 局部更新（如仅切换状态、仅换 Key）把已有值洗成 0。
|
||
if req.Priority != nil {
|
||
data.Priority = *req.Priority
|
||
}
|
||
if req.Weight != nil {
|
||
data.Weight = *req.Weight
|
||
}
|
||
if req.MaxConcurrency != nil {
|
||
data.MaxConcurrency = *req.MaxConcurrency
|
||
}
|
||
if req.TestModel != "" {
|
||
data.TestModel = req.TestModel
|
||
}
|
||
if req.Remark != "" {
|
||
data.Remark = req.Remark
|
||
}
|
||
if req.Status != "" {
|
||
data.Status = req.Status
|
||
}
|
||
if req.Tier != "" {
|
||
data.Tier = req.Tier
|
||
}
|
||
if req.StrictCapacity != nil {
|
||
data.StrictCapacity = *req.StrictCapacity
|
||
}
|
||
if req.IsVIP != nil {
|
||
data.IsVip = *req.IsVIP
|
||
}
|
||
if req.SharingThreshold != nil {
|
||
data.SharingThreshold = *req.SharingThreshold
|
||
}
|
||
if req.PreemptionThreshold != nil {
|
||
data.PreemptionThreshold = *req.PreemptionThreshold
|
||
}
|
||
if req.BorrowingCooldownSeconds != nil {
|
||
data.BorrowingCooldownSeconds = *req.BorrowingCooldownSeconds
|
||
}
|
||
|
||
// Update use_proxy / debug_log_* in settings JSONB。
|
||
// 必须单块合并处理（读一次旧值 → 同时应用全部字段 → 回写一次）：
|
||
// 若写成多个同构块，各自从 DB 读旧值再各自 marshal，同请求多字段时会互相覆盖。
|
||
if req.UseProxy != nil || req.WebSearchToGoogleSearch != nil || req.DebugLogEnabled != nil ||
|
||
req.DebugLogTenantID != nil || req.DebugLogUserID != nil || req.DebugLogApiKeyID != nil {
|
||
// 注意：读单列字符串必须用 Value()——gdb 的 Model.Scan 只接受 struct 系列，
|
||
// 传入 *string 会直接报错；此前用 _ = Scan(&string) 吞错导致永远读到空串，
|
||
// 每次局部更新都会把未提交的 settings 字段静默重置为默认值
|
||
var currentSettings string
|
||
if v, err := dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Fields("settings").Value(); err == nil {
|
||
currentSettings = v.String()
|
||
}
|
||
settings := relay.ParseChannelSettings(currentSettings)
|
||
if req.UseProxy != nil {
|
||
settings.UseProxy = *req.UseProxy
|
||
}
|
||
if req.WebSearchToGoogleSearch != nil {
|
||
settings.WebSearchToGoogleSearch = *req.WebSearchToGoogleSearch
|
||
}
|
||
if req.DebugLogEnabled != nil {
|
||
settings.DebugLogEnabled = *req.DebugLogEnabled
|
||
}
|
||
if req.DebugLogTenantID != nil {
|
||
settings.DebugLogTenantID = *req.DebugLogTenantID
|
||
}
|
||
if req.DebugLogUserID != nil {
|
||
settings.DebugLogUserID = *req.DebugLogUserID
|
||
}
|
||
if req.DebugLogApiKeyID != nil {
|
||
settings.DebugLogApiKeyID = *req.DebugLogApiKeyID
|
||
}
|
||
if settingsJSON, err := json.Marshal(settings); err == nil {
|
||
data.Settings = string(settingsJSON)
|
||
}
|
||
}
|
||
|
||
_, err := dao.ChnChannels.Ctx(ctx).
|
||
Where("id", req.ID).
|
||
Data(data).
|
||
Update()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 更新 Key（如果提供了新的 API Key）
|
||
if req.ApiKey != nil && *req.ApiKey != "" {
|
||
encKey := relay.GetEncryptionKey()
|
||
encrypted, err := uc.EncryptString(encKey, *req.ApiKey)
|
||
if err != nil {
|
||
return nil, gerror.Wrapf(err, "encrypt api key failed")
|
||
}
|
||
// 用 g.Map 而非 do 结构体：需要把 token_expires_at 写回 NULL（do 的零值会被 gf 忽略）。
|
||
// 由 OAuth 改回静态 Key 时旧的 access_token 过期时间必须清掉，否则详情页会显示
|
||
// 一个与当前 Key 无关的过期时间；last_error 同理属于旧 Key 的历史。
|
||
_, err = dao.ChnChannelKeys.Ctx(ctx).
|
||
Where("channel_id", req.ID).
|
||
Data(g.Map{
|
||
"encrypted_key": encrypted,
|
||
"key_type": "apikey",
|
||
"token_expires_at": nil,
|
||
"last_error": "",
|
||
}).
|
||
Update()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 换 Key 必须解除凭证冷却：冷却按 keyID 打标而这里是原地改同一行（keyID 不变），
|
||
// 不解除的话新 Key 会被调度器整段跳过直到 TTL 到期，期间连试都不试。
|
||
if n := dispatchadapter.ClearChannelCredentialCooldown(ctx, req.ID); n > 0 {
|
||
g.Log().Infof(ctx, "[Channel] 渠道 %d 更换 Key，已解除 %d 个 Key 的凭证冷却", req.ID, n)
|
||
}
|
||
}
|
||
dispatchadapter.InvalidateChannel(ctx, req.ID)
|
||
InvalidateChannelOptionsCache(ctx)
|
||
if req.Status == "active" {
|
||
// 手动启用/恢复：复位熔断并开启爬坡窗口，恢复初期小流量验证（rampFactor）
|
||
dispatchadapter.MarkChannelRecovered(ctx, req.ID)
|
||
}
|
||
|
||
return nil, nil
|
||
}
|
||
|
||
// DeleteChannel 删除渠道
|
||
func (s *sAdmin) DeleteChannel(ctx context.Context, req *v1.ChannelDeleteReq) (*v1.ChannelDeleteRes, error) {
|
||
err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||
if _, err := dao.ChnAbilities.Ctx(ctx).Where("channel_id", req.ID).Delete(); err != nil {
|
||
return err
|
||
}
|
||
if _, err := dao.ChnChannelKeys.Ctx(ctx).Where("channel_id", req.ID).Delete(); err != nil {
|
||
return err
|
||
}
|
||
if _, err := dao.ChnHealthScores.Ctx(ctx).Where("channel_id", req.ID).Delete(); err != nil {
|
||
return err
|
||
}
|
||
if _, err := dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Delete(); err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
dispatchadapter.InvalidateChannel(ctx, req.ID)
|
||
InvalidateChannelOptionsCache(ctx)
|
||
|
||
return nil, nil
|
||
}
|
||
|
||
// GetChannelDetail 获取渠道详情
|
||
func (s *sAdmin) GetChannelDetail(ctx context.Context, req *v1.ChannelDetailReq) (*v1.ChannelDetailRes, error) {
|
||
var ch *struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Type int `json:"type"`
|
||
BaseURL string `json:"base_url"`
|
||
Status string `json:"status"`
|
||
Priority int `json:"priority"`
|
||
Weight int `json:"weight"`
|
||
MaxConcurrency int `json:"max_concurrency"`
|
||
Tier string `json:"tier"`
|
||
StrictCapacity bool `json:"strict_capacity"`
|
||
TestModel string `json:"test_model"`
|
||
Remark string `json:"remark"`
|
||
IsVIP bool `json:"is_vip"`
|
||
Settings string `json:"settings"`
|
||
SharingThreshold *float64 `json:"sharing_threshold"`
|
||
PreemptionThreshold *float64 `json:"preemption_threshold"`
|
||
BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds"`
|
||
CreatedAt *gtime.Time `json:"created_at"`
|
||
UpdatedAt *gtime.Time `json:"updated_at"`
|
||
HealthScore *float64 `json:"health_score"`
|
||
}
|
||
|
||
err := dao.ChnChannels.Ctx(ctx).
|
||
LeftJoin("chn_health_scores h ON chn_channels.id = h.channel_id").
|
||
Fields("chn_channels.id, chn_channels.name, chn_channels.type, chn_channels.base_url, chn_channels.status, chn_channels.priority, chn_channels.weight, chn_channels.max_concurrency, chn_channels.tier, chn_channels.strict_capacity, chn_channels.test_model, chn_channels.remark, chn_channels.is_vip, chn_channels.settings, chn_channels.sharing_threshold, chn_channels.preemption_threshold, chn_channels.borrowing_cooldown_seconds, chn_channels.created_at, chn_channels.updated_at, h.health_score").
|
||
Where("chn_channels.id", req.ID).
|
||
Scan(&ch)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if ch == nil {
|
||
return nil, common.NewBusinessError(404, "渠道不存在")
|
||
}
|
||
|
||
typeName := providerTypeNames[ch.Type]
|
||
if typeName == "" {
|
||
typeName = fmt.Sprintf("Unknown(%d)", ch.Type)
|
||
}
|
||
|
||
// 查询 Key 信息
|
||
var keyInfo *struct {
|
||
ID int64 `json:"id"`
|
||
KeyType string `json:"key_type"`
|
||
Status string `json:"status"`
|
||
Name string `json:"name"`
|
||
TokenExpiresAt *gtime.Time `json:"token_expires_at"`
|
||
}
|
||
err = dao.ChnChannelKeys.Ctx(ctx).
|
||
Where("channel_id", req.ID).
|
||
Fields("id, key_type, status, name, token_expires_at").
|
||
Scan(&keyInfo)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if keyInfo == nil {
|
||
return nil, common.NewNotFoundError("密钥")
|
||
}
|
||
|
||
settings := relay.ParseChannelSettings(ch.Settings)
|
||
|
||
res := &v1.ChannelDetailRes{
|
||
ID: ch.ID,
|
||
Name: ch.Name,
|
||
Type: ch.Type,
|
||
TypeName: typeName,
|
||
BaseURL: ch.BaseURL,
|
||
Status: ch.Status,
|
||
Priority: ch.Priority,
|
||
Weight: ch.Weight,
|
||
MaxConcurrency: ch.MaxConcurrency,
|
||
Tier: ch.Tier,
|
||
StrictCapacity: ch.StrictCapacity,
|
||
TestModel: ch.TestModel,
|
||
Remark: ch.Remark,
|
||
IsVIP: ch.IsVIP,
|
||
UseProxy: settings.UseProxy,
|
||
WebSearchToGoogleSearch: settings.WebSearchToGoogleSearch,
|
||
DebugLogEnabled: settings.DebugLogEnabled,
|
||
DebugLogTenantID: settings.DebugLogTenantID,
|
||
DebugLogUserID: settings.DebugLogUserID,
|
||
DebugLogApiKeyID: settings.DebugLogApiKeyID,
|
||
SharingThreshold: ch.SharingThreshold,
|
||
PreemptionThreshold: ch.PreemptionThreshold,
|
||
BorrowingCooldownSeconds: ch.BorrowingCooldownSeconds,
|
||
CreatedAt: ch.CreatedAt.String(),
|
||
UpdatedAt: ch.UpdatedAt.String(),
|
||
HealthScore: ch.HealthScore,
|
||
KeyType: keyInfo.KeyType,
|
||
KeyStatus: keyInfo.Status,
|
||
KeyName: keyInfo.Name,
|
||
KeyCooldownRemaining: dispatchadapter.CredentialCooldownRemaining(ctx, keyInfo.ID),
|
||
}
|
||
if rs, ok := channelRuntimeStates()[req.ID]; ok {
|
||
res.BreakerState = int(rs.Breaker)
|
||
res.BreakerModels = rs.BreakerModels
|
||
res.WorstModel = rs.WorstModel
|
||
res.WorstModelScore = rs.WorstModelScore
|
||
}
|
||
if res.KeyType == "" {
|
||
res.KeyType = "apikey"
|
||
}
|
||
if keyInfo.TokenExpiresAt != nil {
|
||
res.TokenExpiresAt = keyInfo.TokenExpiresAt.String()
|
||
}
|
||
|
||
return res, nil
|
||
}
|
||
|
||
// AddChannelKey 添加渠道 API Key（已废弃：每渠道仅支持一个 Key）
|
||
func (s *sAdmin) AddChannelKey(ctx context.Context, req *v1.ChannelKeyCreateReq) (*v1.ChannelKeyCreateRes, error) {
|
||
return nil, common.NewBusinessError(400, "每渠道仅支持一个 Key，请通过编辑渠道更新 Key")
|
||
}
|
||
|
||
// DeleteChannelKey 删除渠道 API Key
|
||
func (s *sAdmin) DeleteChannelKey(ctx context.Context, req *v1.ChannelKeyDeleteReq) (*v1.ChannelKeyDeleteRes, error) {
|
||
_, err := dao.ChnChannelKeys.Ctx(ctx).
|
||
Where("id", req.KeyID).
|
||
Where("channel_id", req.ChannelID).
|
||
Delete()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
dispatchadapter.InvalidateChannel(ctx, req.ChannelID)
|
||
return nil, nil
|
||
}
|
||
|
||
// SetChannelAbilities 设置渠道模型能力
|
||
func (s *sAdmin) SetChannelAbilities(ctx context.Context, req *v1.ChannelAbilityBatchReq) (*v1.ChannelAbilityBatchRes, error) {
|
||
err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||
if _, err := dao.ChnAbilities.Ctx(ctx).Where("channel_id", req.ChannelID).Delete(); err != nil {
|
||
return err
|
||
}
|
||
for _, ab := range req.Abilities {
|
||
costRatio := ab.CostRatio
|
||
if costRatio <= 0 {
|
||
costRatio = 1.0
|
||
}
|
||
if _, err := dao.ChnAbilities.Ctx(ctx).Insert(do.ChnAbilities{
|
||
ChannelId: req.ChannelID,
|
||
ModelName: ab.ModelName,
|
||
UpstreamModel: ab.UpstreamModel,
|
||
Enabled: ab.Enabled,
|
||
CostRatio: costRatio,
|
||
SupportsResponses: ab.SupportsResponses,
|
||
ChatViaResponses: ab.ChatViaResponses,
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
dispatchadapter.InvalidateChannel(ctx, req.ChannelID)
|
||
|
||
return nil, nil
|
||
}
|
||
|
||
// GetChannelKeys 获取渠道 Key 列表
|
||
func (s *sAdmin) GetChannelKeys(ctx context.Context, req *v1.ChannelKeyListReq) (*v1.ChannelKeyListRes, error) {
|
||
var keys []struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Status string `json:"status"`
|
||
KeyType string `json:"key_type"`
|
||
TokenExpiresAt *gtime.Time `json:"token_expires_at"`
|
||
LastUsedAt *gtime.Time `json:"last_used_at"`
|
||
LastError string `json:"last_error"`
|
||
CreatedAt *gtime.Time `json:"created_at"`
|
||
}
|
||
|
||
err := dao.ChnChannelKeys.Ctx(ctx).
|
||
Where("channel_id", req.ChannelID).
|
||
Fields("id, name, status, key_type, token_expires_at, last_used_at, last_error, created_at").
|
||
OrderDesc("id").
|
||
Scan(&keys)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
list := make([]v1.ChannelKeyItem, len(keys))
|
||
for i, k := range keys {
|
||
item := v1.ChannelKeyItem{
|
||
ID: k.ID,
|
||
Name: k.Name,
|
||
Status: k.Status,
|
||
KeyType: k.KeyType,
|
||
CreatedAt: k.CreatedAt.String(),
|
||
CooldownRemaining: dispatchadapter.CredentialCooldownRemaining(ctx, k.ID),
|
||
}
|
||
if item.KeyType == "" {
|
||
item.KeyType = "apikey"
|
||
}
|
||
if k.TokenExpiresAt != nil {
|
||
item.TokenExpiresAt = k.TokenExpiresAt.String()
|
||
}
|
||
list[i] = item
|
||
}
|
||
|
||
return &v1.ChannelKeyListRes{List: list}, nil
|
||
}
|
||
|
||
// GetChannelAbilities 获取渠道模型能力列表
|
||
func (s *sAdmin) GetChannelAbilities(ctx context.Context, req *v1.ChannelAbilitiesGetReq) (*v1.ChannelAbilitiesGetRes, error) {
|
||
var abilities []struct {
|
||
ID int64 `json:"id" orm:"id"`
|
||
ModelName string `json:"model_name" orm:"model_name"`
|
||
UpstreamModel string `json:"upstream_model" orm:"upstream_model"`
|
||
Enabled bool `json:"enabled" orm:"enabled"`
|
||
CostRatio float64 `json:"cost_ratio" orm:"cost_ratio"`
|
||
SupportsResponses bool `json:"supports_responses" orm:"supports_responses"`
|
||
ChatViaResponses bool `json:"chat_via_responses" orm:"chat_via_responses"`
|
||
}
|
||
|
||
err := dao.ChnAbilities.Ctx(ctx).
|
||
Where("channel_id", req.ChannelID).
|
||
OrderAsc("model_name").
|
||
Scan(&abilities)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 转换为具名类型以传递给 batchGetModelHealth
|
||
abilityRecords := make([]abilityRecord, len(abilities))
|
||
for i, a := range abilities {
|
||
abilityRecords[i] = abilityRecord{
|
||
ID: a.ID,
|
||
ModelName: a.ModelName,
|
||
UpstreamModel: a.UpstreamModel,
|
||
Enabled: a.Enabled,
|
||
CostRatio: a.CostRatio,
|
||
SupportsResponses: a.SupportsResponses,
|
||
ChatViaResponses: a.ChatViaResponses,
|
||
}
|
||
}
|
||
|
||
// 批量读取模型级健康状态（从 Redis）
|
||
healthMap := s.batchGetModelHealth(ctx, req.ChannelID, abilityRecords)
|
||
|
||
list := make([]v1.AbilityItem, len(abilities))
|
||
for i, a := range abilities {
|
||
item := v1.AbilityItem{
|
||
ID: a.ID,
|
||
ModelName: a.ModelName,
|
||
UpstreamModel: a.UpstreamModel,
|
||
Enabled: a.Enabled,
|
||
CostRatio: a.CostRatio,
|
||
SupportsResponses: a.SupportsResponses,
|
||
ChatViaResponses: a.ChatViaResponses,
|
||
}
|
||
|
||
// 填充健康数据
|
||
if h, ok := healthMap[a.ModelName]; ok {
|
||
item.HealthScore = h.HealthScore
|
||
item.SuccEWMA = h.SuccEWMA
|
||
item.LatencyEWMA = h.LatencyEWMA
|
||
item.BreakerState = h.BreakerState
|
||
}
|
||
|
||
list[i] = item
|
||
}
|
||
|
||
return &v1.ChannelAbilitiesGetRes{List: list}, nil
|
||
}
|
||
|
||
// ModelHealthData 模型健康数据
|
||
type ModelHealthData struct {
|
||
HealthScore *float64
|
||
SuccEWMA *float64
|
||
LatencyEWMA *float64
|
||
BreakerState *int
|
||
}
|
||
|
||
// abilityRecord 能力记录（用于类型匹配）
|
||
type abilityRecord struct {
|
||
ID int64
|
||
ModelName string
|
||
UpstreamModel string
|
||
Enabled bool
|
||
CostRatio float64
|
||
SupportsResponses bool
|
||
ChatViaResponses bool
|
||
}
|
||
|
||
// batchGetModelHealth 批量读取模型级健康状态（从 Redis）
|
||
// Redis key 模式：
|
||
// - dispatch:v1:health:<channelID>:<model> (HASH) → succ_ewma, lat_ewma
|
||
// - dispatch:v1:breaker:<channelID>:<model> (HASH) → state, half_open_at, first_opened_ms
|
||
func (s *sAdmin) batchGetModelHealth(ctx context.Context, channelID int64, abilities []abilityRecord) map[string]*ModelHealthData {
|
||
healthMap := make(map[string]*ModelHealthData)
|
||
if len(abilities) == 0 {
|
||
return healthMap
|
||
}
|
||
|
||
redis := g.Redis()
|
||
// 健康分折算口径必须与调度 healthFactor / 维护快照落盘一致（succ^α×100），
|
||
// α 取当前生效的路由策略而非硬编码——策略支持热更新。
|
||
alpha := dispatchadapter.HealthAlpha()
|
||
for _, ab := range abilities {
|
||
healthKey := fmt.Sprintf("dispatch:v1:health:%d:%s", channelID, ab.ModelName)
|
||
breakerKey := fmt.Sprintf("dispatch:v1:breaker:%d:%s", channelID, ab.ModelName)
|
||
|
||
data := &ModelHealthData{}
|
||
|
||
// 读取健康分数（HGETALL health key）
|
||
healthHash, err := redis.HGetAll(ctx, healthKey)
|
||
if err == nil && !healthHash.IsEmpty() {
|
||
hashMap := healthHash.Map()
|
||
|
||
// succ_ewma: 成功率 EWMA (0-1)
|
||
if succEWMAVal, ok := hashMap["succ_ewma"]; ok && succEWMAVal != nil {
|
||
if succStr, ok := succEWMAVal.(string); ok {
|
||
var succEWMA float64
|
||
if _, err := fmt.Sscanf(succStr, "%f", &succEWMA); err == nil && succEWMA >= 0 {
|
||
data.SuccEWMA = &succEWMA
|
||
// 健康分 = succ_ewma^α × 100（与调度 healthFactor 同源）
|
||
healthScore := math.Pow(succEWMA, alpha) * 100
|
||
data.HealthScore = &healthScore
|
||
}
|
||
}
|
||
}
|
||
|
||
// lat_ewma: 延迟 EWMA (ms)
|
||
if latEWMAVal, ok := hashMap["lat_ewma"]; ok && latEWMAVal != nil {
|
||
if latStr, ok := latEWMAVal.(string); ok {
|
||
var latEWMA float64
|
||
if _, err := fmt.Sscanf(latStr, "%f", &latEWMA); err == nil && latEWMA >= 0 {
|
||
data.LatencyEWMA = &latEWMA
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 读取熔断状态（HGET breaker key state）
|
||
stateVal, err := redis.HGet(ctx, breakerKey, "state")
|
||
if err == nil && !stateVal.IsNil() {
|
||
if stateStr := stateVal.String(); stateStr != "" {
|
||
var state int
|
||
if _, err := fmt.Sscanf(stateStr, "%d", &state); err == nil {
|
||
// 0=CLOSED(正常), 1=OPEN(熔断), 2=HALF_OPEN(半开)
|
||
data.BreakerState = &state
|
||
}
|
||
}
|
||
}
|
||
|
||
// 只有至少有一个健康指标时才加入 map
|
||
if data.HealthScore != nil || data.SuccEWMA != nil || data.LatencyEWMA != nil || data.BreakerState != nil {
|
||
healthMap[ab.ModelName] = data
|
||
}
|
||
}
|
||
|
||
return healthMap
|
||
}
|
||
|
||
// GetProviderDefaultURLs 获取供应商默认 API 地址
|
||
func (s *sAdmin) GetProviderDefaultURLs(ctx context.Context, _ *v1.ProviderDefaultURLReq) (*v1.ProviderDefaultURLRes, error) {
|
||
return &v1.ProviderDefaultURLRes{URLs: defaultProviderURLs}, nil
|
||
}
|
||
|
||
// defaultProviderURLs 供应商默认 API 地址
|
||
var defaultProviderURLs = map[int]string{
|
||
1: "https://api.openai.com",
|
||
2: "https://api.anthropic.com",
|
||
3: "https://generativelanguage.googleapis.com",
|
||
// 阿里云百炼填裸域名：adaptor 会按 relay 模式自行拼 /compatible-mode/v1/...（OpenAI 兼容）、
|
||
// /apps/anthropic/v1/messages（Claude）、/api/v1/services/aigc/...（图像），
|
||
// 这里若带上 /compatible-mode 会导致路径重复而 404。
|
||
4: "https://dashscope.aliyuncs.com",
|
||
5: "https://aip.baidubce.com",
|
||
6: "https://hunyuan.tencentcloudapi.com",
|
||
7: "https://open.bigmodel.cn",
|
||
8: "https://api.deepseek.com",
|
||
9: "https://api.moonshot.cn",
|
||
10: "https://ark.cn-beijing.volces.com/api",
|
||
11: "https://bedrock-runtime.us-east-1.amazonaws.com",
|
||
12: "", // Azure 需要自行填写，格式 https://{resource}.openai.azure.com
|
||
13: "https://us-central1-aiplatform.googleapis.com",
|
||
14: "https://api.cohere.com",
|
||
15: "https://api.mistral.ai",
|
||
16: "https://api.x.ai",
|
||
// 零一万物 OpenAI 兼容端点，adaptor 自拼 /v1/chat/completions
|
||
18: "https://api.lingyiwanwu.com",
|
||
// 百度千帆 v2：qianfan.baidubce.com/v2/chat/completions
|
||
19: "https://qianfan.baidubce.com",
|
||
// Cloudflare 填裸域名即可：adaptor 自拼 /client/v4/accounts/{id}/ai/v1，
|
||
// account_id 从 ApiKey（token|accountid 格式）解析，无需写在地址里
|
||
20: "https://api.cloudflare.com",
|
||
// Ollama 本地服务，adaptor 自拼 /api/chat
|
||
22: "http://localhost:11434",
|
||
// 硅基流动（国内站；国际站为 api.siliconflow.com）
|
||
25: "https://api.siliconflow.cn",
|
||
// 讯飞 OpenAI 兼容端点是 spark-api-open（注意 -open 后缀），
|
||
// 旧版 spark-api.xf-yun.com 是 WebSocket 协议主机，走这里会 404
|
||
26: "https://spark-api-open.xf-yun.com",
|
||
// OpenRouter：openrouter.ai/api/v1/chat/completions
|
||
27: "https://openrouter.ai/api",
|
||
// XInference 默认本地端口
|
||
28: "http://localhost:9997",
|
||
// MiniMax 国内站（海外站为 api.minimaxi.chat），adaptor 自拼 /v1/text/chatcompletion_v2
|
||
29: "https://api.minimax.chat",
|
||
// 32 = Coze、33 = Dify：渠道已移除，编号永久保留不复用
|
||
// 即梦：适配器拼 /v2/images/generations，无官方对齐端点可验证，需自行填写
|
||
34: "",
|
||
// Codex 直连 chatgpt.com/backend-api/codex/responses（对齐 Codex CLI）
|
||
35: "https://chatgpt.com",
|
||
41: "", // New API 自引用渠道，需自行填写实例地址
|
||
42: "", // Sub2API 渠道，需自行填写实例地址
|
||
}
|
||
|
||
// defaultProviderURL 返回供应商类型的默认 API 地址
|
||
func defaultProviderURL(t int) string {
|
||
return defaultProviderURLs[t]
|
||
}
|
||
|
||
// GetChannelHealthTrend 获取渠道健康趋势数据（健康度、延迟在 SQL 层四舍五入取整，避免展示小数）
|
||
// 不再返回 stability_score / consecutive_failures：健康体系重写后这两列已停止写入，
|
||
// 返回恒定值只会让前端画出两条毫无意义的常量线。
|
||
func (s *sAdmin) GetChannelHealthTrend(ctx context.Context, req *v1.ChannelHealthTrendReq) (*v1.ChannelHealthTrendRes, error) {
|
||
var points []v1.HealthTrendPoint
|
||
err := dao.ChnHealthSnapshots.Ctx(ctx).
|
||
Fields("snapshot_at, ROUND(health_score)::int AS health_score, success_rate, ROUND(latency_ms)::int AS latency_ms").
|
||
Where("channel_id", req.ID).
|
||
Where("snapshot_at >= ?", gtime.Now().Add(-time.Duration(req.Hours)*time.Hour)).
|
||
OrderAsc("snapshot_at").
|
||
Scan(&points)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if points == nil {
|
||
points = []v1.HealthTrendPoint{}
|
||
}
|
||
|
||
return &v1.ChannelHealthTrendRes{Points: points}, nil
|
||
}
|
||
|
||
// ResetChannelHealth 重置渠道健康度（渠道可能已修复、重新可用）：落库健康分 80（展示层），
|
||
// 并复位熔断 + 成功率 EWMA（调度层），使渠道立即恢复被调度选择的能力。
|
||
func (s *sAdmin) ResetChannelHealth(ctx context.Context, req *v1.ChannelResetHealthReq) (*v1.ChannelResetHealthRes, error) {
|
||
// 渠道必须存在
|
||
count, err := dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Count()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if count == 0 {
|
||
return nil, common.NewBusinessError(404, "渠道不存在")
|
||
}
|
||
|
||
// 如果指定了模型名，只重置该模型的健康度
|
||
if req.ModelName != "" {
|
||
// 验证模型能力是否存在
|
||
abilityCount, err := dao.ChnAbilities.Ctx(ctx).
|
||
Where("channel_id", req.ID).
|
||
Where("model_name", req.ModelName).
|
||
Count()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if abilityCount == 0 {
|
||
return nil, common.NewBusinessError(404, fmt.Sprintf("模型 %s 不存在于该渠道", req.ModelName))
|
||
}
|
||
|
||
// 调度层：只复位指定模型的熔断 + 成功率恢复
|
||
dispatchadapter.ResetModelHealth(ctx, req.ID, req.ModelName)
|
||
// 单模型重置不直接写 chn_health_scores（渠道分是全模型聚合值，不能按单模型拍值），
|
||
// 改为请求重算——否则展示分永远停在旧值，管理员会以为重置没生效
|
||
dispatchadapter.RequestHealthSnapshot(req.ID)
|
||
|
||
return nil, nil
|
||
}
|
||
|
||
// 未指定模型名，重置整个渠道（含所有模型）
|
||
|
||
// 展示层：健康分落库为 80（调度不读此表，仅供仪表盘/审计展示；随后的按需重算/
|
||
// 维护快照会按 Redis 实际读值重算，此处只是让界面立即有反馈）
|
||
// stability_score / consecutive_failures 已废弃，不再写入
|
||
affected, err := dao.ChnHealthScores.Ctx(ctx).
|
||
Where("channel_id", req.ID).
|
||
Data(do.ChnHealthScores{
|
||
SuccessRate: 90.00,
|
||
LatencyMs: 0,
|
||
HealthScore: 80.00,
|
||
CalculatedAt: gtime.Now(),
|
||
}).
|
||
UpdateAndGetAffected()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 健康分记录缺失时兜底插入（正常情况下创建渠道时已由 InitHealthScore 初始化）
|
||
if affected == 0 {
|
||
_, err = dao.ChnHealthScores.Ctx(ctx).Insert(do.ChnHealthScores{
|
||
ChannelId: req.ID,
|
||
SuccessRate: 90.00,
|
||
LatencyMs: 0,
|
||
HealthScore: 80.00,
|
||
CalculatedAt: gtime.Now(),
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
// 调度层：复位渠道级/模型级熔断 + 成功率恢复，渠道立即恢复被选择能力
|
||
dispatchadapter.ResetChannelHealth(ctx, req.ID)
|
||
// 凭证冷却同属"渠道不可用"状态，且只靠 TTL 自愈、无其它人工出口：
|
||
// 管理员点重置的语义就是"这个渠道我修好了"，只复位熔断不解除冷却等于重置了一半
|
||
// ——渠道会继续以「凭证全部冷却」被整体跳过，界面上却显示健康已恢复。
|
||
if n := dispatchadapter.ClearChannelCredentialCooldown(ctx, req.ID); n > 0 {
|
||
g.Log().Infof(ctx, "[Channel] 渠道 %d 重置健康度，已解除 %d 个 Key 的凭证冷却", req.ID, n)
|
||
}
|
||
// 请求按 Redis 实际读值重算落盘：上面写死的 80 只是调度引擎未组装时的兜底，
|
||
// 重算后会收敛到与调度同源的 0.9^α×100
|
||
dispatchadapter.RequestHealthSnapshot(req.ID)
|
||
|
||
return nil, nil
|
||
}
|
||
|
||
// ExportChannels exports channel list to CSV or Excel.
|
||
func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) (*v1.ChannelExportRes, error) {
|
||
channelFields := "chn_channels.id, chn_channels.name, chn_channels.type, chn_channels.status, chn_channels.priority, chn_channels.weight, chn_channels.created_at, h.health_score"
|
||
|
||
config := export.Config{
|
||
Format: req.Format,
|
||
Filename: "渠道_" + gtime.Now().Format("Ymd_His"),
|
||
Columns: []export.Column{
|
||
{Field: "id", Header: "ID"},
|
||
{Field: "name", Header: "名称"},
|
||
{Field: "type_name", Header: "供应商类型"},
|
||
{Field: "status", Header: "状态"},
|
||
{Field: "priority", Header: "优先级"},
|
||
{Field: "weight", Header: "权重"},
|
||
{Field: "health_score", Header: "健康分数"},
|
||
{Field: "created_at", Header: "创建时间"},
|
||
},
|
||
}
|
||
|
||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||
// keyset 翻页：导出按优先级倒序展示，游标取 (priority, id) 复合键 ——
|
||
// priority 大量重复，必须带 id 作 tiebreaker，否则同优先级的行会被跳过或重复。
|
||
// PostgreSQL 行值比较 (a, b) < (?, ?) 按字典序逐列比较，与排序方向一致。
|
||
var (
|
||
hasCursor bool
|
||
cursorPriority int
|
||
cursorID int64
|
||
)
|
||
for {
|
||
query := dao.ChnChannels.Ctx(ctx).
|
||
LeftJoin("chn_health_scores h ON chn_channels.id = h.channel_id")
|
||
if req.Type > 0 {
|
||
query = query.Where("chn_channels.type", req.Type)
|
||
}
|
||
if req.Status != "" {
|
||
query = query.Where("chn_channels.status", req.Status)
|
||
}
|
||
if req.Search != "" {
|
||
query = query.Where("chn_channels.name LIKE ? OR chn_channels.remark LIKE ?", "%"+req.Search+"%", "%"+req.Search+"%")
|
||
}
|
||
if req.ID > 0 {
|
||
query = query.Where("chn_channels.id", req.ID)
|
||
}
|
||
if req.BaseURL != "" {
|
||
query = query.Where("chn_channels.base_url LIKE ?", "%"+req.BaseURL+"%")
|
||
}
|
||
if req.Model != "" {
|
||
// 与 ListChannels 保持一致：EXISTS 相关子查询避免 JOIN 重复行
|
||
query = query.Where("EXISTS (SELECT 1 FROM chn_abilities WHERE chn_abilities.channel_id = chn_channels.id AND (chn_abilities.model_name LIKE ? OR chn_abilities.upstream_model LIKE ?))", "%"+req.Model+"%", "%"+req.Model+"%")
|
||
}
|
||
if hasCursor {
|
||
query = query.Where("(chn_channels.priority, chn_channels.id) < (?, ?)", cursorPriority, cursorID)
|
||
}
|
||
var batch []struct {
|
||
ID int64 `json:"id"`
|
||
Name string `json:"name"`
|
||
Type int `json:"type"`
|
||
Status string `json:"status"`
|
||
Priority int `json:"priority"`
|
||
Weight int `json:"weight"`
|
||
CreatedAt *gtime.Time `json:"created_at"`
|
||
HealthScore *float64 `json:"health_score"`
|
||
}
|
||
if err := query.Fields(channelFields).OrderDesc("chn_channels.priority").OrderDesc("chn_channels.id").Limit(1000).Scan(&batch); err != nil {
|
||
g.Log().Errorf(ctx, "ExportChannels: query batch failed (cursor priority=%d id=%d): %v", cursorPriority, cursorID, err)
|
||
return
|
||
}
|
||
for _, ch := range batch {
|
||
typeName := providerTypeNames[ch.Type]
|
||
if typeName == "" {
|
||
typeName = fmt.Sprintf("Unknown(%d)", ch.Type)
|
||
}
|
||
healthScore := ""
|
||
if ch.HealthScore != nil {
|
||
healthScore = fmt.Sprintf("%.2f", *ch.HealthScore)
|
||
}
|
||
if !yield(map[string]any{
|
||
"id": ch.ID,
|
||
"name": ch.Name,
|
||
"type_name": typeName,
|
||
"status": ch.Status,
|
||
"priority": ch.Priority,
|
||
"weight": ch.Weight,
|
||
"health_score": healthScore,
|
||
"created_at": ch.CreatedAt.String(),
|
||
}) {
|
||
return
|
||
}
|
||
cursorPriority, cursorID = ch.Priority, ch.ID
|
||
}
|
||
if len(batch) < 1000 {
|
||
break
|
||
}
|
||
hasCursor = true
|
||
}
|
||
})
|
||
}
|