Feature: Exec action (#7)

* feat: implement exec action

* chore: change logs to trace

* refactor: extract from action
This commit is contained in:
Ludvig Lundgren 2021-08-15 02:40:38 +02:00 committed by GitHub
parent 4c19187a2f
commit 9eccc6b5e2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 340 additions and 157 deletions

57
internal/action/exec.go Normal file
View file

@ -0,0 +1,57 @@
package action
import (
"os/exec"
"strings"
"time"
"github.com/autobrr/autobrr/internal/domain"
"github.com/rs/zerolog/log"
)
func (s *service) execCmd(announce domain.Announce, action domain.Action, torrentFile string) {
log.Trace().Msgf("action EXEC: release: %v", announce.TorrentName)
// check if program exists
cmd, err := exec.LookPath(action.ExecCmd)
if err != nil {
log.Error().Err(err).Msgf("exec failed, could not find program: %v", action.ExecCmd)
return
}
// handle args and replace vars
m := Macro{
TorrentName: announce.TorrentName,
TorrentPathName: torrentFile,
TorrentUrl: announce.TorrentUrl,
}
// parse and replace values in argument string before continuing
parsedArgs, err := m.Parse(action.ExecArgs)
if err != nil {
log.Error().Err(err).Msgf("exec failed, could not parse arguments: %v", action.ExecCmd)
return
}
// we need to split on space into a string slice, so we can spread the args into exec
args := strings.Split(parsedArgs, " ")
start := time.Now()
// setup command and args
command := exec.Command(cmd, args...)
// execute command
output, err := command.CombinedOutput()
if err != nil {
// everything other than exit 0 is considered an error
log.Error().Err(err).Msgf("command: %v args: %v failed, torrent: %v", cmd, parsedArgs, torrentFile)
}
log.Trace().Msgf("executed command: '%v'", string(output))
duration := time.Since(start)
log.Info().Msgf("executed command: '%v', args: '%v' %v,%v, total time %v", cmd, parsedArgs, announce.TorrentName, announce.Site, duration)
}

30
internal/action/macros.go Normal file
View file

@ -0,0 +1,30 @@
package action
import (
"bytes"
"text/template"
)
type Macro struct {
TorrentName string
TorrentPathName string
TorrentUrl string
}
// Parse takes a string and replaces valid vars
func (m Macro) Parse(text string) (string, error) {
// setup template
tmpl, err := template.New("macro").Parse(text)
if err != nil {
return "", err
}
var tpl bytes.Buffer
err = tmpl.Execute(&tpl, m)
if err != nil {
return "", err
}
return tpl.String(), nil
}

View file

@ -0,0 +1,90 @@
package action
import (
"testing"
)
func TestMacros_Parse(t *testing.T) {
type fields struct {
TorrentName string
TorrentPathName string
TorrentUrl string
}
type args struct {
text string
}
tests := []struct {
name string
fields fields
args args
want string
wantErr bool
}{
{
name: "test_ok",
fields: fields{TorrentPathName: "/tmp/a-temporary-file.torrent"},
args: args{text: "Print mee {{.TorrentPathName}}"},
want: "Print mee /tmp/a-temporary-file.torrent",
wantErr: false,
},
{
name: "test_bad",
fields: fields{TorrentPathName: "/tmp/a-temporary-file.torrent"},
args: args{text: "Print mee {{TorrentPathName}}"},
want: "",
wantErr: true,
},
{
name: "test_program_arg",
fields: fields{TorrentPathName: "/tmp/a-temporary-file.torrent"},
args: args{text: "add {{.TorrentPathName}} --category test"},
want: "add /tmp/a-temporary-file.torrent --category test",
wantErr: false,
},
{
name: "test_program_arg_bad",
fields: fields{TorrentPathName: "/tmp/a-temporary-file.torrent"},
args: args{text: "add {{.TorrenttPathName}} --category test"},
want: "",
wantErr: true,
},
{
name: "test_program_arg",
fields: fields{
TorrentName: "This movie 2021",
TorrentPathName: "/tmp/a-temporary-file.torrent",
},
args: args{text: "add {{.TorrentPathName}} --category test --other {{.TorrentName}}"},
want: "add /tmp/a-temporary-file.torrent --category test --other This movie 2021",
wantErr: false,
},
{
name: "test_args_long",
fields: fields{
TorrentName: "This movie 2021",
TorrentUrl: "https://some.site/download/fakeid",
},
args: args{text: "{{.TorrentName}} {{.TorrentUrl}} SOME_LONG_TOKEN"},
want: "This movie 2021 https://some.site/download/fakeid SOME_LONG_TOKEN",
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
m := Macro{
TorrentPathName: tt.fields.TorrentPathName,
TorrentUrl: tt.fields.TorrentUrl,
TorrentName: tt.fields.TorrentName,
}
got, err := m.Parse(tt.args.text)
if (err != nil) != tt.wantErr {
t.Errorf("Parse() error = %v, wantErr %v", err, tt.wantErr)
return
}
if got != tt.want {
t.Errorf("Parse() got = %v, want %v", got, tt.want)
}
})
}
}

