diff --git a/backend/internal/http/handler/channel_admin.go b/backend/internal/http/handler/channel_admin.go index deeafc8..be459d5 100644 --- a/backend/internal/http/handler/channel_admin.go +++ b/backend/internal/http/handler/channel_admin.go @@ -88,6 +88,25 @@ func (h *ChannelAdminHandler) FetchModels(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"data": items}) } +// SyncModels manually re-runs the full upstream model sync on a saved channel +// (new models bound, disappeared ones unbound, orphans deleted). Fetch +// failures are errors here — unlike the save path which degrades to a warning. +func (h *ChannelAdminHandler) SyncModels(c *gin.Context) { + var body struct { + ID string `json:"id"` + } + if err := c.ShouldBindJSON(&body); err != nil || strings.TrimSpace(body.ID) == "" { + c.JSON(http.StatusBadRequest, gin.H{"detail": "channel id required"}) + return + } + msg, err := h.channels.SyncModels(c.Request.Context(), body.ID) + if err != nil { + writeUpstreamErr(c, err) + return + } + c.JSON(http.StatusOK, gin.H{"ok": true, "message": msg}) +} + // ImportModels binds the ticked upstream models to the channel (creating local // model rows on first import). func (h *ChannelAdminHandler) ImportModels(c *gin.Context) { diff --git a/backend/internal/http/router/router.go b/backend/internal/http/router/router.go index c9d4dfd..1efcd71 100644 --- a/backend/internal/http/router/router.go +++ b/backend/internal/http/router/router.go @@ -146,6 +146,7 @@ func New(cfg *config.Config, auth *service.AuthService, handlers Handlers) *gin. authed.POST("/channels", handlers.ChannelAdmin.Save) authed.POST("/channels/delete", handlers.ChannelAdmin.Delete) 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) diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 1522f72..b5e0bc2 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -194,7 +194,7 @@ type Channel struct { BaseURL string `gorm:"size:500;not null"` APIKeyEnc string `gorm:"type:text;not null"` // AES-GCM encrypted upstream API key Enabled bool `gorm:"not null;default:true;index"` - // Concurrency caps simultaneous in-flight jobs on this channel. 0 = default. + // Concurrency caps simultaneous in-flight jobs on this channel. 0 = unlimited. Concurrency int `gorm:"not null;default:0"` // Weight biases weighted-random channel order: weight 2 is picked first at // roughly 2x the rate of weight 1. diff --git a/backend/internal/service/channel.go b/backend/internal/service/channel.go index dbb229d..496d507 100644 --- a/backend/internal/service/channel.go +++ b/backend/internal/service/channel.go @@ -11,6 +11,7 @@ import ( "errors" "fmt" "io" + "math" "strings" "sync" "time" @@ -132,6 +133,9 @@ func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Cha if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") { return nil, "", errors.New("base_url must start with http:// or https://") } + if d.Concurrency < 0 || d.Concurrency > math.MaxInt32 { + return nil, "", errors.New("并发数需在 0-2147483647 之间（0 = 不限制）") + } weight := d.Weight if weight < 1 { weight = 1 @@ -159,7 +163,7 @@ func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Cha if err := s.channels.Create(ctx, ch); err != nil { return nil, "", err } - return ch, s.syncChannelModels(ctx, ch.ID), nil + return ch, syncWarn(s.syncChannelModels(ctx, ch.ID)), nil } patch := map[string]any{ "name": name, @@ -179,31 +183,58 @@ func (s *ChannelService) Save(ctx context.Context, d SaveChannelDto) (*model.Cha if err != nil { return nil, "", err } - return ch, s.syncChannelModels(ctx, d.ID), nil + return ch, syncWarn(s.syncChannelModels(ctx, d.ID)), nil +} + +// syncWarn folds the sync error into the save-path warning (vivid parity: a +// failed upstream fetch degrades to a non-fatal warning instead of blocking +// the channel save). +func syncWarn(warn string, err error) string { + if err != nil { + return err.Error() + } + return warn +} + +// SyncModels manually triggers a full upstream-model sync for a saved channel +// (same truth semantics as the save path). Unlike Save, an unreachable upstream +// or empty model list is a hard error, not a warning. +func (s *ChannelService) SyncModels(ctx context.Context, channelID string) (string, error) { + if _, err := s.channels.Get(ctx, strings.TrimSpace(channelID)); err != nil { + return "", err + } + warn, err := s.syncChannelModels(ctx, channelID) + if err != nil { + return "", err + } + if warn != "" { + return warn, nil + } + return "已同步上游模型", nil } // syncChannelModels runs after Save: upstream /v1/models is the truth — every // sellable upstream model is upserted locally (alias = upstream id) and bound; // bindings on this channel not in the upstream list are removed, and orphan -// local models are deleted. An unreachable upstream or empty/anomalous list -// only produces a warning — the DB is left untouched. -func (s *ChannelService) syncChannelModels(ctx context.Context, channelID string) string { +// local models are deleted. An unreachable upstream or empty list is an error +// (the save path degrades it to a warning); warn carries non-fatal skips. +func (s *ChannelService) syncChannelModels(ctx context.Context, channelID string) (string, error) { ch, err := s.channels.Get(ctx, channelID) if err != nil { - return "模型自动同步失败：" + err.Error() + return "", fmt.Errorf("模型同步失败：%w", err) } key := s.decrypt(ch.APIKeyEnc) if key == "" { - return "模型自动同步失败：无法解密上游密钥" + return "", errors.New("模型同步失败：无法解密上游密钥") } fctx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() models, err := s.up.Models(fctx, ch.BaseURL, key) if err != nil { - return "模型自动同步失败：" + err.Error() + return "", fmt.Errorf("模型同步失败：%w", err) } if len(models) == 0 { - return "上游未返回任何模型，已跳过同步" + return "", errors.New("上游未返回任何模型，已跳过同步") } specs := make([]ChannelModelSpec, 0, len(models)) upstreamIDs := make([]string, 0, len(models)) @@ -218,13 +249,13 @@ func (s *ChannelService) syncChannelModels(ctx context.Context, channelID string upstreamIDs = append(upstreamIDs, spec.ID) } if len(upstreamIDs) == 0 { - return fmt.Sprintf("上游模型均规格异常，已跳过同步：%s", strings.Join(skipped, "；")) + return "", fmt.Errorf("上游模型均规格异常，已跳过同步：%s", strings.Join(skipped, "；")) } if err := s.ImportModels(ctx, channelID, specs); err != nil { - return "模型自动同步失败：" + err.Error() + return "", fmt.Errorf("模型同步失败：%w", err) } if err := s.channels.UnbindExcept(ctx, channelID, upstreamIDs); err != nil { - return "模型自动同步失败：" + err.Error() + return "", fmt.Errorf("模型同步失败：%w", err) } if orphans, oerr := s.channels.OrphanModelIDs(ctx); oerr == nil { for _, mid := range orphans { @@ -232,9 +263,9 @@ func (s *ChannelService) syncChannelModels(ctx context.Context, channelID string } } if len(skipped) > 0 { - return fmt.Sprintf("已同步上游模型，跳过 %d 个异常项：%s", len(skipped), strings.Join(skipped, "；")) + return fmt.Sprintf("已同步上游模型，跳过 %d 个异常项：%s", len(skipped), strings.Join(skipped, "；")), nil } - return "" + return "", nil } // normalizeRemoteModel validates an upstream /v1/models item into an import @@ -419,7 +450,11 @@ func (s *ChannelService) ImportModels(ctx context.Context, channelID string, spe if upID == "" { continue } - localID := strings.TrimSpace(spec.Alias) + alias := strings.TrimSpace(spec.Alias) + if alias == upID { + return fmt.Errorf("「%s」别名不能与上游模型同名，留空即可", upID) + } + localID := alias if localID == "" { localID = upID } @@ -459,8 +494,8 @@ func (s *ChannelService) ImportModels(ctx context.Context, channelID string, spe } // modelPatchFromSpec maps an upstream model spec onto model_configs columns. -// Prices/enabled/alias are intentionally untouched on update so admin pricing -// survives re-import. +// Prices/alias are intentionally untouched on update so admin pricing survives +// re-import. func modelPatchFromSpec(spec ChannelModelSpec, existing *model.ModelConfig) map[string]any { patch := map[string]any{ "type": spec.Kind, diff --git a/backend/internal/service/v1.go b/backend/internal/service/v1.go index 19da6ac..27f79db 100644 --- a/backend/internal/service/v1.go +++ b/backend/internal/service/v1.go @@ -64,8 +64,8 @@ const videoGenBudget = 30 * time.Minute // imageGenBudget caps one image generation end-to-end. const imageGenBudget = 8 * time.Minute -// defaultChannelConcurrency applies when a channel's Concurrency is 0. -const defaultChannelConcurrency = 5 +// Channel concurrency semantics: 0 = unlimited (the Lua slot script passes +// max=0 straight through), matching vivid's channel design. // maxChannelFailover caps how many channels one request may burn on // retryable/temporary errors before giving up, so an upstream-wide blip can't @@ -100,11 +100,11 @@ type V1Service struct { conc *ConcurrencyService } -// chanAcquire takes one per-channel slot (capped at max; 0 = default), tagged +// chanAcquire takes one per-channel slot (capped at max; 0 = unlimited), tagged // with the generation's eventID. func (s *V1Service) chanAcquire(ctx context.Context, channelID, eventID string, max int) bool { - if max < 1 { - max = defaultChannelConcurrency + if max < 0 { + max = 0 } return s.conc.Acquire(ctx, "conc:c:"+channelID, max, eventID) } diff --git a/frontend/src/views/ChannelsView.vue b/frontend/src/views/ChannelsView.vue index c2d94a1..c7489aa 100644 --- a/frontend/src/views/ChannelsView.vue +++ b/frontend/src/views/ChannelsView.vue @@ -10,6 +10,8 @@ const channels = ref([]) const loading = ref(false) const err = ref('') const warn = ref('') // 保存后自动同步的非致命告警 +const okMsg = ref('') // 手动同步成功提示 +const syncing = ref(false) // create/edit form const form = ref(null) @@ -31,12 +33,6 @@ async function load() { loading.value = false } -function modelSummary(c) { - const names = c.models || [] - if (!names.length) return '' - return names.length <= 2 ? names.join('、') : `${names[0]}、${names[1]} 等` -} - function openAdd() { formErr.value = '' form.value = { id: '', name: '', base_url: '', api_key: '', concurrency: 5, weight: 1, enabled: true } @@ -54,6 +50,7 @@ async function save() { if (saving.value) return saving.value = true formErr.value = '' + okMsg.value = '' const r = await api('/channels', jsonBody('POST', { id: f.id, name: f.name, base_url: f.base_url, api_key: f.api_key, enabled: !!f.enabled, concurrency: f.concurrency || 0, weight: f.weight || 1, @@ -83,13 +80,34 @@ async function del(c) { channels.value = channels.value.filter((x) => x.id !== c.id) } -// 表单内拉模型：编辑态走库内渠道（密钥不用重填）；新渠道用表单里的地址+key 预览。 +// 已保存渠道手动「同步上游模型」：直接全量同步（新模型入库绑定，上游消失的 +// 解绑/清孤儿），拉取失败直接报错；新渠道则只用表单地址+key 预览不落库。 +async function syncModels(c) { + if (syncing.value) return + syncing.value = true + err.value = '' + warn.value = '' + okMsg.value = '' + try { + const r = await api('/channels/sync-models', jsonBody('POST', { id: c.id })) + if (!r.ok) { err.value = r.data?.detail || '同步失败'; return } + const msg = r.data?.message || '已同步上游模型' + if (msg === '已同步上游模型') okMsg.value = msg + else warn.value = msg + load() + } finally { + syncing.value = false + } +} + +// 表单内拉模型：编辑态直接全量同步；新渠道用表单里的地址+key 预览（导入时自动保存）。 function fetchFromForm() { const f = form.value if (!f) return + if (f.id) { syncModels(f); return } if (!/^https?:\/\/\S+$/.test((f.base_url || '').trim())) { formErr.value = '上游地址需以 http(s):// 开头'; return } - if (!f.id && !f.api_key) { formErr.value = '请填写 API Key'; return } - openFetch({ id: f.id || '', name: f.name || f.base_url, base_url: f.base_url, api_key: f.api_key, concurrency: f.concurrency, weight: f.weight, enabled: !!f.enabled }) + if (!f.api_key) { formErr.value = '请填写 API Key'; return } + openFetch({ id: '', name: f.name || f.base_url, base_url: f.base_url, api_key: f.api_key, concurrency: f.concurrency, weight: f.weight, enabled: !!f.enabled }) } async function openFetch(c) { @@ -109,6 +127,10 @@ async function openFetch(c) { // 新渠道尚未入库时，第一次勾选会先自动保存渠道再绑定。 async function toggleBind(s) { if (s.busy) return + if (!s.bound && s.alias && s.alias === s.id) { + fetchErr.value = `「${s.id}」别名不能与上游模型同名，留空即可` + return + } s.busy = true fetchErr.value = '' try { @@ -172,6 +194,7 @@ onMounted(load)

