diff --git a/internal/service/subscription.go b/internal/service/subscription.go index f0ae941..f884834 100644 --- a/internal/service/subscription.go +++ b/internal/service/subscription.go @@ -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{}{} - } -} diff --git a/internal/service/subscription_rss_run.go b/internal/service/subscription_rss_run.go new file mode 100644 index 0000000..a41a911 --- /dev/null +++ b/internal/service/subscription_rss_run.go @@ -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{}{} + } +}