From 8f64f961ac1563b8f46c029358b5daec447bdd55 Mon Sep 17 00:00:00 2001
From: truewhile <62226914+truewhile@users.noreply.github.com>
Date: Sat, 12 Sep 2026 11:02:24 +0800
Subject: [PATCH] =?UTF-8?q?=E7=BC=93=E5=AD=98=E5=BC=B9=E5=B9=95=E6=8A=93?=
=?UTF-8?q?=E5=8F=96=E7=BB=93=E6=9E=9C=E5=B9=B6=E5=90=88=E5=B9=B6=E5=B9=B6?=
=?UTF-8?q?=E5=8F=91=E8=AF=B7=E6=B1=82?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
go.mod | 2 +-
internal/service/danmaku_match_test.go | 35 +++++
internal/service/danmaku_service.go | 168 ++++++++++++++++++++++-
internal/service/danmaku_service_test.go | 78 +++++++++++
4 files changed, 277 insertions(+), 6 deletions(-)
diff --git a/go.mod b/go.mod
index 0fe802c..8a57a16 100644
--- a/go.mod
+++ b/go.mod
@@ -20,6 +20,7 @@ require (
go.uber.org/zap v1.27.0
golang.org/x/crypto v0.49.0
golang.org/x/image v0.37.0
+ golang.org/x/sync v0.20.0
golang.org/x/sys v0.42.0
golang.org/x/time v0.15.0
gopkg.in/yaml.v3 v3.0.1
@@ -88,7 +89,6 @@ require (
golang.org/x/arch v0.25.0 // indirect
golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
golang.org/x/net v0.52.0 // indirect
- golang.org/x/sync v0.20.0 // indirect
golang.org/x/text v0.35.0 // indirect
google.golang.org/protobuf v1.36.11 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
diff --git a/internal/service/danmaku_match_test.go b/internal/service/danmaku_match_test.go
index b31881e..5afca8b 100644
--- a/internal/service/danmaku_match_test.go
+++ b/internal/service/danmaku_match_test.go
@@ -824,3 +824,38 @@ func TestDanmakuFetchEmbyRemoteStreamFailedFallsBackToSearch(t *testing.T) {
require.Equal(t, "降级搜索番剧", res.AnimeTitle)
require.Contains(t, res.Raw, "降级搜索弹幕")
}
+
+// 即使刮削元数据完整，也应先走准确率最高的 hash 层。
+func TestDanmakuFetchPrefersHashOverCompleteMetadata(t *testing.T) {
+ videoPath, wantHash := writeDanmakuTestVideo(t, "测试动画.第01话.mkv")
+ var seen string
+ official := danmakuOfficialServer(t,
+ `{"success":true,"isMatched":true,"matches":[{"episodeId":25484,"animeId":1001,"animeTitle":"官方测试动画","episodeTitle":"第1话"}]}`,
+ `hash命中`,
+ &seen)
+ overrideDanmakuOfficialBase(t, official.URL)
+
+ source := newDanmakuSourceServerWithSearch(t,
+ `{"hasMore":false,"animes":[{"animeId":2002,"animeTitle":"官方测试动画","episodes":[{"episodeId":25484,"episodeTitle":"第1话"}]}]}`)
+
+ svc := newDanmakuTestService(t)
+ ctx := context.Background()
+ require.NoError(t, svc.repo.Setting.Set(ctx, DanmakuSourceKey, source.URL()))
+
+ m := model.Media{
+ Title: "测试动画",
+ Path: videoPath,
+ SizeBytes: 123,
+ EpisodeNum: 1,
+ EpisodeTitle: "起始与终结的序章",
+ Year: 2011,
+ }
+ m.ID = "hash-first"
+ require.NoError(t, svc.repo.DB.Create(&m).Error)
+
+ res, err := svc.Fetch(ctx, "hash-first", "", "")
+ require.NoError(t, err)
+ require.Equal(t, "hash", res.MatchMode)
+ require.NotEmpty(t, res.Raw)
+ require.Contains(t, seen, wantHash)
+}
diff --git a/internal/service/danmaku_service.go b/internal/service/danmaku_service.go
index 9df30a4..959502c 100644
--- a/internal/service/danmaku_service.go
+++ b/internal/service/danmaku_service.go
@@ -20,6 +20,8 @@ import (
"time"
"unicode"
+ "golang.org/x/sync/singleflight"
+
"go.uber.org/zap"
"github.com/truewhile/MeBox/internal/model"
@@ -136,6 +138,12 @@ type DanmakuService struct {
hashCacheMu sync.Mutex
hashCache map[string]string // stamp → 16MB-prefix MD5
+
+ // resultCache 缓存整条弹幕抓取结果，避免同一集重复播放时重走
+ // 「16MB 哈希 + 上游搜索 + 评论拉取」这条高延迟链路。
+ resultCacheMu sync.Mutex
+ resultCache map[string]danmakuResultCacheEntry
+ fetchGroup singleflight.Group
}
func danmakuHTTPClient() *http.Client {
@@ -147,10 +155,11 @@ func NewDanmakuService(log *zap.Logger, repo *repository.Container) *DanmakuServ
log = zap.NewNop()
}
return &DanmakuService{
- log: log,
- repo: repo,
- client: danmakuHTTPClient(),
- hashCache: make(map[string]string),
+ log: log,
+ repo: repo,
+ client: danmakuHTTPClient(),
+ hashCache: make(map[string]string),
+ resultCache: make(map[string]danmakuResultCacheEntry),
}
}
@@ -170,6 +179,121 @@ func (s *DanmakuService) SetRemoteMediaResolver(resolve DanmakuRemoteMediaResolv
}
}
+// 弹幕抓取结果缓存：同一集在 TTL 内重复播放时直接返回，避免重复执行
+// 「16MB 前置哈希 → 上游搜索 → 评论拉取」这条高延迟链路。弹幕库内容变化
+// 很慢，而用户通常会在短时间内反复切集/回看，因此 TTL 取 6 小时。
+const (
+ danmakuResultCacheTTL = 6 * time.Hour
+ danmakuResultCacheMaxEntries = 64
+ // 候选列表由上游搜索决定，相对稳定但可能随源更新，缓存 10 分钟。
+ danmakuResultCacheCandidateTTL = 10 * time.Minute
+ // 空结果可能只是上游临时抖动，只短暂缓存，避免长时间看不到弹幕。
+ danmakuResultCacheEmptyTTL = time.Minute
+)
+
+type danmakuResultCacheEntry struct {
+ result *DanmakuFetchResult
+ expiresAt time.Time
+ storedAt time.Time
+}
+
+func danmakuResultCacheKey(source, mediaID, keyword, episodeID string, merge bool) string {
+ return strings.Join([]string{
+ strings.TrimSpace(source),
+ mediaID,
+ strings.TrimSpace(keyword),
+ strings.TrimSpace(episodeID),
+ strconv.FormatBool(merge),
+ }, "\x00")
+}
+
+func (s *DanmakuService) resultCacheGet(key string) (*DanmakuFetchResult, bool) {
+ if s == nil || key == "" {
+ return nil, false
+ }
+ now := time.Now()
+ s.resultCacheMu.Lock()
+ defer s.resultCacheMu.Unlock()
+ entry, ok := s.resultCache[key]
+ if !ok {
+ return nil, false
+ }
+ if now.After(entry.expiresAt) {
+ delete(s.resultCache, key)
+ return nil, false
+ }
+ return cloneDanmakuFetchResult(entry.result), true
+}
+
+func (s *DanmakuService) resultCachePut(key string, result *DanmakuFetchResult) {
+ if s == nil || key == "" || result == nil {
+ return
+ }
+ now := time.Now()
+ s.resultCacheMu.Lock()
+ defer s.resultCacheMu.Unlock()
+ if s.resultCache == nil {
+ s.resultCache = make(map[string]danmakuResultCacheEntry)
+ }
+ ttl := danmakuResultCacheTTLFor(result)
+ if ttl <= 0 {
+ return
+ }
+ if _, exists := s.resultCache[key]; !exists && len(s.resultCache) >= danmakuResultCacheMaxEntries {
+ oldestKey := ""
+ var oldest time.Time
+ for k, entry := range s.resultCache {
+ if oldestKey == "" || entry.storedAt.Before(oldest) {
+ oldestKey, oldest = k, entry.storedAt
+ }
+ }
+ delete(s.resultCache, oldestKey)
+ }
+ s.resultCache[key] = danmakuResultCacheEntry{
+ result: cloneDanmakuFetchResult(result),
+ expiresAt: now.Add(ttl),
+ storedAt: now,
+ }
+}
+
+// danmakuResultCacheTTLFor 按结果完整性选择缓存时长：拿到弹幕正文才值得
+// 长缓存；只有候选列表时短缓存；空结果只缓存一分钟。
+func danmakuResultCacheTTLFor(result *DanmakuFetchResult) time.Duration {
+ if result == nil {
+ return 0
+ }
+ if strings.TrimSpace(result.Raw) != "" {
+ return danmakuResultCacheTTL
+ }
+ if len(result.Candidates) > 0 {
+ return danmakuResultCacheCandidateTTL
+ }
+ return danmakuResultCacheEmptyTTL
+}
+
+// cloneDanmakuFetchResult 深拷贝切片字段，避免缓存命中后调用方修改共享数据。
+func cloneDanmakuFetchResult(in *DanmakuFetchResult) *DanmakuFetchResult {
+ if in == nil {
+ return nil
+ }
+ out := *in
+ out.Candidates = cloneDanmakuAnimeList(in.Candidates)
+ out.Alternatives = cloneDanmakuAnimeList(in.Alternatives)
+ return &out
+}
+
+func cloneDanmakuAnimeList(in []DanmakuAnime) []DanmakuAnime {
+ if in == nil {
+ return nil
+ }
+ out := make([]DanmakuAnime, len(in))
+ for i, anime := range in {
+ out[i] = anime
+ out[i].Episodes = append([]DanmakuEpisode(nil), anime.Episodes...)
+ }
+ return out
+}
+
// Config reads danmaku settings from the runtime settings table.
func (s *DanmakuService) Config(ctx context.Context) DanmakuRenderConfig {
cfg := DanmakuRenderConfig{
@@ -257,8 +381,42 @@ func (s *DanmakuService) Fetch(ctx context.Context, mediaID, keyword, episodeID
return s.FetchWithOptions(ctx, mediaID, keyword, episodeID, DanmakuFetchOptions{})
}
-// FetchWithOptions 是 Fetch 的带偏好版本。
+// FetchWithOptions 是 Fetch 的带偏好版本。结果按「配置源 + 媒体 + 关键词 +
+// 指定集 + 合并开关」缓存，并用 singleflight 合并并发请求，避免同一集被重复抓取。
func (s *DanmakuService) FetchWithOptions(ctx context.Context, mediaID, keyword, episodeID string, opts DanmakuFetchOptions) (*DanmakuFetchResult, error) {
+ if s == nil {
+ return nil, errors.New("danmaku service unavailable")
+ }
+ cfg := s.Config(ctx)
+ if !cfg.Enabled {
+ return &DanmakuFetchResult{DanmakuRenderConfig: cfg, SourceType: "auto"}, nil
+ }
+ key := danmakuResultCacheKey(cfg.Source, mediaID, keyword, episodeID, opts.MergeSources)
+ if cached, ok := s.resultCacheGet(key); ok {
+ return cached, nil
+ }
+ value, err, _ := s.fetchGroup.Do(key, func() (any, error) {
+ // 等待期间可能已有同一 key 的请求写入缓存。
+ if cached, ok := s.resultCacheGet(key); ok {
+ return cached, nil
+ }
+ res, err := s.fetchWithOptionsUncached(ctx, mediaID, keyword, episodeID, opts)
+ if err != nil {
+ // 与原实现一致：失败时仍把已填充的渲染配置/匹配信息交给调用方。
+ return res, err
+ }
+ s.resultCachePut(key, res)
+ return res, nil
+ })
+ res, _ := value.(*DanmakuFetchResult)
+ if err != nil {
+ return cloneDanmakuFetchResult(res), err
+ }
+ return cloneDanmakuFetchResult(res), nil
+}
+
+// fetchWithOptionsUncached 是未命中缓存时执行的原始抓取流程。
+func (s *DanmakuService) fetchWithOptionsUncached(ctx context.Context, mediaID, keyword, episodeID string, opts DanmakuFetchOptions) (*DanmakuFetchResult, error) {
res := &DanmakuFetchResult{DanmakuRenderConfig: s.Config(ctx), SourceType: "auto"}
if !res.Enabled {
return res, nil
diff --git a/internal/service/danmaku_service_test.go b/internal/service/danmaku_service_test.go
index 2d7a390..48f48a8 100644
--- a/internal/service/danmaku_service_test.go
+++ b/internal/service/danmaku_service_test.go
@@ -6,6 +6,8 @@ import (
"net/http"
"net/http/httptest"
"path/filepath"
+ "sync"
+ "sync/atomic"
"testing"
"github.com/glebarez/sqlite"
@@ -326,3 +328,79 @@ func TestDanmakuFetchDetectsJSONSource(t *testing.T) {
require.Equal(t, "xml", res2.SourceType)
require.Contains(t, res2.Raw, "弹幕A")
}
+
+// 同一集在缓存 TTL 内重复请求时不应再访问上游（搜索和评论都只发一次）。
+func TestDanmakuFetchCachesRepeatedRequests(t *testing.T) {
+ var searchCalls, commentCalls int32
+ mux := http.NewServeMux()
+ mux.HandleFunc("/api/v2/search/episodes", func(w http.ResponseWriter, r *http.Request) {
+ atomic.AddInt32(&searchCalls, 1)
+ w.Header().Set("Content-Type", "application/json")
+ fmt.Fprint(w, `{"hasMore":false,"animes":[{"animeId":1001,"animeTitle":"测试动画","episodes":[{"episodeId":25484,"episodeTitle":"第1话"}]}]}`)
+ })
+ mux.HandleFunc("/api/v2/comment/25484", func(w http.ResponseWriter, r *http.Request) {
+ atomic.AddInt32(&commentCalls, 1)
+ w.Header().Set("Content-Type", "application/xml")
+ fmt.Fprint(w, `缓存测试`)
+ })
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ svc := newDanmakuTestService(t)
+ ctx := context.Background()
+ require.NoError(t, svc.repo.Setting.Set(ctx, DanmakuSourceKey, srv.URL))
+ seedDanmakuMedia(t, svc, "cache-media", "测试动画", "", 1)
+
+ first, err := svc.Fetch(ctx, "cache-media", "", "")
+ require.NoError(t, err)
+ require.NotEmpty(t, first.Raw)
+
+ second, err := svc.Fetch(ctx, "cache-media", "", "")
+ require.NoError(t, err)
+ require.Equal(t, first.Raw, second.Raw)
+ require.EqualValues(t, 1, atomic.LoadInt32(&searchCalls))
+ require.EqualValues(t, 1, atomic.LoadInt32(&commentCalls))
+}
+
+// 并发的同一集请求应被 singleflight 合并，上游只被访问一次。
+func TestDanmakuFetchCoalescesConcurrentRequests(t *testing.T) {
+ var searchCalls int32
+ mux := http.NewServeMux()
+ mux.HandleFunc("/api/v2/search/episodes", func(w http.ResponseWriter, r *http.Request) {
+ atomic.AddInt32(&searchCalls, 1)
+ w.Header().Set("Content-Type", "application/json")
+ fmt.Fprint(w, `{"hasMore":false,"animes":[{"animeId":1001,"animeTitle":"测试动画","episodes":[{"episodeId":25484,"episodeTitle":"第1话"}]}]}`)
+ })
+ mux.HandleFunc("/api/v2/comment/25484", func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/xml")
+ fmt.Fprint(w, `并发测试`)
+ })
+ srv := httptest.NewServer(mux)
+ defer srv.Close()
+
+ svc := newDanmakuTestService(t)
+ ctx := context.Background()
+ require.NoError(t, svc.repo.Setting.Set(ctx, DanmakuSourceKey, srv.URL))
+ seedDanmakuMedia(t, svc, "concurrent-media", "测试动画", "", 1)
+
+ const workers = 8
+ start := make(chan struct{})
+ errs := make(chan error, workers)
+ var wg sync.WaitGroup
+ for i := 0; i < workers; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ _, err := svc.Fetch(ctx, "concurrent-media", "", "")
+ errs <- err
+ }()
+ }
+ close(start)
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ require.NoError(t, err)
+ }
+ require.EqualValues(t, 1, atomic.LoadInt32(&searchCalls))
+}