diff --git a/.gitignore b/.gitignore index 65f4593..b616070 100644 --- a/.gitignore +++ b/.gitignore @@ -191,6 +191,7 @@ test/launcher-config-local.json # E2E 测试输出（录制文件等） /test-output/ +test-output*appdata/ # npm 锁文件（项目统一使用 yarn） package-lock.json diff --git a/src/configs/config.go b/src/configs/config.go index ddce413..4945cd1 100644 --- a/src/configs/config.go +++ b/src/configs/config.go @@ -79,7 +79,7 @@ func (f *Feature) GetEffectiveDownloaderType() DownloaderType { type DanmakuConfig struct { FontSize int `yaml:"font_size" json:"font_size"` // 字体大小 (12~120) FontName string `yaml:"font_name" json:"font_name"` // 字体名称 - ScrollArea string `yaml:"scroll_area" json:"scroll_area"` // 滚动区域: full(全屏), top(顶部), bottom(底部) + ScrollArea string `yaml:"scroll_area" json:"scroll_area"` // 滚动区域: full(全屏), top(顶部半屏), bottom(底部半屏), quarter(1/4屏), three-quarter(3/4屏) ScrollTime int `yaml:"scroll_time" json:"scroll_time"` // 弹幕滚过屏幕的秒数 (5~20) Resolution string `yaml:"resolution" json:"resolution"` // 播放分辨率 Outline *int `yaml:"outline,omitempty" json:"outline,omitempty"` // 描边粗细 (0~4)，nil 表示使用默认值 @@ -115,9 +115,11 @@ var defaultDanmakuConfig = DanmakuConfig{ // validScrollAreas 支持的滚动区域 var validScrollAreas = map[string]bool{ - "full": true, // 全屏滚动 - "top": true, // 仅在屏幕上半部分滚动 - "bottom": true, // 仅在屏幕下半部分滚动 + "full": true, // 全屏滚动 + "top": true, // 仅在屏幕上半部分滚动 + "bottom": true, // 仅在屏幕下半部分滚动 + "quarter": true, // 仅在屏幕上1/4部分滚动 + "three-quarter": true, // 仅在屏幕上3/4部分滚动 } // validResolutions 支持的分辨率列表 @@ -221,7 +223,7 @@ func (d *DanmakuConfig) ValidateWithPlatform(platformKey string) error { return fmt.Errorf("字体名称不能为空") } if !validScrollAreas[d.ScrollArea] { - return fmt.Errorf("不支持的滚动区域: %s，可选值: full, top, bottom", d.ScrollArea) + return fmt.Errorf("不支持的滚动区域: %s，可选值: full, top, bottom, quarter, three-quarter", d.ScrollArea) } if d.ScrollTime < 5 || d.ScrollTime > 20 { return fmt.Errorf("滚动时间必须在 5~20 秒之间，当前值: %d", d.ScrollTime) @@ -649,38 +651,51 @@ func UpdateTransient(mutator func(c *Config) error) (*Config, error) { } func updateImpl(mutator func(c *Config) error, persist bool) (*Config, error) { - updateMu.Lock() - defer updateMu.Unlock() - old := GetCurrentConfig() - // 若当前尚未设置配置，则以默认配置为基础 - var base *Config - if old == nil { - base = NewConfig() - } else { - base = CloneConfigShallow(old) - } - if err := mutator(base); err != nil { - return nil, err - } - // 维护派生字段 - base.RefreshLiveRoomIndexCache() - // 版本号自增 - if old == nil { - base.Version = 1 - } else { - base.Version = old.Version + 1 - } - newCfg := base + var newCfg *Config + var updateErr error - if persist && newCfg.File != "" { - if err := newCfg.Marshal(); err != nil { - // 如果持久化失败，我们选择记录错误但不阻止内存更新 - // 或者返回错误？这里选择返回错误，因为用户期望保存成功。 - return nil, fmt.Errorf("failed to save config: %w", err) + func() { + updateMu.Lock() + defer updateMu.Unlock() + old := GetCurrentConfig() + // 若当前尚未设置配置，则以默认配置为基础 + var base *Config + if old == nil { + base = NewConfig() + } else { + base = CloneConfigShallow(old) } + if err := mutator(base); err != nil { + updateErr = err + return + } + // 维护派生字段 + base.RefreshLiveRoomIndexCache() + // 版本号自增 + if old == nil { + base.Version = 1 + } else { + base.Version = old.Version + 1 + } + newCfg = base + + // 持久化在锁内执行，保证内存与磁盘一致性 + if persist && newCfg.File != "" { + if err := newCfg.Marshal(); err != nil { + updateErr = fmt.Errorf("failed to save config: %w", err) + return + } + } + SetCurrentConfig(newCfg) + }() + + if updateErr != nil { + return nil, updateErr + } + if newCfg == nil { + return nil, errors.New("config update failed") } - SetCurrentConfig(newCfg) return newCfg, nil } @@ -691,38 +706,52 @@ func UpdateCAS(expectedVersion int64, mutator func(c *Config) error) (*Config, e } func updateCASImpl(expectedVersion int64, mutator func(c *Config) error, persist bool) (*Config, error) { - updateMu.Lock() - defer updateMu.Unlock() - cur := GetCurrentConfig() - // 校验版本 - var curVersion int64 - if cur != nil { - curVersion = cur.Version - } - if curVersion != expectedVersion { - return nil, ErrConfigVersionConflict - } - // 克隆并修改 - var base *Config - if cur == nil { - base = NewConfig() - } else { - base = CloneConfigShallow(cur) - } - if err := mutator(base); err != nil { - return nil, err - } - base.RefreshLiveRoomIndexCache() - base.Version = expectedVersion + 1 + var newCfg *Config + var updateErr error - if persist && base.File != "" { - if err := base.Marshal(); err != nil { - return nil, fmt.Errorf("failed to save config: %w", err) + func() { + updateMu.Lock() + defer updateMu.Unlock() + cur := GetCurrentConfig() + // 校验版本 + var curVersion int64 + if cur != nil { + curVersion = cur.Version } + if curVersion != expectedVersion { + updateErr = ErrConfigVersionConflict + return + } + // 克隆并修改 + var base *Config + if cur == nil { + base = NewConfig() + } else { + base = CloneConfigShallow(cur) + } + if err := mutator(base); err != nil { + updateErr = err + return + } + base.RefreshLiveRoomIndexCache() + base.Version = expectedVersion + 1 + newCfg = base + + // 持久化在锁内执行，保证内存与磁盘一致性 + if persist && newCfg.File != "" { + if err := newCfg.Marshal(); err != nil { + updateErr = fmt.Errorf("failed to save config: %w", err) + return + } + } + SetCurrentConfig(newCfg) + }() + + if updateErr != nil { + return nil, updateErr } - SetCurrentConfig(base) - return base, nil + return newCfg, nil } // UpdateWithRetry 在读取-修改-提交之间做乐观锁重试，避免调用方自行实现重试逻辑 @@ -811,6 +840,14 @@ func AppendLiveRoom(room LiveRoom) (*Config, error) { }, 3, 10*time.Millisecond) } +// AppendLiveRoomTransient 追加一个 LiveRoom（仅更新内存，不持久化）。 +func AppendLiveRoomTransient(room LiveRoom) (*Config, error) { + return UpdateWithRetryTransient(func(c *Config) error { + c.LiveRooms = append(c.LiveRooms, room) + return nil + }, 3, 10*time.Millisecond) +} + // RemoveLiveRoomByUrl 从配置中移除指定 URL 的房间 func RemoveLiveRoomByUrl(url string) (*Config, error) { return UpdateWithRetry(func(c *Config) error { @@ -849,6 +886,18 @@ func SetLiveRoomId(url string, id types.LiveID) (*Config, error) { }, 3, 10*time.Millisecond) } +// Persist 将当前内存中的配置持久化到磁盘。 +// 用于批量 Transient 更新后统一刷盘的场景。 +func Persist() error { + updateMu.Lock() + defer updateMu.Unlock() + cfg := GetCurrentConfig() + if cfg == nil { + return errors.New("config not initialized") + } + return cfg.Marshal() +} + type LiveRoom struct { Url string `yaml:"url" json:"url"` IsListening bool `yaml:"is_listening" json:"is_listening"` diff --git a/src/recorders/danmaku/ass.go b/src/recorders/danmaku/ass.go index 1e54003..85251e4 100644 --- a/src/recorders/danmaku/ass.go +++ b/src/recorders/danmaku/ass.go @@ -28,7 +28,7 @@ type AssWriter struct { laneEnd int // last usable lane index (exclusive) laneNum int // total lanes in the usable range nextLane int - laneLast []int64 // last end time (centiseconds) per lane + laneLast []int64 // last tail-clear time (centiseconds) per lane } func parseResolution(res string) (int, int) { @@ -72,6 +72,10 @@ func NewAssWriter(filePath string, startAt time.Time, cfg configs.DanmakuConfig, laneEnd = totalLanes / 2 case "bottom": laneStart = totalLanes / 2 + case "quarter": + laneEnd = totalLanes / 4 + case "three-quarter": + laneEnd = totalLanes * 3 / 4 } laneNum := laneEnd - laneStart if laneNum < 1 { @@ -84,14 +88,14 @@ func NewAssWriter(filePath string, startAt time.Time, cfg configs.DanmakuConfig, cfg: cfg, title: title, resX: resX, - resY: resY, + resY: resY, scrollTimeMs: scrollTimeMs, - bannerSpeed: bannerSpeed, - laneStart: laneStart, - laneEnd: laneEnd, - laneNum: laneNum, - nextLane: 0, - laneLast: make([]int64, laneNum), + bannerSpeed: bannerSpeed, + laneStart: laneStart, + laneEnd: laneEnd, + laneNum: laneNum, + nextLane: 0, + laneLast: make([]int64, laneNum), } if err := w.writeHeader(); err != nil { @@ -251,7 +255,12 @@ func (w *AssWriter) AddDanmaku(recvAt time.Time, username, text string, color in } endCS := startCS + durationCS - lane := w.assignLane(startCS, endCS) + lane, adjustedStartCS := w.assignLane(startCS, textWidth) + // 基于文字宽度的防重叠：使用调整后的起始时间 + if adjustedStartCS != startCS { + startCS = adjustedStartCS + endCS = startCS + durationCS + } laneHeight := w.cfg.FontSize + 4 marginV := (lane + w.laneStart) * laneHeight @@ -298,7 +307,12 @@ func (w *AssWriter) AddGift(recvAt time.Time, username, giftName string, num int } endCS := startCS + durationCS - lane := w.assignLane(startCS, endCS) + lane, adjustedStartCS := w.assignLane(startCS, textWidth) + // 基于文字宽度的防重叠：使用调整后的起始时间 + if adjustedStartCS != startCS { + startCS = adjustedStartCS + endCS = startCS + durationCS + } laneHeight := w.cfg.FontSize + 4 marginV := (lane + w.laneStart) * laneHeight @@ -373,24 +387,36 @@ func (w *AssWriter) AddSuperChat(recvAt time.Time, username, text string, price } } -func (w *AssWriter) assignLane(startCS, endCS int64) int { +// assignLane 分配一个空闲 lane，基于文字宽度的防重叠。 +// 参数：startCS 弹幕开始时间，textWidth 文字像素宽度。 +// 返回值：lane 索引、调整后的 startCS（可能延迟）。 +func (w *AssWriter) assignLane(startCS int64, textWidth int) (int, int64) { + // 安全间距：防止因字体渲染差异导致的重叠 + safeTextWidth := textWidth + w.cfg.FontSize + // 优先找空闲 lane（前一条弹幕的尾部已离开屏幕右侧） for i := 0; i < w.laneNum; i++ { idx := (w.nextLane + i) % w.laneNum if w.laneLast[idx] <= startCS { - w.laneLast[idx] = endCS + // 存储该弹幕尾部离开右侧的时间点（用于下一条弹幕判断） + tailClearCS := startCS + int64(w.scrollTimeMs)*int64(safeTextWidth)/int64(w.resX)/10 + w.laneLast[idx] = tailClearCS w.nextLane = (idx + 1) % w.laneNum - return idx + return idx, startCS } } + // 所有 lane 都被占用，延迟到最早可用的时间点 earliest := 0 for i := 1; i < w.laneNum; i++ { if w.laneLast[i] < w.laneLast[earliest] { earliest = i } } - w.laneLast[earliest] = endCS + newStartCS := w.laneLast[earliest] + // 存储新弹幕尾部离开右侧的时间点 + tailClearCS := newStartCS + int64(w.scrollTimeMs)*int64(safeTextWidth)/int64(w.resX)/10 + w.laneLast[earliest] = tailClearCS w.nextLane = (earliest + 1) % w.laneNum - return earliest + return earliest, newStartCS } func (w *AssWriter) OutputPath() string { diff --git a/src/servers/handler.go b/src/servers/handler.go index 6ace7eb..bf6d550 100644 --- a/src/servers/handler.go +++ b/src/servers/handler.go @@ -19,6 +19,7 @@ import ( "text/template" "time" + "github.com/google/uuid" "github.com/gorilla/mux" "github.com/hr3lxphr6j/requests" "github.com/tidwall/gjson" @@ -678,8 +679,9 @@ func addLives(writer http.ResponseWriter, r *http.Request) { inst := instance.GetInstance(r.Context()) b, err := io.ReadAll(r.Body) if err != nil { - writeJSON(writer, map[string]any{ - "error": err.Error(), + writeJsonWithStatusCode(writer, http.StatusBadRequest, commonResp{ + ErrNo: http.StatusBadRequest, + ErrMsg: err.Error(), }) return } @@ -687,8 +689,9 @@ func addLives(writer http.ResponseWriter, r *http.Request) { errorMessages := make([]string, 0, 4) gjson.ParseBytes(b).ForEach(func(key, value gjson.Result) bool { isListen := value.Get("listen").Bool() + notifyOnly := value.Get("notify_only").Bool() urlStr := strings.Trim(value.Get("url").String(), " ") - if retInfo, err := addLiveImpl(inst.Ctx, urlStr, isListen); err != nil { + if retInfo, err := addLiveImpl(inst.Ctx, urlStr, isListen, notifyOnly, true); err != nil { msg := urlStr + ": " + err.Error() applog.GetLogger().Error(msg) errorMessages = append(errorMessages, msg) @@ -699,11 +702,10 @@ func addLives(writer http.ResponseWriter, r *http.Request) { return true }) sort.Sort(info) - // TODO return error messages too writeJSON(writer, info) } -func addLiveImpl(ctx context.Context, urlStr string, isListen bool) (info *live.Info, err error) { +func addLiveImpl(ctx context.Context, urlStr string, isListen bool, notifyOnly bool, persist bool) (info *live.Info, err error) { if !strings.HasPrefix(urlStr, "http://") && !strings.HasPrefix(urlStr, "https://") { urlStr = "https://" + urlStr } @@ -718,6 +720,7 @@ func addLiveImpl(ctx context.Context, urlStr string, isListen bool) (info *live. liveRoom = &configs.LiveRoom{ Url: u.String(), IsListening: isListen, + NotifyOnly: notifyOnly, } needAppend = true } @@ -725,33 +728,171 @@ func addLiveImpl(ctx context.Context, urlStr string, isListen bool) (info *live. if err != nil { return nil, err } + if !inst.Lives.SetIfAbsent(newLive.GetLiveId(), newLive) { + return nil, errors.New("直播间已存在") + } // 记录 LiveId 到全局配置（并发安全） configs.SetLiveRoomId(u.String(), newLive.GetLiveId()) - if inst.Lives.SetIfAbsent(newLive.GetLiveId(), newLive) { - if isListen { - inst.ListenerManager.(listeners.Manager).AddListener(ctx, newLive) + // 先将房间写入配置，再启动 Listener，确保 sendLiveNotification 能读到 NotifyOnly 等字段 + if needAppend { + if liveRoom == nil { + return nil, errors.New("liveRoom is nil, cannot append to LiveRooms") } - info = parseInfo(ctx, newLive) - - if needAppend { - if liveRoom == nil { - return nil, errors.New("liveRoom is nil, cannot append to LiveRooms") - } - // 使用统一的 Update 接口做 COW 并原子替换 + if persist { if _, err := configs.AppendLiveRoom(*liveRoom); err != nil { return nil, err } + } else { + if _, err := configs.AppendLiveRoomTransient(*liveRoom); err != nil { + return nil, err + } } - // 广播直播间列表变更事件 - GetSSEHub().BroadcastListChange(newLive.GetLiveId(), "room_added", map[string]interface{}{ - "live_id": string(newLive.GetLiveId()), - "url": urlStr, - "listening": isListen, - }) } + if isListen { + inst.ListenerManager.(listeners.Manager).AddListener(ctx, newLive) + } + info = parseInfo(ctx, newLive) + // 广播直播间列表变更事件 + GetSSEHub().BroadcastListChange(newLive.GetLiveId(), "room_added", map[string]interface{}{ + "live_id": string(newLive.GetLiveId()), + "url": urlStr, + "listening": isListen, + }) return info, nil } +func batchAddLives(writer http.ResponseWriter, r *http.Request) { + inst := instance.GetInstance(r.Context()) + b, err := io.ReadAll(r.Body) + if err != nil { + writeJsonWithStatusCode(writer, http.StatusBadRequest, commonResp{ + ErrNo: http.StatusBadRequest, + ErrMsg: "failed to read request body", + }) + return + } + + var req batchAddRequest + if err := json.Unmarshal(b, &req); err != nil { + writeJsonWithStatusCode(writer, http.StatusBadRequest, commonResp{ + ErrNo: http.StatusBadRequest, + ErrMsg: "invalid request body", + }) + return + } + + // 过滤空 URL，只保留有效条目 + validURLs := make([]string, 0, len(req.URLs)) + for _, u := range req.URLs { + if s := strings.TrimSpace(u); s != "" { + validURLs = append(validURLs, s) + } + } + + if len(validURLs) == 0 { + writeJsonWithStatusCode(writer, http.StatusBadRequest, commonResp{ + ErrNo: http.StatusBadRequest, + ErrMsg: "urls is empty", + }) + return + } + + const maxBatchSize = 50 + if len(validURLs) > maxBatchSize { + writeJsonWithStatusCode(writer, http.StatusBadRequest, commonResp{ + ErrNo: http.StatusBadRequest, + ErrMsg: fmt.Sprintf("too many urls: %d (max %d)", len(validURLs), maxBatchSize), + }) + return + } + + // 使用客户端提供的 batch_id，或生成新的 + batchID := req.BatchID + if batchID == "" { + batchID = "batch_" + uuid.New().String() + } + + // 立即返回 batch_id 和有效总数 + writeJSON(writer, batchAddResponse{ + BatchID: batchID, + Total: len(validURLs), + }) + + // 异步处理批量添加 + go func() { + defer func() { + if r := recover(); r != nil { + applog.GetLogger().Errorf("batch add: panic recovered: %v", r) + GetSSEHub().BroadcastBatchComplete(batchID, batchCompleteEvent{ + Total: len(validURLs), + SuccessCount: 0, + FailCount: len(validURLs), + Timestamp: time.Now(), + }) + } + }() + hub := GetSSEHub() + successCount := 0 + failCount := 0 + + for i, urlStr := range validURLs { + // 检查重复房间，避免调用 addLiveImpl 做无用的网络请求 + checkURL := urlStr + if !strings.HasPrefix(checkURL, "http://") && !strings.HasPrefix(checkURL, "https://") { + checkURL = "https://" + checkURL + } + if u, err := url.Parse(checkURL); err == nil { + if _, err := configs.GetCurrentConfig().GetLiveRoomByUrl(u.String()); err == nil { + event := batchProgressEvent{ + Index: i, + Total: len(validURLs), + URL: urlStr, + Success: false, + Error: "直播间已存在", + } + failCount++ + hub.BroadcastBatchProgress(batchID, event) + continue + } + } + + retInfo, err := addLiveImpl(inst.Ctx, urlStr, req.Listen, req.NotifyOnly, false) + event := batchProgressEvent{ + Index: i, + Total: len(validURLs), + URL: urlStr, + Success: err == nil, + } + if err != nil { + event.Error = err.Error() + failCount++ + } else { + event.Info = retInfo + successCount++ + } + + hub.BroadcastBatchProgress(batchID, event) + } + + // 批量完成后统一持久化一次配置 + var persistErr string + if successCount > 0 { + if err := configs.Persist(); err != nil { + persistErr = err.Error() + applog.GetLogger().Errorf("batch add: failed to persist config: %v", err) + } + } + + hub.BroadcastBatchComplete(batchID, batchCompleteEvent{ + Total: len(validURLs), + SuccessCount: successCount, + FailCount: failCount, + PersistError: persistErr, + Timestamp: time.Now(), + }) + }() +} + func removeLive(writer http.ResponseWriter, r *http.Request) { inst := instance.GetInstance(r.Context()) vars := mux.Vars(r) @@ -888,7 +1029,7 @@ func applyLiveRoomsByConfig(ctx context.Context, oldConfig *configs.Config, newC newUrlMap[newRoom.Url] = newRoom if room, err := oldConfig.GetLiveRoomByUrl(newRoom.Url); err != nil { // add live - if _, err := addLiveImpl(ctx, newRoom.Url, newRoom.IsListening); err != nil { + if _, err := addLiveImpl(ctx, newRoom.Url, newRoom.IsListening, newRoom.NotifyOnly, true); err != nil { return err } } else { diff --git a/src/servers/models.go b/src/servers/models.go index b8a5d3a..dadad99 100644 --- a/src/servers/models.go +++ b/src/servers/models.go @@ -1,6 +1,8 @@ package servers import ( + "time" + "github.com/bililive-go/bililive-go/src/live" ) @@ -21,3 +23,36 @@ func (c liveSlice) Swap(i, j int) { func (c liveSlice) Less(i, j int) bool { return c[i].Live.GetLiveId() < c[j].Live.GetLiveId() } + +// batchAddRequest 批量添加直播间请求 +type batchAddRequest struct { + URLs []string `json:"urls"` + Listen bool `json:"listen"` + NotifyOnly bool `json:"notify_only,omitempty"` + BatchID string `json:"batch_id,omitempty"` +} + +// batchAddResponse 批量添加直播间响应（立即返回） +type batchAddResponse struct { + BatchID string `json:"batch_id"` + Total int `json:"total"` +} + +// batchProgressEvent SSE 进度事件数据 +type batchProgressEvent struct { + Index int `json:"index"` + Total int `json:"total"` + URL string `json:"url"` + Success bool `json:"success"` + Error string `json:"error,omitempty"` + Info *live.Info `json:"live_info,omitempty"` +} + +// batchCompleteEvent SSE 完成事件数据 +type batchCompleteEvent struct { + Total int `json:"total"` + SuccessCount int `json:"success_count"` + FailCount int `json:"fail_count"` + PersistError string `json:"persist_error,omitempty"` + Timestamp time.Time `json:"timestamp"` +} diff --git a/src/servers/osrp_handler.go b/src/servers/osrp_handler.go index 0ac063f..0daf591 100644 --- a/src/servers/osrp_handler.go +++ b/src/servers/osrp_handler.go @@ -260,7 +260,7 @@ func osrpAddTask(w http.ResponseWriter, r *http.Request) { } // 添加直播间 - info, err := addLiveImpl(ctx, req.URL, req.AutoStart) + info, err := addLiveImpl(ctx, req.URL, req.AutoStart, false, true) if err != nil { osrpWriteError(w, http.StatusBadRequest, "ADD_FAILED", err.Error()) return diff --git a/src/servers/server.go b/src/servers/server.go index 20b750b..45abc56 100644 --- a/src/servers/server.go +++ b/src/servers/server.go @@ -84,6 +84,7 @@ func initMux(ctx context.Context) *mux.Router { apiRoute.HandleFunc("/raw-config", putRawConfig).Methods("PUT") apiRoute.HandleFunc("/lives", getAllLives).Methods("GET") apiRoute.HandleFunc("/lives", addLives).Methods("POST") + apiRoute.HandleFunc("/lives/batch", batchAddLives).Methods("POST") apiRoute.HandleFunc("/lives/{id}", getLive).Methods("GET") apiRoute.HandleFunc("/lives/{id}", removeLive).Methods("DELETE") apiRoute.HandleFunc("/lives/{id}/logs", getLiveLogs).Methods("GET") diff --git a/src/servers/sse.go b/src/servers/sse.go index 0514c09..6581fc7 100644 --- a/src/servers/sse.go +++ b/src/servers/sse.go @@ -46,6 +46,10 @@ const ( SSEEventDanmaku SSEEventType = "danmaku" // SSEEventFFmpegStatus FFmpeg 就绪状态变更 SSEEventFFmpegStatus SSEEventType = "ffmpeg_status" + // SSEEventBatchProgress 批量添加进度 + SSEEventBatchProgress SSEEventType = "batch_progress" + // SSEEventBatchComplete 批量添加完成 + SSEEventBatchComplete SSEEventType = "batch_complete" ) // SSEMessage SSE 消息结构 @@ -109,6 +113,28 @@ func (h *SSEHub) Broadcast(msg SSEMessage) { } } +// BroadcastCritical 广播关键消息，channel 满时丢弃最旧消息腾出位置，保证送达。 +func (h *SSEHub) BroadcastCritical(msg SSEMessage) { + h.mu.RLock() + defer h.mu.RUnlock() + for ch := range h.clients { + select { + case ch <- msg: + default: + // channel 满，丢弃最旧的消息为关键事件腾出位置 + select { + case <-ch: + default: + } + select { + case ch <- msg: + default: + // 仍然满（极端情况），放弃 + } + } + } +} + // BroadcastLiveUpdate 广播直播间状态更新 func (h *SSEHub) BroadcastLiveUpdate(roomID types.LiveID, data interface{}) { h.Broadcast(SSEMessage{ @@ -239,6 +265,24 @@ func (h *SSEHub) BroadcastDanmaku(roomID types.LiveID, data interface{}) { }) } +// BroadcastBatchProgress 广播批量添加进度 +func (h *SSEHub) BroadcastBatchProgress(batchID string, data interface{}) { + h.Broadcast(SSEMessage{ + Type: SSEEventBatchProgress, + RoomID: batchID, + Data: data, + }) +} + +// BroadcastBatchComplete 广播批量添加完成（关键事件，保证送达） +func (h *SSEHub) BroadcastBatchComplete(batchID string, data interface{}) { + h.BroadcastCritical(SSEMessage{ + Type: SSEEventBatchComplete, + RoomID: batchID, + Data: data, + }) +} + // ClientCount 获取当前连接的客户端数量 func (h *SSEHub) ClientCount() int { h.mu.RLock() diff --git a/src/webapp/src/component/add-room-dialog/index.tsx b/src/webapp/src/component/add-room-dialog/index.tsx deleted file mode 100644 index 346eb7f..0000000 --- a/src/webapp/src/component/add-room-dialog/index.tsx +++ /dev/null @@ -1,86 +0,0 @@ -import { Modal, Input } from 'antd'; -import React from 'react'; -import API from '../../utils/api'; - -const api = new API(); - -interface Props { - refresh?: any; - children?: React.ReactNode; -} - -class AddRoomDialog extends React.Component { - state = { - ModalText: '请输入直播间的URL地址', - visible: false, - confirmLoading: false, - textView: '' - }; - - showModal = () => { - this.setState({ - ModalText: '请输入直播间的URL地址', - visible: true, - confirmLoading: false, - }); - }; - - handleOk = () => { - this.setState({ - ModalText: '正在添加直播间......', - confirmLoading: true, - }); - - api.addNewRoom(this.state.textView) - .then((rsp) => { - // 保存设置 - api.saveSettingsInBackground(); - this.setState({ - visible: false, - confirmLoading: false, - textView: '' - }); - this.props.refresh(); - }) - .catch(err => { - alert(`添加直播间失败:\n${err}`); - this.setState({ - visible: false, - confirmLoading: false, - textView: '' - }); - }) - }; - - handleCancel = () => { - this.setState({ - visible: false, - textView: '' - }); - }; - - textChange = (e: any) => { - this.setState({ - textView: e.target.value - }) - } - - render() { - const { visible, confirmLoading, ModalText, textView } = this.state; - return ( -
- -

