scanner: Add a new FileTask for scanning music files

This task is also responsible for checking if the file already exists in
the database before scanning.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

scanner/task.py

@@ -1,6 +1,7 @@
 # Copyright 2021 (c) Anna Schumaker.
 import db
 from gi.repository import GObject
+from . import metadata
 
 class Task(GObject.GObject):
     def run_task(self): raise NotImplementedError
@@ -8,3 +9,26 @@ class Task(GObject.GObject):
 
 class CommitTask(Task):
     def run_task(self): db.commit()
+
+
+class FileTask(Task):
+    def __init__(self, library, filepath):
+        Task.__init__(self)
+        self.library  = library
+        self.filepath = filepath
+
+    def run_task(self):
+        if db.track.Table.lookup(self.filepath):
+            return
+
+        with metadata.Metadata(self.filepath) as meta:
+            artist = db.artist.Table.find(meta.artist(), meta.artistsort())
+            album  = db.album.Table.find(artist, meta.album())
+            disc   = db.disc.Table.find(album, meta.discnumber(), meta.discsubtitle())
+            year   = db.year.Table.find(meta.year())
+            track  = db.track.Table.insert(self.library, artist, album, disc, year,
+                                           meta.tracknumber(), meta.length(),
+                                           meta.title(), self.filepath)
+            for genre in meta.genres():
+                db.genre.Map.insert(db.genre.Table.find(genre), track)
+            db.playlist.TempMap.insert(db.playlist.Table.find("New Tracks"), track)

scanner/test_task.py

@@ -1,8 +1,13 @@
 # Copyright 2021 (c) Anna Schumaker.
+import db
+import pathlib
 import unittest
 from gi.repository import GObject
 from . import task
 
+test_tracks  = pathlib.Path("./data/Test Album")
+test_track01 = test_tracks / "01 - Test Track.ogg"
+
 class TestScannerTask(unittest.TestCase):
     def test_scanner_task_init(self):
         t = task.Task()
@@ -16,3 +21,40 @@ class TestScannerCommitTask(unittest.TestCase):
         ct = task.CommitTask()
         self.assertIsInstance(ct, task.Task)
         self.assertEqual(ct.run_task(), None)
+
+
+class TestScannerFileTask(unittest.TestCase):
+    def setUp(self):
+        db.reset()
+
+    def test_scanner_file_task(self):
+        lib = db.library.Table.find(test_tracks)
+        ft = task.FileTask(lib, test_track01)
+
+        self.assertIsInstance(ft, task.Task)
+        self.assertEqual(ft.library,  lib)
+        self.assertEqual(ft.filepath, test_track01)
+
+        self.assertIsNone(ft.run_task())
+        self.assertIsNone(ft.run_task())
+
+        artist = db.artist.Table.lookup("Test Artist")
+        self.assertEqual(artist.sort, "artist, test")
+
+        album = db.album.Table.lookup(artist, "Test Album")
+        self.assertIsNotNone(album)
+
+        self.assertIsNotNone(db.disc.Table.lookup(album, 1))
+        self.assertIsNotNone(db.year.Table.lookup(2019))
+
+        track = db.track.Table.lookup(test_track01)
+        self.assertEqual(track.number, 1)
+        self.assertEqual(track.length, 10)
+        self.assertEqual(track.title, "Test Track")
+        self.assertEqual(track.path, test_track01)
+
+        genre = db.genre.Table.lookup("Test")
+        self.assertEqual(db.genre.Map.lookup_tracks(genre), [ track ])
+
+        new = db.playlist.Table.find("New Tracks")
+        self.assertEqual(db.playlist.TempMap.lookup_tracks(new), [ track ])