View file

@ -0,0 +1,148 @@
package action
import (
"strconv"
"time"
"github.com/autobrr/autobrr/internal/domain"
"github.com/autobrr/autobrr/pkg/qbittorrent"
"github.com/rs/zerolog/log"
)
func (s *service) qbittorrent(action domain.Action, hash string, torrentFile string) error {
log.Trace().Msgf("action QBITTORRENT: %v", torrentFile)
// get client for action
client, err := s.clientSvc.FindByID(action.ClientID)
if err != nil {
log.Error().Err(err).Msgf("error finding client: %v", action.ClientID)
return err
}
if client == nil {
return err
}
qbtSettings := qbittorrent.Settings{
Hostname: client.Host,
Port: uint(client.Port),
Username: client.Username,
Password: client.Password,
SSL: client.SSL,
}
qbt := qbittorrent.NewClient(qbtSettings)
// save cookies?
err = qbt.Login()
if err != nil {
log.Error().Err(err).Msgf("error logging into client: %v", action.ClientID)
return err
}
// TODO check for active downloads and other rules
options := map[string]string{}
if action.Paused {
options["paused"] = "true"
}
if action.SavePath != "" {
options["savepath"] = action.SavePath
options["autoTMM"] = "false"
}
if action.Category != "" {
options["category"] = action.Category
}
if action.Tags != "" {
options["tags"] = action.Tags
}
if action.LimitUploadSpeed > 0 {
options["upLimit"] = strconv.FormatInt(action.LimitUploadSpeed, 10)
}
if action.LimitDownloadSpeed > 0 {
options["dlLimit"] = strconv.FormatInt(action.LimitDownloadSpeed, 10)
}
err = qbt.AddTorrentFromFile(torrentFile, options)
if err != nil {
log.Error().Err(err).Msgf("error sending to client: %v", action.ClientID)
return err
}
if !action.Paused && hash != "" {
err = checkTrackerStatus(*qbt, hash)
if err != nil {
log.Error().Err(err).Msgf("could not get tracker status for torrent: %v", hash)
return err
}
}
log.Debug().Msgf("torrent %v successfully added to: %v", hash, client.Name)
return nil
}
func checkTrackerStatus(qb qbittorrent.Client, hash string) error {
announceOK := false
attempts := 0
for attempts < REANNOUNCE_MAX_ATTEMPTS {
log.Debug().Msgf("RE-ANNOUNCE %v attempt: %v", hash, attempts)
// initial sleep to give tracker a head start
time.Sleep(REANNOUNCE_INTERVAL * time.Millisecond)
trackers, err := qb.GetTorrentTrackers(hash)
if err != nil {
log.Error().Err(err).Msgf("could not get trackers of torrent: %v", hash)
return err
}
// check if status not working or something else
_, working := findTrackerStatus(trackers, qbittorrent.TrackerStatusOK)
if !working {
err = qb.ReAnnounceTorrents([]string{hash})
if err != nil {
log.Error().Err(err).Msgf("could not get re-announce torrent: %v", hash)
return err
}
attempts++
continue
} else {
log.Debug().Msgf("RE-ANNOUNCE %v OK", hash)
announceOK = true
break
}
}
if !announceOK {
log.Debug().Msgf("RE-ANNOUNCE %v took too long, deleting torrent", hash)
err := qb.DeleteTorrents([]string{hash}, false)
if err != nil {
log.Error().Err(err).Msgf("could not delete torrent: %v", hash)
return err
}
}
return nil
}
// Check if status not working or something else
// https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API-(qBittorrent-4.1)#get-torrent-trackers
// 0 Tracker is disabled (used for DHT, PeX, and LSD)
// 1 Tracker has not been contacted yet
// 2 Tracker has been contacted and is working
// 3 Tracker is updating
// 4 Tracker has been contacted, but it is not working (or doesn't send proper replies)
func findTrackerStatus(slice []qbittorrent.TorrentTracker, status qbittorrent.TrackerStatus) (int, bool) {
for i, item := range slice {
if item.Status == status {
return i, true
}
}
return -1, false
}