{{ err }}

{{ warn }}

+

{{ okMsg }}

@@ -179,8 +202,6 @@ onMounted(load) 渠道 - 上游地址 - 模型 并发 配比 成功调用 @@ -197,14 +218,7 @@ onMounted(load)
{{ c.name }}
{{ c.id }}
- -
{{ c.base_url }}
- - -
{{ (c.models || []).length }}
-
{{ modelSummary(c) }}
- - {{ c.concurrency || '默认' }} + {{ c.concurrency || '不限' }} {{ c.weight || 1 }} {{ (c.success_total || 0).toLocaleString() }} {{ (c.used_credits || 0).toLocaleString() }} @@ -218,13 +232,13 @@ onMounted(load) {{ c.last_used_at ? fmtTs(c.last_used_at) : '-' }} - + - 正在加载渠道… - 没有渠道，点右上角「添加渠道」接入上游实例 + 正在加载渠道… + 没有渠道，点右上角「添加渠道」接入上游实例
@@ -255,8 +269,8 @@ onMounted(load)
- -

0 = 默认 5

+ +

0 = 不限

@@ -276,7 +290,7 @@ onMounted(load) - {{ form.id ? '用已保存的地址与密钥拉取' : '用上方的地址与 API Key 拉取；导入时会先自动保存渠道' }} + {{ form.id ? '直接全量同步上游模型到模型列表' : '用上方的地址与 API Key 预览；导入时会先自动保存渠道' }}

{{ formErr }}

diff --git a/frontend/src/views/LogsView.vue b/frontend/src/views/LogsView.vue index 4970b31..3954a24 100644 --- a/frontend/src/views/LogsView.vue +++ b/frontend/src/views/LogsView.vue @@ -278,7 +278,7 @@ const sourcePill = (s) => ({ — - + {{ statusLabel(e.status) }}