From 4632b1ab04ae5f36e6a53315215a744c8e391dcd Mon Sep 17 00:00:00 2001 From: chiyi Date: Mon, 19 Oct 2026 19:14:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E5=AF=B9=E9=BD=90=20vivid=20?= =?UTF-8?q?=E6=9C=80=E6=96=B0=E6=B8=A0=E9=81=93/=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E8=AE=BE=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 渠道管理(对齐 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> --- backend/internal/bootstrap/app.go | 2 +- .../internal/http/handler/channel_admin.go | 18 -- backend/internal/http/router/router.go | 1 - backend/internal/repo/event_repo.go | 23 +++ backend/internal/service/admin_write.go | 2 + backend/internal/service/channel.go | 55 +++--- backend/internal/upstream/client.go | 7 +- frontend/src/components/ModelFormModal.vue | 4 + frontend/src/views/ChannelsView.vue | 176 +++++++----------- 9 files changed, 129 insertions(+), 159 deletions(-) diff --git a/backend/internal/bootstrap/app.go b/backend/internal/bootstrap/app.go index 37a9e4d..7b97227 100644 --- a/backend/internal/bootstrap/app.go +++ b/backend/internal/bootstrap/app.go @@ -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 { diff --git a/backend/internal/http/handler/channel_admin.go b/backend/internal/http/handler/channel_admin.go index be459d5..0e6046f 100644 --- a/backend/internal/http/handler/channel_admin.go +++ b/backend/internal/http/handler/channel_admin.go @@ -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) { diff --git a/backend/internal/http/router/router.go b/backend/internal/http/router/router.go index 1efcd71..7d6ed52 100644 --- a/backend/internal/http/router/router.go +++ b/backend/internal/http/router/router.go @@ -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) diff --git a/backend/internal/repo/event_repo.go b/backend/internal/repo/event_repo.go index 8d77d06..816464c 100644 --- a/backend/internal/repo/event_repo.go +++ b/backend/internal/repo/event_repo.go @@ -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) { diff --git a/backend/internal/service/admin_write.go b/backend/internal/service/admin_write.go index 6a6ea71..e0ef5d0 100644 --- a/backend/internal/service/admin_write.go +++ b/backend/internal/service/admin_write.go @@ -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(), diff --git a/backend/internal/service/channel.go b/backend/internal/service/channel.go index 496d507..154dd67 100644 --- a/backend/internal/service/channel.go +++ b/backend/internal/service/channel.go @@ -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 diff --git a/backend/internal/upstream/client.go b/backend/internal/upstream/client.go index 07f0e61..2810fcf 100644 --- a/backend/internal/upstream/client.go +++ b/backend/internal/upstream/client.go @@ -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 diff --git a/frontend/src/components/ModelFormModal.vue b/frontend/src/components/ModelFormModal.vue index bf7e45d..fd33e35 100644 --- a/frontend/src/components/ModelFormModal.vue +++ b/frontend/src/components/ModelFormModal.vue @@ -167,6 +167,10 @@ async function save() { duration_prices_agent, alias: alias.value.trim(), max_reference_images: e.max_reference_images || 0, + max_reference_videos: e.max_reference_videos || 0, + max_reference_audios: e.max_reference_audios || 0, + max_ref_video_seconds: e.max_ref_video_seconds || 0, + max_ref_audio_seconds: e.max_ref_audio_seconds || 0, reference_mode: e.reference_mode || 'none', weight: Number(weight.value) || 0, } diff --git a/frontend/src/views/ChannelsView.vue b/frontend/src/views/ChannelsView.vue index c7489aa..1f4865e 100644 --- a/frontend/src/views/ChannelsView.vue +++ b/frontend/src/views/ChannelsView.vue @@ -10,19 +10,20 @@ const channels = ref([]) const loading = ref(false) const err = ref('') const warn = ref('') // 保存后自动同步的非致命告警 -const okMsg = ref('') // 手动同步成功提示 -const syncing = ref(false) +const okMsg = ref('') // 同步模型成功提示 // create/edit form const form = ref(null) const saving = ref(false) const formErr = ref('') -// fetch-models → 勾选即绑定（选中自动导入到模型列表，取消即解绑） -const fetchTarget = ref(null) // channel snapshot being imported into +// 同步上游模型弹窗：开关选择 + 别名，批量导入（保存渠道已自动全量同步） +const fetchTarget = ref(null) // saved channel being synced const fetching = ref(false) const fetchErr = ref('') const specs = ref([]) +const picked = ref(new Set()) +const importing = ref(false) async function load() { loading.value = true @@ -80,84 +81,44 @@ async function del(c) { channels.value = channels.value.filter((x) => x.id !== c.id) } -// 已保存渠道手动「同步上游模型」：直接全量同步（新模型入库绑定，上游消失的 -// 解绑/清孤儿），拉取失败直接报错；新渠道则只用表单地址+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.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) { fetchTarget.value = c specs.value = [] + picked.value = new Set() fetchErr.value = '' fetching.value = true - const r = await api('/channels/fetch-models', jsonBody('POST', { - id: c.id || '', base_url: c.base_url || '', api_key: c.api_key || '', - })) + const r = await api('/channels/fetch-models', jsonBody('POST', { id: c.id })) fetching.value = false if (!r.ok) { fetchErr.value = r.data?.detail || '拉取失败'; return } - specs.value = (r.data?.data || []).map((s) => ({ ...s, alias: s.alias || '', busy: false })) + specs.value = (r.data?.data || []).map((s) => ({ ...s, alias: '' })) + picked.value = new Set(specs.value.map((s) => s.id)) } -// 勾选 = 立即导入绑定；取消勾选 = 解绑（模型没有其他渠道接入时本地行一并删除）。 -// 新渠道尚未入库时，第一次勾选会先自动保存渠道再绑定。 -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 +function togglePick(id) { + const next = new Set(picked.value) + next.has(id) ? next.delete(id) : next.add(id) + picked.value = next +} + +async function doImport() { + if (importing.value || !picked.value.size) return + const dup = specs.value.find((s) => picked.value.has(s.id) && s.alias && s.alias === s.id) + if (dup) { fetchErr.value = `${dup.id}的别名与上游模型名相同，别名只在需要改名时设置，同名请留空`; return } + const models = specs.value.filter((s) => picked.value.has(s.id)) + .map((s) => { const { bound, ...rest } = s; return rest }) + importing.value = true fetchErr.value = '' try { - const target = fetchTarget.value - if (!s.bound) { - if (!target.id) { - const sr = await api('/channels', jsonBody('POST', { - id: '', name: target.name || target.base_url, base_url: target.base_url, api_key: target.api_key, - enabled: target.enabled !== false, concurrency: target.concurrency || 0, weight: target.weight || 1, - })) - if (!sr.ok) { fetchErr.value = sr.data?.detail || '渠道保存失败'; return } - if (sr.data?.warning) warn.value = sr.data.warning - target.id = sr.data?.id - form.value = null - } - const { bound, busy, ...rest } = s - const r = await api('/channels/import-models', jsonBody('POST', { id: target.id, models: [rest] })) - if (!r.ok) { fetchErr.value = r.data?.detail || '导入失败'; return } - s.bound = true - } else { - const r = await api('/channels/unbind-model', jsonBody('POST', { id: target.id, model: s.id })) - if (!r.ok) { fetchErr.value = r.data?.detail || '解绑失败'; return } - s.bound = false - } + const r = await api('/channels/import-models', jsonBody('POST', { id: fetchTarget.value.id, models })) + if (!r.ok) { fetchErr.value = r.data?.detail || '导入失败'; return } + okMsg.value = `已导入 ${models.length} 个模型` + fetchTarget.value = null load() } finally { - s.busy = false + importing.value = false } } @@ -204,7 +165,9 @@ onMounted(load) 渠道 并发 配比 - 成功调用 + 运行中 + 成功数 + 失败数 累计消耗 上游余额 状态 @@ -220,7 +183,9 @@ onMounted(load) {{ c.concurrency || '不限' }} {{ c.weight || 1 }} + {{ c.running_count || 0 }} {{ (c.success_total || 0).toLocaleString() }} + {{ (c.fail_count || 0).toLocaleString() }} {{ (c.used_credits || 0).toLocaleString() }} {{ c.upstream_balance == null ? '-' : c.upstream_balance.toLocaleString() }} @@ -232,13 +197,13 @@ onMounted(load) {{ c.last_used_at ? fmtTs(c.last_used_at) : '-' }} - + - 正在加载渠道… - 没有渠道，点右上角「添加渠道」接入上游实例 + 正在加载渠道… + 没有渠道，点右上角「添加渠道」接入上游实例 @@ -286,12 +251,6 @@ onMounted(load) {{ form.enabled ? '已启用' : '已停用' }} -
- - {{ form.id ? '直接全量同步上游模型到模型列表' : '用上方的地址与 API Key 预览；导入时会先自动保存渠道' }} -

{{ formErr }}

@@ -301,42 +260,40 @@ onMounted(load)
- +
-

上游模型 · {{ fetchTarget.name || fetchTarget.base_url }}

+

同步上游模型 · {{ fetchTarget.name || fetchTarget.base_url }}

正在从上游拉取模型列表…