mirror of
https://github.com/idanoo/autobrr
synced 2025-07-23 08:49:13 +00:00
feat: add torznab feed support (#246)
* feat(torznab): initial impl * feat: torznab processing * feat: torznab more scheduling * feat: feeds web * feat(feeds): create on indexer create * feat(feeds): update migration * feat(feeds): restart on update * feat(feeds): set cron schedule * feat(feeds): use basic empty state * chore: remove duplicate migrations * feat: parse release size from torznab * chore: cleanup unused code
This commit is contained in:
Ludvig Lundgren
GitHub
parent
d4d864cd2c
commit
bb62e724a1
34 changed files with 2408 additions and 361 deletions
277
internal/feed/service.go
Normal file
277
internal/feed/service.go
Normal file
|
|
@ -0,0 +1,277 @@
|
|||
package feed
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
"github.com/autobrr/autobrr/internal/scheduler"
|
||||
"github.com/autobrr/autobrr/pkg/torznab"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
type Service interface {
|
||||
FindByID(ctx context.Context, id int) (*domain.Feed, error)
|
||||
FindByIndexerIdentifier(ctx context.Context, indexer string) (*domain.Feed, error)
|
||||
Find(ctx context.Context) ([]domain.Feed, error)
|
||||
Store(ctx context.Context, feed *domain.Feed) error
|
||||
Update(ctx context.Context, feed *domain.Feed) error
|
||||
ToggleEnabled(ctx context.Context, id int, enabled bool) error
|
||||
Delete(ctx context.Context, id int) error
|
||||
|
||||
Start() error
|
||||
}
|
||||
|
||||
type feedInstance struct {
|
||||
Name string
|
||||
IndexerIdentifier string
|
||||
URL string
|
||||
ApiKey string
|
||||
Implementation string
|
||||
CronSchedule string
|
||||
}
|
||||
|
||||
type service struct {
|
||||
jobs map[string]int
|
||||
|
||||
repo domain.FeedRepo
|
||||
cacheRepo domain.FeedCacheRepo
|
||||
releaseSvc release.Service
|
||||
scheduler scheduler.Service
|
||||
}
|
||||
|
||||
func NewService(repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service, scheduler scheduler.Service) Service {
|
||||
return &service{
|
||||
jobs: map[string]int{},
|
||||
repo: repo,
|
||||
cacheRepo: cacheRepo,
|
||||
releaseSvc: releaseSvc,
|
||||
scheduler: scheduler,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *service) FindByID(ctx context.Context, id int) (*domain.Feed, error) {
|
||||
return s.repo.FindByID(ctx, id)
|
||||
}
|
||||
|
||||
func (s *service) FindByIndexerIdentifier(ctx context.Context, indexer string) (*domain.Feed, error) {
|
||||
return s.repo.FindByIndexerIdentifier(ctx, indexer)
|
||||
}
|
||||
|
||||
func (s *service) Find(ctx context.Context) ([]domain.Feed, error) {
|
||||
return s.repo.Find(ctx)
|
||||
}
|
||||
|
||||
func (s *service) Store(ctx context.Context, feed *domain.Feed) error {
|
||||
return s.repo.Store(ctx, feed)
|
||||
}
|
||||
|
||||
func (s *service) Update(ctx context.Context, feed *domain.Feed) error {
|
||||
return s.update(ctx, feed)
|
||||
}
|
||||
|
||||
func (s *service) Delete(ctx context.Context, id int) error {
|
||||
return s.delete(ctx, id)
|
||||
}
|
||||
|
||||
func (s *service) ToggleEnabled(ctx context.Context, id int, enabled bool) error {
|
||||
return s.toggleEnabled(ctx, id, enabled)
|
||||
}
|
||||
|
||||
func (s *service) update(ctx context.Context, feed *domain.Feed) error {
|
||||
if err := s.repo.Update(ctx, feed); err != nil {
|
||||
log.Error().Err(err).Msg("feed.Update: error updating feed")
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.restartJob(feed); err != nil {
|
||||
log.Error().Err(err).Msg("feed.Update: error restarting feed")
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) delete(ctx context.Context, id int) error {
|
||||
f, err := s.repo.FindByID(ctx, id)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("feed.ToggleEnabled: error finding feed")
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.stopTorznabJob(f.Indexer); err != nil {
|
||||
log.Error().Err(err).Msg("feed.Delete: error stopping torznab job")
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.repo.Delete(ctx, id); err != nil {
|
||||
log.Error().Err(err).Msg("feed.Delete: error deleting feed")
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug().Msgf("feed.Delete: stopping and removing feed: %v", f.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) toggleEnabled(ctx context.Context, id int, enabled bool) error {
|
||||
if err := s.repo.ToggleEnabled(ctx, id, enabled); err != nil {
|
||||
log.Error().Err(err).Msg("feed.ToggleEnabled: error toggle enabled")
|
||||
return err
|
||||
}
|
||||
|
||||
f, err := s.repo.FindByID(ctx, id)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("feed.ToggleEnabled: error finding feed")
|
||||
return err
|
||||
}
|
||||
|
||||
if !enabled {
|
||||
if err := s.stopTorznabJob(f.Indexer); err != nil {
|
||||
log.Error().Err(err).Msg("feed.ToggleEnabled: error stopping torznab job")
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug().Msgf("feed.ToggleEnabled: stopping feed: %v", f.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := s.startJob(*f); err != nil {
|
||||
log.Error().Err(err).Msg("feed.ToggleEnabled: error starting torznab job")
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug().Msgf("feed.ToggleEnabled: started feed: %v", f.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) Start() error {
|
||||
// get all torznab indexer definitions
|
||||
feeds, err := s.repo.Find(context.TODO())
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("feed.Start: error getting feeds")
|
||||
return err
|
||||
}
|
||||
|
||||
for _, i := range feeds {
|
||||
if err := s.startJob(i); err != nil {
|
||||
log.Error().Err(err).Msg("feed.Start: failed to initialize torznab job")
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) restartJob(f *domain.Feed) error {
|
||||
// stop feed
|
||||
if err := s.stopTorznabJob(f.Indexer); err != nil {
|
||||
log.Error().Err(err).Msg("feed.restartJob: error stopping torznab job")
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug().Msgf("feed.restartJob: stopping feed: %v", f.Name)
|
||||
|
||||
if f.Enabled {
|
||||
if err := s.startJob(*f); err != nil {
|
||||
log.Error().Err(err).Msg("feed.restartJob: error starting torznab job")
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug().Msgf("feed.restartJob: restarted feed: %v", f.Name)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) startJob(f domain.Feed) error {
|
||||
// get all torznab indexer definitions
|
||||
if !f.Enabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
// get torznab_url from settings
|
||||
if f.URL == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
// cron schedule to run every X minutes
|
||||
schedule := fmt.Sprintf("*/%d * * * *", f.Interval)
|
||||
|
||||
fi := feedInstance{
|
||||
Name: f.Name,
|
||||
IndexerIdentifier: f.Indexer,
|
||||
Implementation: f.Type,
|
||||
URL: f.URL,
|
||||
ApiKey: f.ApiKey,
|
||||
CronSchedule: schedule,
|
||||
}
|
||||
|
||||
switch fi.Implementation {
|
||||
case string(domain.FeedTypeTorznab):
|
||||
if err := s.addTorznabJob(fi); err != nil {
|
||||
log.Error().Err(err).Msg("feed.startJob: failed to initialize feed")
|
||||
return err
|
||||
}
|
||||
//case "rss":
|
||||
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) addTorznabJob(f feedInstance) error {
|
||||
if f.URL == "" {
|
||||
return errors.New("torznab feed requires URL")
|
||||
}
|
||||
if f.CronSchedule == "" {
|
||||
f.CronSchedule = "*/15 * * * *"
|
||||
}
|
||||
|
||||
// setup logger
|
||||
l := log.With().Str("feed_name", f.Name).Logger()
|
||||
|
||||
// setup torznab Client
|
||||
c := torznab.NewClient(f.URL, f.ApiKey)
|
||||
|
||||
// create job
|
||||
job := &TorznabJob{
|
||||
Name: f.Name,
|
||||
IndexerIdentifier: f.IndexerIdentifier,
|
||||
Client: c,
|
||||
Log: l,
|
||||
Repo: s.cacheRepo,
|
||||
ReleaseSvc: s.releaseSvc,
|
||||
URL: f.URL,
|
||||
}
|
||||
|
||||
// schedule job
|
||||
id, err := s.scheduler.AddJob(job, f.CronSchedule, f.IndexerIdentifier)
|
||||
if err != nil {
|
||||
return fmt.Errorf("feed.AddTorznabJob: add job failed: %w", err)
|
||||
}
|
||||
job.JobID = id
|
||||
|
||||
// add to job map
|
||||
s.jobs[f.IndexerIdentifier] = id
|
||||
|
||||
log.Debug().Msgf("feed.AddTorznabJob: %v", f.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) stopTorznabJob(indexer string) error {
|
||||
// remove job from scheduler
|
||||
if err := s.scheduler.RemoveJobByIdentifier(indexer); err != nil {
|
||||
return fmt.Errorf("feed.stopTorznabJob: stop job failed: %w", err)
|
||||
}
|
||||
|
||||
log.Debug().Msgf("feed.stopTorznabJob: %v", indexer)
|
||||
|
||||
return nil
|
||||
}
|
||||
134
internal/feed/torznab.go
Normal file
134
internal/feed/torznab.go
Normal file
|
|
@ -0,0 +1,134 @@
|
|||
package feed
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
"github.com/autobrr/autobrr/pkg/torznab"
|
||||
|
||||
"github.com/rs/zerolog"
|
||||
)
|
||||
|
||||
type TorznabJob struct {
|
||||
Name string
|
||||
IndexerIdentifier string
|
||||
Log zerolog.Logger
|
||||
URL string
|
||||
Client *torznab.Client
|
||||
Repo domain.FeedCacheRepo
|
||||
ReleaseSvc release.Service
|
||||
|
||||
attempts int
|
||||
errors []error
|
||||
|
||||
JobID int
|
||||
}
|
||||
|
||||
func (j *TorznabJob) Run() {
|
||||
err := j.process()
|
||||
if err != nil {
|
||||
j.Log.Err(err).Int("attempts", j.attempts).Msg("torznab process error")
|
||||
|
||||
j.errors = append(j.errors, err)
|
||||
}
|
||||
|
||||
j.attempts = 0
|
||||
j.errors = j.errors[:0]
|
||||
}
|
||||
|
||||
func (j *TorznabJob) process() error {
|
||||
// get feed
|
||||
items, err := j.getFeed()
|
||||
if err != nil {
|
||||
j.Log.Error().Err(err).Msgf("torznab.process: error fetching feed items")
|
||||
return fmt.Errorf("torznab.process: error getting feed items: %w", err)
|
||||
}
|
||||
|
||||
if len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
j.Log.Debug().Msgf("torznab.process: refreshing feed: %v, found (%d) new items to check", j.Name, len(items))
|
||||
|
||||
releases := make([]*domain.Release, 0)
|
||||
|
||||
for _, item := range items {
|
||||
rls, err := domain.NewRelease(item.Title, "")
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
rls.TorrentName = item.Title
|
||||
rls.TorrentURL = item.GUID
|
||||
rls.Implementation = domain.ReleaseImplementationTorznab
|
||||
rls.Indexer = j.IndexerIdentifier
|
||||
|
||||
// parse size bytes string
|
||||
rls.ParseSizeBytesString(item.Size)
|
||||
|
||||
if err := rls.Parse(); err != nil {
|
||||
j.Log.Error().Err(err).Msgf("torznab.process: error parsing release")
|
||||
continue
|
||||
}
|
||||
|
||||
releases = append(releases, rls)
|
||||
}
|
||||
|
||||
// process all new releases
|
||||
go j.ReleaseSvc.ProcessMultiple(releases)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (j *TorznabJob) getFeed() ([]torznab.FeedItem, error) {
|
||||
// get feed
|
||||
feedItems, err := j.Client.GetFeed()
|
||||
if err != nil {
|
||||
j.Log.Error().Err(err).Msgf("torznab.getFeed: error fetching feed items")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
j.Log.Trace().Msgf("torznab getFeed: refreshing feed: %v, found (%d) items", j.Name, len(feedItems))
|
||||
|
||||
items := make([]torznab.FeedItem, 0)
|
||||
if len(feedItems) == 0 {
|
||||
return items, nil
|
||||
}
|
||||
|
||||
sort.SliceStable(feedItems, func(i, j int) bool {
|
||||
return feedItems[i].PubDate.After(feedItems[j].PubDate.Time)
|
||||
})
|
||||
|
||||
for _, i := range feedItems {
|
||||
if i.GUID == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
//if cacheValue, err := j.Repo.Get(j.Name, i.GUID); err == nil {
|
||||
// j.Log.Trace().Msgf("torznab getFeed: cacheValue: %v", cacheValue)
|
||||
//}
|
||||
|
||||
if exists, err := j.Repo.Exists(j.Name, i.GUID); err == nil {
|
||||
if exists {
|
||||
j.Log.Trace().Msg("torznab getFeed: cache item exists, skip")
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// do something more
|
||||
|
||||
items = append(items, i)
|
||||
|
||||
ttl := (24 * time.Hour) * 28
|
||||
|
||||
if err := j.Repo.Put(j.Name, i.GUID, []byte("test"), ttl); err != nil {
|
||||
j.Log.Error().Err(err).Str("guid", i.GUID).Msg("torznab getFeed: cache.Put: error storing item in cache")
|
||||
}
|
||||
}
|
||||
|
||||
// send to filters
|
||||
return items, nil
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue