diff --git a/cmd/server/main.go b/cmd/server/main.go index cc78097..6072c7c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -78,7 +78,6 @@ func main() { if err := services.Auth.SeedAdmin(context.Background()); err != nil { logger.Warn("seed admin failed", zap.Error(err)) } - services.Boot() router := buildRouter(cfg, logger, services) @@ -88,27 +87,29 @@ func main() { ReadHeaderTimeout: 15 * time.Second, } + ln, err := net.Listen("tcp", srv.Addr) + if err != nil { + logger.Fatal("listen failed", zap.String("addr", srv.Addr), zap.Error(err)) + } + localIP := getLocalIP() + logger.Info("server is ready", + zap.String("local", fmt.Sprintf("http://%s:%d", localIP, cfg.App.Port)), + zap.String("listen", srv.Addr), + ) go func() { - localIP := getLocalIP() - publicIP := getPublicIP(3 * time.Second) - if publicIP != "" { - logger.Info("server is ready", - zap.String("local", fmt.Sprintf("http://%s:%d", localIP, cfg.App.Port)), - zap.String("public", fmt.Sprintf("http://%s:%d", publicIP, cfg.App.Port)), - zap.String("listen", srv.Addr), - ) - } else { - logger.Info("server is ready", - zap.String("local", fmt.Sprintf("http://%s:%d", localIP, cfg.App.Port)), - zap.String("listen", srv.Addr), - ) - } - // 自动启动 Telegram 长轮询（无需公网 Webhook） - services.TelegramBot.StartPolling(context.Background()) - if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { logger.Fatal("listen failed", zap.Error(err)) } }() + go func() { + if publicIP := getPublicIP(3 * time.Second); publicIP != "" { + logger.Info("server public endpoint", + zap.String("public", fmt.Sprintf("http://%s:%d", publicIP, cfg.App.Port)), + ) + } + }() + go services.Boot() + go services.TelegramBot.StartPolling(context.Background()) // Graceful shutdown. stop := make(chan os.Signal, 1) @@ -200,6 +201,13 @@ func setNoCacheHeaders(c *gin.Context) { func shouldBypassSPAFallback(path string) bool { lower := strings.ToLower(path) + for _, exact := range []string{ + "/emby", + } { + if lower == exact { + return true + } + } for _, prefix := range []string{ "/api/", "/emby/", @@ -213,6 +221,10 @@ func shouldBypassSPAFallback(path string) bool { "/displaypreferences/", "/branding/", "/localization/", + "/startup/", + "/quickconnect/", + "/socket", + "/embywebsocket", } { if strings.HasPrefix(lower, prefix) { return true diff --git a/cmd/server/main_test.go b/cmd/server/main_test.go index 0472eaf..56a78db 100644 --- a/cmd/server/main_test.go +++ b/cmd/server/main_test.go @@ -72,14 +72,23 @@ func TestServeSPAServesAssetsImmutableAndBypassesAPIRoutes(t *testing.T) { t.Fatalf("asset Cache-Control = %q, want immutable", got) } - apiReq := httptest.NewRequest(http.MethodGet, "/api/missing", nil) - apiResp := httptest.NewRecorder() - router.ServeHTTP(apiResp, apiReq) - if apiResp.Code != http.StatusNotFound { - t.Fatalf("api fallback status = %d, want 404", apiResp.Code) - } - if strings.Contains(apiResp.Body.String(), "index") { - t.Fatalf("api route should not serve SPA index: %q", apiResp.Body.String()) + for _, path := range []string{ + "/api/missing", + "/emby", + "/emby/missing", + "/Startup/Configuration", + "/QuickConnect/Enabled", + "/embywebsocket", + } { + req := httptest.NewRequest(http.MethodGet, path, nil) + resp := httptest.NewRecorder() + router.ServeHTTP(resp, req) + if resp.Code != http.StatusNotFound { + t.Fatalf("%s fallback status = %d, want 404", path, resp.Code) + } + if strings.Contains(resp.Body.String(), "index") { + t.Fatalf("%s should not serve SPA index: %q", path, resp.Body.String()) + } } } diff --git a/internal/database/database.go b/internal/database/database.go index 75d4421..96f69e1 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -78,7 +78,10 @@ func AutoMigrate(db *gorm.DB) error { if err := enforceTelegramBindingOneToOne(db); err != nil { return err } - return ensurePerformanceIndexes(db) + if err := ensurePerformanceIndexes(db); err != nil { + return err + } + return ensureMediaSearchIndex(db) } func ensurePerformanceIndexes(db *gorm.DB) error { @@ -86,6 +89,8 @@ func ensurePerformanceIndexes(db *gorm.DB) error { `CREATE INDEX IF NOT EXISTS idx_media_library_created_active ON media(library_id, created_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_media_library_episode_active ON media(library_id, season_num, episode_num, created_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_media_series_active ON media(series_id, season_num, episode_num) WHERE deleted_at IS NULL`, + `CREATE INDEX IF NOT EXISTS idx_media_title_active ON media(title COLLATE NOCASE) WHERE deleted_at IS NULL`, + `CREATE INDEX IF NOT EXISTS idx_media_original_name_active ON media(original_name COLLATE NOCASE) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_favorites_user_media_active ON favorites(user_id, media_id) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_playback_histories_user_media_active ON playback_histories(user_id, media_id, watched_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_playback_histories_resume_active ON playback_histories(user_id, completed, watched_at DESC) WHERE deleted_at IS NULL`, @@ -99,6 +104,36 @@ func ensurePerformanceIndexes(db *gorm.DB) error { return nil } +func ensureMediaSearchIndex(db *gorm.DB) error { + if mediaSearchIndexNeedsRebuild(db) { + _ = db.Exec(`DROP TABLE IF EXISTS media_search_fts`).Error + } + if err := db.Exec(`CREATE VIRTUAL TABLE IF NOT EXISTS media_search_fts USING fts5(media_id UNINDEXED, title, original_name, path, genres, tokenize='trigram')`).Error; err != nil { + if fallbackErr := db.Exec(`CREATE VIRTUAL TABLE IF NOT EXISTS media_search_fts USING fts5(media_id UNINDEXED, title, original_name, path, genres, tokenize='unicode61')`).Error; fallbackErr != nil { + // FTS is an acceleration path. Some embedded SQLite builds may omit + // FTS5; keep startup working and let repository queries fall back to + // LIKE-based Chinese fuzzy search. + return nil + } + } + return nil +} + +func mediaSearchIndexNeedsRebuild(db *gorm.DB) bool { + var cols []struct { + Name string + } + if err := db.Raw(`PRAGMA table_info(media_search_fts)`).Scan(&cols).Error; err != nil || len(cols) == 0 { + return false + } + for _, col := range cols { + if col.Name == "genres" { + return false + } + } + return true +} + func enforceTelegramBindingOneToOne(db *gorm.DB) error { if !db.Migrator().HasTable(&model.TelegramBinding{}) { return nil diff --git a/internal/handler/cloud.go b/internal/handler/cloud.go index f100bd1..18b5f31 100644 --- a/internal/handler/cloud.go +++ b/internal/handler/cloud.go @@ -74,7 +74,7 @@ func cloudMountHandler(svc *service.Container) gin.HandlerFunc { path := service.BuildCloudLibraryPath(typ, in.Dir, in.DirPath) name := strings.TrimSpace(in.Name) if name == "" { - name = cloudMountLibraryName(typ, strings.TrimSpace(in.Dir)) + name = cloudMountLibraryName(typ, strings.TrimSpace(in.Dir), strings.TrimSpace(in.DirPath)) } mediaType := strings.TrimSpace(in.MediaType) if mediaType == "" || strings.EqualFold(mediaType, "auto") { @@ -120,7 +120,8 @@ func cloudMountHandler(svc *service.Container) gin.HandlerFunc { updates["type"] = mediaType lib.Type = mediaType } - if name != "" && name != lib.Name && !strings.Contains(lib.Name, " · ") { + currentDisplayName, _ := service.CloudLibraryDisplayName(*lib) + if name != "" && name != lib.Name && (currentDisplayName == "" || currentDisplayName != name || strings.Contains(lib.Name, " · ")) { updates["name"] = name lib.Name = name } @@ -211,22 +212,31 @@ func cloudScanStatusHandler(svc *service.Container) gin.HandlerFunc { } } -func cloudMountLibraryName(typ, dir string) string { - base := typ - switch typ { - case cloud.TypeQuark: - base = "夸克网盘" - case cloud.Type115: - base = "115 网盘" - case cloud.TypeCloudDrive2: - base = "CloudDrive2" - case cloud.TypeOpenList: - base = "OpenList" +func cloudMountLibraryName(typ, dir, displayDir string) string { + base := service.CloudMountProviderLabel(typ) + displayDir = strings.Trim(strings.TrimSpace(strings.ReplaceAll(displayDir, "\\", "/")), "/") + if displayDir != "" { + parts := strings.Split(displayDir, "/") + for i := len(parts) - 1; i >= 0; i-- { + if part := strings.TrimSpace(parts[i]); part != "" { + return part + } + } } if dir == "" || dir == "0" { return base } - return base + " · " + dir + dir = strings.Trim(strings.TrimSpace(strings.ReplaceAll(dir, "\\", "/")), "/") + if dir == "" { + return base + } + parts := strings.Split(dir, "/") + for i := len(parts) - 1; i >= 0; i-- { + if part := strings.TrimSpace(parts[i]); part != "" { + return part + } + } + return base } // cloud115QRStartHandler begins a 115 QR-code login and returns the session + @@ -329,17 +339,29 @@ func serveCloudResolvedLink(svc *service.Container, c *gin.Context, typ, ref str if rng := c.GetHeader("Range"); rng != "" { req.Header.Set("Range", rng) } + if accept := c.GetHeader("Accept"); accept != "" { + req.Header.Set("Accept", accept) + } + if c.GetHeader("Accept-Encoding") == "" { + req.Header.Set("Accept-Encoding", "identity") + } resp, err := http.DefaultClient.Do(req) if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } defer resp.Body.Close() - for _, h := range []string{"Content-Type", "Content-Length", "Content-Range", "Accept-Ranges"} { + for _, h := range []string{"Content-Type", "Content-Length", "Content-Range", "Accept-Ranges", "ETag", "Last-Modified"} { if v := resp.Header.Get(h); v != "" { c.Header(h, v) } } + if c.Writer.Header().Get("Accept-Ranges") == "" { + c.Header("Accept-Ranges", "bytes") + } + if resp.StatusCode >= 400 { + c.Header("Cache-Control", "no-store") + } c.Status(resp.StatusCode) if c.Request.Method != http.MethodHead { _, _ = io.Copy(c.Writer, resp.Body) diff --git a/internal/handler/cloud_test.go b/internal/handler/cloud_test.go new file mode 100644 index 0000000..87e1f74 --- /dev/null +++ b/internal/handler/cloud_test.go @@ -0,0 +1,26 @@ +package handler + +import "testing" + +func TestCloudMountLibraryNameDefaultsToDirectoryBaseName(t *testing.T) { + tests := []struct { + name string + provider string + dir string + displayDir string + want string + }{ + {name: "openlist directory", provider: "openlist", dir: "/国产剧", displayDir: "/国产剧", want: "国产剧"}, + {name: "nested directory", provider: "openlist", dir: "id-123", displayDir: "剧集/国产剧", want: "国产剧"}, + {name: "provider root", provider: "openlist", dir: "", displayDir: "", want: "OpenList"}, + {name: "115 root id", provider: "cloud115", dir: "0", displayDir: "", want: "115 网盘"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := cloudMountLibraryName(tt.provider, tt.dir, tt.displayDir); got != tt.want { + t.Fatalf("cloudMountLibraryName() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/internal/handler/emby.go b/internal/handler/emby.go index b00d411..cff79b2 100644 --- a/internal/handler/emby.go +++ b/internal/handler/emby.go @@ -104,6 +104,30 @@ func embyPingHandler(_ *service.Container) gin.HandlerFunc { } } +func embyRootHandler(svc *service.Container) gin.HandlerFunc { + return func(c *gin.Context) { + c.JSON(http.StatusOK, embyPublicSystemInfoPayload(c, svc)) + } +} + +func embyPublicSystemInfoPayload(c *gin.Context, svc *service.Container) map[string]any { + if svc != nil && svc.Emby != nil { + return embyWithRequestAddress(c, svc.Emby.SystemInfoPublic()) + } + return embyWithRequestAddress(c, map[string]any{ + "Id": "mediastation-go-001", + "ServerId": "mediastation-go-001", + "ServerName": "MediaStationGo", + "Version": "4.8.10.0", + "ServerVersion": "4.8.10.0", + "ProductName": "Emby Server", + "OperatingSystem": "Windows", + "SupportsHttps": false, + "SupportsAutoDiscovery": true, + "StartupWizardCompleted": true, + }) +} + // ─── Users / Auth ──────────────────────────────────────────────────────────── type embyAuthByNameReq struct { @@ -446,6 +470,7 @@ func embyVirtualFoldersHandler(svc *service.Container) gin.HandlerFunc { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } + libs = service.FilterDisplayCloudLibraries(c.Request.Context(), svc.Repo, libs) uid := embyUserID(c) visibility := service.UserDefaultMediaVisibility(c.Request.Context(), svc.Repo, uid) out := make([]gin.H, 0, len(libs)) @@ -986,6 +1011,40 @@ func embyServerConfigurationHandler(_ *service.Container) gin.HandlerFunc { } } +func embyPublicServerConfigurationHandler(_ *service.Container) gin.HandlerFunc { + return func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{ + "IsStartupWizardCompleted": true, + "EnableRemoteAccess": true, + "EnableUPnP": false, + "EnableHttps": false, + "RequireHttps": false, + "LocalNetworkSubnets": []string{}, + "LocalNetworkAddresses": []string{}, + "RemoteClientBitrateLimit": 0, + }) + } +} + +func embyStartupConfigurationHandler(_ *service.Container) gin.HandlerFunc { + return func(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{ + "IsStartupWizardCompleted": true, + "StartupWizardCompleted": true, + "EnableRemoteAccess": true, + "UICulture": "zh-CN", + "MetadataCountryCode": "CN", + "PreferredMetadataLanguage": "zh-CN", + }) + } +} + +func embyQuickConnectEnabledHandler(_ *service.Container) gin.HandlerFunc { + return func(c *gin.Context) { + c.JSON(http.StatusOK, false) + } +} + func embyEmptyItemsHandler(_ *service.Container) gin.HandlerFunc { return func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"Items": []any{}, "TotalRecordCount": 0}) @@ -1002,6 +1061,12 @@ func embyBrandingConfigHandler(_ *service.Container) gin.HandlerFunc { } } +func embyBrandingCSSHandler(_ *service.Container) gin.HandlerFunc { + return func(c *gin.Context) { + c.Data(http.StatusOK, "text/css; charset=utf-8", []byte("")) + } +} + func embyLocalizationOptionsHandler(_ *service.Container) gin.HandlerFunc { return func(c *gin.Context) { c.JSON(http.StatusOK, []map[string]any{ @@ -1022,6 +1087,13 @@ func registerEmbyRoutes(r *gin.Engine, jwtSecret string, svc *service.Container) c.Next() }) + if prefix == "/emby" { + grp.GET("", embyRootHandler(svc)) + grp.HEAD("", embyRootHandler(svc)) + grp.GET("/", embyRootHandler(svc)) + grp.HEAD("/", embyRootHandler(svc)) + } + // 公开端点 for _, path := range []string{"/System/Info/Public", "/system/info/public"} { grp.GET(path, embySystemInfoPublicHandler(svc)) @@ -1034,11 +1106,29 @@ func registerEmbyRoutes(r *gin.Engine, jwtSecret string, svc *service.Container) for _, path := range []string{"/System/Endpoint", "/system/endpoint"} { grp.GET(path, embySystemEndpointHandler(svc)) } + for _, path := range []string{"/System/Configuration/Public", "/system/configuration/public"} { + grp.GET(path, embyPublicServerConfigurationHandler(svc)) + grp.HEAD(path, embyPublicServerConfigurationHandler(svc)) + } + for _, path := range []string{"/Startup/Configuration", "/startup/configuration"} { + grp.GET(path, embyStartupConfigurationHandler(svc)) + grp.HEAD(path, embyStartupConfigurationHandler(svc)) + } + for _, path := range []string{"/Startup/Complete", "/startup/complete"} { + grp.POST(path, embyNoContentHandler(svc)) + } + for _, path := range []string{"/QuickConnect/Enabled", "/quickconnect/enabled"} { + grp.GET(path, embyQuickConnectEnabledHandler(svc)) + grp.HEAD(path, embyQuickConnectEnabledHandler(svc)) + } for _, path := range []string{"/System/Ping", "/system/ping"} { grp.GET(path, embyPingHandler(svc)) grp.HEAD(path, embyPingHandler(svc)) grp.POST(path, embyPingHandler(svc)) } + for _, path := range []string{"/Sessions/Capabilities", "/Sessions/Capabilities/Full", "/sessions/capabilities", "/sessions/capabilities/full"} { + grp.POST(path, embyNoContentHandler(svc)) + } // 30/min per IP: many Emby clients sit behind a single NAT/reverse-proxy // IP, so a low limit would throttle legitimate logins into 429s. embyLoginLimiter := middleware.NewRateLimiter(30, 1*time.Minute) @@ -1051,6 +1141,10 @@ func registerEmbyRoutes(r *gin.Engine, jwtSecret string, svc *service.Container) for _, path := range []string{"/Branding/Configuration", "/branding/configuration"} { grp.GET(path, embyBrandingConfigHandler(svc)) } + for _, path := range []string{"/Branding/Css", "/branding/css"} { + grp.GET(path, embyBrandingCSSHandler(svc)) + grp.HEAD(path, embyBrandingCSSHandler(svc)) + } for _, path := range []string{"/Localization/Options", "/localization/options"} { grp.GET(path, embyLocalizationOptionsHandler(svc)) } @@ -1109,8 +1203,6 @@ func registerEmbyRoutes(r *gin.Engine, jwtSecret string, svc *service.Container) auth.POST("/Sessions/Playing", embyPlayingProgressHandler(svc)) auth.POST("/Sessions/Playing/Progress", embyPlayingProgressHandler(svc)) auth.POST("/Sessions/Playing/Stopped", embyPlayingProgressHandler(svc)) - auth.POST("/Sessions/Capabilities", embyNoContentHandler(svc)) - auth.POST("/Sessions/Capabilities/Full", embyNoContentHandler(svc)) auth.POST("/Users/:userId/FavoriteItems/:itemId", embyFavoriteHandler(svc, true)) auth.DELETE("/Users/:userId/FavoriteItems/:itemId", embyFavoriteHandler(svc, false)) @@ -1175,8 +1267,6 @@ func registerLowercaseEmbyAuthRoutes(auth *gin.RouterGroup, svc *service.Contain auth.POST("/sessions/playing", embyPlayingProgressHandler(svc)) auth.POST("/sessions/playing/progress", embyPlayingProgressHandler(svc)) auth.POST("/sessions/playing/stopped", embyPlayingProgressHandler(svc)) - auth.POST("/sessions/capabilities", embyNoContentHandler(svc)) - auth.POST("/sessions/capabilities/full", embyNoContentHandler(svc)) auth.POST("/users/:userId/favoriteitems/:itemId", embyFavoriteHandler(svc, true)) auth.DELETE("/users/:userId/favoriteitems/:itemId", embyFavoriteHandler(svc, false)) diff --git a/internal/handler/emby_test.go b/internal/handler/emby_test.go index e141323..22f191d 100644 --- a/internal/handler/emby_test.go +++ b/internal/handler/emby_test.go @@ -179,6 +179,60 @@ func TestEmbyPublicSystemInfoLooksLikeModernEmbyServer(t *testing.T) { } } +func TestEmbySenPlayerDiscoveryRoutesReturnProtocolResponses(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatalf("open db: %v", err) + } + cfg := &config.Config{} + cfg.App.Port = 9011 + repos := repository.New(db) + router := gin.New() + registerEmbyRoutes(router, "test-secret", &service.Container{ + Repo: repos, + Emby: service.NewEmbyService(cfg, zap.NewNop(), repos), + }) + + tests := []struct { + path string + contentType string + contains string + }{ + {path: "/emby", contentType: "application/json", contains: "Emby Server"}, + {path: "/emby/", contentType: "application/json", contains: "Emby Server"}, + {path: "/Startup/Configuration", contentType: "application/json", contains: "StartupWizardCompleted"}, + {path: "/emby/Startup/Configuration", contentType: "application/json", contains: "StartupWizardCompleted"}, + {path: "/System/Configuration/Public", contentType: "application/json", contains: "IsStartupWizardCompleted"}, + {path: "/emby/System/Configuration/Public", contentType: "application/json", contains: "IsStartupWizardCompleted"}, + {path: "/QuickConnect/Enabled", contentType: "application/json", contains: "false"}, + {path: "/emby/QuickConnect/Enabled", contentType: "application/json", contains: "false"}, + {path: "/Branding/Css", contentType: "text/css", contains: ""}, + {path: "/emby/Branding/Css", contentType: "text/css", contains: ""}, + } + + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, tt.path, nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", w.Code, w.Body.String()) + } + if contentType := w.Header().Get("Content-Type"); !strings.Contains(contentType, tt.contentType) { + t.Fatalf("Content-Type = %q, want %q", contentType, tt.contentType) + } + if tt.contains != "" && !strings.Contains(w.Body.String(), tt.contains) { + t.Fatalf("body = %q, want contains %q", w.Body.String(), tt.contains) + } + if strings.Contains(w.Body.String(), " 0 { + return items, nil + } + } + } + return r.searchFilteredLIKE(ctx, query, limit, filter) +} + +func (r *MediaRepository) searchFilteredFTS(ctx context.Context, query string, limit int, filter MediaQueryFilter) ([]model.Media, bool) { + if !r.searchIndexEnabled(ctx) { + return nil, false + } + ftsQuery := mediaFTSQuery(query) + if ftsQuery == "" { + return nil, false + } + var items []model.Media + q := r.db.WithContext(ctx). + Table("media"). + Select("media.*"). + Joins("JOIN media_search_fts ON media_search_fts.media_id = media.id"). + Where("media.deleted_at IS NULL"). + Where("media_search_fts MATCH ?", ftsQuery) + q = applyQualifiedMediaQueryFilter(q, filter) + err := q.Order("bm25(media_search_fts), media.created_at DESC").Limit(limit).Find(&items).Error + if err != nil { + return nil, false + } + return items, true +} + +func (r *MediaRepository) searchFilteredLIKE(ctx context.Context, query string, limit int, filter MediaQueryFilter) ([]model.Media, error) { var items []model.Media q := r.db.WithContext(ctx).Model(&model.Media{}).Limit(limit) q = applyMediaQueryFilter(q, filter) - if query != "" { - like := "%" + query + "%" - q = q.Where("title LIKE ? OR original_name LIKE ?", like, like) + terms := mediaSearchTerms(query) + for _, term := range terms { + like := "%" + escapeLike(term) + "%" + q = q.Where( + "(title LIKE ? ESCAPE '\\' OR original_name LIKE ? ESCAPE '\\' OR path LIKE ? ESCAPE '\\' OR genres LIKE ? ESCAPE '\\')", + like, like, like, like, + ) } - err := q.Order("created_at desc").Find(&items).Error + if query != "" { + prefix := escapeLike(query) + "%" + exact := query + q = q.Order(gorm.Expr( + "CASE WHEN title = ? THEN 0 WHEN original_name = ? THEN 1 WHEN title LIKE ? ESCAPE '\\' THEN 2 WHEN original_name LIKE ? ESCAPE '\\' THEN 3 ELSE 4 END, created_at desc", + exact, exact, prefix, prefix, + )) + } else { + q = q.Order("created_at desc") + } + err := q.Find(&items).Error return items, err } +func applyQualifiedMediaQueryFilter(q *gorm.DB, filter MediaQueryFilter) *gorm.DB { + if !filter.IncludeNSFW { + q = q.Where("media.nsfw = ?", false) + } + if len(filter.HiddenLibraryIDs) > 0 { + q = q.Where("media.library_id NOT IN ?", filter.HiddenLibraryIDs) + } + if len(filter.AllowedLibraryIDs) > 0 { + q = q.Where("media.library_id IN ?", filter.AllowedLibraryIDs) + } + return q +} + +func mediaFTSQuery(query string) string { + terms := mediaSearchTerms(query) + if len(terms) == 0 { + return "" + } + quoted := make([]string, 0, len(terms)) + for _, term := range terms { + term = strings.ReplaceAll(term, `"`, `""`) + if term != "" { + quoted = append(quoted, `"`+term+`"`) + } + } + return strings.Join(quoted, " AND ") +} + +func mediaSearchTerms(query string) []string { + query = strings.TrimSpace(query) + if query == "" { + return nil + } + fields := strings.FieldsFunc(query, func(r rune) bool { + return unicode.IsSpace(r) || unicode.IsPunct(r) || unicode.IsSymbol(r) + }) + out := make([]string, 0, len(fields)) + seen := map[string]struct{}{} + for _, field := range fields { + field = strings.TrimSpace(field) + if field == "" { + continue + } + lower := strings.ToLower(field) + if _, ok := seen[lower]; ok { + continue + } + seen[lower] = struct{}{} + out = append(out, field) + } + return out +} + +func escapeLike(value string) string { + value = strings.ReplaceAll(value, `\`, `\\`) + value = strings.ReplaceAll(value, `%`, `\%`) + value = strings.ReplaceAll(value, `_`, `\_`) + return value +} + +func (r *MediaRepository) refreshSearchIndex(ctx context.Context, mediaID string) error { + if strings.TrimSpace(mediaID) == "" { + return nil + } + if !r.searchIndexEnabled(ctx) { + return nil + } + tx := r.db.WithContext(ctx) + _ = tx.Exec(`DELETE FROM media_search_fts WHERE media_id = ?`, mediaID).Error + return tx.Exec(` +INSERT INTO media_search_fts(media_id, title, original_name, path, genres) +SELECT id, COALESCE(title, ''), COALESCE(original_name, ''), COALESCE(path, ''), COALESCE(genres, '') +FROM media +WHERE id = ? AND deleted_at IS NULL +`, mediaID).Error +} + +func (r *MediaRepository) BackfillSearchIndex(ctx context.Context, batchLimit int) (int64, error) { + if batchLimit <= 0 { + batchLimit = 1000 + } + if !r.searchIndexEnabled(ctx) { + return 0, nil + } + res := r.db.WithContext(ctx).Exec(` +INSERT INTO media_search_fts(media_id, title, original_name, path, genres) +SELECT m.id, COALESCE(m.title, ''), COALESCE(m.original_name, ''), COALESCE(m.path, ''), COALESCE(m.genres, '') +FROM media AS m +WHERE m.deleted_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM media_search_fts AS f WHERE f.media_id = m.id + ) +ORDER BY m.created_at DESC +LIMIT ? +`, batchLimit) + return res.RowsAffected, res.Error +} + +func (r *MediaRepository) searchIndexEnabled(ctx context.Context) bool { + if r == nil || r.db == nil { + return false + } + r.searchIndexOnce.Do(func() { + var count int64 + err := r.db.WithContext(ctx). + Raw(`SELECT COUNT(*) FROM sqlite_master WHERE name = 'media_search_fts'`). + Scan(&count).Error + r.searchIndexAvailable = err == nil && count > 0 + }) + return r.searchIndexAvailable +} + // DeleteByLibrary purges all media tied to a library. func (r *MediaRepository) DeleteByLibrary(ctx context.Context, libraryID string) error { + if r.searchIndexEnabled(ctx) { + _ = r.db.WithContext(ctx).Exec(`DELETE FROM media_search_fts WHERE media_id IN (SELECT id FROM media WHERE library_id = ?)`, libraryID).Error + } return r.db.WithContext(ctx).Where("library_id = ?", libraryID).Delete(&model.Media{}).Error } // PurgeByLibrary permanently removes media tied to a library. Used for virtual // cloud mounts where "remove mount" must not populate the recycle bin. func (r *MediaRepository) PurgeByLibrary(ctx context.Context, libraryID string) error { + if r.searchIndexEnabled(ctx) { + _ = r.db.WithContext(ctx).Exec(`DELETE FROM media_search_fts WHERE media_id IN (SELECT id FROM media WHERE library_id = ?)`, libraryID).Error + } return r.db.WithContext(ctx).Unscoped().Where("library_id = ?", libraryID).Delete(&model.Media{}).Error } diff --git a/internal/repository/repository_test.go b/internal/repository/repository_test.go new file mode 100644 index 0000000..ac5b18e --- /dev/null +++ b/internal/repository/repository_test.go @@ -0,0 +1,117 @@ +package repository + +import ( + "testing" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + + "github.com/ShukeBta/MediaStationGo/internal/database" + "github.com/ShukeBta/MediaStationGo/internal/model" +) + +func TestMediaSearchFilteredSupportsChineseFuzzyTerms(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := database.AutoMigrate(db); err != nil { + t.Fatalf("migrate: %v", err) + } + repos := New(db) + lib := model.Library{Name: "国产剧", Path: "/media/国产剧", Type: "tv", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatalf("create library: %v", err) + } + rows := []model.Media{ + { + Base: model.Base{ID: "m-ferry"}, + LibraryID: lib.ID, + Title: "灵魂摆渡·十年", + OriginalName: "The Ferry Man 10th Anniversary", + Path: "/media/国产剧/灵魂摆渡·十年/S01E01.mkv", + Genres: "悬疑,奇幻", + }, + { + Base: model.Base{ID: "m-ashes"}, + LibraryID: lib.ID, + Title: "翘楚", + OriginalName: "Ashes to Crown", + Path: "/media/国产剧/翘楚/S01E01.mkv", + Genres: "剧情", + }, + } + for i := range rows { + if err := repos.Media.Upsert(t.Context(), &rows[i]); err != nil { + t.Fatalf("upsert media: %v", err) + } + } + + items, err := repos.Media.SearchFiltered(t.Context(), "灵魂 十年", 10, MediaQueryFilter{IncludeNSFW: true}) + if err != nil { + t.Fatalf("search chinese terms: %v", err) + } + if len(items) == 0 || items[0].ID != "m-ferry" { + t.Fatalf("chinese fuzzy search missed target: %#v", items) + } + + items, err = repos.Media.SearchFiltered(t.Context(), "Ferry", 10, MediaQueryFilter{IncludeNSFW: true}) + if err != nil { + t.Fatalf("search original name: %v", err) + } + if len(items) == 0 || items[0].ID != "m-ferry" { + t.Fatalf("original-name search missed target: %#v", items) + } + + items, err = repos.Media.SearchFiltered(t.Context(), "悬疑", 10, MediaQueryFilter{IncludeNSFW: true}) + if err != nil { + t.Fatalf("search genre: %v", err) + } + if len(items) == 0 || items[0].ID != "m-ferry" { + t.Fatalf("genre search missed target: %#v", items) + } +} + +func TestMediaSearchIndexBackfillRunsInBatches(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := database.AutoMigrate(db); err != nil { + t.Fatalf("migrate: %v", err) + } + repos := New(db) + lib := model.Library{Name: "电影", Path: "/media/movie", Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + if err := repos.DB.Create(&model.Media{ + Base: model.Base{ID: "m-backfill"}, + LibraryID: lib.ID, + Title: "后台索引", + Path: "/media/movie/后台索引.mkv", + }).Error; err != nil { + t.Fatal(err) + } + var before int64 + if err := repos.DB.Raw(`SELECT COUNT(*) FROM media_search_fts`).Scan(&before).Error; err != nil { + t.Fatal(err) + } + if before != 0 { + t.Fatalf("startup migrate should not synchronously backfill FTS, got %d rows", before) + } + n, err := repos.Media.BackfillSearchIndex(t.Context(), 1) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("backfilled rows = %d, want 1", n) + } + var after int64 + if err := repos.DB.Raw(`SELECT COUNT(*) FROM media_search_fts`).Scan(&after).Error; err != nil { + t.Fatal(err) + } + if after != 1 { + t.Fatalf("fts rows = %d, want 1", after) + } +} diff --git a/internal/service/boot_cloud.go b/internal/service/boot_cloud.go index 046f8f6..47e8886 100644 --- a/internal/service/boot_cloud.go +++ b/internal/service/boot_cloud.go @@ -7,21 +7,32 @@ import ( "go.uber.org/zap" "github.com/ShukeBta/MediaStationGo/internal/model" + "github.com/ShukeBta/MediaStationGo/internal/repository" ) -// BootCloudLibraries 在系统启动后自动扫描所有云盘媒体库，使媒体对所有用户立即可见。 -// 避免每个用户首次访问时都触发扫描。 +// BootCloudLibraries optionally scans cloud libraries after startup. It is +// disabled by default for huge cloud mounts; normal automatic refresh is handled +// by the nightly cloud_sync scheduler window, and operators can still scan +// manually at any time. func (c *Container) BootCloudLibraries(ctx context.Context) { if c == nil || c.Repo == nil || c.Scan == nil { return } + if !bootCloudLibraryScanEnabled(ctx, c.Repo) { + c.Log.Info("boot: cloud library scans disabled; use manual scan or nightly cloud sync") + return + } libs, err := c.Repo.Library.List(ctx) if err != nil { c.Log.Warn("boot cloud libraries: list failed", zap.Error(err)) return } + libs = FilterScannableCloudLibraries(ctx, c.Repo, libs) cloudLibs := make([]model.Library, 0) for _, lib := range libs { + if !lib.Enabled { + continue + } if _, ok := ParseCloudLibraryMount(lib.Path); ok { cloudLibs = append(cloudLibs, lib) } @@ -32,19 +43,32 @@ func (c *Container) BootCloudLibraries(ctx context.Context) { c.Log.Info("boot: scheduling cloud library scans", zap.Int("count", len(cloudLibs))) // 延迟3秒后启动，避免和系统初始化任务冲突 time.AfterFunc(3*time.Second, func() { - for _, lib := range cloudLibs { - libID := lib.ID - libName := lib.Name - go func() { - scanCtx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) - defer cancel() - c.Log.Info("boot: scanning cloud library", zap.String("id", libID), zap.String("name", libName)) - if _, err := c.Scan.ScanLibraryWithoutAutoScrape(scanCtx, libID); err != nil { - c.Log.Warn("boot: cloud library scan failed", zap.String("id", libID), zap.String("name", libName), zap.Error(err)) - } else { - c.Log.Info("boot: cloud library scan completed", zap.String("id", libID), zap.String("name", libName)) - } - }() - } + go c.runBootCloudLibraryScanQueue(cloudLibs) }) } + +func (c *Container) runBootCloudLibraryScanQueue(cloudLibs []model.Library) { + for _, lib := range cloudLibs { + libID := lib.ID + libName := lib.Name + scanCtx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) + c.Log.Info("boot: scanning cloud library", zap.String("id", libID), zap.String("name", libName)) + if _, err := c.Scan.ScanLibraryWithoutAutoScrape(scanCtx, libID); err != nil { + c.Log.Warn("boot: cloud library scan failed", zap.String("id", libID), zap.String("name", libName), zap.Error(err)) + } else { + c.Log.Info("boot: cloud library scan completed", zap.String("id", libID), zap.String("name", libName)) + } + cancel() + } +} + +func bootCloudLibraryScanEnabled(ctx context.Context, repo *repository.Container) bool { + if repo == nil || repo.Setting == nil { + return false + } + value, err := repo.Setting.Get(ctx, "cloud.boot_scan_enabled") + if err != nil { + return false + } + return parseBoolSetting(value, false) +} diff --git a/internal/service/cloud/clouddrive2.go b/internal/service/cloud/clouddrive2.go index a600365..8ef3eeb 100644 --- a/internal/service/cloud/clouddrive2.go +++ b/internal/service/cloud/clouddrive2.go @@ -1,8 +1,10 @@ package cloud import ( + "bytes" "context" "encoding/base64" + "encoding/json" "encoding/xml" "fmt" "io" @@ -27,6 +29,7 @@ type cloudDrive2Provider struct { password string token string ua string + apiBase *url.URL client *http.Client proxy bool } @@ -42,6 +45,10 @@ func newOpenList(cfg map[string]any, client *http.Client) *cloudDrive2Provider { func newCloudDAVProvider(typ, name string, cfg map[string]any, client *http.Client, defaultDAVPath string) *cloudDrive2Provider { rawURL := webDAVURLFromConfig(cfg, defaultDAVPath) u, _ := url.Parse(strings.TrimRight(rawURL, "/")) + var apiBase *url.URL + if typ == TypeOpenList { + apiBase = openListAPIBaseFromConfig(cfg, rawURL, defaultDAVPath) + } ua := str(cfg["ua"]) if ua == "" { ua = defaultUA @@ -58,6 +65,7 @@ func newCloudDAVProvider(typ, name string, cfg map[string]any, client *http.Clie password: str(cfg["password"]), token: str(cfg["token"]), ua: ua, + apiBase: apiBase, client: client, proxy: proxy, } @@ -74,6 +82,11 @@ func (p *cloudDrive2Provider) List(ctx context.Context, dir string) ([]FileEntry if err := p.validate(); err != nil { return nil, err } + if p.typ == TypeOpenList && p.apiBase != nil && strings.TrimSpace(p.token) != "" { + if entries, err := p.listOpenListAPI(ctx, dir); err == nil { + return entries, nil + } + } target := normalizeCloudDAVPath(dir) req, err := http.NewRequestWithContext(ctx, "PROPFIND", p.urlFor(target), strings.NewReader(cloudDAVPropfindBody)) if err != nil { @@ -121,6 +134,75 @@ func (p *cloudDrive2Provider) List(ctx context.Context, dir string) ([]FileEntry return out, nil } +func (p *cloudDrive2Provider) listOpenListAPI(ctx context.Context, dir string) ([]FileEntry, error) { + const pageSize = 500 + target := normalizeCloudDAVPath(dir) + out := make([]FileEntry, 0, pageSize) + for pageNum := 1; ; pageNum++ { + payload := map[string]any{ + "path": target, + "password": "", + "page": pageNum, + "per_page": pageSize, + "refresh": false, + } + body, _ := json.Marshal(payload) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.openListAPIURL("/api/fs/list"), bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", p.ua) + if p.token != "" { + req.Header.Set("Authorization", p.token) + } + resp, err := p.client.Do(req) + if err != nil { + return nil, decorateDAVTransportError(p.name, p.openListAPIURL("/api/fs/list"), err) + } + var decoded openListListResponse + decodeErr := json.NewDecoder(io.LimitReader(resp.Body, 32<<20)).Decode(&decoded) + resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("%s: api list %s returned http %d", p.name, target, resp.StatusCode) + } + if decodeErr != nil { + return nil, fmt.Errorf("%s: decode api list: %w", p.name, decodeErr) + } + if decoded.Code != 0 && decoded.Code != 200 { + msg := strings.TrimSpace(decoded.Message) + if msg == "" { + msg = fmt.Sprintf("code %d", decoded.Code) + } + return nil, fmt.Errorf("%s: api list %s failed: %s", p.name, target, msg) + } + for _, item := range decoded.Data.Content { + name := strings.TrimSpace(item.Name) + if name == "" || name == "." || name == "/" { + continue + } + out = append(out, FileEntry{ + ID: joinOpenListAPIPath(target, name), + Name: name, + IsDir: item.IsDir, + Size: item.Size, + }) + } + total := decoded.Data.Total + if total > 0 { + if len(out) >= total || len(decoded.Data.Content) == 0 { + break + } + continue + } + if len(decoded.Data.Content) == 0 || len(decoded.Data.Content) < pageSize { + break + } + } + return out, nil +} + func (p *cloudDrive2Provider) Resolve(ctx context.Context, fileRef string) (*DirectLink, error) { if err := p.validate(); err != nil { return nil, err @@ -184,6 +266,51 @@ func defaultWebDAVURL(server, defaultDAVPath string) string { return server + davPath } +func openListAPIBaseFromConfig(cfg map[string]any, webDAVURL, defaultDAVPath string) *url.URL { + raw := str(cfg["server"]) + if raw == "" { + raw = firstNonEmpty(str(cfg["api_url"]), webDAVURL) + } + raw = strings.TrimRight(strings.TrimSpace(raw), "/") + if raw == "" { + return nil + } + u, err := url.Parse(raw) + if err != nil || u.Scheme == "" || u.Host == "" { + return nil + } + davPath := strings.Trim(strings.TrimSpace(defaultDAVPath), "/") + if davPath != "" { + pathParts := strings.Split(strings.TrimRight(u.Path, "/"), "/") + if len(pathParts) > 0 && strings.EqualFold(pathParts[len(pathParts)-1], davPath) { + u.Path = strings.Join(pathParts[:len(pathParts)-1], "/") + if u.Path == "" { + u.Path = "/" + } + } + } + u.RawPath = "" + u.RawQuery = "" + u.Fragment = "" + return u +} + +func (p *cloudDrive2Provider) openListAPIURL(apiPath string) string { + if p.apiBase == nil { + return "" + } + u := *p.apiBase + u.RawPath = "" + basePath := strings.TrimRight(u.Path, "/") + apiPath = "/" + strings.TrimLeft(apiPath, "/") + if basePath == "" || basePath == "/" { + u.Path = apiPath + } else { + u.Path = basePath + apiPath + } + return u.String() +} + func ensureDefaultDAVPath(rawURL, defaultDAVPath string) string { rawURL = strings.TrimRight(strings.TrimSpace(rawURL), "/") if rawURL == "" { @@ -322,6 +449,21 @@ type cloudDAVResourceType struct { Collection *struct{} `xml:"collection"` } +type openListListResponse struct { + Code int `json:"code"` + Message string `json:"message"` + Data struct { + Content []openListListItem `json:"content"` + Total int `json:"total"` + } `json:"data"` +} + +type openListListItem struct { + Name string `json:"name"` + Size int64 `json:"size"` + IsDir bool `json:"is_dir"` +} + func normalizeCloudDAVPath(p string) string { p = strings.ReplaceAll(strings.TrimSpace(p), "\\", "/") if p == "" || p == "." { @@ -341,6 +483,15 @@ func sameCloudDAVPath(a, b string) bool { return strings.TrimRight(normalizeCloudDAVPath(a), "/") == strings.TrimRight(normalizeCloudDAVPath(b), "/") } +func joinOpenListAPIPath(dir, name string) string { + dir = strings.TrimRight(normalizeCloudDAVPath(dir), "/") + name = strings.Trim(strings.ReplaceAll(name, "\\", "/"), "/") + if dir == "" || dir == "/" { + return normalizeCloudDAVPath(name) + } + return normalizeCloudDAVPath(dir + "/" + name) +} + func parseDAVSize(raw string) int64 { n, _ := strconv.ParseInt(strings.TrimSpace(raw), 10, 64) return n diff --git a/internal/service/cloud_mount.go b/internal/service/cloud_mount.go index 825061b..b36c18e 100644 --- a/internal/service/cloud_mount.go +++ b/internal/service/cloud_mount.go @@ -1,10 +1,12 @@ package service import ( + "context" "net/url" "strings" "github.com/ShukeBta/MediaStationGo/internal/model" + "github.com/ShukeBta/MediaStationGo/internal/repository" "github.com/ShukeBta/MediaStationGo/internal/service/cloud" ) @@ -137,6 +139,364 @@ func FilterShadowedCloudLibraries(libs []model.Library) []model.Library { return out } +func FilterDisplayCloudLibraries(ctx context.Context, repo *repository.Container, libs []model.Library) []model.Library { + if len(libs) == 0 { + return libs + } + counts := cloudLibraryMediaCounts(ctx, repo, libs) + collapsed := make([]model.Library, 0, len(libs)) + byKey := make(map[string]int, len(libs)) + for _, lib := range libs { + key, ok := cloudLibraryDisplayKey(lib) + if !ok { + collapsed = append(collapsed, lib) + continue + } + if prevIndex, exists := byKey[key]; exists { + if betterDisplayCloudLibrary(lib, collapsed[prevIndex], counts) { + collapsed[prevIndex] = lib + } + continue + } + byKey[key] = len(collapsed) + collapsed = append(collapsed, lib) + } + collapsed = FilterShadowedCloudLibraries(collapsed) + return mergeDisplayCloudLibraries(collapsed) +} + +func FilterScannableCloudLibraries(ctx context.Context, repo *repository.Container, libs []model.Library) []model.Library { + if len(libs) == 0 { + return libs + } + counts := cloudLibraryMediaCounts(ctx, repo, libs) + collapsed := make([]model.Library, 0, len(libs)) + byKey := make(map[string]int, len(libs)) + for _, lib := range libs { + key, ok := cloudLibraryDisplayKey(lib) + if !ok { + collapsed = append(collapsed, lib) + continue + } + if prevIndex, exists := byKey[key]; exists { + if betterDisplayCloudLibrary(lib, collapsed[prevIndex], counts) { + collapsed[prevIndex] = lib + } + continue + } + byKey[key] = len(collapsed) + collapsed = append(collapsed, lib) + } + return FilterShadowedCloudLibraries(collapsed) +} + +func NormalizeCloudLibraryDisplayNames(libs []model.Library) []model.Library { + out := make([]model.Library, 0, len(libs)) + for _, lib := range libs { + if displayName, ok := CloudLibraryDisplayName(lib); ok && displayName != "" { + lib.Name = displayName + } + out = append(out, lib) + } + return out +} + +func cloudLibraryMediaCounts(ctx context.Context, repo *repository.Container, libs []model.Library) map[string]int64 { + counts := make(map[string]int64, len(libs)) + if repo == nil || repo.DB == nil || len(libs) == 0 { + return counts + } + ids := make([]string, 0, len(libs)) + for _, lib := range libs { + if _, ok := ParseCloudLibraryMount(lib.Path); ok { + ids = append(ids, lib.ID) + } + } + if len(ids) == 0 { + return counts + } + var rows []struct { + LibraryID string + Count int64 + } + if err := repo.DB.WithContext(ctx). + Model(&model.Media{}). + Select("library_id, COUNT(*) AS count"). + Where("library_id IN ? AND deleted_at IS NULL", ids). + Group("library_id"). + Scan(&rows).Error; err != nil { + return counts + } + for _, row := range rows { + counts[row.LibraryID] = row.Count + } + return counts +} + +func cloudLibraryDisplayKey(lib model.Library) (string, bool) { + info, ok := ParseCloudLibraryMount(lib.Path) + if !ok { + return "", false + } + dir := firstNonEmpty(info.DisplayDir, info.ScanDir) + return info.Provider + "\x00" + dir, true +} + +func betterDisplayCloudLibrary(candidate, current model.Library, counts map[string]int64) bool { + candidateCount := counts[candidate.ID] + currentCount := counts[current.ID] + if (candidateCount > 0) != (currentCount > 0) { + return candidateCount > 0 + } + if candidate.Enabled != current.Enabled { + return candidate.Enabled + } + candidateCanonical := cloudLibraryPathIsCanonical(candidate) + currentCanonical := cloudLibraryPathIsCanonical(current) + if candidateCanonical != currentCanonical { + return candidateCanonical + } + if !candidate.CreatedAt.Equal(current.CreatedAt) { + return candidate.CreatedAt.After(current.CreatedAt) + } + return candidate.ID > current.ID +} + +func cloudLibraryPathIsCanonical(lib model.Library) bool { + info, ok := ParseCloudLibraryMount(lib.Path) + if !ok { + return false + } + return BuildCloudLibraryPath(info.Provider, info.ScanDir, info.DisplayDir) == strings.TrimSpace(lib.Path) +} + +func mergeDisplayCloudLibraries(libs []model.Library) []model.Library { + if len(libs) == 0 { + return libs + } + localByKey := make(map[string]struct{}, len(libs)) + for _, lib := range libs { + if _, ok := ParseCloudLibraryMount(lib.Path); ok || !lib.Enabled { + continue + } + if key, ok := CloudLibraryMergeKey(lib); ok { + localByKey[key] = struct{}{} + } + } + out := make([]model.Library, 0, len(libs)) + for _, lib := range libs { + if displayName, ok := CloudLibraryDisplayName(lib); ok && displayName != "" { + lib.Name = displayName + if key, ok := CloudLibraryMergeKey(lib); ok { + if _, exists := localByKey[key]; exists { + continue + } + } + } + out = append(out, lib) + } + return out +} + +func CloudLibraryDisplayName(lib model.Library) (string, bool) { + info, ok := ParseCloudLibraryMount(lib.Path) + if !ok { + return "", false + } + name := stripCloudProviderDisplayPrefix(strings.TrimSpace(lib.Name), info.Provider) + dir := firstNonEmpty(info.DisplayDir, info.ScanDir) + if name == "" || strings.EqualFold(name, CloudMountProviderLabel(info.Provider)) { + if base := cloudMountDirBase(dir); base != "" { + name = base + } + } + if name == "" { + name = CloudMountProviderLabel(info.Provider) + } + return name, true +} + +func CloudLibraryMergeKey(lib model.Library) (string, bool) { + name := strings.TrimSpace(lib.Name) + if displayName, ok := CloudLibraryDisplayName(lib); ok { + name = displayName + } + name = normalizeLibraryMergeName(name) + if name == "" { + return "", false + } + return strings.ToLower(strings.TrimSpace(lib.Type)) + "\x00" + name, true +} + +func MergedLibraryIDsForLibrary(ctx context.Context, repo *repository.Container, libraryID string) ([]string, error) { + libraryID = strings.TrimSpace(libraryID) + if libraryID == "" || repo == nil || repo.Library == nil { + return []string{libraryID}, nil + } + lib, err := repo.Library.FindByID(ctx, libraryID) + if err != nil { + return nil, err + } + if lib == nil { + return []string{libraryID}, nil + } + libs, err := repo.Library.List(ctx) + if err != nil { + return nil, err + } + return MergedLibraryIDs(libs, *lib), nil +} + +func MergedLibraryIDs(libs []model.Library, lib model.Library) []string { + ids := []string{} + seen := map[string]struct{}{} + add := func(id string) { + id = strings.TrimSpace(id) + if id == "" { + return + } + if _, ok := seen[id]; ok { + return + } + seen[id] = struct{}{} + ids = append(ids, id) + } + add(lib.ID) + key, ok := CloudLibraryMergeKey(lib) + if !ok { + return ids + } + _, libIsCloud := ParseCloudLibraryMount(lib.Path) + for _, candidate := range libs { + if candidate.ID == lib.ID || !candidate.Enabled { + continue + } + candidateKey, ok := CloudLibraryMergeKey(candidate) + if !ok || candidateKey != key { + continue + } + _, candidateIsCloud := ParseCloudLibraryMount(candidate.Path) + if !libIsCloud && !candidateIsCloud { + continue + } + add(candidate.ID) + } + return ids +} + +func ExpandMediaVisibilityForMergedCloudLibraries(ctx context.Context, repo *repository.Container, visibility MediaVisibility) MediaVisibility { + if repo == nil || repo.Library == nil { + return visibility + } + if len(visibility.AllowedLibraryIDs) > 0 { + visibility.AllowedLibraryIDs = expandMergedLibraryIDs(ctx, repo, visibility.AllowedLibraryIDs) + } + if len(visibility.HiddenLibraryIDs) > 0 { + visibility.HiddenLibraryIDs = expandMergedLibraryIDs(ctx, repo, visibility.HiddenLibraryIDs) + } + return visibility +} + +func expandMergedLibraryIDs(ctx context.Context, repo *repository.Container, ids []string) []string { + if len(ids) == 0 { + return ids + } + libs, err := repo.Library.List(ctx) + if err != nil { + return ids + } + byID := make(map[string]model.Library, len(libs)) + for _, lib := range libs { + byID[lib.ID] = lib + } + out := make([]string, 0, len(ids)) + seen := map[string]struct{}{} + add := func(id string) { + id = strings.TrimSpace(id) + if id == "" { + return + } + if _, ok := seen[id]; ok { + return + } + seen[id] = struct{}{} + out = append(out, id) + } + for _, id := range ids { + lib, ok := byID[id] + if !ok { + add(id) + continue + } + for _, mergedID := range MergedLibraryIDs(libs, lib) { + add(mergedID) + } + } + return out +} + +func CloudMountProviderLabel(provider string) string { + switch strings.TrimSpace(provider) { + case cloud.TypeQuark: + return "夸克网盘" + case cloud.Type115: + return "115 网盘" + case cloud.TypeCloudDrive2: + return "CloudDrive2" + case cloud.TypeOpenList: + return "OpenList" + default: + if strings.TrimSpace(provider) == "" { + return "网盘" + } + return strings.TrimSpace(provider) + } +} + +func stripCloudProviderDisplayPrefix(name, provider string) string { + name = strings.TrimSpace(name) + if name == "" { + return "" + } + for _, label := range []string{CloudMountProviderLabel(provider), strings.TrimSpace(provider)} { + label = strings.TrimSpace(label) + if label == "" || len(name) < len(label) || !strings.EqualFold(name[:len(label)], label) { + continue + } + rest := strings.TrimSpace(name[len(label):]) + rest = strings.TrimLeft(rest, " \t\r\n·・-—–|｜:/\\") + if rest != "" { + return strings.TrimSpace(rest) + } + if strings.EqualFold(name, label) { + return "" + } + } + return name +} + +func cloudMountDirBase(dir string) string { + dir = strings.Trim(strings.TrimSpace(strings.ReplaceAll(dir, "\\", "/")), "/") + if dir == "" { + return "" + } + parts := strings.Split(dir, "/") + for i := len(parts) - 1; i >= 0; i-- { + if part := strings.TrimSpace(parts[i]); part != "" { + return part + } + } + return "" +} + +func normalizeLibraryMergeName(name string) string { + name = strings.ToLower(strings.TrimSpace(name)) + if name == "" { + return "" + } + return strings.Join(strings.Fields(name), " ") +} + func ShadowedCloudLibraryIDSet(libs []model.Library) map[string]bool { out := make(map[string]bool) for _, lib := range libs { diff --git a/internal/service/cloud_mount_filter_test.go b/internal/service/cloud_mount_filter_test.go new file mode 100644 index 0000000..59e3e69 --- /dev/null +++ b/internal/service/cloud_mount_filter_test.go @@ -0,0 +1,192 @@ +package service + +import ( + "slices" + "testing" + "time" + + "github.com/ShukeBta/MediaStationGo/internal/config" + "github.com/glebarez/sqlite" + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/ShukeBta/MediaStationGo/internal/model" + "github.com/ShukeBta/MediaStationGo/internal/repository" +) + +func TestFilterDisplayCloudLibrariesPrefersPopulatedCanonicalDuplicate(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + now := time.Now() + oldEmpty := model.Library{ + Base: model.Base{ID: "old-empty", CreatedAt: now.Add(-time.Hour)}, + Name: "OpenList · 国产剧", + Path: "cloud://openlist/%2F国产剧", + Type: "tv", + Enabled: true, + } + newPopulated := model.Library{ + Base: model.Base{ID: "new-populated", CreatedAt: now}, + Name: "OpenList · 国产剧", + Path: BuildCloudLibraryPath("openlist", "/国产剧", "/国产剧"), + Type: "tv", + Enabled: true, + } + if err := repos.Library.Create(t.Context(), &oldEmpty); err != nil { + t.Fatal(err) + } + if err := repos.Library.Create(t.Context(), &newPopulated); err != nil { + t.Fatal(err) + } + if err := repos.DB.Create(&model.Media{ + LibraryID: newPopulated.ID, + Title: "剧集", + Path: "cloud://openlist/国产剧/剧集.mkv", + }).Error; err != nil { + t.Fatal(err) + } + + filtered := FilterDisplayCloudLibraries(t.Context(), repos, []model.Library{oldEmpty, newPopulated}) + if len(filtered) != 1 || filtered[0].ID != newPopulated.ID { + t.Fatalf("filtered = %#v, want only populated canonical duplicate", filtered) + } + + scanner := NewScannerService(nil, zap.NewNop(), repos, nil, nil, nil) + if conflict := scanner.shadowedCloudLibrary(t.Context(), &oldEmpty); conflict == nil || conflict.Library.ID != newPopulated.ID { + t.Fatalf("old duplicate scan conflict = %#v, want populated canonical library", conflict) + } +} + +func TestFilterDisplayCloudLibrariesMergesCloudMountIntoExistingLibrary(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + local := model.Library{Name: "国产剧", Path: "/media/国产剧", Type: "tv", Enabled: true} + cloud := model.Library{Name: "OpenList · 国产剧", Path: BuildCloudLibraryPath("openlist", "/国产剧", "/国产剧"), Type: "tv", Enabled: true} + movieCloud := model.Library{Name: "OpenList · 国产剧", Path: BuildCloudLibraryPath("openlist", "/电影/国产剧", "/电影/国产剧"), Type: "movie", Enabled: true} + for _, lib := range []*model.Library{&local, &cloud, &movieCloud} { + if err := repos.Library.Create(t.Context(), lib); err != nil { + t.Fatal(err) + } + } + + filtered := FilterDisplayCloudLibraries(t.Context(), repos, []model.Library{local, cloud, movieCloud}) + if got := libraryNames(filtered); !slices.Equal(got, []string{"国产剧", "国产剧"}) { + t.Fatalf("filtered names = %#v, want local tv plus stripped movie cloud", got) + } + if filtered[0].ID != local.ID { + t.Fatalf("first filtered library = %s, want existing local library %s", filtered[0].ID, local.ID) + } + if filtered[1].ID != movieCloud.ID { + t.Fatalf("movie cloud library should stay separate when type differs: %#v", filtered) + } + + merged := MergedLibraryIDs([]model.Library{local, cloud, movieCloud}, local) + if !slices.Equal(merged, []string{local.ID, cloud.ID}) { + t.Fatalf("merged ids = %#v, want local+same-type cloud", merged) + } +} + +func TestListMediaVisibleIncludesMergedCloudLibraryItems(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + local := model.Library{Name: "国产剧", Path: "/media/国产剧", Type: "tv", Enabled: true} + cloud := model.Library{Name: "OpenList · 国产剧", Path: BuildCloudLibraryPath("openlist", "/国产剧", "/国产剧"), Type: "tv", Enabled: true} + other := model.Library{Name: "欧美剧", Path: BuildCloudLibraryPath("openlist", "/欧美剧", "/欧美剧"), Type: "tv", Enabled: true} + for _, lib := range []*model.Library{&local, &cloud, &other} { + if err := repos.Library.Create(t.Context(), lib); err != nil { + t.Fatal(err) + } + } + if err := repos.DB.Create(&[]model.Media{ + {LibraryID: local.ID, Title: "本地剧", Path: "/media/国产剧/local.mkv"}, + {LibraryID: cloud.ID, Title: "云盘剧", Path: "cloud://openlist/国产剧/cloud.mkv"}, + {LibraryID: other.ID, Title: "其他剧", Path: "cloud://openlist/欧美剧/other.mkv"}, + }).Error; err != nil { + t.Fatal(err) + } + svc := NewMediaService(&config.Config{}, zap.NewNop(), repos) + + items, total, err := svc.ListMediaVisible(t.Context(), local.ID, 1, 20, MediaVisibility{IncludeNSFW: true}) + if err != nil { + t.Fatal(err) + } + if total != 2 { + t.Fatalf("total = %d, want merged local+cloud items", total) + } + if got := mediaTitles(items); !slices.Equal(got, []string{"云盘剧", "本地剧"}) { + t.Fatalf("items = %#v, want local+cloud only", got) + } + + items, total, err = svc.ListMediaVisible(t.Context(), local.ID, 1, 20, MediaVisibility{ + IncludeNSFW: true, + AllowedLibraryIDs: []string{local.ID}, + }) + if err != nil { + t.Fatal(err) + } + if total != 2 || !slices.Equal(mediaTitles(items), []string{"云盘剧", "本地剧"}) { + t.Fatalf("profile-limited merged list total=%d items=%#v", total, mediaTitles(items)) + } +} + +func TestStartAllCloudLibraryScansIncludesMergedCloudMounts(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + local := model.Library{Name: "国产剧", Path: "/media/国产剧", Type: "tv", Enabled: true} + cloud := model.Library{Name: "OpenList · 国产剧", Path: BuildCloudLibraryPath("openlist", "/国产剧", "/国产剧"), Type: "tv", Enabled: true} + for _, lib := range []*model.Library{&local, &cloud} { + if err := repos.Library.Create(t.Context(), lib); err != nil { + t.Fatal(err) + } + } + scanner := NewScannerService(&config.Config{}, zap.NewNop(), repos, NewHub(zap.NewNop()), nil, nil) + + statuses, err := scanner.StartAllCloudLibraryScans() + if err != nil { + t.Fatal(err) + } + if len(statuses) != 1 || statuses[0].LibraryID != cloud.ID { + t.Fatalf("scan-all statuses = %#v, want merged cloud library queued", statuses) + } +} + +func libraryNames(libs []model.Library) []string { + out := make([]string, 0, len(libs)) + for _, lib := range libs { + out = append(out, lib.Name) + } + return out +} + +func mediaTitles(items []model.Media) []string { + out := make([]string, 0, len(items)) + for _, item := range items { + out = append(out, item.Title) + } + slices.Sort(out) + return out +} diff --git a/internal/service/emby_compat.go b/internal/service/emby_compat.go index db54f47..af7b4a9 100644 --- a/internal/service/emby_compat.go +++ b/internal/service/emby_compat.go @@ -17,6 +17,7 @@ import ( "encoding/hex" "errors" "fmt" + "net/url" "path/filepath" "regexp" "sort" @@ -31,6 +32,7 @@ import ( "github.com/ShukeBta/MediaStationGo/internal/config" "github.com/ShukeBta/MediaStationGo/internal/model" "github.com/ShukeBta/MediaStationGo/internal/repository" + "github.com/ShukeBta/MediaStationGo/internal/service/cloud" ) // 用一个固定的 ServerId 字符串。Emby 客户端会缓存这个 id，第一次见到 @@ -54,9 +56,11 @@ const PlaybackDirectOnlySettingKey = "playback.direct_only" // EmbyService produces Emby-shaped JSON. type EmbyService struct { - cfg *config.Config - log *zap.Logger - repo *repository.Container + cfg *config.Config + log *zap.Logger + repo *repository.Container + storage cloudPlaybackResolver + probe cloudPlaybackProber virtualMu sync.RWMutex virtualSeries map[string]embySeriesCacheEntry @@ -67,11 +71,27 @@ type EmbyService struct { visibilityCache map[string]embyVisibilityCacheEntry } +type cloudPlaybackResolver interface { + CloudResolve(ctx context.Context, typ, fileRef, clientUA string) (*cloud.DirectLink, error) +} + +type cloudPlaybackProber interface { + ProbeHTTP(ctx context.Context, rawURL string, headers map[string]string) (*ProbeResult, error) +} + // NewEmbyService is the constructor. func NewEmbyService(cfg *config.Config, log *zap.Logger, repo *repository.Container) *EmbyService { return &EmbyService{cfg: cfg, log: log, repo: repo} } +func (e *EmbyService) SetCloudProbe(storage cloudPlaybackResolver, probe cloudPlaybackProber) { + if e == nil { + return + } + e.storage = storage + e.probe = probe +} + // ─── System ────────────────────────────────────────────────────────────────── // SystemInfo returns the full Emby identity payload. @@ -200,7 +220,7 @@ func (e *EmbyService) Views(ctx context.Context, userID string) (map[string]any, if err != nil { return nil, err } - libs = FilterShadowedCloudLibraries(libs) + libs = FilterDisplayCloudLibraries(ctx, e.repo, libs) visibility := e.mediaVisibility(ctx, userID) items := make([]map[string]any, 0, len(libs)) for _, l := range libs { @@ -386,7 +406,7 @@ func (e *EmbyService) mediaItems(ctx context.Context, p ItemsParams) (map[string q := e.repo.DB.WithContext(ctx).Model(&model.Media{}) q = e.applyUserMediaVisibility(ctx, q, p.UserID) if p.ParentID != "" { - q = q.Where("library_id = ? OR series_id = ?", p.ParentID, p.ParentID) + q = q.Where("library_id IN ? OR series_id = ?", e.mergedLibraryIDs(ctx, p.ParentID), p.ParentID) } if p.SearchTerm != "" { q = q.Where("title LIKE ? OR original_name LIKE ?", "%"+p.SearchTerm+"%", "%"+p.SearchTerm+"%") @@ -595,7 +615,7 @@ func (e *EmbyService) latestSeriesItemsForLibrary(ctx context.Context, userID, l rowLimit = embySeriesGroupingLimit } q := e.repo.DB.WithContext(ctx).Model(&model.Media{}). - Where("library_id = ? AND (season_num > 0 OR episode_num > 0)", libraryID) + Where("library_id IN ? AND (season_num > 0 OR episode_num > 0)", e.mergedLibraryIDs(ctx, libraryID)) q = e.applyUserMediaVisibility(ctx, q, userID) var rows []model.Media if err := q.Order("created_at desc").Limit(rowLimit).Find(&rows).Error; err != nil { @@ -741,7 +761,7 @@ func (e *EmbyService) seriesItemsForLibrary(ctx context.Context, libraryID strin q := e.repo.DB.WithContext(ctx).Model(&model.Media{}).Where("season_num > 0 OR episode_num > 0") q = e.applyUserMediaVisibility(ctx, q, p.UserID) if libraryID != "" { - q = q.Where("library_id = ?", libraryID) + q = q.Where("library_id IN ?", e.mergedLibraryIDs(ctx, libraryID)) } if p.SearchTerm != "" { q = q.Where("title LIKE ? OR original_name LIKE ?", "%"+p.SearchTerm+"%", "%"+p.SearchTerm+"%") @@ -781,7 +801,7 @@ func (e *EmbyService) libraryIsEpisodic(ctx context.Context, libraryID string) ( } var count int64 err := e.repo.DB.WithContext(ctx).Model(&model.Media{}). - Where("library_id = ? AND (season_num > 0 OR episode_num > 0)", libraryID). + Where("library_id IN ? AND (season_num > 0 OR episode_num > 0)", e.mergedLibraryIDs(ctx, libraryID)). Count(&count).Error return count > 0, err } @@ -1348,6 +1368,7 @@ func (e *EmbyService) mediaVisibility(ctx context.Context, userID string) MediaV if !visibility.IncludeNSFW { visibility.HiddenLibraryIDs = e.hiddenLibraryIDs(ctx, visibility) } + visibility = ExpandMediaVisibilityForMergedCloudLibraries(ctx, e.repo, visibility) visibility = cloneMediaVisibility(visibility) e.visibilityMu.Lock() @@ -1366,6 +1387,14 @@ func (e *EmbyService) mediaVisibility(ctx context.Context, userID string) MediaV return visibility } +func (e *EmbyService) mergedLibraryIDs(ctx context.Context, libraryID string) []string { + ids, err := MergedLibraryIDsForLibrary(ctx, e.repo, libraryID) + if err != nil || len(ids) == 0 { + return []string{libraryID} + } + return ids +} + func cloneMediaVisibility(visibility MediaVisibility) MediaVisibility { if visibility.AllowedLibraryIDs != nil { visibility.AllowedLibraryIDs = append([]string(nil), visibility.AllowedLibraryIDs...) @@ -1440,12 +1469,99 @@ func (e *EmbyService) PlaybackInfo(ctx context.Context, mediaID, userID string) if err != nil || m == nil { return nil, err } + e.ensureCloudTrackMetadata(ctx, m) return map[string]any{ "MediaSources": []map[string]any{e.mediaSource(m, false, e.directPlayOnly(ctx))}, "PlaySessionId": fmt.Sprintf("%s-%d", m.ID, time.Now().Unix()), }, nil } +func (e *EmbyService) ensureCloudTrackMetadata(ctx context.Context, m *model.Media) { + if e == nil || m == nil || e.storage == nil || e.probe == nil || !mediaTrackMetadataMissing(m) { + return + } + typ, ref, ok := parseCloudMediaPlaybackURL(m.STRMURL) + if !ok { + return + } + probeCtx, cancel := context.WithTimeout(ctx, 8*time.Second) + defer cancel() + link, err := e.storage.CloudResolve(probeCtx, typ, ref, "") + if err != nil { + if e.log != nil { + e.log.Debug("resolve cloud media for playback probe failed", zap.String("media_id", m.ID), zap.Error(err)) + } + return + } + probe, err := e.probe.ProbeHTTP(probeCtx, link.URL, link.Headers) + if err != nil { + if e.log != nil { + e.log.Debug("playback cloud ffprobe failed", zap.String("media_id", m.ID), zap.Error(err)) + } + return + } + updates := probeResultUpdates(probe) + if len(updates) == 0 { + return + } + if err := e.repo.DB.WithContext(ctx).Model(&model.Media{}).Where("id = ?", m.ID).Updates(updates).Error; err != nil && e.log != nil { + e.log.Debug("persist playback cloud probe failed", zap.String("media_id", m.ID), zap.Error(err)) + } + applyProbeResultToMediaValue(m, probe) +} + +func mediaTrackMetadataMissing(m *model.Media) bool { + return m.DurationSec <= 0 || + m.Width <= 0 || + m.Height <= 0 || + strings.TrimSpace(m.VideoCodec) == "" || + strings.TrimSpace(m.AudioCodec) == "" +} + +func parseCloudMediaPlaybackURL(raw string) (string, string, bool) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", "", false + } + u, err := url.Parse(raw) + if err != nil { + return "", "", false + } + path := strings.Trim(u.Path, "/") + const prefix = "api/cloud/play/" + idx := strings.Index(strings.ToLower(path), prefix) + if idx < 0 { + return "", "", false + } + typ := strings.TrimSpace(path[idx+len(prefix):]) + ref := strings.TrimSpace(u.Query().Get("ref")) + return typ, ref, typ != "" && ref != "" +} + +func applyProbeResultToMediaValue(m *model.Media, probe *ProbeResult) { + if m == nil || probe == nil { + return + } + if probe.DurationSec > 0 { + m.DurationSec = probe.DurationSec + } + if probe.Width > 0 { + m.Width = probe.Width + } + if probe.Height > 0 { + m.Height = probe.Height + } + if strings.TrimSpace(probe.VideoCodec) != "" { + m.VideoCodec = probe.VideoCodec + } + if strings.TrimSpace(probe.AudioCodec) != "" { + m.AudioCodec = probe.AudioCodec + } + if strings.TrimSpace(probe.Container) != "" { + m.Container = probe.Container + } +} + // directPlayOnly reports whether the admin enabled「客户端直连解码」mode. // In that mode the host never transcodes; clients must direct-play. func (e *EmbyService) directPlayOnly(ctx context.Context) bool { @@ -1494,6 +1610,13 @@ func (e *EmbyService) mediaSource(m *model.Media, asEmbedded, directOnly bool) m container = "strm" } isCloud := strings.TrimSpace(m.STRMURL) != "" + if isCloud { + // Cloud/WebDAV media is already a direct/proxy stream. Advertising HLS + // transcoding makes some Emby clients pick /master.m3u8, forcing this + // lightweight server to pull remote bytes through ffmpeg and often + // surfacing as "network/playback failed". Keep cloud media direct-only. + directOnly = true + } src := map[string]any{ "Id": m.ID, "Name": m.Title, @@ -1671,5 +1794,3 @@ func intToStr(v int) string { } return strconv.Itoa(v) } - - diff --git a/internal/service/emby_compat_test.go b/internal/service/emby_compat_test.go index f4b39d0..6b93b38 100644 --- a/internal/service/emby_compat_test.go +++ b/internal/service/emby_compat_test.go @@ -11,6 +11,7 @@ import ( "github.com/ShukeBta/MediaStationGo/internal/config" "github.com/ShukeBta/MediaStationGo/internal/model" "github.com/ShukeBta/MediaStationGo/internal/repository" + "github.com/ShukeBta/MediaStationGo/internal/service/cloud" ) func TestEmbyItemsExposeSeriesSeasonEpisodeHierarchy(t *testing.T) { @@ -370,6 +371,67 @@ func TestEmbyPlaybackInfoKeepsSTRMBehindStreamEndpoint(t *testing.T) { } } +func TestEmbyPlaybackInfoProbesMissingCloudTrackMetadata(t *testing.T) { + svc := newTestEmbyService(t) + lib := model.Library{Name: "OpenList", Path: `cloud://openlist/Movies`, Type: "movie", Enabled: true} + if err := svc.repo.Library.Create(t.Context(), &lib); err != nil { + t.Fatalf("create library: %v", err) + } + media := model.Media{ + Base: model.Base{ID: "cloud-probe-1"}, + LibraryID: lib.ID, + Title: "云盘电影", + Path: `cloud://openlist/Movies/Movie.mkv`, + STRMURL: `http://nas.local/api/cloud/play/openlist?ref=%2FMovies%2FMovie.mkv`, + } + if err := svc.repo.DB.Create(&media).Error; err != nil { + t.Fatalf("create media: %v", err) + } + resolver := &fakeCloudPlaybackResolver{ + link: &cloud.DirectLink{ + URL: "http://cdn.example.test/Movie.mkv", + Headers: map[string]string{"Authorization": "Bearer probe-token"}, + }, + } + prober := &fakeCloudPlaybackProber{ + probe: &ProbeResult{ + DurationSec: 3661, + Width: 3840, + Height: 2160, + VideoCodec: "hevc", + AudioCodec: "eac3", + Container: "matroska,webm", + }, + } + svc.SetCloudProbe(resolver, prober) + + pb, err := svc.PlaybackInfo(t.Context(), "cloud-probe-1", "user-1") + if err != nil { + t.Fatalf("playback info: %v", err) + } + if resolver.typ != "openlist" || resolver.ref != "/Movies/Movie.mkv" { + t.Fatalf("resolver called with typ=%q ref=%q", resolver.typ, resolver.ref) + } + if prober.rawURL != "http://cdn.example.test/Movie.mkv" || prober.headers["Authorization"] != "Bearer probe-token" { + t.Fatalf("probe called with url=%q headers=%#v", prober.rawURL, prober.headers) + } + src := pb["MediaSources"].([]map[string]any)[0] + if src["RunTimeTicks"] != int64(3661)*10_000_000 { + t.Fatalf("runtime ticks not filled from probe: %#v", src) + } + streams := src["MediaStreams"].([]map[string]any) + if len(streams) != 2 || streams[0]["Codec"] != "hevc" || streams[1]["Codec"] != "eac3" { + t.Fatalf("media streams not filled from probe: %#v", streams) + } + var persisted model.Media + if err := svc.repo.DB.First(&persisted, "id = ?", "cloud-probe-1").Error; err != nil { + t.Fatalf("reload media: %v", err) + } + if persisted.DurationSec != 3661 || persisted.Width != 3840 || persisted.Height != 2160 || persisted.VideoCodec != "hevc" || persisted.AudioCodec != "eac3" { + t.Fatalf("probe metadata not persisted: %#v", persisted) + } +} + func newTestEmbyService(t *testing.T) *EmbyService { t.Helper() db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) @@ -382,3 +444,29 @@ func newTestEmbyService(t *testing.T) *EmbyService { repos := repository.New(db) return NewEmbyService(&config.Config{}, zap.NewNop(), repos) } + +type fakeCloudPlaybackResolver struct { + link *cloud.DirectLink + typ string + ref string + ua string +} + +func (f *fakeCloudPlaybackResolver) CloudResolve(_ context.Context, typ, fileRef, clientUA string) (*cloud.DirectLink, error) { + f.typ = typ + f.ref = fileRef + f.ua = clientUA + return f.link, nil +} + +type fakeCloudPlaybackProber struct { + probe *ProbeResult + rawURL string + headers map[string]string +} + +func (f *fakeCloudPlaybackProber) ProbeHTTP(_ context.Context, rawURL string, headers map[string]string) (*ProbeResult, error) { + f.rawURL = rawURL + f.headers = headers + return f.probe, nil +} diff --git a/internal/service/ffprobe.go b/internal/service/ffprobe.go index 486d025..21e8852 100644 --- a/internal/service/ffprobe.go +++ b/internal/service/ffprobe.go @@ -101,6 +101,45 @@ func (f *FFprobeService) Probe(ctx context.Context, path string) (*ProbeResult, return f.probeWithFFmpeg(ctx, path) } +// ProbeHTTP runs ffprobe against a remote HTTP(S) media URL. Headers are +// passed to ffprobe/ffmpeg so WebDAV/OpenList/115 links that require cookies, +// authorization, or a provider-specific User-Agent can still expose stream +// metadata without downloading the whole file. +func (f *FFprobeService) ProbeHTTP(ctx context.Context, rawURL string, headers map[string]string) (*ProbeResult, error) { + if f == nil { + return nil, errors.New("ffprobe service nil") + } + rawURL = strings.TrimSpace(rawURL) + if rawURL == "" { + return nil, errors.New("empty probe url") + } + token, err := f.acquire(ctx) + if err != nil { + return nil, err + } + defer f.release(token) + headerText := ffmpegHeaderText(headers) + if bin, err := resolveLocalExecutable(f.cfg.App.FFprobePath, "ffprobe"); err == nil { + f.cfg.App.FFprobePath = bin + probeCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + args := []string{"-v", "error"} + if headerText != "" { + args = append(args, "-headers", headerText) + } + args = append(args, "-print_format", "json", "-show_format", "-show_streams", rawURL) + cmd := exec.CommandContext(probeCtx, bin, args...) // #nosec G204 -- bin is resolved by resolveLocalExecutable before execution. + out, err := cmd.Output() + if err == nil { + return parseProbeJSON(out) + } + if f.log != nil { + f.log.Debug("remote ffprobe failed, trying ffmpeg fallback", zap.Error(err)) + } + } + return f.probeHTTPWithFFmpeg(ctx, rawURL, headerText) +} + func (f *FFprobeService) acquire(ctx context.Context) (chan struct{}, error) { f.mu.RLock() limiter := f.limiter @@ -140,6 +179,44 @@ func (f *FFprobeService) probeWithFFmpeg(ctx context.Context, path string) (*Pro return res, nil } +func (f *FFprobeService) probeHTTPWithFFmpeg(ctx context.Context, rawURL, headerText string) (*ProbeResult, error) { + bin, err := resolveLocalExecutable(f.cfg.App.FFmpegPath, "ffmpeg") + if err != nil { + return nil, fmt.Errorf("ffprobe/ffmpeg unavailable: %w", err) + } + f.cfg.App.FFmpegPath = bin + args := []string{"-hide_banner"} + if headerText != "" { + args = append(args, "-headers", headerText) + } + args = append(args, "-i", rawURL) + out, _ := commandOutput(ctx, 30*time.Second, bin, args...) + res := parseFFmpegProbeText(string(out)) + if res.VideoCodec == "" && res.AudioCodec == "" && res.DurationSec == 0 { + return nil, fmt.Errorf("remote ffmpeg probe: no stream metadata parsed") + } + return res, nil +} + +func ffmpegHeaderText(headers map[string]string) string { + if len(headers) == 0 { + return "" + } + var b strings.Builder + for k, v := range headers { + k = strings.TrimSpace(k) + v = strings.TrimSpace(v) + if k == "" || strings.ContainsAny(k, "\r\n") || strings.ContainsAny(v, "\r\n") { + continue + } + b.WriteString(k) + b.WriteString(": ") + b.WriteString(v) + b.WriteString("\r\n") + } + return b.String() +} + // rawProbe mirrors the relevant fields of `ffprobe -show_format -show_streams`. type rawProbe struct { Format struct { diff --git a/internal/service/media.go b/internal/service/media.go index dc064a6..5356778 100644 --- a/internal/service/media.go +++ b/internal/service/media.go @@ -261,7 +261,12 @@ func (s *MediaService) ListMediaVisible(ctx context.Context, libraryID string, p if page < 1 { page = 1 } - return s.repo.Media.ListByLibraryFiltered(ctx, libraryID, (page-1)*pageSize, pageSize, repository.MediaQueryFilter{ + visibility = ExpandMediaVisibilityForMergedCloudLibraries(ctx, s.repo, visibility) + libraryIDs, err := MergedLibraryIDsForLibrary(ctx, s.repo, libraryID) + if err != nil { + return nil, 0, err + } + return s.repo.Media.ListByLibrariesFiltered(ctx, libraryIDs, (page-1)*pageSize, pageSize, repository.MediaQueryFilter{ IncludeNSFW: visibility.IncludeNSFW, AllowedLibraryIDs: visibility.AllowedLibraryIDs, HiddenLibraryIDs: visibility.HiddenLibraryIDs, @@ -279,6 +284,7 @@ func (s *MediaService) SearchMediaVisible(ctx context.Context, query string, lim } else if limit > 2000 { limit = 2000 } + visibility = ExpandMediaVisibilityForMergedCloudLibraries(ctx, s.repo, visibility) return s.repo.Media.SearchFiltered(ctx, query, limit, repository.MediaQueryFilter{ IncludeNSFW: visibility.IncludeNSFW, AllowedLibraryIDs: visibility.AllowedLibraryIDs, diff --git a/internal/service/scanner.go b/internal/service/scanner.go index 24135fb..701b8a8 100644 --- a/internal/service/scanner.go +++ b/internal/service/scanner.go @@ -17,6 +17,7 @@ import ( "net/url" "os" "path/filepath" + "sort" "strings" "sync" "time" @@ -65,6 +66,11 @@ type ScannerService struct { cloudImagePrefetchQueue chan cloudImagePrefetchTask cloudImagePrefetchMu sync.Mutex cloudImagePrefetching map[string]struct{} + cloudMediaProbeOnce sync.Once + cloudMediaProbeQueue chan cloudMediaProbeTask + cloudMediaProbeMu sync.Mutex + cloudMediaProbing map[string]struct{} + cloudMediaProbeBackoff map[string]time.Time } // NewScannerService is the constructor. @@ -84,6 +90,9 @@ func NewScannerService( cloudSlots: make(chan struct{}, 1), cloudImagePrefetchQueue: make(chan cloudImagePrefetchTask, 256), cloudImagePrefetching: make(map[string]struct{}), + cloudMediaProbeQueue: make(chan cloudMediaProbeTask, 1024), + cloudMediaProbing: make(map[string]struct{}), + cloudMediaProbeBackoff: make(map[string]time.Time), } } @@ -92,6 +101,14 @@ func NewScannerService( // while the scanner is needed earlier by watcher/download services. func (s *ScannerService) SetStorageConfig(storage *StorageConfigService) { s.storage = storage + if storage != nil && s.probe != nil { + s.cloudMediaProbeOnce.Do(func() { + workers := normalizeFFprobeMaxConcurrent(s.cfg.App.FFprobeMaxConcurrent) + for i := 0; i < workers; i++ { + go s.cloudMediaProbeWorker() + } + }) + } } // SetImageProxy lets cloud scans warm sidecar poster/backdrop files into the @@ -248,6 +265,10 @@ type ScanResult struct { var ErrCloudScanAlreadyRunning = errors.New("cloud scan already running") +const maxCloudMediaProbeQueuePerScan = 32 + +const cloudMediaProbeFailureBackoff = 6 * time.Hour + // CloudScanStatus is the operator-facing state for long-running cloud scans. type CloudScanStatus struct { LibraryID string `json:"library_id"` @@ -281,6 +302,86 @@ type cloudImagePrefetchTask struct { stableKey string } +type cloudMediaProbeTask struct { + typ string + ref string + path string +} + +type existingCloudMedia struct { + SizeBytes int64 + DurationSec int + Width int + Height int + VideoCodec string + AudioCodec string + Container string + PosterURL string + BackdropURL string + STRMURL string +} + +func (s *ScannerService) cloudMediaProbeWorker() { + for task := range s.cloudMediaProbeQueue { + s.probeCloudMediaAsync(task) + } +} + +func (s *ScannerService) queueCloudMediaProbe(typ, ref, path string) bool { + if s == nil || s.storage == nil || s.probe == nil { + return false + } + typ = strings.TrimSpace(typ) + ref = strings.TrimSpace(ref) + path = strings.TrimSpace(path) + if typ == "" || ref == "" || path == "" { + return false + } + s.cloudMediaProbeMu.Lock() + if until, ok := s.cloudMediaProbeBackoff[path]; ok { + if time.Now().Before(until) { + s.cloudMediaProbeMu.Unlock() + return false + } + delete(s.cloudMediaProbeBackoff, path) + } + if _, ok := s.cloudMediaProbing[path]; ok { + s.cloudMediaProbeMu.Unlock() + return false + } + s.cloudMediaProbing[path] = struct{}{} + s.cloudMediaProbeMu.Unlock() + + task := cloudMediaProbeTask{typ: typ, ref: ref, path: path} + select { + case s.cloudMediaProbeQueue <- task: + return true + default: + s.cloudMediaProbeMu.Lock() + delete(s.cloudMediaProbing, path) + s.cloudMediaProbeMu.Unlock() + if s.log != nil { + s.log.Warn("cloud media probe queue full", zap.String("provider", typ), zap.String("path", path)) + } + return false + } +} + +func (s *ScannerService) queueCloudMediaProbeWithBudget(typ, ref, path string, budget *int) bool { + if budget != nil { + if *budget <= 0 { + return false + } + } + if !s.queueCloudMediaProbe(typ, ref, path) { + return false + } + if budget != nil { + *budget-- + } + return true +} + func (s *ScannerService) beginCloudScan(ctx context.Context, lib *model.Library, mount CloudMountInfo) (context.Context, func(*ScanResult, error), error) { if s == nil || lib == nil { return ctx, func(*ScanResult, error) {}, nil @@ -549,7 +650,7 @@ func (s *ScannerService) StartAllCloudLibraryScans() ([]CloudScanStatus, error) if err != nil { return nil, err } - libs = FilterShadowedCloudLibraries(libs) + libs = FilterScannableCloudLibraries(context.Background(), s.repo, libs) statuses := make([]CloudScanStatus, 0, len(libs)) for _, lib := range libs { if !lib.Enabled { @@ -697,7 +798,7 @@ func (s *ScannerService) scanCloudLibrary(ctx context.Context, lib *model.Librar if s.storage == nil { return res, fmt.Errorf("cloud storage service unavailable") } - + // 验证存储配置是否存在且已启用 cfg, err := s.repo.StorageConfig.Get(ctx, mount.Provider) if err != nil || cfg == nil { @@ -842,11 +943,27 @@ func (s *ScannerService) scanCloudLibrary(ctx context.Context, lib *model.Librar if err := walkCloud(rootDir, rootDisplayDir, nil); err != nil { return res, err } - existingPaths, err := s.existingCloudMediaPaths(ctx, lib.ID) + existingMedia, err := s.existingCloudMediaSnapshot(ctx, lib.ID) if err != nil { - s.log.Warn("load existing cloud media paths failed", zap.String("library_id", lib.ID), zap.Error(err)) - existingPaths = nil + s.log.Warn("load existing cloud media snapshot failed", zap.String("library_id", lib.ID), zap.Error(err)) + existingMedia = nil } + if existingMedia != nil { + priority := func(candidate cloudCandidate) int { + existing, ok := existingMedia[candidate.path] + if !ok { + return 2 + } + if cloudTrackMetadataMissing(existing) || cloudMetadataNeedsRefresh(existing, candidate.localMeta) { + return 0 + } + return 1 + } + sort.SliceStable(candidates, func(i, j int) bool { + return priority(candidates[i]) < priority(candidates[j]) + }) + } + probeBudget := maxCloudMediaProbeQueuePerScan for _, candidate := range candidates { select { case <-ctx.Done(): @@ -854,7 +971,7 @@ func (s *ScannerService) scanCloudLibrary(ctx context.Context, lib *model.Librar default: } seen[candidate.path] = struct{}{} - s.ingestCloudFile(ctx, lib, typ, candidate.ref, candidate.path, candidate.name, candidate.size, candidate.localMeta, existingPaths, res) + s.ingestCloudFile(ctx, lib, typ, candidate.ref, candidate.path, candidate.name, candidate.size, candidate.localMeta, existingMedia, &probeBudget, res) publishProgress("importing", res.Visited == 1 || res.Visited%100 == 0) } removed, err := s.pruneMissingCloudMedia(ctx, lib.ID, seen) @@ -892,21 +1009,42 @@ func (s *ScannerService) startAutoScrape(ctx context.Context, libraryID string) }() } -func (s *ScannerService) existingCloudMediaPaths(ctx context.Context, libraryID string) (map[string]struct{}, error) { +func (s *ScannerService) existingCloudMediaSnapshot(ctx context.Context, libraryID string) (map[string]existingCloudMedia, error) { var rows []struct { - Path string + Path string + SizeBytes int64 + DurationSec int + Width int + Height int + VideoCodec string + AudioCodec string + Container string + PosterURL string + BackdropURL string + STRMURL string } if err := s.repo.DB.WithContext(ctx). Model(&model.Media{}). - Select("path"). + Select("path, size_bytes, duration_sec, width, height, video_codec, audio_codec, container, poster_url, backdrop_url, strm_url"). Where("library_id = ? AND path LIKE ?", libraryID, "cloud://%"). Find(&rows).Error; err != nil { return nil, err } - out := make(map[string]struct{}, len(rows)) + out := make(map[string]existingCloudMedia, len(rows)) for _, row := range rows { if row.Path != "" { - out[row.Path] = struct{}{} + out[row.Path] = existingCloudMedia{ + SizeBytes: row.SizeBytes, + DurationSec: row.DurationSec, + Width: row.Width, + Height: row.Height, + VideoCodec: row.VideoCodec, + AudioCodec: row.AudioCodec, + Container: row.Container, + PosterURL: row.PosterURL, + BackdropURL: row.BackdropURL, + STRMURL: row.STRMURL, + } } } return out, nil @@ -918,10 +1056,34 @@ func (s *ScannerService) shadowedCloudLibrary(ctx context.Context, lib *model.Li s.log.Warn("list libraries for cloud shadow check failed", zap.String("library_id", lib.ID), zap.Error(err)) return nil } + visible := FilterScannableCloudLibraries(ctx, s.repo, libs) + for _, kept := range visible { + if kept.ID == lib.ID { + return nil + } + } + current, ok := ParseCloudLibraryMount(lib.Path) + if ok { + currentKey, _ := cloudLibraryDisplayKey(*lib) + for _, kept := range visible { + info, ok := ParseCloudLibraryMount(kept.Path) + if !ok || info.Provider != current.Provider { + continue + } + keptKey, _ := cloudLibraryDisplayKey(kept) + exact := currentKey != "" && currentKey == keptKey + return &CloudMountConflict{ + Library: kept, + Exact: exact, + Nested: !exact, + ExistingIsAncestor: cloudMountAncestor(info.DisplayDir, current.DisplayDir), + } + } + } return CloudLibraryShadowed(libs, *lib) } -func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library, typ, ref, path, name string, size int64, localMeta *LocalMetadata, existingPaths map[string]struct{}, res *ScanResult) { +func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library, typ, ref, path, name string, size int64, localMeta *LocalMetadata, existingMedia map[string]existingCloudMedia, probeBudget *int, res *ScanResult) { res.Visited++ ext := strings.ToLower(filepath.Ext(name)) title, year := CleanQuery(name) @@ -940,10 +1102,20 @@ func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library } } } + expectedSTRMURL := BuildPublicAPIURL(ctx, s.repo, s.cfg, "/api/cloud/play/"+typ, url.Values{"ref": []string{ref}}) isNewMedia := false - if existingPaths != nil { - _, exists := existingPaths[path] + needsTrackProbe := true + if existingMedia != nil { + existing, exists := existingMedia[path] isNewMedia = !exists + needsTrackProbe = !exists || cloudTrackMetadataMissing(existing) + if exists && existing.SizeBytes == size && existing.STRMURL == expectedSTRMURL && !cloudMetadataNeedsRefresh(existing, localMeta) { + if needsTrackProbe && ext != ".strm" { + s.queueCloudMediaProbeWithBudget(typ, ref, path, probeBudget) + } + res.Skipped++ + return + } } else { isNewMedia = !s.mediaPathExists(ctx, path) } @@ -954,7 +1126,7 @@ func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library Path: path, SizeBytes: size, Container: strings.TrimPrefix(ext, "."), - STRMURL: BuildPublicAPIURL(ctx, s.repo, s.cfg, "/api/cloud/play/"+typ, url.Values{"ref": []string{ref}}), + STRMURL: expectedSTRMURL, ScrapeStatus: "pending", } if ext == ".strm" { @@ -976,6 +1148,9 @@ func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library s.log.Warn("upsert cloud media failed", zap.String("path", path), zap.Error(err)) return } + if needsTrackProbe && ext != ".strm" { + s.queueCloudMediaProbeWithBudget(typ, ref, path, probeBudget) + } if isNewMedia { res.Added++ } else { @@ -993,6 +1168,113 @@ func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library } } +func (s *ScannerService) probeCloudMediaAsync(task cloudMediaProbeTask) { + defer func() { + s.cloudMediaProbeMu.Lock() + delete(s.cloudMediaProbing, task.path) + s.cloudMediaProbeMu.Unlock() + }() + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + probe, err := s.probeCloudFileMetadata(ctx, task.typ, task.ref) + if err != nil { + if s.log != nil { + s.log.Debug("cloud media async probe failed", zap.String("provider", task.typ), zap.String("path", task.path), zap.Error(err)) + } + s.cloudMediaProbeMu.Lock() + if s.cloudMediaProbeBackoff == nil { + s.cloudMediaProbeBackoff = make(map[string]time.Time) + } + s.cloudMediaProbeBackoff[task.path] = time.Now().Add(cloudMediaProbeFailureBackoff) + s.cloudMediaProbeMu.Unlock() + return + } + updates := probeResultUpdates(probe) + if len(updates) == 0 { + return + } + if err := s.repo.DB.WithContext(ctx).Model(&model.Media{}).Where("path = ?", task.path).Updates(updates).Error; err != nil { + if s.log != nil { + s.log.Debug("update cloud media track metadata failed", zap.String("path", task.path), zap.Error(err)) + } + return + } + s.cloudMediaProbeMu.Lock() + delete(s.cloudMediaProbeBackoff, task.path) + s.cloudMediaProbeMu.Unlock() + if s.hub != nil { + s.hub.Publish("scan", map[string]any{ + "path": task.path, + "cloud": true, + "track_probed": true, + "duration_sec": probe.DurationSec, + "video_codec": probe.VideoCodec, + "audio_codec": probe.AudioCodec, + "width": probe.Width, + "height": probe.Height, + "probe_message": "云盘媒体轨道元数据已后台补齐", + }) + } +} + +func (s *ScannerService) probeCloudFileMetadata(ctx context.Context, typ, ref string) (*ProbeResult, error) { + if s == nil || s.probe == nil || s.storage == nil { + return nil, errors.New("cloud probe unavailable") + } + link, err := s.storage.CloudResolve(ctx, typ, ref, "") + if err != nil { + return nil, err + } + return s.probe.ProbeHTTP(ctx, link.URL, link.Headers) +} + +func probeResultUpdates(probe *ProbeResult) map[string]any { + updates := map[string]any{} + if probe == nil { + return updates + } + if probe.DurationSec > 0 { + updates["duration_sec"] = probe.DurationSec + } + if probe.Width > 0 { + updates["width"] = probe.Width + } + if probe.Height > 0 { + updates["height"] = probe.Height + } + if strings.TrimSpace(probe.VideoCodec) != "" { + updates["video_codec"] = probe.VideoCodec + } + if strings.TrimSpace(probe.AudioCodec) != "" { + updates["audio_codec"] = probe.AudioCodec + } + if probe.Container != "" { + updates["container"] = probe.Container + } + return updates +} + +func cloudMetadataNeedsRefresh(existing existingCloudMedia, localMeta *LocalMetadata) bool { + if localMeta == nil { + return false + } + if strings.TrimSpace(localMeta.PosterURL) != "" && strings.TrimSpace(existing.PosterURL) == "" { + return true + } + if strings.TrimSpace(localMeta.BackdropURL) != "" && strings.TrimSpace(existing.BackdropURL) == "" { + return true + } + return false +} + +func cloudTrackMetadataMissing(existing existingCloudMedia) bool { + return existing.DurationSec <= 0 || + existing.Width <= 0 || + existing.Height <= 0 || + strings.TrimSpace(existing.VideoCodec) == "" || + strings.TrimSpace(existing.AudioCodec) == "" +} + func cloudSeriesTitleFromMediaPath(mediaPath string) (string, int) { displayPath := strings.TrimSpace(mediaPath) if strings.HasPrefix(strings.ToLower(displayPath), "cloud://") { @@ -1437,5 +1719,3 @@ func (s *ScannerService) maybeGenerateSTRMAfterScan(libraryID string) { } }() } - - diff --git a/internal/service/scanner_cloud_test.go b/internal/service/scanner_cloud_test.go index ab71a40..0570569 100644 --- a/internal/service/scanner_cloud_test.go +++ b/internal/service/scanner_cloud_test.go @@ -1,6 +1,8 @@ package service import ( + "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -96,8 +98,8 @@ func TestScanCloudLibraryImportsRecursivePlayableMedia(t *testing.T) { if err != nil { t.Fatalf("rescan same cloud: %v", err) } - if res.Added != 0 || res.Updated != 2 { - t.Fatalf("same cloud rescan should update existing rows only, got %#v", res) + if res.Added != 0 || res.Updated != 0 || res.Skipped != 2 { + t.Fatalf("same cloud rescan should skip unchanged rows, got %#v", res) } empty = true @@ -493,6 +495,181 @@ func TestScanCloudLibraryReadsRemoteNFOAndArtwork(t *testing.T) { } } +func TestScanOpenListCloudLibraryUsesAPIPaginationBeyondFirstPage(t *testing.T) { + const totalFiles = 125 + requestedPages := map[int]bool{} + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/fs/list" { + t.Fatalf("unexpected path %s", r.URL.Path) + } + if r.Method != http.MethodPost { + t.Fatalf("unexpected method %s", r.Method) + } + if r.Header.Get("Authorization") != "openlist-token" { + t.Fatalf("missing openlist token: %q", r.Header.Get("Authorization")) + } + var in struct { + Path string `json:"path"` + Page int `json:"page"` + PerPage int `json:"per_page"` + } + if err := json.NewDecoder(r.Body).Decode(&in); err != nil { + t.Fatalf("decode request: %v", err) + } + if in.Path != "/Movies" { + t.Fatalf("path = %q, want /Movies", in.Path) + } + if in.PerPage <= 100 { + t.Fatalf("per_page = %d, want API pagination larger than legacy 100", in.PerPage) + } + requestedPages[in.Page] = true + effectivePageSize := in.PerPage + if effectivePageSize > 100 { + effectivePageSize = 100 + } + start := (in.Page - 1) * effectivePageSize + content := []map[string]any{} + for idx := start; idx < totalFiles && idx < start+effectivePageSize; idx++ { + content = append(content, map[string]any{ + "name": fmt.Sprintf("Movie.%03d.mkv", idx+1), + "size": int64(1024 + idx), + "is_dir": false, + }) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "code": 200, + "message": "success", + "data": map[string]any{ + "content": content, + "total": totalFiles, + }, + }) + })) + defer upstream.Close() + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}, &model.Setting{}, &model.StorageConfig{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + log := zap.NewNop() + storage := NewStorageConfigService(log, repos, NewCryptoService("", log)) + if _, err := storage.Save(t.Context(), StorageInput{ + Type: "openlist", + Config: map[string]any{ + "server": upstream.URL, + "token": "openlist-token", + }, + }); err != nil { + t.Fatal(err) + } + lib := model.Library{Name: "OpenList · Movies", Path: BuildCloudLibraryPath("openlist", "/Movies", "/Movies"), Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + scanner := NewScannerService(&config.Config{}, log, repos, NewHub(log), nil, nil) + scanner.SetStorageConfig(storage) + + res, err := scanner.ScanLibrary(t.Context(), lib.ID) + if err != nil { + t.Fatalf("scan openlist: %v", err) + } + if res.Added != totalFiles { + t.Fatalf("scan result = %#v, want added=%d", res, totalFiles) + } + if got := countMedia(t, repos); got != totalFiles { + t.Fatalf("media count = %d, want %d", got, totalFiles) + } + if !requestedPages[1] || !requestedPages[2] { + t.Fatalf("expected pagination beyond the first 100 entries, got pages %#v", requestedPages) + } +} + +func TestScanCloudLibraryQueuesMissingExistingTrackMetadataBeforeNewFiles(t *testing.T) { + const newFiles = maxCloudMediaProbeQueuePerScan + 5 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/file/sort" || r.URL.Query().Get("pdir_fid") != "0" { + t.Fatalf("unexpected cloud list request %s?%s", r.URL.Path, r.URL.RawQuery) + } + var b strings.Builder + b.WriteString(`{"status":200,"code":0,"data":{"list":[`) + for i := 0; i < newFiles; i++ { + if i > 0 { + b.WriteByte(',') + } + _, _ = fmt.Fprintf(&b, `{"fid":"new-%02d","file_name":"New.Movie.%02d.mkv","dir":false,"size":%d}`, i, i, 1000+i) + } + _, _ = fmt.Fprintf(&b, `,{"fid":"existing","file_name":"Existing.Show.S01E01.mkv","dir":false,"size":2048}]}}`) + _, _ = w.Write([]byte(b.String())) + })) + defer upstream.Close() + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}, &model.Setting{}, &model.StorageConfig{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + log := zap.NewNop() + storage := NewStorageConfigService(log, repos, NewCryptoService("", log)) + if _, err := storage.Save(t.Context(), StorageInput{ + Type: "quark", + Config: map[string]any{ + "cookie": "kps=test", + "base": upstream.URL, + }, + }); err != nil { + t.Fatal(err) + } + lib := model.Library{Name: "夸克网盘", Path: "cloud://quark/0", Type: "tv", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + existingPath := "cloud://quark/Existing.Show.S01E01.mkv" + if err := repos.DB.Create(&model.Media{ + LibraryID: lib.ID, + Title: "Existing Show", + Path: existingPath, + SizeBytes: 2048, + Container: "mkv", + STRMURL: "/api/cloud/play/quark?ref=existing", + SeasonNum: 1, + EpisodeNum: 1, + }).Error; err != nil { + t.Fatal(err) + } + scanner := NewScannerService(&config.Config{}, log, repos, NewHub(log), NewFFprobeService(&config.Config{}, log), nil) + scanner.storage = storage + + res, err := scanner.ScanLibrary(t.Context(), lib.ID) + if err != nil { + t.Fatalf("scan cloud: %v", err) + } + if res.Added != newFiles || res.Skipped != 1 { + t.Fatalf("scan result = %#v, want new files added and existing skipped", res) + } + foundExistingProbe := false + for { + select { + case task := <-scanner.cloudMediaProbeQueue: + if task.path == existingPath { + foundExistingProbe = true + } + default: + if !foundExistingProbe { + t.Fatal("existing media missing track metadata did not receive probe budget before new files") + } + return + } + } +} + func TestParseCloudImagePlaybackURL(t *testing.T) { typ, ref, ok := parseCloudImagePlaybackURL("http://nas.local/api/cloud/play/openlist?ref=%2FAnime%2FJianLai%2Fposter.jpg") if !ok || typ != "openlist" || ref != "/Anime/JianLai/poster.jpg" { diff --git a/internal/service/scheduler.go b/internal/service/scheduler.go index 1ec404c..d47a9d9 100644 --- a/internal/service/scheduler.go +++ b/internal/service/scheduler.go @@ -47,6 +47,7 @@ type SchedulerService struct { storageCfg *StorageConfigService hub *Hub cacheDir string + now func() time.Time mu sync.Mutex stopCh chan struct{} @@ -64,6 +65,15 @@ type scheduledJob struct { type schedulerManualRunKey struct{} +const ( + cloudAutoSyncEnabledKey = "cloud.auto_sync_enabled" + cloudSyncIntervalSecondsKey = "cloud.sync_interval_seconds" + cloudLastAutoSyncDateKey = "cloud.last_auto_sync_date" + cloudAutoSyncWindowStartHour = 19 + cloudAutoSyncWindowEndHour = 21 + cloudAutoSyncCompletedDateForm = "2006-01-02" +) + // NewSchedulerService is the constructor. func NewSchedulerService( log *zap.Logger, @@ -84,6 +94,7 @@ func NewSchedulerService( storageCfg: storageCfg, hub: hub, cacheDir: cacheDir, + now: time.Now, stopCh: make(chan struct{}), } } @@ -214,7 +225,7 @@ func (s *SchedulerService) loopWithInitialDelay(ctx context.Context, j *schedule func (s *SchedulerService) runOnce(ctx context.Context, j *scheduledJob) error { err := j.run(ctx) s.mu.Lock() - j.lastRun = time.Now() + j.lastRun = s.currentTime() if err != nil { j.lastErr = err.Error() } else { @@ -336,24 +347,35 @@ func (s *SchedulerService) cloudUploadInterval(ctx context.Context) time.Duratio // /api/cloud/play 302/proxy. func (s *SchedulerService) jobSyncCloudLibraries(ctx context.Context) error { manual, _ := ctx.Value(schedulerManualRunKey{}).(bool) - if s.scanner == nil || (!manual && !s.autoCloudSyncEnabled(ctx)) { + if s.scanner == nil || (!manual && !s.autoCloudSyncDue(ctx, s.currentTime())) { return nil } libs, err := s.repo.Library.List(ctx) if err != nil { return err } + libs = FilterScannableCloudLibraries(ctx, s.repo, libs) + var firstErr error for _, l := range libs { if !l.Enabled { continue } - if _, _, ok := parseCloudLibraryPath(l.Path); !ok { + if _, ok := ParseCloudLibraryMount(l.Path); !ok { continue } if _, err := s.scanner.ScanLibraryWithoutAutoScrape(ctx, l.ID); err != nil { s.log.Warn("cloud sync failed", zap.String("library", l.ID), zap.Error(err)) + if firstErr == nil { + firstErr = err + } } } + if firstErr != nil { + return firstErr + } + if !manual { + _ = s.markCloudAutoSyncCompleted(ctx, s.currentTime()) + } return nil } @@ -361,19 +383,45 @@ func (s *SchedulerService) autoCloudSyncEnabled(ctx context.Context) bool { if s.repo == nil || s.repo.Setting == nil { return false } - v, err := s.repo.Setting.Get(ctx, "cloud.auto_sync_enabled") + v, err := s.repo.Setting.Get(ctx, cloudAutoSyncEnabledKey) if err != nil { return false } return parseBoolSetting(v, false) } +func (s *SchedulerService) autoCloudSyncDue(ctx context.Context, now time.Time) bool { + if !s.autoCloudSyncEnabled(ctx) || !cloudAutoSyncInWindow(now) { + return false + } + if s.repo == nil || s.repo.Setting == nil { + return true + } + last, err := s.repo.Setting.Get(ctx, cloudLastAutoSyncDateKey) + if err != nil { + return true + } + return strings.TrimSpace(last) != now.Format(cloudAutoSyncCompletedDateForm) +} + +func cloudAutoSyncInWindow(now time.Time) bool { + hour := now.In(time.Local).Hour() + return hour >= cloudAutoSyncWindowStartHour && hour < cloudAutoSyncWindowEndHour +} + +func (s *SchedulerService) markCloudAutoSyncCompleted(ctx context.Context, now time.Time) error { + if s.repo == nil || s.repo.Setting == nil { + return nil + } + return s.repo.Setting.Set(ctx, cloudLastAutoSyncDateKey, now.Format(cloudAutoSyncCompletedDateForm)) +} + func (s *SchedulerService) cloudSyncInterval(ctx context.Context) time.Duration { const fallback = 30 * time.Minute if s.repo == nil || s.repo.Setting == nil { return fallback } - v, err := s.repo.Setting.Get(ctx, "cloud.sync_interval_seconds") + v, err := s.repo.Setting.Get(ctx, cloudSyncIntervalSecondsKey) if err != nil { return fallback } @@ -387,6 +435,13 @@ func (s *SchedulerService) cloudSyncInterval(ctx context.Context) time.Duration return time.Duration(seconds) * time.Second } +func (s *SchedulerService) currentTime() time.Time { + if s != nil && s.now != nil { + return s.now() + } + return time.Now() +} + // periodicScanEnabled reports whether the operator opted into periodic full // library re-scans. Defaults to false so the incremental watcher is the only // thing touching the disk under normal operation. diff --git a/internal/service/scheduler_test.go b/internal/service/scheduler_test.go index 5478e95..fe27579 100644 --- a/internal/service/scheduler_test.go +++ b/internal/service/scheduler_test.go @@ -144,7 +144,11 @@ func TestSchedulerCloudSyncImportsMountedCloudLibrary(t *testing.T) { }); err != nil { t.Fatal(err) } - lib := model.Library{Name: "夸克网盘", Path: "cloud://quark/0", Type: "movie", Enabled: true} + local := model.Library{Name: "电影", Path: "/media/电影", Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &local); err != nil { + t.Fatal(err) + } + lib := model.Library{Name: "夸克网盘 · 电影", Path: "cloud://quark/0", Type: "movie", Enabled: true} if err := repos.Library.Create(t.Context(), &lib); err != nil { t.Fatal(err) } @@ -154,6 +158,7 @@ func TestSchedulerCloudSyncImportsMountedCloudLibrary(t *testing.T) { scanner := NewScannerService(&config.Config{}, log, repos, NewHub(log), nil, nil) scanner.SetStorageConfig(storage) scheduler := NewSchedulerService(log, repos, scanner, nil, nil, storage, NewHub(log), "") + scheduler.now = func() time.Time { return fixedNightlySyncTime() } if err := scheduler.jobSyncCloudLibraries(t.Context()); err != nil { t.Fatalf("cloud sync: %v", err) @@ -167,6 +172,134 @@ func TestSchedulerCloudSyncImportsMountedCloudLibrary(t *testing.T) { } } +func TestSchedulerCloudSyncRunsOnlyOnceInsideNightlyWindow(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if r.URL.Path != "/file/sort" || r.URL.Query().Get("pdir_fid") != "0" { + t.Fatalf("unexpected cloud list request %s?%s", r.URL.Path, r.URL.RawQuery) + } + _, _ = w.Write([]byte(`{"status":200,"code":0,"data":{"list":[ + {"fid":"f1","file_name":"Nightly.Cloud.Movie.2026.mkv","dir":false,"size":1024} + ]}}`)) + })) + defer upstream.Close() + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}, &model.Setting{}, &model.StorageConfig{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + log := zap.NewNop() + storage := NewStorageConfigService(log, repos, NewCryptoService("", log)) + if _, err := storage.Save(t.Context(), StorageInput{ + Type: "quark", + Config: map[string]any{ + "cookie": "kps=test", + "base": upstream.URL, + }, + }); err != nil { + t.Fatal(err) + } + lib := model.Library{Name: "夸克网盘", Path: "cloud://quark/0", Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + if err := repos.Setting.Set(t.Context(), cloudAutoSyncEnabledKey, "true"); err != nil { + t.Fatal(err) + } + scanner := NewScannerService(&config.Config{}, log, repos, NewHub(log), nil, nil) + scanner.SetStorageConfig(storage) + scheduler := NewSchedulerService(log, repos, scanner, nil, nil, storage, NewHub(log), "") + + scheduler.now = func() time.Time { + return time.Date(2026, 6, 11, 18, 30, 0, 0, time.Local) + } + if err := scheduler.jobSyncCloudLibraries(t.Context()); err != nil { + t.Fatalf("cloud sync outside window: %v", err) + } + if got := requests.Load(); got != 0 { + t.Fatalf("outside nightly window made %d requests, want 0", got) + } + + scheduler.now = func() time.Time { return fixedNightlySyncTime() } + if err := scheduler.jobSyncCloudLibraries(t.Context()); err != nil { + t.Fatalf("cloud sync inside window: %v", err) + } + if got := requests.Load(); got != 1 { + t.Fatalf("inside nightly window requests = %d, want 1", got) + } + if got := countMedia(t, repos); got != 1 { + t.Fatalf("media count = %d, want 1", got) + } + + scheduler.now = func() time.Time { + return time.Date(2026, 6, 11, 20, 15, 0, 0, time.Local) + } + if err := scheduler.jobSyncCloudLibraries(t.Context()); err != nil { + t.Fatalf("second cloud sync same day: %v", err) + } + if got := requests.Load(); got != 1 { + t.Fatalf("same-day auto sync should not rerun, requests = %d", got) + } +} + +func TestSchedulerRunNowCloudSyncBypassesNightlyWindow(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + _, _ = w.Write([]byte(`{"status":200,"code":0,"data":{"list":[ + {"fid":"f1","file_name":"Manual.Cloud.Movie.2026.mkv","dir":false,"size":1024} + ]}}`)) + })) + defer upstream.Close() + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Library{}, &model.Media{}, &model.Setting{}, &model.StorageConfig{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + log := zap.NewNop() + storage := NewStorageConfigService(log, repos, NewCryptoService("", log)) + if _, err := storage.Save(t.Context(), StorageInput{ + Type: "quark", + Config: map[string]any{ + "cookie": "kps=test", + "base": upstream.URL, + }, + }); err != nil { + t.Fatal(err) + } + lib := model.Library{Name: "夸克网盘", Path: "cloud://quark/0", Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + scanner := NewScannerService(&config.Config{}, log, repos, NewHub(log), nil, nil) + scanner.SetStorageConfig(storage) + scheduler := NewSchedulerService(log, repos, scanner, nil, nil, storage, NewHub(log), "") + scheduler.now = func() time.Time { + return time.Date(2026, 6, 11, 10, 0, 0, 0, time.Local) + } + scheduler.jobs = []*scheduledJob{{ + name: "cloud_sync", + interval: time.Minute, + run: scheduler.jobSyncCloudLibraries, + }} + + if err := scheduler.RunNow(t.Context(), "cloud_sync"); err != nil { + t.Fatalf("manual cloud sync: %v", err) + } + if got := requests.Load(); got != 1 { + t.Fatalf("manual cloud sync requests = %d, want 1", got) + } +} + func TestSchedulerCloudSyncDisabledByDefault(t *testing.T) { var requests atomic.Int32 upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -215,6 +348,10 @@ func TestSchedulerCloudSyncDisabledByDefault(t *testing.T) { } } +func fixedNightlySyncTime() time.Time { + return time.Date(2026, 6, 11, 19, 30, 0, 0, time.Local) +} + func TestSchedulerLoopWaitsIntervalAfterSlowRun(t *testing.T) { scheduler := NewSchedulerService(zap.NewNop(), nil, nil, nil, nil, nil, nil, "") ctx, cancel := context.WithCancel(t.Context()) diff --git a/internal/service/service.go b/internal/service/service.go index 747ed53..fee4398 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -73,7 +73,8 @@ type Container struct { Device *DeviceService stopCtx context.Context - stopCancel context.CancelFunc} + stopCancel context.CancelFunc +} // New 构建服务容器。 func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Container { @@ -116,6 +117,7 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont storageCfg := NewStorageConfigService(log, repos, crypto) strmSvc := NewSTRMService(log, repos, cfg) scanner.SetStorageConfig(storageCfg) + emby.SetCloudProbe(storageCfg, probe) downloadClients := NewDownloadClientService(log, repos) assistant := NewAssistantService(log, repos, ai) douban := NewDoubanProvider(cfg, log) @@ -239,6 +241,7 @@ func (c *Container) Boot() { if err := c.NormalizeCloudLibraryTypes(c.stopCtx); err != nil { c.Log.Warn("normalize cloud library types failed", zap.Error(err)) } + go c.warmMediaSearchIndex(c.stopCtx) // 加载所有已配置的下载客户端 if err := c.DownloadMgr.LoadAll(c.stopCtx); err != nil { @@ -261,6 +264,39 @@ func (c *Container) Boot() { } } +func (c *Container) warmMediaSearchIndex(ctx context.Context) { + if c == nil || c.Repo == nil || c.Repo.Media == nil { + return + } + const batchSize = 1000 + const pause = 100 * time.Millisecond + total := int64(0) + for { + select { + case <-ctx.Done(): + return + default: + } + n, err := c.Repo.Media.BackfillSearchIndex(ctx, batchSize) + if err != nil { + c.Log.Debug("media search index warmup stopped", zap.Error(err)) + return + } + if n == 0 { + if total > 0 { + c.Log.Info("media search index warmed", zap.Int64("indexed", total)) + } + return + } + total += n + select { + case <-ctx.Done(): + return + case <-time.After(pause): + } + } +} + func (c *Container) NormalizeCloudLibraryTypes(ctx context.Context) error { if c == nil || c.Repo == nil || c.Repo.Library == nil || c.Repo.DB == nil { return nil @@ -340,9 +376,3 @@ func (c *Container) Close() { // unused guard var _ = time.Now - - - - - - diff --git a/internal/service/stats.go b/internal/service/stats.go index 2fa56fd..43c0bb0 100644 --- a/internal/service/stats.go +++ b/internal/service/stats.go @@ -62,7 +62,7 @@ func (s *StatsService) Compute(ctx context.Context, dataDir string) (*Snapshot, if err != nil { return nil, err } - libs = FilterShadowedCloudLibraries(libs) + libs = FilterDisplayCloudLibraries(ctx, s.repo, libs) activeLibraryIDs := make([]string, 0, len(libs)) for _, lib := range libs { if !lib.Enabled { diff --git a/internal/service/storage_config.go b/internal/service/storage_config.go index 17ec640..f49f330 100644 --- a/internal/service/storage_config.go +++ b/internal/service/storage_config.go @@ -15,6 +15,7 @@ import ( "net/url" "strconv" "strings" + "sync" "time" "go.uber.org/zap" @@ -26,19 +27,42 @@ import ( // StorageConfigService encrypts + persists external storage configs. type StorageConfigService struct { - log *zap.Logger - repo *repository.Container - crypto *CryptoService - client *http.Client + log *zap.Logger + repo *repository.Container + crypto *CryptoService + client *http.Client + resolveMu sync.Mutex + resolveCache map[string]cloudResolveCacheEntry + resolveFlight map[string]*cloudResolveCall } +type cloudResolveCacheEntry struct { + link *cloud.DirectLink + expiresAt time.Time + hits int + lastHit time.Time +} + +type cloudResolveCall struct { + done chan struct{} + link *cloud.DirectLink + err error +} + +const ( + cloudResolveHotHitThreshold = 3 + cloudResolveBackgroundRefreshMax = 30 * time.Second +) + // NewStorageConfigService is the constructor. func NewStorageConfigService(log *zap.Logger, repo *repository.Container, crypto *CryptoService) *StorageConfigService { return &StorageConfigService{ - log: log, - repo: repo, - crypto: crypto, - client: &http.Client{Timeout: 120 * time.Second}, + log: log, + repo: repo, + crypto: crypto, + client: &http.Client{Timeout: 120 * time.Second}, + resolveCache: make(map[string]cloudResolveCacheEntry), + resolveFlight: make(map[string]*cloudResolveCall), } } @@ -119,6 +143,7 @@ func (s *StorageConfigService) Save(ctx context.Context, in StorageInput) (*Stor if err := s.repo.StorageConfig.Upsert(ctx, row); err != nil { return nil, err } + s.clearResolveCacheForType(in.Type) return s.Get(ctx, in.Type) } @@ -263,6 +288,211 @@ func (s *StorageConfigService) CloudList(ctx context.Context, typ, dirID string) // points at a link the client can fetch directly (true offload). When clientUA // is empty the provider's default UA is used. func (s *StorageConfigService) CloudResolve(ctx context.Context, typ, fileRef, clientUA string) (*cloud.DirectLink, error) { + if s == nil { + return nil, errors.New("storage config service unavailable") + } + cacheKey := s.resolveCacheKey(typ, fileRef, clientUA) + if link, ok, refresh := s.cachedResolve(cacheKey, typ); ok { + if refresh { + s.refreshResolveInBackground(cacheKey, typ, fileRef, clientUA) + } + return link, nil + } + if call, owner := s.beginResolve(cacheKey); !owner { + select { + case <-call.done: + if call.err != nil { + return nil, call.err + } + return cloneDirectLink(call.link), nil + case <-ctx.Done(): + return nil, ctx.Err() + } + } else { + defer s.finishResolve(cacheKey, call) + p, err := s.cloudProviderWithUA(ctx, typ, clientUA) + if err != nil { + call.err = err + return nil, err + } + link, err := p.Resolve(ctx, fileRef) + if err != nil { + call.err = err + return nil, err + } + call.link = cloneDirectLink(link) + s.storeResolvedLink(cacheKey, typ, link) + return cloneDirectLink(link), nil + } +} + +func (s *StorageConfigService) resolveCacheKey(typ, fileRef, clientUA string) string { + return strings.TrimSpace(typ) + "\x00" + strings.TrimSpace(fileRef) + "\x00" + strings.TrimSpace(clientUA) +} + +func (s *StorageConfigService) cachedResolve(key, typ string) (*cloud.DirectLink, bool, bool) { + s.resolveMu.Lock() + defer s.resolveMu.Unlock() + if s.resolveCache == nil { + s.resolveCache = make(map[string]cloudResolveCacheEntry) + return nil, false, false + } + entry, ok := s.resolveCache[key] + now := time.Now() + if !ok || now.After(entry.expiresAt) { + if ok { + delete(s.resolveCache, key) + } + return nil, false, false + } + entry.hits++ + entry.lastHit = now + s.resolveCache[key] = entry + refreshWindow := cloudResolveHotRefreshWindow(cloudResolveCacheTTL(typ)) + shouldRefresh := entry.hits >= cloudResolveHotHitThreshold && + refreshWindow > 0 && + now.Add(refreshWindow).After(entry.expiresAt) + return cloneDirectLink(entry.link), true, shouldRefresh +} + +func (s *StorageConfigService) beginResolve(key string) (*cloudResolveCall, bool) { + s.resolveMu.Lock() + defer s.resolveMu.Unlock() + if s.resolveFlight == nil { + s.resolveFlight = make(map[string]*cloudResolveCall) + } + if call := s.resolveFlight[key]; call != nil { + return call, false + } + call := &cloudResolveCall{done: make(chan struct{})} + s.resolveFlight[key] = call + return call, true +} + +func (s *StorageConfigService) finishResolve(key string, call *cloudResolveCall) { + s.resolveMu.Lock() + if current := s.resolveFlight[key]; current == call { + delete(s.resolveFlight, key) + } + s.resolveMu.Unlock() + close(call.done) +} + +func (s *StorageConfigService) refreshResolveInBackground(key, typ, fileRef, clientUA string) { + if s == nil { + return + } + go func() { + call, owner := s.beginResolve(key) + if !owner { + return + } + defer s.finishResolve(key, call) + ctx, cancel := context.WithTimeout(context.Background(), cloudResolveBackgroundRefreshMax) + defer cancel() + p, err := s.cloudProviderWithUA(ctx, typ, clientUA) + if err != nil { + call.err = err + if s.log != nil { + s.log.Debug("refresh cloud direct link failed", zap.String("provider", typ), zap.Error(err)) + } + return + } + link, err := p.Resolve(ctx, fileRef) + if err != nil { + call.err = err + if s.log != nil { + s.log.Debug("refresh cloud direct link failed", zap.String("provider", typ), zap.Error(err)) + } + return + } + call.link = cloneDirectLink(link) + s.storeResolvedLink(key, typ, link) + }() +} + +func (s *StorageConfigService) storeResolvedLink(key, typ string, link *cloud.DirectLink) { + if link == nil || strings.TrimSpace(link.URL) == "" { + return + } + ttl := cloudResolveCacheTTL(typ) + if ttl <= 0 { + return + } + s.resolveMu.Lock() + defer s.resolveMu.Unlock() + if s.resolveCache == nil { + s.resolveCache = make(map[string]cloudResolveCacheEntry) + } + now := time.Now() + hits := 0 + if existing, ok := s.resolveCache[key]; ok { + hits = existing.hits + } + s.resolveCache[key] = cloudResolveCacheEntry{link: cloneDirectLink(link), expiresAt: now.Add(ttl), hits: hits, lastHit: now} +} + +func cloudResolveHotRefreshWindow(ttl time.Duration) time.Duration { + if ttl <= 0 { + return 0 + } + window := ttl / 4 + if window < 15*time.Second { + window = 15 * time.Second + } + if window > 2*time.Minute { + window = 2 * time.Minute + } + return window +} + +func cloudResolveCacheTTL(typ string) time.Duration { + switch typ { + case cloud.TypeQuark, cloud.Type115: + return 2 * time.Minute + case cloud.TypeCloudDrive2, cloud.TypeOpenList: + return 15 * time.Minute + default: + return 5 * time.Minute + } +} + +func cloneDirectLink(link *cloud.DirectLink) *cloud.DirectLink { + if link == nil { + return nil + } + out := &cloud.DirectLink{ + URL: link.URL, + Headers: make(map[string]string, len(link.Headers)), + Proxy: link.Proxy, + } + for k, v := range link.Headers { + out.Headers[k] = v + } + return out +} + +func (s *StorageConfigService) clearResolveCacheForType(typ string) { + typ = strings.TrimSpace(typ) + if typ == "" { + return + } + prefix := typ + "\x00" + s.resolveMu.Lock() + defer s.resolveMu.Unlock() + for key := range s.resolveCache { + if strings.HasPrefix(key, prefix) { + delete(s.resolveCache, key) + } + } + for key, call := range s.resolveFlight { + if strings.HasPrefix(key, prefix) && call != nil { + call.err = fmt.Errorf("%s storage config changed", typ) + } + } +} + +func (s *StorageConfigService) CloudResolveUncached(ctx context.Context, typ, fileRef, clientUA string) (*cloud.DirectLink, error) { p, err := s.cloudProviderWithUA(ctx, typ, clientUA) if err != nil { return nil, err @@ -496,7 +726,6 @@ func strr(v any) string { return strings.TrimSpace(fmt.Sprint(v)) } - // DeleteStorage 删除存储配置并清理关联数据 func (s *StorageConfigService) DeleteStorage(ctx context.Context, storageType string) error { // 查找配置 diff --git a/internal/service/storage_config_cache_test.go b/internal/service/storage_config_cache_test.go new file mode 100644 index 0000000..ff82e14 --- /dev/null +++ b/internal/service/storage_config_cache_test.go @@ -0,0 +1,81 @@ +package service + +import ( + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func TestCloudResolveHotCacheRefreshesInBackground(t *testing.T) { + var resolves atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/file/download" { + t.Fatalf("unexpected path %s", r.URL.Path) + } + n := resolves.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"status":200,"code":0,"data":[{"fid":"f1","download_url":"http://cdn.local/%d.mkv"}]}`, n) + })) + defer upstream.Close() + + _, storage := newStorageUploadTestService(t) + if _, err := storage.Save(t.Context(), StorageInput{ + Type: "quark", + Config: map[string]any{ + "cookie": "kps=test", + "base": upstream.URL, + }, + }); err != nil { + t.Fatal(err) + } + + link, err := storage.CloudResolve(t.Context(), "quark", "f1", "Player/1") + if err != nil { + t.Fatal(err) + } + if link.URL != "http://cdn.local/1.mkv" || resolves.Load() != 1 { + t.Fatalf("first resolve link=%#v resolves=%d", link, resolves.Load()) + } + for i := 0; i < cloudResolveHotHitThreshold-1; i++ { + link, err = storage.CloudResolve(t.Context(), "quark", "f1", "Player/1") + if err != nil { + t.Fatal(err) + } + if link.URL != "http://cdn.local/1.mkv" || resolves.Load() != 1 { + t.Fatalf("cached resolve link=%#v resolves=%d", link, resolves.Load()) + } + } + + key := storage.resolveCacheKey("quark", "f1", "Player/1") + storage.resolveMu.Lock() + entry := storage.resolveCache[key] + entry.hits = cloudResolveHotHitThreshold + entry.expiresAt = time.Now().Add(5 * time.Second) + storage.resolveCache[key] = entry + storage.resolveMu.Unlock() + + link, err = storage.CloudResolve(t.Context(), "quark", "f1", "Player/1") + if err != nil { + t.Fatal(err) + } + if link.URL != "http://cdn.local/1.mkv" { + t.Fatalf("hot hit should return cached link immediately, got %s", link.URL) + } + deadline := time.Now().Add(2 * time.Second) + for resolves.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if resolves.Load() < 2 { + t.Fatalf("background refresh did not run, resolves=%d", resolves.Load()) + } + link, err = storage.CloudResolve(t.Context(), "quark", "f1", "Player/1") + if err != nil { + t.Fatal(err) + } + if link.URL != "http://cdn.local/2.mkv" { + t.Fatalf("refreshed link = %s, want second URL", link.URL) + } +} diff --git a/internal/service/telegram_bot.go b/internal/service/telegram_bot.go index fed3c9c..8126d9c 100644 --- a/internal/service/telegram_bot.go +++ b/internal/service/telegram_bot.go @@ -673,7 +673,7 @@ func (s *TelegramBotService) activeTelegramStatsLibraries(ctx context.Context) ( if err != nil { return nil, err } - libs = FilterShadowedCloudLibraries(libs) + libs = FilterDisplayCloudLibraries(ctx, s.repo, libs) out := libs[:0] for _, lib := range libs { if lib.Enabled { diff --git a/web/src/pages/SettingsPage.tsx b/web/src/pages/SettingsPage.tsx index f5fb209..c7fc851 100644 --- a/web/src/pages/SettingsPage.tsx +++ b/web/src/pages/SettingsPage.tsx @@ -302,18 +302,25 @@ const GROUPS: SettingGroup[] = [ items: [ { key: 'cloud.auto_sync_enabled', - label: '自动同步网盘媒体库', + label: '夜间自动同步网盘媒体库', type: 'toggle', - hint: '默认关闭，避免 NAS 反复递归读取大型网盘目录。需要定时刷新时再开启；手动扫描仍可在外部存储页面执行。', + hint: '默认关闭。开启后仅在每天 19:00-21:00 自动补齐缺失媒体并扫描新增；手动扫描仍可随时执行。', defaultValue: 'false', }, { key: 'cloud.sync_interval_seconds', - label: '网盘媒体库同步间隔秒数', + label: '夜间窗口检查间隔秒数', type: 'number', - hint: '最小 300 秒，建议 1800 秒或更高；手动可在任务调度中运行 cloud_sync。', + hint: '最小 300 秒，建议 1800 秒；同一天成功同步后不会重复全量扫，避免大型网盘反复递归。', defaultValue: '1800', }, + { + key: 'cloud.boot_scan_enabled', + label: '启动后立即扫描网盘', + type: 'toggle', + hint: '默认关闭。仅排障或小型网盘建议开启；大型库请使用手动扫描或夜间自动同步。', + defaultValue: 'false', + }, { key: 'cloud.upload_auto_enabled', label: '启用自动转存', diff --git a/web/src/pages/StorageConfigPage.tsx b/web/src/pages/StorageConfigPage.tsx index fb6b373..9f4bc39 100644 --- a/web/src/pages/StorageConfigPage.tsx +++ b/web/src/pages/StorageConfigPage.tsx @@ -568,7 +568,7 @@ function CloudBrowser({ type }: { type: StorageType }) { setMounting(true) try { const label = TYPE_LABEL[type] ?? type - const name = cur.id ? `${label} · ${cur.name}` : label + const name = cur.id ? cur.name : label const res = await cloudAPI.mount(type, cur.id, name, mountMediaType, currentMountPath()) handleMountResult(res, cur.name) await loadMounts() @@ -589,10 +589,9 @@ function CloudBrowser({ type }: { type: StorageType }) { let ok = 0 let skipped = 0 let failed = 0 - const label = TYPE_LABEL[type] ?? type for (const dir of dirs) { try { - const result = await cloudAPI.mount(type, dir.id, `${label} · ${dir.name}`, 'auto', childMountPath(dir)) + const result = await cloudAPI.mount(type, dir.id, dir.name, 'auto', childMountPath(dir)) const state = handleMountResult(result, dir.name) if (state === 'skipped') skipped += 1 else ok += 1 @@ -733,6 +732,7 @@ function CloudBrowser({ type }: { type: StorageType }) {

挂载后不会复制网盘文件；后台会递归读取该目录里的子文件夹和媒体文件，扫描到的影片会自动加入对应媒体库。小目录通常几十秒，大目录取决于网盘接口速度。 + 如果已有同名同类型媒体库，会在首页和 Emby/SenPlayer 中自动归并显示。