feat(channel): 对齐 vivid 最新渠道/模型管理设计
渠道管理(对齐 9db9971/9f6eeca/b607a8d/268a8f7/7f8a87e): - 列表新增「运行中」(Redis 并发槽位实时计数,与调度同源)和「失败数」 (event_logs 按渠道聚合)两列 - 行内「同步上游模型」恢复弹窗:默认全开、右侧开关、彩点+单行别名, 底部「同步模型」批量导入;编辑表单去掉获取模型列表按钮(保存已自动 全量同步);去掉勾选即解绑语义与 unbind-model 端点 - 导入加固:别名与上游同名报错、同批别名复用拒绝、本地类型冲突拒绝、 单次上限 50 模型管理(对齐 5545195): - 模型表单/CreateModel 补齐 max_reference_videos/audios 与 max_ref_video/audio_seconds 透传 产物(对齐 2253e58): - 下载结果仅信任 image/*/video/* 响应头,text/plain 等乱报一律按魔数 嗅探定 MIME Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
@@ -118,7 +118,7 @@ func NewApp(ctx context.Context) (*App, error) {
|
||||
// rows when CHANNEL_SECRET isn't set.
|
||||
channelSecret = cfg.PostgresDSN
|
||||
}
|
||||
channelSvc := service.NewChannelService(channelRepo, modelRepo, upClient, channelSecret)
|
||||
channelSvc := service.NewChannelService(channelRepo, modelRepo, eventRepo, concSvc, upClient, channelSecret)
|
||||
// Sweep provider-era model rows: anything not bound to a channel can never
|
||||
// be routed under the channel scheduler.
|
||||
if n, err := channelSvc.PurgeOrphanModels(ctx); err != nil {
|
||||
|
||||
@@ -125,24 +125,6 @@ func (h *ChannelAdminHandler) ImportModels(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
// UnbindModel detaches one upstream model from a channel; the local model row
|
||||
// is deleted too when no other channel still binds it.
|
||||
func (h *ChannelAdminHandler) UnbindModel(c *gin.Context) {
|
||||
var body struct {
|
||||
ID string `json:"id"`
|
||||
Model string `json:"model"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil || strings.TrimSpace(body.ID) == "" || strings.TrimSpace(body.Model) == "" {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"detail": "id and model required"})
|
||||
return
|
||||
}
|
||||
if err := h.channels.UnbindModel(c.Request.Context(), body.ID, body.Model); err != nil {
|
||||
writeUpstreamErr(c, err)
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
// writeUpstreamErr translates upstream-protocol errors into sensible admin
|
||||
// responses (auth → 400 with the upstream's message, temporary → 502).
|
||||
func writeUpstreamErr(c *gin.Context, err error) {
|
||||
|
||||
@@ -148,7 +148,6 @@ func New(cfg *config.Config, auth *service.AuthService, handlers Handlers) *gin.
|
||||
authed.POST("/channels/fetch-models", handlers.ChannelAdmin.FetchModels)
|
||||
authed.POST("/channels/sync-models", handlers.ChannelAdmin.SyncModels)
|
||||
authed.POST("/channels/import-models", handlers.ChannelAdmin.ImportModels)
|
||||
authed.POST("/channels/unbind-model", handlers.ChannelAdmin.UnbindModel)
|
||||
authed.GET("/providers", handlers.AdminRead.Providers)
|
||||
authed.GET("/images", handlers.AdminRead.Images)
|
||||
authed.DELETE("/images", handlers.AdminRead.DeleteImage)
|
||||
|
||||
@@ -668,6 +668,29 @@ func (r *EventRepository) SetAccount(ctx context.Context, eventID, accountID, ac
|
||||
Updates(map[string]any{"account_id": accountID, "account_email": accountEmail}).Error
|
||||
}
|
||||
|
||||
// FailedByAccount counts failed events grouped by account_id (channel id) —
|
||||
// the channel list's 失败数 column.
|
||||
func (r *EventRepository) FailedByAccount(ctx context.Context) (map[string]int64, error) {
|
||||
type row struct {
|
||||
AccountID string `gorm:"column:account_id"`
|
||||
Count int64 `gorm:"column:count"`
|
||||
}
|
||||
var rows []row
|
||||
if err := r.db.WithContext(ctx).
|
||||
Model(&model.EventLog{}).
|
||||
Select("account_id, COUNT(*) AS count").
|
||||
Where("status = ? AND account_id <> ''", "failed").
|
||||
Group("account_id").
|
||||
Scan(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make(map[string]int64, len(rows))
|
||||
for _, item := range rows {
|
||||
out[item.AccountID] = item.Count
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// InFlightByAccount counts pending (in-flight) events grouped by account_id, for
|
||||
// the accounts view's live "in-flight" column.
|
||||
func (r *EventRepository) InFlightByAccount(ctx context.Context) (map[string]int64, error) {
|
||||
|
||||
@@ -402,6 +402,8 @@ func (s *AdminWriteService) CreateModel(ctx context.Context, body map[string]any
|
||||
MaxReferenceImages: intValue(body["max_reference_images"]),
|
||||
MaxReferenceVideos: intValue(body["max_reference_videos"]),
|
||||
MaxReferenceAudios: intValue(body["max_reference_audios"]),
|
||||
MaxRefVideoSeconds: intValue(body["max_ref_video_seconds"]),
|
||||
MaxRefAudioSeconds: intValue(body["max_ref_audio_seconds"]),
|
||||
ReferenceMode: defaultString(strings.TrimSpace(stringValue(body["reference_mode"])), "none"),
|
||||
Weight: intValue(body["weight"]),
|
||||
CreatedAt: time.Now(),
|
||||
|
||||
@@ -30,13 +30,15 @@ import (
|
||||
type ChannelService struct {
|
||||
channels *repo.ChannelRepository
|
||||
models *repo.ModelRepository
|
||||
events *repo.EventRepository
|
||||
conc *ConcurrencyService
|
||||
up *upstream.Client
|
||||
secret []byte
|
||||
}
|
||||
|
||||
func NewChannelService(channels *repo.ChannelRepository, models *repo.ModelRepository, up *upstream.Client, secret string) *ChannelService {
|
||||
func NewChannelService(channels *repo.ChannelRepository, models *repo.ModelRepository, events *repo.EventRepository, conc *ConcurrencyService, up *upstream.Client, secret string) *ChannelService {
|
||||
sum := sha256.Sum256([]byte("image2api-channel-keys|" + secret))
|
||||
return &ChannelService{channels: channels, models: models, up: up, secret: sum[:]}
|
||||
return &ChannelService{channels: channels, models: models, events: events, conc: conc, up: up, secret: sum[:]}
|
||||
}
|
||||
|
||||
// ChannelView is the admin list row: local stats + bound model names + a live
|
||||
@@ -50,6 +52,10 @@ type ChannelView struct {
|
||||
Concurrency int `json:"concurrency"`
|
||||
Weight int `json:"weight"`
|
||||
Models []string `json:"models"`
|
||||
// RunningCount 运行中 = Redis 并发槽位实时计数(与调度器同源);
|
||||
// FailCount 失败数 = event_logs 实时聚合(channel 数少,无需预计算)。
|
||||
RunningCount int64 `json:"running_count"`
|
||||
FailCount int64 `json:"fail_count"`
|
||||
SuccessTotal int64 `json:"success_total"`
|
||||
UsedCredits float64 `json:"used_credits"`
|
||||
UpstreamBalance *int64 `json:"upstream_balance"`
|
||||
@@ -63,6 +69,7 @@ func (s *ChannelService) List(ctx context.Context) ([]ChannelView, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fails, _ := s.events.FailedByAccount(ctx)
|
||||
out := make([]ChannelView, len(items))
|
||||
var wg sync.WaitGroup
|
||||
for i, ch := range items {
|
||||
@@ -73,6 +80,8 @@ func (s *ChannelService) List(ctx context.Context) ([]ChannelView, error) {
|
||||
Enabled: ch.Enabled,
|
||||
Concurrency: ch.Concurrency,
|
||||
Weight: ch.Weight,
|
||||
RunningCount: int64(s.conc.Count(ctx, "conc:c:"+ch.ID)),
|
||||
FailCount: fails[ch.ID],
|
||||
SuccessTotal: ch.SuccessTotal,
|
||||
UsedCredits: ch.UsedCredits,
|
||||
CreatedAt: ch.CreatedAt.Unix(),
|
||||
@@ -354,37 +363,6 @@ func (s *ChannelService) FetchModels(ctx context.Context, d FetchModelsDto) ([]u
|
||||
return items, nil
|
||||
}
|
||||
|
||||
// UnbindModel detaches one upstream model from a channel; when the local model
|
||||
// row ends up bound to no channel at all it is deleted too (same orphan
|
||||
// semantics as deleting a whole channel).
|
||||
func (s *ChannelService) UnbindModel(ctx context.Context, channelID, upstreamModelID string) error {
|
||||
upID := strings.TrimSpace(upstreamModelID)
|
||||
if upID == "" {
|
||||
return errors.New("model id required")
|
||||
}
|
||||
m, err := s.channels.UpstreamModelMap(ctx, channelID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var localID string
|
||||
for mid, up := range m {
|
||||
if up == upID {
|
||||
localID = mid
|
||||
break
|
||||
}
|
||||
}
|
||||
if localID == "" {
|
||||
return nil // not bound — idempotent
|
||||
}
|
||||
if err := s.channels.Unbind(ctx, channelID, localID); err != nil {
|
||||
return err
|
||||
}
|
||||
if rest, _ := s.channels.ChannelIDsByModel(ctx, localID); len(rest) == 0 {
|
||||
_, _ = s.models.Delete(ctx, localID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PurgeOrphanModels sweeps model_configs rows left over from the provider era:
|
||||
// an unbound model can never be routed, so any row that didn't come through a
|
||||
// channel import (provider != "channel") is deleted at startup. Channel-typed
|
||||
@@ -445,6 +423,10 @@ func (s *ChannelService) ImportModels(ctx context.Context, channelID string, spe
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(specs) > 50 {
|
||||
return errors.New("单次最多导入 50 个模型")
|
||||
}
|
||||
seenAliases := make(map[string]string, len(specs)) // localID → 上游 id,别名冲突当场拒绝
|
||||
for _, spec := range specs {
|
||||
upID := strings.TrimSpace(spec.ID)
|
||||
if upID == "" {
|
||||
@@ -458,10 +440,17 @@ func (s *ChannelService) ImportModels(ctx context.Context, channelID string, spe
|
||||
if localID == "" {
|
||||
localID = upID
|
||||
}
|
||||
if prev, dup := seenAliases[localID]; dup && prev != upID {
|
||||
return fmt.Errorf("别名 %s 被多个上游模型复用", localID)
|
||||
}
|
||||
seenAliases[localID] = upID
|
||||
existing, gerr := s.models.Get(ctx, localID)
|
||||
if gerr != nil && !errors.Is(gerr, gorm.ErrRecordNotFound) {
|
||||
existing = nil
|
||||
}
|
||||
if existing != nil && existing.Type != "" && spec.Kind != "" && existing.Type != spec.Kind {
|
||||
return fmt.Errorf("模型 %s 已存在且类型不同(%s)", localID, existing.Type)
|
||||
}
|
||||
var local *model.ModelConfig
|
||||
if existing != nil && existing.ID == localID {
|
||||
local = existing
|
||||
|
||||
@@ -422,8 +422,13 @@ func (c *Client) Download(ctx context.Context, url string) ([]byte, string, erro
|
||||
if len(data) == 0 {
|
||||
return nil, "", fmt.Errorf("%w: empty result", ErrRetryable)
|
||||
}
|
||||
// Content-Type fallback chain: only trust image/* video/* headers (upstreams
|
||||
// often mislabel results text/plain or octet-stream), else sniff magic bytes.
|
||||
ct := strings.TrimSpace(resp.Header.Get("Content-Type"))
|
||||
if ct == "" || ct == "application/octet-stream" {
|
||||
if i := strings.IndexByte(ct, ';'); i >= 0 {
|
||||
ct = strings.TrimSpace(ct[:i])
|
||||
}
|
||||
if !strings.HasPrefix(ct, "image/") && !strings.HasPrefix(ct, "video/") {
|
||||
ct = http.DetectContentType(data)
|
||||
}
|
||||
return data, ct, nil
|
||||
|
||||
Reference in New Issue
Block a user