From 277540ca170a67c8136c7f66782f9ae9e96a8b23 Mon Sep 17 00:00:00 2001 From: Magic-Xin Date: Tue, 22 Sep 2026 18:27:07 -0400 Subject: [PATCH 1/2] fix(bili): restore live room parsing and protect login data --- plugins/bili.go | 142 +++++++++++++++++++++++++++++-------------- plugins/bili_test.go | 121 ++++++++++++++++++++++++++++++++++++ 2 files changed, 219 insertions(+), 44 deletions(-) create mode 100644 plugins/bili_test.go diff --git a/plugins/bili.go b/plugins/bili.go index 9dd05a0..b5b8e33 100644 --- a/plugins/bili.go +++ b/plugins/bili.go @@ -17,6 +17,7 @@ import ( "net/http" "net/url" "os" + "path/filepath" "regexp" "strconv" "strings" @@ -206,65 +207,94 @@ func (*Bili) iToS(i int64) string { return fmt.Sprintf("%d", i) } -func (b *Bili) getLiveData(roomId string) *LiveData { - const api = "https://api.live.bilibili.com/xlive/web-room/v1/index/getInfoByRoom?room_id=" - req, err := http.NewRequest("GET", api+roomId, nil) +func (b *Bili) getLiveData(roomID string) *LiveData { + return b.getLiveDataFromAPI(roomID, "https://api.live.bilibili.com") +} + +type biliAPIResponse[T any] struct { + Code int `json:"code"` + Message string `json:"message"` + Data T `json:"data"` +} + +type biliRoomInfo struct { + UID int64 `json:"uid"` + RoomID int64 `json:"room_id"` + ShortID int64 `json:"short_id"` + Title string `json:"title"` + Keyframe string `json:"keyframe"` + AreaName string `json:"area_name"` + ParentAreaName string `json:"parent_area_name"` + LiveStatus int `json:"live_status"` + Online int64 `json:"online"` +} + +type biliMasterInfo struct { + Info struct { + Name string `json:"uname"` + } `json:"info"` +} + +func fetchBiliLiveAPI[T any](endpoint string) (T, error) { + var result biliAPIResponse[T] + req, err := http.NewRequest(http.MethodGet, endpoint, nil) if err != nil { - log.Printf("Live url parser request error: %v", err) - return nil + return result.Data, err } + req.Header.Set("User-Agent", "MacArthurGo/1.0") resp, err := essentials.HTTPClient.Do(req) if err != nil { - log.Printf("Live url parser response error: %v", err) - return nil + return result.Data, err } - defer func(Body io.ReadCloser) { - err := Body.Close() - if err != nil { - log.Printf("Live url parser close error: %v", err) - } - }(resp.Body) - body, err := io.ReadAll(resp.Body) - if err != nil { - log.Printf("Live url parser read body error: %v", err) - return nil + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return result.Data, fmt.Errorf("HTTP %d", resp.StatusCode) + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return result.Data, err } + if result.Code != 0 { + return result.Data, fmt.Errorf("Bilibili code %d: %s", result.Code, result.Message) + } + return result.Data, nil +} - var i any - err = json.Unmarshal(body, &i) +func (b *Bili) getLiveDataFromAPI(roomID, apiBase string) *LiveData { + room, err := fetchBiliLiveAPI[biliRoomInfo](apiBase + "/room/v1/Room/get_info?room_id=" + url.QueryEscape(roomID)) if err != nil { - log.Printf("Live url parser unmarshal error: %v", err) + log.Printf("Live room parser error: %v", err) return nil } - ctx := i.(map[string]any) - if ctx["code"].(float64) != 0 { + if room.RoomID == 0 || room.Title == "" { + log.Printf("Live room parser error: incomplete room data for %s", roomID) return nil } - data := &LiveData{ - Title: ctx["data"].(map[string]any)["room_info"].(map[string]any)["title"].(string), - User: ctx["data"].(map[string]any)["anchor_info"].(map[string]any)["base_info"].(map[string]any)["uname"].(string), - ThumbnailUrl: ctx["data"].(map[string]any)["room_info"].(map[string]any)["keyframe"].(string), + data := &LiveData{Title: room.Title, ThumbnailUrl: room.Keyframe} + if room.UID != 0 { + master, err := fetchBiliLiveAPI[biliMasterInfo](apiBase + "/live_user/v1/Master/info?uid=" + strconv.FormatInt(room.UID, 10)) + if err != nil { + log.Printf("Live master parser error: %v", err) + } else { + data.User = master.Info.Name + } } - if shortId := ctx["data"].(map[string]any)["room_info"].(map[string]any)["short_id"].(float64); shortId != 0 { - data.RoomId = fmt.Sprintf("短号: %d", int64(shortId)) - data.Url = "https://live.bilibili.com/" + fmt.Sprintf("%d", int64(shortId)) + if room.ShortID != 0 { + data.RoomId = fmt.Sprintf("短号: %d", room.ShortID) + data.Url = "https://live.bilibili.com/" + strconv.FormatInt(room.ShortID, 10) } else { - data.RoomId = fmt.Sprintf("房间号: %d", int64(ctx["data"].(map[string]any)["room_info"].(map[string]any)["room_id"].(float64))) - data.Url = "https://live.bilibili.com/" + fmt.Sprintf("%d", int64(ctx["data"].(map[string]any)["room_info"].(map[string]any)["room_id"].(float64))) + data.RoomId = fmt.Sprintf("房间号: %d", room.RoomID) + data.Url = "https://live.bilibili.com/" + strconv.FormatInt(room.RoomID, 10) } - - areaName := ctx["data"].(map[string]any)["room_info"].(map[string]any)["area_name"].(string) - parentAreaName := ctx["data"].(map[string]any)["room_info"].(map[string]any)["parent_area_name"].(string) - if areaName != parentAreaName { - data.AreaName = parentAreaName + "-" + areaName - } else { - data.AreaName = parentAreaName + data.AreaName = room.AreaName + if room.AreaName == "" { + data.AreaName = room.ParentAreaName + } else if room.ParentAreaName != "" && room.ParentAreaName != room.AreaName { + data.AreaName = room.ParentAreaName + "-" + room.AreaName } - - if ctx["data"].(map[string]any)["room_info"].(map[string]any)["live_status"].(float64) == 1 { - data.Status = "直播中 " + b.iToS(int64(ctx["data"].(map[string]any)["room_info"].(map[string]any)["online"].(float64))) + "人气" + if room.LiveStatus == 1 { + data.Status = "直播中\t" + b.iToS(room.Online) + "人气" } else { data.Status = "未开播" } @@ -375,13 +405,33 @@ func (a *AISummarize) SaveLoginInfo() { return } - err = os.WriteFile("bili_info.dat", data, 0644) + err = saveBiliLoginInfo("bili_info.dat", data) if err != nil { log.Printf("Bili Save Login Info Error: %s", err) } } +func saveBiliLoginInfo(path string, data []byte) error { + file, err := os.CreateTemp(filepath.Dir(path), ".bili_info-*") + if err != nil { + return err + } + defer os.Remove(file.Name()) + if _, err := file.Write(data); err != nil { + file.Close() + return err + } + if err := file.Close(); err != nil { + return err + } + return os.Rename(file.Name(), path) +} + func (a *AISummarize) LoadLoginInfo() { + if err := os.Chmod("bili_info.dat", 0600); err != nil && !os.IsNotExist(err) { + log.Printf("Bili Login Info Permission Error: %s", err) + return + } data, err := os.ReadFile("bili_info.dat") if err != nil { a.loginMsg = "未找到 B 站登录信息，请先登录" @@ -562,9 +612,13 @@ func (v *VideoData) ToArrayMessage() []cqcode.ArrayMessage { func (l *LiveData) ToArrayMessage() []cqcode.ArrayMessage { var messageArray []cqcode.ArrayMessage - messageArray = append(messageArray, *cqcode.Image(l.ThumbnailUrl + "\n")) + if l.ThumbnailUrl != "" { + messageArray = append(messageArray, *cqcode.Image(l.ThumbnailUrl)) + } messageArray = append(messageArray, *cqcode.Text(l.Title + "\n")) - messageArray = append(messageArray, *cqcode.Text("主播: " + l.User + "\n")) + if l.User != "" { + messageArray = append(messageArray, *cqcode.Text("主播: " + l.User + "\n")) + } messageArray = append(messageArray, *cqcode.Text(l.RoomId + "\n")) messageArray = append(messageArray, *cqcode.Text("分区: " + l.AreaName + "\n")) messageArray = append(messageArray, *cqcode.Text(l.Status + "\n")) diff --git a/plugins/bili_test.go b/plugins/bili_test.go new file mode 100644 index 0000000..0d744a3 --- /dev/null +++ b/plugins/bili_test.go @@ -0,0 +1,121 @@ +package plugins + +import ( + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" +) + +func TestBiliLiveData(t *testing.T) { + tests := []struct { + name string + roomReply string + masterCode int + wantRoom string + wantURL string + wantStatus string + wantName string + wantImage bool + }{ + { + name: "live short room", + roomReply: `{"code":0,"data":{"uid":1472906636,"room_id":22727121,"short_id":6655,"title":"直播标题","keyframe":"https://example.com/cover.jpg","area_name":"游戏","parent_area_name":"娱乐","live_status":1,"online":12601}}`, + wantRoom: "短号: 6655", + wantURL: "https://live.bilibili.com/6655", + wantStatus: "直播中\t1.3万人气", + wantName: "ywwuyi", + wantImage: true, + }, + { + name: "offline room with unavailable master", + roomReply: `{"code":0,"data":{"uid":1472906636,"room_id":22727121,"title":"直播标题","area_name":"游戏","parent_area_name":"游戏","live_status":0}}`, + masterCode: http.StatusServiceUnavailable, + wantRoom: "房间号: 22727121", + wantURL: "https://live.bilibili.com/22727121", + wantStatus: "未开播", + }, + { + name: "room API rejects request", + roomReply: `{"code":-352,"message":"-352"}`, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("User-Agent") != "MacArthurGo/1.0" { + t.Errorf("User-Agent = %q", r.Header.Get("User-Agent")) + } + switch r.URL.Path { + case "/room/v1/Room/get_info": + if got := r.URL.Query().Get("room_id"); got != "6655" { + t.Errorf("room_id = %q", got) + } + fmt.Fprint(w, tt.roomReply) + case "/live_user/v1/Master/info": + if got := r.URL.Query().Get("uid"); got != "1472906636" { + t.Errorf("uid = %q", got) + } + if tt.masterCode != 0 { + w.WriteHeader(tt.masterCode) + return + } + fmt.Fprint(w, `{"code":0,"data":{"info":{"uname":"ywwuyi"}}}`) + default: + t.Errorf("unexpected path %q", r.URL.Path) + } + })) + defer server.Close() + + got := (&Bili{}).getLiveDataFromAPI("6655", server.URL) + if tt.wantRoom == "" { + if got != nil { + t.Fatalf("getLiveDataFromAPI() = %#v, want nil", got) + } + return + } + if got == nil { + t.Fatal("getLiveDataFromAPI() = nil") + } + if got.RoomId != tt.wantRoom || got.Url != tt.wantURL || got.Status != tt.wantStatus || got.User != tt.wantName { + t.Errorf("live data = %#v", got) + } + messages := got.ToArrayMessage() + imageCount := 0 + for _, message := range messages { + if message.Type == "image" { + imageCount++ + } + } + if (imageCount == 1) != tt.wantImage { + t.Errorf("image count = %d, want image = %t", imageCount, tt.wantImage) + } + }) + } +} + +func TestSaveBiliLoginInfoRestrictsExistingFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "bili_info.dat") + if err := os.WriteFile(path, []byte("old"), 0644); err != nil { + t.Fatal(err) + } + if err := saveBiliLoginInfo(path, []byte("new")); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if got := info.Mode().Perm(); got != 0600 { + t.Errorf("file permissions = %#o, want 0600", got) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(data) != "new" { + t.Errorf("file contents = %q", data) + } +} From 833689aac2f25e0b3e671a19c4bd89ff7b0e26c2 Mon Sep 17 00:00:00 2001 From: Magic-Xin Date: Tue, 22 Sep 2026 18:27:39 -0400 Subject: [PATCH 2/2] fix: preserve stats updates and unblock shutdown callbacks --- client/ws_client.go | 17 ++++++++++---- client/ws_client_test.go | 50 ++++++++++++++++++++++++++++++++++++++++ plugins/statics.go | 36 ++++++++++++++++++----------- plugins/statics_test.go | 21 +++++++++++++++++ 4 files changed, 107 insertions(+), 17 deletions(-) create mode 100644 client/ws_client_test.go diff --git a/client/ws_client.go b/client/ws_client.go index 07ec4d5..d2ea1a9 100644 --- a/client/ws_client.go +++ b/client/ws_client.go @@ -107,10 +107,19 @@ func (c *Client) Run(ctx context.Context) error { workersDone := c.startWorkers(workerCtx) defer func() { cancelWorkers() - select { - case <-workersDone: - case <-time.After(writeTimeout): - log.Printf("Timed out waiting for event workers to stop") + timer := time.NewTimer(writeTimeout) + defer timer.Stop() + for { + select { + case <-workersDone: + return + case <-c.outbound: + // A callback may be sending while the WebSocket writer has stopped. + // Drain queued actions so it can release its plugin lock. + case <-timer.C: + log.Printf("Timed out waiting for event workers to stop") + return + } } }() diff --git a/client/ws_client_test.go b/client/ws_client_test.go new file mode 100644 index 0000000..7554c9c --- /dev/null +++ b/client/ws_client_test.go @@ -0,0 +1,50 @@ +package client + +import ( + "context" + "testing" + "time" +) + +func TestRunDrainsOutboundDuringShutdown(t *testing.T) { + started := make(chan struct{}) + completed := make(chan struct{}) + bot := New(Options{ + Address: "ws://[", + QueueSize: 1, + Workers: 1, + ReconnectMin: time.Hour, + ReconnectMax: time.Hour, + EventHandler: func(_ []byte, send chan<- []byte) error { + send <- []byte("first") + close(started) + send <- []byte("second") + close(completed) + return nil + }, + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + runDone := make(chan error, 1) + go func() { runDone <- bot.Run(ctx) }() + bot.events <- []byte("event") + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("event handler did not start") + } + cancel() + select { + case <-completed: + case <-time.After(time.Second): + t.Fatal("event handler remained blocked on outbound send") + } + select { + case err := <-runDone: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("Run remained blocked on event worker") + } +} diff --git a/plugins/statics.go b/plugins/statics.go index e48a251..cc98af4 100644 --- a/plugins/statics.go +++ b/plugins/statics.go @@ -52,8 +52,9 @@ type staticsStore struct { root string retention int data map[string]map[int64]*groupStats - dirtyDates map[string]bool + dirtyDates map[string]uint64 mu sync.RWMutex + flushMu sync.Mutex flushOnce sync.Once } @@ -428,7 +429,7 @@ func (s *staticsStore) incrementMessage(dateKey string, groupId int64, hour int) defer s.mu.Unlock() stats := s.ensureGroupStats(dateKey, groupId) stats.Hourly[hour]++ - s.dirtyDates[dateKey] = true + s.dirtyDates[dateKey]++ } func (s *staticsStore) addWordCounts(dateKey string, groupId int64, freq map[string]int) { @@ -444,7 +445,7 @@ func (s *staticsStore) addWordCounts(dateKey string, groupId int64, freq map[str for word, count := range freq { stats.Words[word] += int64(count) } - s.dirtyDates[dateKey] = true + s.dirtyDates[dateKey]++ } func (s *staticsStore) getHourly(dateKey string, groupId int64) [24]int64 { @@ -500,7 +501,7 @@ func newStaticsStore(dir string, retention int) (*staticsStore, error) { root: dir, retention: retention, data: make(map[string]map[int64]*groupStats), - dirtyDates: make(map[string]bool), + dirtyDates: make(map[string]uint64), } if err := store.loadRecent(); err != nil { return nil, err @@ -595,13 +596,15 @@ func (s *staticsStore) snapshotDirtyDates() []string { } func (s *staticsStore) flushDate(date string) error { + s.flushMu.Lock() + defer s.flushMu.Unlock() + s.mu.RLock() groups := s.data[date] + version := s.dirtyDates[date] if len(groups) == 0 { s.mu.RUnlock() - s.mu.Lock() - delete(s.dirtyDates, date) - s.mu.Unlock() + s.clearDirtyDate(date, version) return nil } snapshot := make(map[int64]*groupStats, len(groups)) @@ -613,9 +616,7 @@ func (s *staticsStore) flushDate(date string) error { } s.mu.RUnlock() if len(snapshot) == 0 { - s.mu.Lock() - delete(s.dirtyDates, date) - s.mu.Unlock() + s.clearDirtyDate(date, version) return nil } serializable := make(map[string]*groupStats, len(snapshot)) @@ -634,13 +635,22 @@ func (s *staticsStore) flushDate(date string) error { if err := os.Rename(tmpPath, finalPath); err != nil { return err } - s.mu.Lock() - delete(s.dirtyDates, date) - s.mu.Unlock() + s.clearDirtyDate(date, version) return nil } +func (s *staticsStore) clearDirtyDate(date string, version uint64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.dirtyDates[date] == version { + delete(s.dirtyDates, date) + } +} + func (s *staticsStore) cleanupExpired() { + s.flushMu.Lock() + defer s.flushMu.Unlock() + cutoff := time.Now().AddDate(0, 0, -s.retention) s.mu.Lock() for date := range s.data { diff --git a/plugins/statics_test.go b/plugins/statics_test.go index b94f75b..6608e1f 100644 --- a/plugins/statics_test.go +++ b/plugins/statics_test.go @@ -23,3 +23,24 @@ func TestSegmentWordsFiltersShortWordsAndParticles(t *testing.T) { } } } + +func TestStaticsStoreKeepsUpdatesAfterSnapshot(t *testing.T) { + store, err := newStaticsStore(t.TempDir(), 2) + if err != nil { + t.Fatal(err) + } + const date = "2026-09-22" + store.incrementMessage(date, 1, 12) + versionAtSnapshot := store.dirtyDates[date] + store.addWordCounts(date, 1, map[string]int{"测试": 1}) + store.clearDirtyDate(date, versionAtSnapshot) + if _, ok := store.dirtyDates[date]; !ok { + t.Fatal("new update was cleared by an older snapshot") + } + if err := store.flushDate(date); err != nil { + t.Fatal(err) + } + if _, ok := store.dirtyDates[date]; ok { + t.Fatal("completed flush left date dirty") + } +}