sima/lib/db.py

   1 # Copyright (c) 2009-2013, 2019-2021 kaliko <kaliko@azylum.org>
   2 #
   3 #  This file is part of sima
   4 #
   5 #  sima is free software: you can redistribute it and/or modify
   6 #  it under the terms of the GNU General Public License as published by
   7 #  the Free Software Foundation, either version 3 of the License, or
   8 #  (at your option) any later version.
   9 #
  10 #  sima is distributed in the hope that it will be useful,
  11 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #  GNU General Public License for more details.
  14 #
  15 #  You should have received a copy of the GNU General Public License
  16 #  along with sima.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 #
  19 """SQlite database library
  20
  21 https://stackoverflow.com/questions/62818662/sqlite-foreign-key-reverse-cascade-delete
  22 """
  23
  24 __DB_VERSION__ = 4
  25 __HIST_DURATION__ = int(30 * 24)  # in hours
  26
  27 import sqlite3
  28
  29 from datetime import (datetime, timedelta)
  30
  31 from sima.lib.meta import Artist, Album
  32 from sima.lib.track import Track
  33
  34
  35 class SimaDB:
  36     "SQLite management"
  37
  38     def __init__(self, db_path=None):
  39         self._db_path = db_path
  40
  41     def get_database_connection(self):
  42         """get database reference"""
  43         connection = sqlite3.connect(
  44             self._db_path, isolation_level=None)
  45         return connection
  46
  47     def close_database_connection(self, connection):
  48         """Close the database connection."""
  49         connection.close()
  50
  51     def create_db(self):
  52         """ Set up a database
  53         """
  54         connection = self.get_database_connection()
  55         connection.execute(
  56             'CREATE TABLE IF NOT EXISTS db_info'
  57             ' (name CHAR(50), value CHAR(50))')
  58         connection.execute('''INSERT INTO db_info (name, value) SELECT ?, ?
  59                            WHERE NOT EXISTS
  60                            ( SELECT 1 FROM db_info WHERE name = ? )''',
  61                            ('DB Version', __DB_VERSION__, 'DB Version'))
  62         connection.execute(  # ARTISTS
  63             'CREATE TABLE IF NOT EXISTS artists (id INTEGER PRIMARY KEY, '
  64             'name VARCHAR(100), mbid CHAR(36))')
  65         connection.execute(  # ALBUMS
  66             'CREATE TABLE IF NOT EXISTS albums (id INTEGER PRIMARY KEY, '
  67             'name VARCHAR(100), mbid CHAR(36))')
  68         connection.execute(  # ALBUMARTISTS
  69             'CREATE TABLE IF NOT EXISTS albumartists (id INTEGER PRIMARY KEY, '
  70             'name VARCHAR(100), mbid CHAR(36))')
  71         connection.execute(  # TRACKS
  72             'CREATE TABLE IF NOT EXISTS tracks (id INTEGER PRIMARY KEY, '
  73             'title VARCHAR(100), artist INTEGER, '
  74             'album INTEGER, albumartist INTEGER, '
  75             'file VARCHAR(500), mbid CHAR(36), '
  76             'FOREIGN KEY(artist)       REFERENCES artists(id), '
  77             'FOREIGN KEY(album)        REFERENCES albums(id), '
  78             'FOREIGN KEY(albumartist)  REFERENCES albumartists(id))')
  79         connection.execute(  # HISTORY
  80             'CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, '
  81             'last_play TIMESTAMP, track integer, '
  82             'FOREIGN KEY(track) REFERENCES tracks(id))')
  83         # Create cleanup triggers:
  84         # Tracks table
  85         connection.execute('''
  86                 CREATE TRIGGER IF NOT EXISTS cleanup_tracks
  87                 AFTER DELETE ON history
  88                 WHEN ((SELECT count(*) FROM history WHERE track=old.id) = 0)
  89                 BEGIN
  90                  DELETE FROM tracks WHERE id = old.id;
  91                 END;
  92                 ''')
  93         # Artists table
  94         connection.execute('''
  95                 CREATE TRIGGER IF NOT EXISTS cleanup_artists
  96                 AFTER DELETE ON tracks
  97                 WHEN ((SELECT count(*) FROM tracks WHERE artist=old.artist) = 0)
  98                 BEGIN
  99                  DELETE FROM artists WHERE id = old.artist;
 100                 END;
 101                 ''')
 102         # Albums table
 103         connection.execute('''
 104                 CREATE TRIGGER IF NOT EXISTS cleanup_albums
 105                 AFTER DELETE ON tracks
 106                 WHEN ((SELECT count(*) FROM tracks WHERE album=old.album) = 0)
 107                 BEGIN
 108                  DELETE FROM albums WHERE id = old.album;
 109                 END;
 110                 ''')
 111         # AlbumArtists table
 112         connection.execute('''
 113                 CREATE TRIGGER IF NOT EXISTS cleanup_albumartists
 114                 AFTER DELETE ON tracks
 115                 WHEN ((SELECT count(*) FROM tracks WHERE albumartist=old.albumartist) = 0)
 116                 BEGIN
 117                  DELETE FROM albumartists WHERE id = old.albumartist;
 118                 END;
 119                 ''')
 120         self.close_database_connection(connection)
 121
 122     def _get_album(self, album, connection):
 123         if album.mbid:
 124             return connection.execute(
 125                 "SELECT id FROM albums WHERE mbid = ?",
 126                 (album.mbid,))
 127         else:
 128             return connection.execute(
 129                 "SELECT id FROM albums WHERE name = ? AND mbid IS NULL",
 130                 (album.name,))
 131
 132     def get_album(self, album, with_connection=None, add=True):
 133         """get album information from the database.
 134         if not in database insert new entry.
 135
 136         :param sima.lib.meta.Album album: album objet
 137         :param sqlite3.Connection with_connection: SQLite connection
 138         """
 139         if with_connection:
 140             connection = with_connection
 141         else:
 142             connection = self.get_database_connection()
 143         rows = self._get_album(album, connection)
 144         for row in rows:
 145             if not with_connection:
 146                 self.close_database_connection(connection)
 147             return row[0]
 148         if not add:
 149             if not with_connection:
 150                 self.close_database_connection(connection)
 151             return None
 152         connection.execute(
 153             "INSERT INTO albums (name, mbid) VALUES (?, ?)",
 154             (album.name, album.mbid))
 155         connection.commit()
 156         rows = self._get_album(album, connection)
 157         for row in rows:
 158             if not with_connection:
 159                 self.close_database_connection(connection)
 160             return row[0]
 161         print('damned: %s' % album.mbid)
 162         if not with_connection:
 163             self.close_database_connection(connection)
 164         return None
 165
 166     def _get_albumartist(self, artist, connection):
 167         if artist.mbid:
 168             return connection.execute(
 169                 "SELECT id FROM albumartists WHERE mbid = ?",
 170                 (artist.mbid,))
 171         else:
 172             return connection.execute(
 173                 "SELECT id FROM albumartists WHERE name = ? AND mbid IS NULL",
 174                 (artist.name,))
 175
 176     def get_albumartist(self, artist, with_connection=None, add=True):
 177         """get albumartist information from the database.
 178         if not in database insert new entry.
 179
 180         :param sima.lib.meta.Artist artist: artist
 181         :param sqlite3.Connection with_connection: SQLite connection
 182         """
 183         if with_connection:
 184             connection = with_connection
 185         else:
 186             connection = self.get_database_connection()
 187         rows = self._get_albumartist(artist, connection)
 188         for row in rows:
 189             if not with_connection:
 190                 self.close_database_connection(connection)
 191             return row[0]
 192         if not add:
 193             if not with_connection:
 194                 self.close_database_connection(connection)
 195             return None
 196         connection.execute(
 197             "INSERT INTO albumartists (name, mbid) VALUES (?, ?)",
 198             (artist.name, artist.mbid))
 199         connection.commit()
 200         rows = self._get_albumartist(artist, connection)
 201         for row in rows:
 202             if not with_connection:
 203                 self.close_database_connection(connection)
 204             return row[0]
 205         if not with_connection:
 206             self.close_database_connection(connection)
 207
 208     def _get_artist(self, artist, connection):
 209         if artist.mbid:
 210             return connection.execute(
 211                 "SELECT id FROM artists WHERE mbid = ?",
 212                 (artist.mbid,))
 213         else:
 214             return connection.execute(
 215                 "SELECT id FROM artists WHERE name = ? AND mbid IS NULL", (artist.name,))
 216
 217     def get_artist(self, artist, with_connection=None, add=True):
 218         """get artist information from the database.
 219         if not in database insert new entry.
 220
 221         :param sima.lib.meta.Artist artist: artist
 222         :param sqlite3.Connection with_connection: SQLite connection
 223         """
 224         if with_connection:
 225             connection = with_connection
 226         else:
 227             connection = self.get_database_connection()
 228         rows = self._get_artist(artist, connection)
 229         for row in rows:
 230             if not with_connection:
 231                 self.close_database_connection(connection)
 232             return row[0]
 233         if not add:
 234             if not with_connection:
 235                 self.close_database_connection(connection)
 236             return None
 237         connection.execute(
 238             "INSERT INTO artists (name, mbid) VALUES (?, ?)",
 239             (artist.name, artist.mbid))
 240         connection.commit()
 241         rows = self._get_artist(artist, connection)
 242         for row in rows:
 243             if not with_connection:
 244                 self.close_database_connection(connection)
 245             return row[0]
 246         if not with_connection:
 247             self.close_database_connection(connection)
 248
 249     def get_track(self, track, with_connection=None, add=True):
 250         """Get a track from Tracks table, add if not existing,
 251         Attention: use Track() object!!
 252         if not in database insert new entry."""
 253         if with_connection:
 254             connection = with_connection
 255         else:
 256             connection = self.get_database_connection()
 257         rows = connection.execute(
 258             "SELECT * FROM tracks WHERE file = ?", (track.file,))
 259         for row in rows:
 260             if not with_connection:
 261                 self.close_database_connection(connection)
 262             return row[0]
 263         if not add:  # Not adding non existing track
 264             return None
 265         # Get an artist record or None
 266         if track.artist:
 267             art = Artist(name=track.artist, mbid=track.musicbrainz_artistid)
 268             art_id = self.get_artist(art, with_connection=connection)
 269         else:
 270             art_id = None
 271         # Get an albumartist record or None
 272         if track.albumartist:
 273             albart = Artist(name=track.albumartist,
 274                             mbid=track.musicbrainz_albumartistid)
 275             albart_id = self.get_albumartist(albart, with_connection=connection)
 276         else:
 277             albart_id = None
 278         # Get an album record or None
 279         if track.album:
 280             alb = Album(name=track.album, mbid=track.musicbrainz_albumid)
 281             alb_id = self.get_album(alb, with_connection=connection)
 282         else:
 283             alb_id = None
 284         connection.execute(
 285             """INSERT INTO tracks (artist, albumartist, album, title, mbid, file)
 286                 VALUES (?, ?, ?, ?, ?, ?)""",
 287             (art_id, albart_id, alb_id, track.title, track.musicbrainz_trackid,
 288                 track.file))
 289         connection.commit()
 290         rows = connection.execute(
 291             "SELECT id FROM tracks WHERE file = ?", (track.file,))
 292         for row in rows:
 293             if not with_connection:
 294                 self.close_database_connection(connection)
 295             return row[0]
 296         if not with_connection:
 297             connection.commit()
 298             self.close_database_connection(connection)
 299         return None
 300
 301     def add_history(self, track, date=None):
 302         """Record last play date of track (ie. not a real exhautive play history).
 303         :param track sima.lib.track.Track: track to add to history"""
 304         if not date:
 305             date = datetime.now()
 306         connection = self.get_database_connection()
 307         track_id = self.get_track(track, with_connection=connection)
 308         rows = connection.execute("SELECT * FROM history WHERE track = ? ",
 309                                   (track_id,))
 310         if not rows.fetchone():
 311             connection.execute("INSERT INTO history (track) VALUES (?)",
 312                                (track_id,))
 313         connection.execute("UPDATE history SET last_play = ? "
 314                            " WHERE track = ?", (date, track_id,))
 315         connection.commit()
 316         self.close_database_connection(connection)
 317
 318     def purge_history(self, duration=__HIST_DURATION__):
 319         """Remove old entries in history
 320         :param duration int: Purge history record older than duration in hours
 321                             (defaults to __HIST_DURATION__)"""
 322         connection = self.get_database_connection()
 323         connection.execute("DELETE FROM history WHERE last_play"
 324                            " < datetime('now', '-%i hours')" % duration)
 325         connection.commit()
 326         self.close_database_connection(connection)
 327
 328     def get_history(self, duration=__HIST_DURATION__):
 329         date = datetime.utcnow() - timedelta(hours=duration)
 330         connection = self.get_database_connection()
 331         connection.row_factory = sqlite3.Row
 332         rows = connection.execute("""
 333                 SELECT tracks.title, tracks.file, artists.name AS artist,
 334                        albumartists.name AS albumartist,
 335                        artists.mbid as musicbrainz_artistid,
 336                        albums.name AS album,
 337                        albums.mbid AS musicbrainz_albumid,
 338                        tracks.mbid as musicbrainz_trackid
 339                 FROM history
 340                 JOIN tracks ON history.track = tracks.id
 341                 LEFT OUTER JOIN artists ON tracks.artist = artists.id
 342                 LEFT OUTER JOIN albumartists ON tracks.albumartist = albumartists.id
 343                 LEFT OUTER JOIN albums ON tracks.album = albums.id
 344                 WHERE history.last_play > ?
 345                 ORDER BY history.last_play DESC""", (date.isoformat(' '),))
 346         hist = list()
 347         for row in rows:
 348             hist.append(Track(**row))
 349         connection.close()
 350         return hist
 351
 352
 353 def main():
 354     db = SimaDB('/dev/shm/test.sqlite')
 355     db.create_db()
 356
 357 # VIM MODLINE
 358 # vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8