Scanner: Task Progress (#866)

* Scanner: Add progress value in worker * Scanner: add progress in /tasks response + set progress for scan task * Scanner: set progress for refresh task * Scanner: set progress for cleaning task
Arthi-chaud · Feb 3, 2025 · 693f617 · 693f617
1 parent a3fa306
commit 693f617
Show file tree

Hide file tree

Showing 6 changed files with 74 additions and 18 deletions.
diff --git a/scanner/app/routes.go b/scanner/app/routes.go
@@ -17,7 +17,10 @@ type ScannerStatus struct {
 }
 
 type ScannerTaskStatus struct {
-	CurrentTask  *string  `json:"current_task"`
+	// Name of the currently running task
+	CurrentTask *string `json:"current_task"`
+	// Progress (0-100) of the running task. Can be null
+	Progress     *int     `json:"progress"`
 	PendingTasks []string `json:"pending_tasks"`
 }
 
@@ -27,6 +30,7 @@ func logTaskAdded(task t.Task) {
 	log.Info().Str("name", task.Name).Msg(TaskAddedtoQueueMessage)
 }
 
+// @Tags        Tasks
 // @Summary		Get Status of Scanner
 // @Produce		json
 // @Success		200	{object}	ScannerStatus
@@ -35,25 +39,30 @@ func (s *ScannerContext) Status(c echo.Context) error {
 	return c.JSON(http.StatusOK, ScannerStatus{Message: "Scanner is alive."})
 }
 
+// @Tags        Tasks
 // @Summary		Get Current + Pending Tasks
 // @Produce		json
 // @Success		200 {object} ScannerTaskStatus
 // @Router	    /tasks [get]
 func (s *ScannerContext) Tasks(c echo.Context) error {
-	currentTask, pendingTasks := s.worker.GetCurrentTasks()
+	currentTask, progressValue, pendingTasks := s.worker.GetCurrentTasks()
 	formattedCurentTask := &currentTask.Name
+	progressPtr := &progressValue
 	if currentTask.Name == "" {
 		formattedCurentTask = nil
+		progressPtr = nil
 	}
 	formattedPendingTasks := internal.Fmap(pendingTasks, func(t t.TaskInfo, _ int) string {
 		return t.Name
 	})
 	return c.JSON(http.StatusOK, ScannerTaskStatus{
 		CurrentTask:  formattedCurentTask,
+		Progress:     progressPtr,
 		PendingTasks: formattedPendingTasks,
 	})
 }
 
+// @Tags        Tasks
 // @Summary		Request a Scan for all libraries
 // @Produce		json
 // @Success		202	{object}	ScannerStatus
@@ -74,6 +83,7 @@ func (s *ScannerContext) ScanAll(c echo.Context) error {
 	return c.JSON(http.StatusAccepted, ScannerStatus{Message: TaskAddedtoQueueMessage})
 }
 
+// @Tags        Tasks
 // @Summary		Request a Scan for a single library
 // @Produce		json
 // @Success		202	{object}	ScannerStatus
@@ -94,6 +104,7 @@ func (s *ScannerContext) Scan(c echo.Context) error {
 	return c.JSON(http.StatusAccepted, ScannerStatus{Message: TaskAddedtoQueueMessage})
 }
 
+// @Tags        Tasks
 // @Summary		Request a Clean
 // @Produce		json
 // @Success		202	{object}	ScannerStatus
@@ -114,6 +125,7 @@ func (s *ScannerContext) Clean(c echo.Context) error {
 	return c.JSON(http.StatusAccepted, ScannerStatus{Message: TaskAddedtoQueueMessage})
 }
 
+// @Tags        Tasks
 // @Summary		Request a Clean for a single library
 // @Produce		json
 // @Success		202	{object}	ScannerStatus
@@ -134,6 +146,7 @@ func (s *ScannerContext) CleanLibrary(c echo.Context) error {
 	return c.JSON(http.StatusAccepted, ScannerStatus{Message: TaskAddedtoQueueMessage})
 }
 
+// @Tags        Tasks
 // @Summary		Refresh Metadata of selected files
 // @Description	Exactly one query parameter must be given
 // @Produce		json

diff --git a/scanner/internal/tasks/clean.go b/scanner/internal/tasks/clean.go
@@ -3,6 +3,7 @@ package tasks
 import (
 	"fmt"
 	"path"
+	"strconv"
 
 	"github.com/Arthi-chaud/Meelo/scanner/internal"
 	"github.com/Arthi-chaud/Meelo/scanner/internal/api"
@@ -21,21 +22,27 @@ func execClean(library api.Library, c config.Config, w *Worker) error {
 	if err != nil {
 		return err
 	}
+	w.SetProgress(25, 100)
 	filesInDir, err := filesystem.GetAllFilesInDirectory(path.Join(c.DataDirectory, library.Path))
 	if err != nil {
 		return err
 	}
+
+	w.SetProgress(50, 100)
 	filesToClean := []api.File{}
 	for _, registeredFile := range registeredFiles {
 		fullRegisteredPath := path.Join(c.DataDirectory, library.Path, registeredFile.Path)
 		if !internal.Contains(filesInDir, fullRegisteredPath) {
 			filesToClean = append(filesToClean, registeredFile)
 		}
 	}
+	w.SetProgress(75, 100)
 	successfulClean := DeleteFilesInApi(filesToClean, c, w)
+	w.SetProgress(100, 100)
 	log.Info().
+		Str("cleaned", strconv.Itoa(successfulClean)).
 		Str("library", library.Slug).
-		Msgf("Cleaned %d files.", successfulClean)
+		Msg("Finished cleaning files")
 	return nil
 }
 

diff --git a/scanner/internal/tasks/common.go b/scanner/internal/tasks/common.go
@@ -1,7 +1,6 @@
 package tasks
 
 import (
-	"fmt"
 	"path"
 
 	"github.com/Arthi-chaud/Meelo/scanner/internal"
@@ -14,7 +13,7 @@ import (
 func pushMetadata(fileFullPath string, m internal.Metadata, c config.Config, w *Worker, updateMethod api.SaveMetadataMethod) error {
 	created, err := api.SaveMetadata(c, m, updateMethod)
 	if err != nil {
-		return fmt.Errorf("saving metadata failed, this might be a bug")
+		return err
 	}
 	if len(m.IllustrationLocation) > 0 {
 		err := SaveIllustration(IllustrationTask{

diff --git a/scanner/internal/tasks/refresh.go b/scanner/internal/tasks/refresh.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"path"
 	"reflect"
+	"strconv"
 
 	"github.com/Arthi-chaud/Meelo/scanner/internal"
 	"github.com/Arthi-chaud/Meelo/scanner/internal/api"
@@ -21,6 +22,8 @@ func NewMetadataRefreshTask(refreshSelector api.FileSelectorDto, force bool, c c
 
 func execRefresh(refreshSelector api.FileSelectorDto, force bool, c config.Config, w *Worker) error {
 	successfulUpdates := 0
+	skippedUpdates := 0
+	failedUpdates := 0
 	libraries, err := api.GetAllLibraries(c)
 	if err != nil {
 		return err
@@ -29,10 +32,13 @@ func execRefresh(refreshSelector api.FileSelectorDto, force bool, c config.Confi
 	if err != nil {
 		return err
 	}
+	selectedFilesCount := len(selectedFiles)
 	for _, selectedFile := range selectedFiles {
+		w.SetProgress(skippedUpdates+failedUpdates+successfulUpdates, selectedFilesCount)
 		selectedFilePath, err := buildFullFileEntryPath(selectedFile, libraries, c)
 		if err != nil {
 			log.Error().Msg(err.Error())
+			failedUpdates++
 			continue
 		}
 		// If force is false, compute checksum,
@@ -42,9 +48,11 @@ func execRefresh(refreshSelector api.FileSelectorDto, force bool, c config.Confi
 			newChecksum, err := internal.ComputeChecksum(selectedFilePath)
 			if err != nil {
 				log.Error().Msg(err.Error())
+				failedUpdates++
 				continue
 			}
 			if newChecksum == selectedFile.Checksum {
+				skippedUpdates++
 				continue
 			}
 		}
@@ -60,16 +68,21 @@ func execRefresh(refreshSelector api.FileSelectorDto, force bool, c config.Confi
 			for _, err := range errs {
 				log.Trace().Msg(err.Error())
 			}
+			failedUpdates++
 			continue
 		}
 		err = pushMetadata(selectedFilePath, m, c, w, api.Update)
 		if err != nil {
 			log.Error().Msg(err.Error())
+			failedUpdates++
 		} else {
-			successfulUpdates = successfulUpdates + 1
+			successfulUpdates++
 		}
 	}
-	log.Info().Msgf("Updated metadata for %d files", successfulUpdates)
+	log.Info().
+		Str("sucess", strconv.Itoa(successfulUpdates)).
+		Str("failed", strconv.Itoa(failedUpdates)).
+		Msg("Finished updating metadata")
 	return nil
 }
 

diff --git a/scanner/internal/tasks/scan.go b/scanner/internal/tasks/scan.go
@@ -60,6 +60,7 @@ func execScan(library api.Library, c config.Config, w *Worker) error {
 func scanAndPostFiles(filePaths []string, c config.Config, w *Worker) int {
 	const chunkSize = 5
 	successfulRegistrations := 0
+	failedRegistration := 0
 	scanResChan := make(chan ScanRes, chunkSize)
 	defer close(scanResChan)
 	fileCount := len(filePaths)
@@ -70,23 +71,26 @@ func scanAndPostFiles(filePaths []string, c config.Config, w *Worker) int {
 		}
 		for range len(filesChunk) {
 			res := <-scanResChan
-			errCount := len(res.errors)
 			baseFile := path.Base(res.filePath)
-			if errCount != 0 {
+			if len(res.errors) != 0 {
 				log.Error().Str("file", baseFile).Msg("Parsing failed")
 				for _, err := range res.errors {
 					log.Trace().Msg(err.Error())
 				}
-				continue
-			}
-			log.Info().Str("file", baseFile).Msg("Parsing successful")
-			err := pushMetadata(res.filePath, res.metadata, c, w, api.Create)
-			if err != nil {
-				log.Error().Msg(err.Error())
+				failedRegistration = failedRegistration + 1
 			} else {
-				successfulRegistrations = successfulRegistrations + 1
+				log.Info().Str("file", baseFile).Msg("Parsing successful")
+				err := pushMetadata(res.filePath, res.metadata, c, w, api.Create)
+				if err != nil {
+					log.Error().Str("file", baseFile).Msg("Could not POST metadata")
+					log.Trace().Msg(err.Error())
+					failedRegistration = failedRegistration + 1
+				} else {
+					successfulRegistrations = successfulRegistrations + 1
+				}
 			}
 		}
+		w.SetProgress(successfulRegistrations+failedRegistration, fileCount)
 	}
 	return successfulRegistrations
 }

diff --git a/scanner/internal/tasks/worker.go b/scanner/internal/tasks/worker.go
@@ -1,6 +1,7 @@
 package tasks
 
 import (
+	"strconv"
 	"sync"
 
 	"github.com/Arthi-chaud/Meelo/scanner/internal"
@@ -13,6 +14,7 @@ type Worker struct {
 	thumbnailQueue chan ThumbnailTask
 	currentTask    Task
 	queuedTasks    []Task
+	progress       int // A number between 0 and 100
 	mu             sync.Mutex
 	wg             sync.WaitGroup
 }
@@ -40,12 +42,29 @@ func (w *Worker) StartWorker(c config.Config) {
 	}()
 }
 
+func (w *Worker) SetProgress(stepsFinished int, stepsCount int) {
+	if stepsCount == 0 {
+		log.Error().Msg("Could not set progress for task. Step count is zero.")
+	}
+	newProgress := int(float64(100*stepsFinished) / float64(stepsCount))
+	if newProgress < 0 || newProgress > 100 {
+		log.Warn().
+			Str("input", strconv.Itoa(newProgress)).
+			Msg("Attempt to set a progress value out of bound")
+		return
+	}
+	w.mu.Lock()
+	w.progress = newProgress
+	w.mu.Unlock()
+}
+
 func (w *Worker) process(task Task) {
 	defer w.wg.Done() // Decrement the WaitGroup counter when the task is done
 
 	w.mu.Lock()
 	w.queuedTasks = removeTask(w.queuedTasks, task.Id)
 	w.currentTask = task
+	w.progress = 0
 	w.mu.Unlock()
 
 	log.Info().Str("task", task.Name).Msgf("Processing task")
@@ -58,6 +77,7 @@ func (w *Worker) process(task Task) {
 	}
 	w.mu.Lock()
 	w.currentTask = Task{}
+	w.progress = 0
 	w.mu.Unlock()
 }
 
@@ -84,11 +104,11 @@ func removeTask(tasks []Task, id string) []Task {
 	return tasks
 }
 
-func (w *Worker) GetCurrentTasks() (TaskInfo, []TaskInfo) {
+func (w *Worker) GetCurrentTasks() (TaskInfo, int, []TaskInfo) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 
-	return w.currentTask.GetInfo(), internal.Fmap(w.queuedTasks, func(task Task, i int) TaskInfo {
+	return w.currentTask.GetInfo(), w.progress, internal.Fmap(w.queuedTasks, func(task Task, i int) TaskInfo {
 		return task.GetInfo()
 	})
 }