View file

@ -1,17 +1,12 @@
package action
import (
"fmt"
"io"
"os"
"strconv"
"strings"
"time"
"path"
"github.com/autobrr/autobrr/internal/domain"
"github.com/autobrr/autobrr/internal/download_client"
"github.com/autobrr/autobrr/pkg/qbittorrent"
"github.com/rs/zerolog/log"
)
@ -19,7 +14,7 @@ const REANNOUNCE_MAX_ATTEMPTS = 30
const REANNOUNCE_INTERVAL = 7000
type Service interface {
RunActions(torrentFile string, hash string, filter domain.Filter) error
RunActions(torrentFile string, hash string, filter domain.Filter, announce domain.Announce) error
Store(action domain.Action) (*domain.Action, error)
Fetch() ([]domain.Action, error)
Delete(actionID int) error
@ -35,14 +30,14 @@ func NewService(repo domain.ActionRepo, clientSvc download_client.Service) Servi
return &service{repo: repo, clientSvc: clientSvc}
}
func (s *service) RunActions(torrentFile string, hash string, filter domain.Filter) error {
func (s *service) RunActions(torrentFile string, hash string, filter domain.Filter, announce domain.Announce) error {
for _, action := range filter.Actions {
if !action.Enabled {
// only run active actions
continue
}
log.Debug().Msgf("process action: %v", action.Name)
log.Trace().Msgf("process action: %v", action.Name)
switch action.Type {
case domain.ActionTypeTest:
@ -59,11 +54,13 @@ func (s *service) RunActions(torrentFile string, hash string, filter domain.Filt
}
}()
case domain.ActionTypeExec:
go s.execCmd(announce, action, torrentFile)
// deluge
// pvr *arr
// exec
default:
panic("implement me")
log.Debug().Msgf("unsupported action: %v type: %v", action.Name, action.Type)
}
}
@ -111,7 +108,7 @@ func (s *service) test(torrentFile string) {
}
func (s *service) watchFolder(dir string, torrentFile string) {
log.Debug().Msgf("action WATCH_FOLDER: %v file: %v", dir, torrentFile)
log.Trace().Msgf("action WATCH_FOLDER: %v file: %v", dir, torrentFile)
// Open original file
original, err := os.Open(torrentFile)
@ -120,8 +117,8 @@ func (s *service) watchFolder(dir string, torrentFile string) {
}
defer original.Close()
tmpFileName := strings.Split(torrentFile, "/")
fullFileName := fmt.Sprintf("%v/%v", dir, tmpFileName[1])
_, tmpFileName := path.Split(torrentFile)
fullFileName := path.Join(dir, tmpFileName)
// Create new file
newFile, err := os.Create(fullFileName)
@ -136,143 +133,5 @@ func (s *service) watchFolder(dir string, torrentFile string) {
log.Fatal().Err(err)
}
log.Info().Msgf("action WATCH_FOLDER: wrote file: %v", fullFileName)
}
func (s *service) qbittorrent(action domain.Action, hash string, torrentFile string) error {
log.Debug().Msgf("action QBITTORRENT: %v", torrentFile)
// get client for action
client, err := s.clientSvc.FindByID(action.ClientID)
if err != nil {
log.Error().Err(err).Msgf("error finding client: %v", action.ClientID)
return err
}
if client == nil {
return err
}
qbtSettings := qbittorrent.Settings{
Hostname: client.Host,
Port: uint(client.Port),
Username: client.Username,
Password: client.Password,
SSL: client.SSL,
}
qbt := qbittorrent.NewClient(qbtSettings)
// save cookies?
err = qbt.Login()
if err != nil {
log.Error().Err(err).Msgf("error logging into client: %v", action.ClientID)
return err
}
// TODO check for active downloads and other rules
options := map[string]string{}
if action.Paused {
options["paused"] = "true"
}
if action.SavePath != "" {
options["savepath"] = action.SavePath
options["autoTMM"] = "false"
}
if action.Category != "" {
options["category"] = action.Category
}
if action.Tags != "" {
options["tags"] = action.Tags
}
if action.LimitUploadSpeed > 0 {
options["upLimit"] = strconv.FormatInt(action.LimitUploadSpeed, 10)
}
if action.LimitDownloadSpeed > 0 {
options["dlLimit"] = strconv.FormatInt(action.LimitDownloadSpeed, 10)
}
err = qbt.AddTorrentFromFile(torrentFile, options)
if err != nil {
log.Error().Err(err).Msgf("error sending to client: %v", action.ClientID)
return err
}
if !action.Paused && hash != "" {
err = checkTrackerStatus(*qbt, hash)
if err != nil {
log.Error().Err(err).Msgf("could not get tracker status for torrent: %v", hash)
return err
}
}
log.Debug().Msgf("torrent %v successfully added to: %v", hash, client.Name)
return nil
}
func checkTrackerStatus(qb qbittorrent.Client, hash string) error {
announceOK := false
attempts := 0
for attempts < REANNOUNCE_MAX_ATTEMPTS {
log.Debug().Msgf("RE-ANNOUNCE %v attempt: %v", hash, attempts)
// initial sleep to give tracker a head start
time.Sleep(REANNOUNCE_INTERVAL * time.Millisecond)
trackers, err := qb.GetTorrentTrackers(hash)
if err != nil {
log.Error().Err(err).Msgf("could not get trackers of torrent: %v", hash)
return err
}
// check if status not working or something else
_, working := findTrackerStatus(trackers, qbittorrent.TrackerStatusOK)
if !working {
err = qb.ReAnnounceTorrents([]string{hash})
if err != nil {
log.Error().Err(err).Msgf("could not get re-announce torrent: %v", hash)
return err
}
attempts++
continue
} else {
log.Debug().Msgf("RE-ANNOUNCE %v OK", hash)
announceOK = true
break
}
}
if !announceOK {
log.Debug().Msgf("RE-ANNOUNCE %v took too long, deleting torrent", hash)
err := qb.DeleteTorrents([]string{hash}, false)
if err != nil {
log.Error().Err(err).Msgf("could not delete torrent: %v", hash)
return err
}
}
return nil
}
// Check if status not working or something else
// https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API-(qBittorrent-4.1)#get-torrent-trackers
// 0 Tracker is disabled (used for DHT, PeX, and LSD)
// 1 Tracker has not been contacted yet
// 2 Tracker has been contacted and is working
// 3 Tracker is updating
// 4 Tracker has been contacted, but it is not working (or doesn't send proper replies)
func findTrackerStatus(slice []qbittorrent.TorrentTracker, status qbittorrent.TrackerStatus) (int, bool) {
for i, item := range slice {
if item.Status == status {
return i, true
}
}
return -1, false
log.Info().Msgf("saved file to watch folder: %v", fullFileName)
}

View file

@ -267,7 +267,7 @@ func (s *service) extractReleaseInfo(varMap map[string]string, releaseName strin
return err
}
log.Debug().Msgf("release: %+v", release)
log.Trace().Msgf("release: %+v", release)
// https://github.com/autodl-community/autodl-irssi/pull/194/files
// year

View file

@ -68,7 +68,7 @@ func (s *service) Parse(announceID string, msg string) error {
// no filter found, lets return
if foundFilter == nil {
log.Debug().Msg("no matching filter found")
log.Trace().Msg("no matching filter found")
return nil
}
announce.Filter = foundFilter

View file

@ -66,8 +66,7 @@ func (s *service) Process(announce domain.Announce) error {
hash := meta.HashInfoBytes().String()
// take action (watchFolder, test, runProgram, qBittorrent, Deluge etc)
// actionService
err = s.actionSvc.RunActions(res.FileName, hash, *announce.Filter)
err = s.actionSvc.RunActions(res.FileName, hash, *announce.Filter, announce)
if err != nil {
log.Error().Err(err).Msgf("error running actions for filter: %v", announce.Filter.Name)
return err