{ModalText}

- - -
- ); - } -} - -export default AddRoomDialog; diff --git a/src/webapp/src/component/batch-add-room-dialog/index.tsx b/src/webapp/src/component/batch-add-room-dialog/index.tsx new file mode 100644 index 0000000..3fcca06 --- /dev/null +++ b/src/webapp/src/component/batch-add-room-dialog/index.tsx @@ -0,0 +1,449 @@ +import React, { useState, useEffect, useRef, useCallback } from 'react'; +import { + Modal, + Input, + Button, + Upload, + Progress, + Space, + Typography, + Radio, + Alert, + Switch, + message as antdMessage, +} from 'antd'; +import { + UploadOutlined, + CheckCircleOutlined, + CloseCircleOutlined, + MinusCircleOutlined, +} from '@ant-design/icons'; +import API from '../../utils/api'; +import { sseManager } from '../../utils/sse'; + +const { TextArea } = Input; +const { Text } = Typography; + +const api = new API(); + +interface BatchResultItem { + url: string; + status: 'pending' | 'success' | 'error'; + error?: string; +} + +interface Props { + visible: boolean; + onClose: () => void; + onSuccess: () => void; +} + +type InputMode = 'single' | 'batch'; +type ImportMode = 'text' | 'file'; + +function generateBatchId(): string { + return 'batch_' + Date.now() + '_' + Math.random().toString(36).substring(2, 10); +} + +const MAX_BATCH_URLS = 50; + +const AddRoomDialog: React.FC = ({ visible, onClose, onSuccess }) => { + const [inputMode, setInputMode] = useState('single'); + const [importMode, setImportMode] = useState('text'); + const [singleUrl, setSingleUrl] = useState(''); + const [textInput, setTextInput] = useState(''); + const [listen, setListen] = useState(true); + const [notifyOnly, setNotifyOnly] = useState(false); + const [processing, setProcessing] = useState(false); + const [current, setCurrent] = useState(0); + const [total, setTotal] = useState(0); + const [results, setResults] = useState([]); + const [completed, setCompleted] = useState(false); + const [successCount, setSuccessCount] = useState(0); + const [failCount, setFailCount] = useState(0); + + const sseSubRef = useRef([]); + const sseTimeoutRef = useRef | null>(null); + + const parseUrls = useCallback((text: string): string[] => { + return text + .split(/[\r\n]+/) + .map(line => line.trim()) + .filter(line => line.length > 0 && !line.startsWith('#') && !line.startsWith('＃')); + }, []); + + const handleFileImport = useCallback((file: File) => { + const reader = new FileReader(); + reader.onload = (e) => { + const text = e.target?.result as string; + setTextInput(text); + setImportMode('text'); + }; + reader.onerror = () => { + antdMessage.error('文件读取失败'); + }; + reader.readAsText(file); + return false; + }, []); + + const cleanupSSE = useCallback(() => { + sseSubRef.current.forEach(id => sseManager.unsubscribe(id)); + sseSubRef.current = []; + if (sseTimeoutRef.current) { + clearTimeout(sseTimeoutRef.current); + sseTimeoutRef.current = null; + } + }, []); + + const subscribeSSE = useCallback((batchId: string) => { + const progressSubId = sseManager.subscribe( + batchId, + 'batch_progress', + (message) => { + const data = message.data; + setCurrent(data.index); + setResults(prev => prev.map((item, i) => + i === data.index + ? { ...item, status: data.success ? 'success' : 'error', error: data.error } + : item + )); + } + ); + + const completeSubId = sseManager.subscribe( + batchId, + 'batch_complete', + (message) => { + const data = message.data; + setProcessing(false); + setCompleted(true); + setSuccessCount(data.success_count); + setFailCount(data.fail_count); + cleanupSSE(); + + if (data.fail_count === 0) { + if (data.persist_error) { + antdMessage.warning('添加成功，但配置保存失败，重启后可能丢失'); + } else { + antdMessage.success(`添加成功`); + } + } else if (data.success_count === 0) { + antdMessage.error(`添加失败`); + } else { + antdMessage.warning(`添加完成，成功 ${data.success_count} 个，失败 ${data.fail_count} 个`); + } + onSuccess(); + } + ); + + sseSubRef.current = [progressSubId, completeSubId]; + + // 5 分钟超时保护：后端异常时避免 UI 永久卡住 + sseTimeoutRef.current = setTimeout(() => { + antdMessage.error('批量添加超时，请刷新页面查看结果'); + setProcessing(false); + setCompleted(true); + cleanupSSE(); + }, 5 * 60 * 1000); + }, [cleanupSSE, onSuccess]); + + useEffect(() => { + return cleanupSSE; + }, [cleanupSSE]); + + const handleStart = async () => { + if (processing) return; + + // 根据模式获取 URL 列表 + let urls: string[]; + if (inputMode === 'single') { + const trimmed = singleUrl.trim(); + if (!trimmed) { + antdMessage.warning('请输入直播间地址'); + return; + } + urls = [trimmed]; + } else { + urls = parseUrls(textInput); + if (urls.length === 0) { + antdMessage.warning('请输入至少一个直播间地址'); + return; + } + if (urls.length > MAX_BATCH_URLS) { + antdMessage.warning(`单次最多添加 ${MAX_BATCH_URLS} 个直播间，当前 ${urls.length} 个`); + return; + } + } + + const batchId = generateBatchId(); + + setProcessing(true); + setCompleted(false); + setCurrent(-1); + setTotal(urls.length); + setResults(urls.map(url => ({ url, status: 'pending' as const }))); + setSuccessCount(0); + setFailCount(0); + + subscribeSSE(batchId); + + try { + await api.batchAddRooms(urls, listen, notifyOnly, batchId); + } catch (err: any) { + antdMessage.error('请求失败: ' + (err?.message || err)); + cleanupSSE(); + setProcessing(false); + } + }; + + const handleClose = () => { + if (processing) { + Modal.confirm({ + title: '确认关闭', + content: '关闭后前端不再显示进度，后端会继续完成剩余添加。确认关闭？', + onOk: () => { + cleanupSSE(); + resetState(); + onClose(); + }, + }); + return; + } + resetState(); + onClose(); + }; + + const resetState = () => { + cleanupSSE(); + setInputMode('single'); + setImportMode('text'); + setSingleUrl(''); + setTextInput(''); + setListen(true); + setNotifyOnly(false); + setProcessing(false); + setCurrent(0); + setTotal(0); + setResults([]); + setCompleted(false); + setSuccessCount(0); + setFailCount(0); + }; + + const progressPercent = total > 0 && current >= 0 ? Math.round(((current + 1) / total) * 100) : 0; + const validUrlCount = inputMode === 'single' + ? (singleUrl.trim() ? 1 : 0) + : parseUrls(textInput).length; + + // 单个模式下，成功/失败后显示简要结果而非进度列表 + const isSingleMode = inputMode === 'single' || total === 1; + + return ( + + + {!completed && ( + + )} + + ) + } + > + {/* 输入区域（仅未处理/未完成时显示） */} + {!processing && !completed && ( + <> + {/* 模式切换 */} +
+ { + setInputMode(e.target.value); + setImportMode('text'); + }} + size="small" + > + 单个添加 + 批量添加 + + {inputMode === 'batch' && ( + setImportMode(e.target.value)} + size="small" + > + 手动输入 + 导入TXT + + )} +
+ + {/* 文件导入按钮 */} + {inputMode === 'batch' && importMode === 'file' && ( + + + + )} + + {/* URL 输入 */} + {inputMode === 'single' ? ( + setSingleUrl(e.target.value)} + placeholder="输入直播间地址，如 https://live.bilibili.com/12345" + onPressEnter={handleStart} + /> + ) : ( + <> + +