aboutsummaryrefslogtreecommitdiff
path: root/worker
diff options
context:
space:
mode:
Diffstat (limited to 'worker')
-rw-r--r--worker/worker.go47
1 files changed, 41 insertions, 6 deletions
diff --git a/worker/worker.go b/worker/worker.go
index 2444e89..525a736 100644
--- a/worker/worker.go
+++ b/worker/worker.go
@@ -1,10 +1,13 @@
package worker
import (
+ "bufio"
+ "bytes"
"context"
+ "fmt"
+ "os/exec"
"git.sr.ht/~gabrielgio/midr/db"
- "git.sr.ht/~gabrielgio/midr/yt"
work "git.sr.ht/~sircmpwn/dowork"
)
@@ -40,6 +43,30 @@ func NewWorkder() Worker {
}
}
+func RunYtDlpProcess(entry *db.Entry) (error, []byte, []byte) {
+ args := []string{entry.Link}
+ var stdout bytes.Buffer
+ var stderr bytes.Buffer
+
+ output_template := fmt.Sprintf("%s/%%(title)s.%%(ext)s", entry.OutputFolder)
+ args = append(args, "-o", output_template)
+
+ downloaded_txt := fmt.Sprintf("%s/downloaded.txt", entry.OutputFolder)
+ args = append(args, "--download-archive", downloaded_txt)
+
+ if len(entry.DateAfter) > 0 {
+ args = append(args, "--dateafter", entry.DateAfter)
+ }
+
+ cmd := exec.Command("yt-dlp", args...)
+ cmd.Stdout = bufio.NewWriter(&stdout)
+ cmd.Stderr = bufio.NewWriter(&stderr)
+
+ err := cmd.Run()
+
+ return err, stdout.Bytes(), stderr.Bytes()
+}
+
func (w *Worker) CanEnqueue(index uint) bool {
v, found := w.jobs[index]
return !found || v == statusNotQueued
@@ -49,23 +76,31 @@ func (w *Worker) RemoveJob(id uint) {
delete(w.jobs, id)
}
-func (w *Worker) SpawnWorker(entry *db.Entry) {
-
+func (w *Worker) RunYtDlpWorker(entry *db.Entry) <-chan []byte {
if !w.CanEnqueue(entry.ID) {
- return
+ return nil
}
+ log := make(chan []byte)
+
w.c <- command{action: commandEnqueue, index: entry.ID}
task := work.NewTask(func(ctx context.Context) error {
w.c <- command{action: commandStart, index: entry.ID}
- yt.RunYtDlpProcess(entry)
- return nil
+ err, stdout, stderr := RunYtDlpProcess(entry)
+
+ log <- stdout
+ log <- stderr
+
+ return err
}).After(func(ctx context.Context, task *work.Task) {
w.c <- command{action: commandDequeue, index: entry.ID}
+ close(log)
})
work.Enqueue(task)
+
+ return log
}
func (w *Worker) startReader() {