split subscription rss run helpers

2026-09-29 03:26:37 +08:00 · 2026-06-27 09:28:56 +08:00
parent b2b7bd6c0e
commit 94e179ce48
2 changed files with 151 additions and 139 deletions
@@ -44,14 +44,6 @@ const (
 	subscriptionStartupDelay        = defaultSubscriptionPollInterval
 )

-type rssSubscriptionRunState struct {
-	seen              []string
-	seenSet           map[string]struct{}
-	availability      LocalAvailability
-	availabilityQuery string
-	washOff           bool
-}
-
 // NewSubscriptionService is the constructor.
 func NewSubscriptionService(cfg *config.Config, log *zap.Logger, repo *repository.Container, downloads *DownloadService, site *SiteService, hub *Hub) *SubscriptionService {
 	return &SubscriptionService{
@@ -249,134 +241,3 @@ func (s *SubscriptionService) runAll(ctx context.Context) {
 		}
 	}
 }
-
-func (s *SubscriptionService) runOne(ctx context.Context, sub *model.Subscription) (int, error) {
-	s.prepareSubscriptionForRun(ctx, sub)
-	if strings.HasPrefix(strings.ToLower(strings.TrimSpace(sub.FeedURL)), "site-search://") {
-		return s.runSiteSearch(ctx, sub)
-	}
-
-	feed, err := s.fetch(ctx, sub.FeedURL)
-	if err != nil {
-		return 0, err
-	}
-
-	filter := compileFilter(sub.Filter)
-	guidKey := fmt.Sprintf("subscription.%s.seen", sub.ID)
-	seenRaw, _ := s.repo.Setting.Get(ctx, guidKey)
-	seen := splitNonEmpty(seenRaw)
-	seenSet := make(map[string]struct{}, len(seen))
-	for _, g := range seen {
-		seenSet[g] = struct{}{}
-	}
-
-	s.updateSubscriptionTotalEpisodes(ctx, sub, s.resolveSubscriptionTotalEpisodes(ctx, sub, inferRSSTotalEpisodes(feed.Channel.Items, sub, filter)))
-	// RSS 和站点搜索统一使用候选规划：先按订阅规则过滤，再按洗版优先级/集数去重择优。
-	// 非洗版订阅成功下载一次即满足，媒体库与下载中任务会作为可用性输入避免重复下载。
-	runState := &rssSubscriptionRunState{
-		seen:              seen,
-		seenSet:           seenSet,
-		availability:      mergeLocalAvailability(SubscriptionLocalAvailability(ctx, s.repo, sub), s.pendingDownloadAvailability(ctx, sub)),
-		availabilityQuery: availabilityQuery(subscriptionName(sub), subscriptionFilter(sub)),
-		washOff:           !sub.WashEnabled,
-	}
-	candidates := selectRSSSubscriptionCandidates(feed.Channel.Items, sub, filter, runState.seenSet, runState.availability)
-	queued := s.enqueueRSSSubscriptionCandidates(ctx, sub, candidates, runState)
-	s.finishRSSSubscriptionRun(ctx, sub, guidKey, runState, queued)
-	return queued, nil
-}
-
-func (s *SubscriptionService) enqueueRSSSubscriptionCandidates(ctx context.Context, sub *model.Subscription, candidates []siteSearchCandidate, state *rssSubscriptionRunState) int {
-	queued := 0
-	for _, candidate := range candidates {
-		if s.enqueueRSSSubscriptionCandidate(ctx, sub, candidate, state) {
-			queued++
-		}
-	}
-	return queued
-}
-
-func (s *SubscriptionService) enqueueRSSSubscriptionCandidate(ctx context.Context, sub *model.Subscription, candidate siteSearchCandidate, state *rssSubscriptionRunState) bool {
-	item := candidate.Item
-	mediaType, mediaCategory := s.classifySubscriptionItem(ctx, sub, item.Title, "")
-	savePath := s.resolveSubscriptionSavePath(ctx, sub, mediaType, mediaCategory)
-	if s.downloadPathHasCandidate(ctx, sub, item.Title, savePath) {
-		state.markTitleAvailable(item.Title)
-		return false
-	}
-	if _, err := s.downloads.AddDownloadWithMeta(ctx, sub.UserID, candidate.Download, savePath, DownloadTaskMeta{
-		SubscriptionID:       sub.ID,
-		Title:                firstNonEmpty(item.Title, sub.Name),
-		PosterURL:            sub.PosterURL,
-		BackdropURL:          sub.BackdropURL,
-		Overview:             sub.Overview,
-		MediaType:            mediaType,
-		MediaCategory:        mediaCategory,
-		AllowExistingLibrary: sub.WashEnabled,
-	}); err != nil {
-		if IsDownloadDedupError(err) {
-			if s.subscriptionCandidateConfirmedAvailable(ctx, sub, candidate) {
-				state.markCandidateAvailable(candidate)
-				return false
-			}
-			if s.log != nil {
-				s.log.Info("subscription dedup candidate not confirmed available",
-					zap.String("title", item.Title),
-					zap.String("media_type", mediaType),
-					zap.String("media_category", mediaCategory),
-					zap.String("save_path", savePath))
-			}
-			return false
-		}
-		s.log.Warn("subscription enqueue failed",
-			zap.String("title", item.Title),
-			zap.String("media_type", mediaType),
-			zap.String("media_category", mediaCategory),
-			zap.String("save_path", savePath),
-			zap.Error(err))
-		return false
-	}
-	state.markTitleAvailable(item.Title)
-	state.markSeen(candidate.GUID)
-	return true
-}
-
-func (s *SubscriptionService) finishRSSSubscriptionRun(ctx context.Context, sub *model.Subscription, guidKey string, state *rssSubscriptionRunState, queued int) {
-	state.availability = s.finalizePendingAvailability(sub, state.availability)
-	// Remember the last 200 GUIDs so the seen set doesn't grow forever.
-	if len(state.seen) > 200 {
-		state.seen = state.seen[len(state.seen)-200:]
-	}
-	_ = s.repo.Setting.Set(ctx, guidKey, strings.Join(state.seen, "\n"))
-
-	now := time.Now()
-	_ = s.repo.DB.Model(sub).Updates(map[string]any{"last_run_at": &now}).Error
-	_ = s.archiveCompletedSubscription(ctx, sub, state.availability)
-	if queued > 0 {
-		s.hub.Publish("subscription", map[string]any{
-			"id":     sub.ID,
-			"name":   sub.Name,
-			"queued": queued,
-		})
-		s.notifySubscriptionHit(sub, queued, nil)
-	}
-}
-
-func (state *rssSubscriptionRunState) markTitleAvailable(title string) {
-	if state.washOff {
-		addAvailabilityTitle(title, state.availabilityQuery, &state.availability)
-	}
-}
-
-func (state *rssSubscriptionRunState) markCandidateAvailable(candidate siteSearchCandidate) {
-	if state.washOff {
-		addSiteSearchCandidateAvailability(candidate, &state.availability)
-	}
-}
-
-func (state *rssSubscriptionRunState) markSeen(guid string) {
-	state.seen = append(state.seen, guid)
-	if state.seenSet != nil {
-		state.seenSet[guid] = struct{}{}
-	}
-}
@@ -0,0 +1,151 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/ShukeBta/MediaStationGo/internal/model"
+)
+
+type rssSubscriptionRunState struct {
+	seen              []string
+	seenSet           map[string]struct{}
+	availability      LocalAvailability
+	availabilityQuery string
+	washOff           bool
+}
+
+func (s *SubscriptionService) runOne(ctx context.Context, sub *model.Subscription) (int, error) {
+	s.prepareSubscriptionForRun(ctx, sub)
+	if strings.HasPrefix(strings.ToLower(strings.TrimSpace(sub.FeedURL)), "site-search://") {
+		return s.runSiteSearch(ctx, sub)
+	}
+
+	feed, err := s.fetch(ctx, sub.FeedURL)
+	if err != nil {
+		return 0, err
+	}
+
+	filter := compileFilter(sub.Filter)
+	guidKey := fmt.Sprintf("subscription.%s.seen", sub.ID)
+	seenRaw, _ := s.repo.Setting.Get(ctx, guidKey)
+	seen := splitNonEmpty(seenRaw)
+	seenSet := make(map[string]struct{}, len(seen))
+	for _, g := range seen {
+		seenSet[g] = struct{}{}
+	}
+
+	s.updateSubscriptionTotalEpisodes(ctx, sub, s.resolveSubscriptionTotalEpisodes(ctx, sub, inferRSSTotalEpisodes(feed.Channel.Items, sub, filter)))
+	// RSS 和站点搜索统一使用候选规划：先按订阅规则过滤，再按洗版优先级/集数去重择优。
+	// 非洗版订阅成功下载一次即满足，媒体库与下载中任务会作为可用性输入避免重复下载。
+	runState := &rssSubscriptionRunState{
+		seen:              seen,
+		seenSet:           seenSet,
+		availability:      mergeLocalAvailability(SubscriptionLocalAvailability(ctx, s.repo, sub), s.pendingDownloadAvailability(ctx, sub)),
+		availabilityQuery: availabilityQuery(subscriptionName(sub), subscriptionFilter(sub)),
+		washOff:           !sub.WashEnabled,
+	}
+	candidates := selectRSSSubscriptionCandidates(feed.Channel.Items, sub, filter, runState.seenSet, runState.availability)
+	queued := s.enqueueRSSSubscriptionCandidates(ctx, sub, candidates, runState)
+	s.finishRSSSubscriptionRun(ctx, sub, guidKey, runState, queued)
+	return queued, nil
+}
+
+func (s *SubscriptionService) enqueueRSSSubscriptionCandidates(ctx context.Context, sub *model.Subscription, candidates []siteSearchCandidate, state *rssSubscriptionRunState) int {
+	queued := 0
+	for _, candidate := range candidates {
+		if s.enqueueRSSSubscriptionCandidate(ctx, sub, candidate, state) {
+			queued++
+		}
+	}
+	return queued
+}
+
+func (s *SubscriptionService) enqueueRSSSubscriptionCandidate(ctx context.Context, sub *model.Subscription, candidate siteSearchCandidate, state *rssSubscriptionRunState) bool {
+	item := candidate.Item
+	mediaType, mediaCategory := s.classifySubscriptionItem(ctx, sub, item.Title, "")
+	savePath := s.resolveSubscriptionSavePath(ctx, sub, mediaType, mediaCategory)
+	if s.downloadPathHasCandidate(ctx, sub, item.Title, savePath) {
+		state.markTitleAvailable(item.Title)
+		return false
+	}
+	if _, err := s.downloads.AddDownloadWithMeta(ctx, sub.UserID, candidate.Download, savePath, DownloadTaskMeta{
+		SubscriptionID:       sub.ID,
+		Title:                firstNonEmpty(item.Title, sub.Name),
+		PosterURL:            sub.PosterURL,
+		BackdropURL:          sub.BackdropURL,
+		Overview:             sub.Overview,
+		MediaType:            mediaType,
+		MediaCategory:        mediaCategory,
+		AllowExistingLibrary: sub.WashEnabled,
+	}); err != nil {
+		if IsDownloadDedupError(err) {
+			if s.subscriptionCandidateConfirmedAvailable(ctx, sub, candidate) {
+				state.markCandidateAvailable(candidate)
+				return false
+			}
+			if s.log != nil {
+				s.log.Info("subscription dedup candidate not confirmed available",
+					zap.String("title", item.Title),
+					zap.String("media_type", mediaType),
+					zap.String("media_category", mediaCategory),
+					zap.String("save_path", savePath))
+			}
+			return false
+		}
+		s.log.Warn("subscription enqueue failed",
+			zap.String("title", item.Title),
+			zap.String("media_type", mediaType),
+			zap.String("media_category", mediaCategory),
+			zap.String("save_path", savePath),
+			zap.Error(err))
+		return false
+	}
+	state.markTitleAvailable(item.Title)
+	state.markSeen(candidate.GUID)
+	return true
+}
+
+func (s *SubscriptionService) finishRSSSubscriptionRun(ctx context.Context, sub *model.Subscription, guidKey string, state *rssSubscriptionRunState, queued int) {
+	state.availability = s.finalizePendingAvailability(sub, state.availability)
+	// Remember the last 200 GUIDs so the seen set doesn't grow forever.
+	if len(state.seen) > 200 {
+		state.seen = state.seen[len(state.seen)-200:]
+	}
+	_ = s.repo.Setting.Set(ctx, guidKey, strings.Join(state.seen, "\n"))
+
+	now := time.Now()
+	_ = s.repo.DB.Model(sub).Updates(map[string]any{"last_run_at": &now}).Error
+	_ = s.archiveCompletedSubscription(ctx, sub, state.availability)
+	if queued > 0 {
+		s.hub.Publish("subscription", map[string]any{
+			"id":     sub.ID,
+			"name":   sub.Name,
+			"queued": queued,
+		})
+		s.notifySubscriptionHit(sub, queued, nil)
+	}
+}
+
+func (state *rssSubscriptionRunState) markTitleAvailable(title string) {
+	if state.washOff {
+		addAvailabilityTitle(title, state.availabilityQuery, &state.availability)
+	}
+}
+
+func (state *rssSubscriptionRunState) markCandidateAvailable(candidate siteSearchCandidate) {
+	if state.washOff {
+		addSiteSearchCandidateAvailability(candidate, &state.availability)
+	}
+}
+
+func (state *rssSubscriptionRunState) markSeen(guid string) {
+	state.seen = append(state.seen, guid)
+	if state.seenSet != nil {
+		state.seenSet[guid] = struct{}{}
+	}
+}