mirror of
https://github.com/idanoo/autobrr
synced 2025-07-23 08:49:13 +00:00
refactor: filter and action flow (#225)
* refactor: fitler and action flow * fix: save release before filters * feat: add action client to notifications * feat: improve filter check logging
This commit is contained in:
Ludvig Lundgren
GitHub
parent
f32379ae76
commit
a3854ecd59
21 changed files with 654 additions and 313 deletions
|
|
@ -2,7 +2,6 @@ package announce
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
|
|
@ -11,9 +10,6 @@ import (
|
|||
"text/template"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/filter"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
|
|
@ -24,17 +20,15 @@ type Processor interface {
|
|||
type announceProcessor struct {
|
||||
indexer domain.IndexerDefinition
|
||||
|
||||
filterSvc filter.Service
|
||||
releaseSvc release.Service
|
||||
announceSvc Service
|
||||
|
||||
queues map[string]chan string
|
||||
}
|
||||
|
||||
func NewAnnounceProcessor(indexer domain.IndexerDefinition, filterSvc filter.Service, releaseSvc release.Service) Processor {
|
||||
func NewAnnounceProcessor(announceSvc Service, indexer domain.IndexerDefinition) Processor {
|
||||
ap := &announceProcessor{
|
||||
indexer: indexer,
|
||||
filterSvc: filterSvc,
|
||||
releaseSvc: releaseSvc,
|
||||
announceSvc: announceSvc,
|
||||
indexer: indexer,
|
||||
}
|
||||
|
||||
// setup queues and consumers
|
||||
|
|
@ -115,52 +109,8 @@ func (a *announceProcessor) processQueue(queue chan string) {
|
|||
continue
|
||||
}
|
||||
|
||||
// send to filter service to take care of the rest
|
||||
|
||||
// find and check filter
|
||||
filterOK, foundFilter, err := a.filterSvc.FindAndCheckFilters(newRelease)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("could not find filter")
|
||||
continue
|
||||
}
|
||||
|
||||
// no foundFilter found, lets return
|
||||
if !filterOK || foundFilter == nil {
|
||||
log.Trace().Msg("no matching filter found")
|
||||
continue
|
||||
|
||||
// TODO check in config for "Save all releases"
|
||||
// Save as rejected
|
||||
//newRelease.FilterStatus = domain.ReleaseStatusFilterRejected
|
||||
//err = s.releaseSvc.Store(ctx, newRelease)
|
||||
//if err != nil {
|
||||
// log.Error().Err(err).Msgf("error writing release to database: %+v", newRelease)
|
||||
// return nil
|
||||
//}
|
||||
//return nil
|
||||
}
|
||||
|
||||
// save release
|
||||
newRelease.Filter = foundFilter
|
||||
newRelease.FilterName = foundFilter.Name
|
||||
newRelease.FilterID = foundFilter.ID
|
||||
|
||||
newRelease.FilterStatus = domain.ReleaseStatusFilterApproved
|
||||
err = a.releaseSvc.Store(context.Background(), newRelease)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("error writing release to database: %+v", newRelease)
|
||||
continue
|
||||
}
|
||||
|
||||
log.Info().Msgf("Matched '%v' (%v) for %v", newRelease.TorrentName, newRelease.Filter.Name, newRelease.Indexer)
|
||||
|
||||
// process release
|
||||
go func(rel *domain.Release) {
|
||||
err = a.releaseSvc.Process(*rel)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("could not process release: %+v", newRelease)
|
||||
}
|
||||
}(newRelease)
|
||||
// process release in a new go routine
|
||||
go a.announceSvc.Process(newRelease)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
132
internal/announce/service.go
Normal file
132
internal/announce/service.go
Normal file
|
|
@ -0,0 +1,132 @@
|
|||
package announce
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/action"
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/filter"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
type Service interface {
|
||||
Process(release *domain.Release)
|
||||
}
|
||||
|
||||
type service struct {
|
||||
actionSvc action.Service
|
||||
filterSvc filter.Service
|
||||
releaseSvc release.Service
|
||||
}
|
||||
|
||||
type actionClientTypeKey struct {
|
||||
Type domain.ActionType
|
||||
ClientID int32
|
||||
}
|
||||
|
||||
func NewService(actionSvc action.Service, filterSvc filter.Service, releaseSvc release.Service) Service {
|
||||
return &service{
|
||||
actionSvc: actionSvc,
|
||||
filterSvc: filterSvc,
|
||||
releaseSvc: releaseSvc,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *service) Process(release *domain.Release) {
|
||||
// TODO check in config for "Save all releases"
|
||||
// TODO cross-seed check
|
||||
// TODO dupe checks
|
||||
|
||||
// get filters by priority
|
||||
filters, err := s.filterSvc.FindByIndexerIdentifier(release.Indexer)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("announce.Service.Process: error finding filters for indexer: %v", release.Indexer)
|
||||
return
|
||||
}
|
||||
|
||||
// keep track of action clients to avoid sending the same thing all over again
|
||||
// save both client type and client id to potentially try another client of same type
|
||||
triedActionClients := map[actionClientTypeKey]struct{}{}
|
||||
|
||||
// save release
|
||||
//release.FilterStatus = domain.ReleaseStatusFilterApproved
|
||||
err = s.releaseSvc.Store(context.Background(), release)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("announce.Service.Process: error writing release to database: %+v", release)
|
||||
return
|
||||
}
|
||||
|
||||
// loop over and check filters
|
||||
for _, f := range filters {
|
||||
// save filter on release
|
||||
release.Filter = &f
|
||||
release.FilterName = f.Name
|
||||
release.FilterID = f.ID
|
||||
|
||||
// TODO filter limit checks
|
||||
|
||||
// test filter
|
||||
match, err := s.filterSvc.CheckFilter(f, release)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("announce.Service.Process: could not find filter")
|
||||
return
|
||||
}
|
||||
|
||||
if !match {
|
||||
log.Trace().Msgf("announce.Service.Process: indexer: %v, filter: %v release: %v, no match", release.Indexer, release.Filter.Name, release.TorrentName)
|
||||
continue
|
||||
}
|
||||
|
||||
log.Info().Msgf("Matched '%v' (%v) for %v", release.TorrentName, release.Filter.Name, release.Indexer)
|
||||
|
||||
var rejections []string
|
||||
|
||||
// run actions (watchFolder, test, exec, qBittorrent, Deluge, arr etc.)
|
||||
for _, a := range release.Filter.Actions {
|
||||
// only run enabled actions
|
||||
if !a.Enabled {
|
||||
log.Trace().Msgf("announce.Service.Process: indexer: %v, filter: %v release: %v action '%v' not enabled, skip", release.Indexer, release.Filter.Name, release.TorrentName, a.Name)
|
||||
continue
|
||||
}
|
||||
|
||||
log.Trace().Msgf("announce.Service.Process: indexer: %v, filter: %v release: %v , run action: %v", release.Indexer, release.Filter.Name, release.TorrentName, a.Name)
|
||||
|
||||
// keep track of action clients to avoid sending the same thing all over again
|
||||
_, tried := triedActionClients[actionClientTypeKey{Type: a.Type, ClientID: a.ClientID}]
|
||||
if tried {
|
||||
log.Trace().Msgf("announce.Service.Process: indexer: %v, filter: %v release: %v action client already tried, skip", release.Indexer, release.Filter.Name, release.TorrentName)
|
||||
continue
|
||||
}
|
||||
|
||||
rejections, err = s.actionSvc.RunAction(a, *release)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("announce.Service.Process: error running actions for filter: %v", release.Filter.Name)
|
||||
continue
|
||||
}
|
||||
|
||||
if len(rejections) > 0 {
|
||||
// if we get a rejection, remember which action client it was from
|
||||
triedActionClients[actionClientTypeKey{Type: a.Type, ClientID: a.ClientID}] = struct{}{}
|
||||
|
||||
// log something and fire events
|
||||
log.Debug().Msgf("announce.Service.Process: indexer: %v, filter: %v release: %v, rejected: %v", release.Indexer, release.Filter.Name, release.TorrentName, strings.Join(rejections, ", "))
|
||||
}
|
||||
|
||||
// if no rejections consider action approved, run next
|
||||
continue
|
||||
}
|
||||
|
||||
// if we have rejections from arr, continue to next filter
|
||||
if len(rejections) > 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// all actions run, decide to stop or continue here
|
||||
break
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue