1 # Copyright (c) 2009-2013, 2019-2021 kaliko <kaliko@azylum.org>
3 # This file is part of sima
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.
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.
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/>.
19 """SQlite database library
21 https://stackoverflow.com/questions/62818662/sqlite-foreign-key-reverse-cascade-delete
25 __HIST_DURATION__ = int(30 * 24) # in hours
29 from datetime import (datetime, timedelta)
31 from sima.lib.meta import Artist, Album
32 from sima.lib.track import Track
35 class SimaDBError(Exception):
44 def __init__(self, db_path=None):
45 self._db_path = db_path
47 def get_database_connection(self):
48 """get database reference"""
49 connection = sqlite3.connect(
50 self._db_path, isolation_level=None)
53 def close_database_connection(self, connection):
54 """Close the database connection."""
60 connection = self.get_database_connection()
62 'CREATE TABLE IF NOT EXISTS db_info'
63 ' (name CHAR(50), value CHAR(50))')
64 connection.execute('''INSERT INTO db_info (name, value) SELECT ?, ?
66 ( SELECT 1 FROM db_info WHERE name = ? )''',
67 ('DB Version', __DB_VERSION__, 'DB Version'))
68 connection.execute( # ARTISTS
69 'CREATE TABLE IF NOT EXISTS artists (id INTEGER PRIMARY KEY, '
70 'name VARCHAR(100), mbid CHAR(36))')
71 connection.execute( # ALBUMS
72 'CREATE TABLE IF NOT EXISTS albums (id INTEGER PRIMARY KEY, '
73 'name VARCHAR(100), mbid CHAR(36))')
74 connection.execute( # ALBUMARTISTS
75 'CREATE TABLE IF NOT EXISTS albumartists (id INTEGER PRIMARY KEY, '
76 'name VARCHAR(100), mbid CHAR(36))')
77 connection.execute( # TRACKS
78 'CREATE TABLE IF NOT EXISTS tracks (id INTEGER PRIMARY KEY, '
79 'title VARCHAR(100), artist INTEGER, '
80 'album INTEGER, albumartist INTEGER, '
81 'file VARCHAR(500), mbid CHAR(36), '
82 'FOREIGN KEY(artist) REFERENCES artists(id), '
83 'FOREIGN KEY(album) REFERENCES albums(id), '
84 'FOREIGN KEY(albumartist) REFERENCES albumartists(id))')
85 connection.execute( # HISTORY
86 'CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, '
87 'last_play TIMESTAMP, track integer, '
88 'FOREIGN KEY(track) REFERENCES tracks(id))')
89 connection.execute( # BLOCKLIST
90 'CREATE TABLE IF NOT EXISTS blocklist (id INTEGER PRIMARY KEY, '
91 'artist INTEGER, album INTEGER, track INTEGER, '
92 'FOREIGN KEY(artist) REFERENCES artists(id), '
93 'FOREIGN KEY(album) REFERENCES albums(id), '
94 'FOREIGN KEY(track) REFERENCES tracks(id))')
95 # Create cleanup triggers:
96 # DELETE history → Tracks table
97 connection.execute('''
98 CREATE TRIGGER IF NOT EXISTS del_history_cleanup_tracks
99 AFTER DELETE ON history
100 WHEN ((SELECT count(*) FROM history WHERE track=old.track) = 0 AND
101 (SELECT count(*) FROM blocklist WHERE track=old.track) = 0)
103 DELETE FROM tracks WHERE id = old.track;
106 # DELETE Tracks → Artists table
107 connection.execute('''
108 CREATE TRIGGER IF NOT EXISTS del_tracks_cleanup_artists
109 AFTER DELETE ON tracks
110 WHEN ((SELECT count(*) FROM tracks WHERE artist=old.artist) = 0 AND
111 (SELECT count(*) FROM blocklist WHERE artist=old.artist) = 0)
113 DELETE FROM artists WHERE id = old.artist;
116 # DELETE Tracks → Albums table
117 connection.execute('''
118 CREATE TRIGGER IF NOT EXISTS del_tracks_cleanup_albums
119 AFTER DELETE ON tracks
120 WHEN ((SELECT count(*) FROM tracks WHERE album=old.album) = 0 AND
121 (SELECT count(*) FROM blocklist WHERE album=old.album) = 0)
123 DELETE FROM albums WHERE id = old.album;
126 # DELETE Tracks → cleanup AlbumArtists table
127 connection.execute('''
128 CREATE TRIGGER IF NOT EXISTS del_tracks_cleanup_albumartists
129 AFTER DELETE ON tracks
130 WHEN ((SELECT count(*) FROM tracks WHERE albumartist=old.albumartist) = 0)
132 DELETE FROM albumartists WHERE id = old.albumartist;
135 # DELETE blocklist → Tracks table
136 connection.execute('''
137 CREATE TRIGGER IF NOT EXISTS del_blocklist_cleanup_tracks
138 AFTER DELETE ON blocklist
139 WHEN ((SELECT count(*) FROM history WHERE track=old.track) = 0 AND
140 (SELECT count(*) FROM blocklist WHERE track=old.track) = 0)
142 DELETE FROM tracks WHERE id = old.track;
145 self.close_database_connection(connection)
148 connection = self.get_database_connection()
149 rows = connection.execute(
150 "SELECT name FROM sqlite_master WHERE type='table'")
151 for r in rows.fetchall():
152 connection.execute(f'DROP TABLE IF EXISTS {r[0]}')
155 def _remove_blocklist_id(self, blid):
157 connection = self.get_database_connection()
158 connection.execute('DELETE FROM blocklist'
159 ' WHERE blocklist.id = ?', (blid,))
161 self.close_database_connection(connection)
163 def _get_album(self, album, connection):
165 return connection.execute(
166 "SELECT id FROM albums WHERE mbid = ?",
169 return connection.execute(
170 "SELECT id FROM albums WHERE name = ? AND mbid IS NULL",
173 def get_album(self, album, with_connection=None, add=True):
174 """get album information from the database.
175 if not in database insert new entry.
177 :param sima.lib.meta.Album album: album objet
178 :param sqlite3.Connection with_connection: SQLite connection
181 connection = with_connection
183 connection = self.get_database_connection()
184 rows = self._get_album(album, connection)
186 if not with_connection:
187 self.close_database_connection(connection)
190 if not with_connection:
191 self.close_database_connection(connection)
194 "INSERT INTO albums (name, mbid) VALUES (?, ?)",
195 (album.name, album.mbid))
197 rows = self._get_album(album, connection)
199 if not with_connection:
200 self.close_database_connection(connection)
202 print('damned: %s' % album.mbid)
203 if not with_connection:
204 self.close_database_connection(connection)
207 def _get_albumartist(self, artist, connection):
209 return connection.execute(
210 "SELECT id FROM albumartists WHERE mbid = ?",
213 return connection.execute(
214 "SELECT id FROM albumartists WHERE name = ? AND mbid IS NULL",
217 def get_albumartist(self, artist, with_connection=None, add=True):
218 """get albumartist information from the database.
219 if not in database insert new entry.
221 :param sima.lib.meta.Artist artist: artist
222 :param sqlite3.Connection with_connection: SQLite connection
225 connection = with_connection
227 connection = self.get_database_connection()
228 rows = self._get_albumartist(artist, connection)
230 if not with_connection:
231 self.close_database_connection(connection)
234 if not with_connection:
235 self.close_database_connection(connection)
238 "INSERT INTO albumartists (name, mbid) VALUES (?, ?)",
239 (artist.name, artist.mbid))
241 rows = self._get_albumartist(artist, connection)
243 if not with_connection:
244 self.close_database_connection(connection)
246 if not with_connection:
247 self.close_database_connection(connection)
249 def _get_artist(self, artist, connection):
251 return connection.execute(
252 "SELECT id FROM artists WHERE mbid = ?",
255 return connection.execute(
256 "SELECT id FROM artists WHERE name = ? AND mbid IS NULL", (artist.name,))
258 def get_artist(self, artist, with_connection=None, add=True):
259 """get artist information from the database.
260 if not in database insert new entry.
262 :param sima.lib.meta.Artist artist: artist
263 :param sqlite3.Connection with_connection: SQLite connection
266 connection = with_connection
268 connection = self.get_database_connection()
269 rows = self._get_artist(artist, connection)
271 if not with_connection:
272 self.close_database_connection(connection)
275 if not with_connection:
276 self.close_database_connection(connection)
279 "INSERT INTO artists (name, mbid) VALUES (?, ?)",
280 (artist.name, artist.mbid))
282 rows = self._get_artist(artist, connection)
284 if not with_connection:
285 self.close_database_connection(connection)
287 if not with_connection:
288 self.close_database_connection(connection)
290 def get_track(self, track, with_connection=None, add=True):
291 """Get a track from Tracks table, add if not existing,
292 Attention: use Track() object!!
293 if not in database insert new entry."""
295 raise SimaDBError('Got a track with no file attribute: %r' % track)
297 connection = with_connection
299 connection = self.get_database_connection()
300 rows = connection.execute(
301 "SELECT * FROM tracks WHERE file = ?", (track.file,))
303 if not with_connection:
304 self.close_database_connection(connection)
306 if not add: # Not adding non existing track
308 # Get an artist record or None
310 art = Artist(name=track.artist, mbid=track.musicbrainz_artistid)
311 art_id = self.get_artist(art, with_connection=connection)
314 # Get an albumartist record or None
315 if track.albumartist:
316 albart = Artist(name=track.albumartist,
317 mbid=track.musicbrainz_albumartistid)
318 albart_id = self.get_albumartist(albart, with_connection=connection)
321 # Get an album record or None
323 alb = Album(name=track.album, mbid=track.musicbrainz_albumid)
324 alb_id = self.get_album(alb, with_connection=connection)
328 """INSERT INTO tracks (artist, albumartist, album, title, mbid, file)
329 VALUES (?, ?, ?, ?, ?, ?)""",
330 (art_id, albart_id, alb_id, track.title, track.musicbrainz_trackid,
333 rows = connection.execute(
334 "SELECT id FROM tracks WHERE file = ?", (track.file,))
336 if not with_connection:
337 self.close_database_connection(connection)
339 if not with_connection:
341 self.close_database_connection(connection)
344 def add_history(self, track, date=None):
345 """Record last play date of track (ie. not a real exhautive play history).
346 :param track sima.lib.track.Track: track to add to history"""
348 date = datetime.now()
349 connection = self.get_database_connection()
350 track_id = self.get_track(track, with_connection=connection)
351 rows = connection.execute("SELECT * FROM history WHERE track = ? ",
353 if not rows.fetchone():
354 connection.execute("INSERT INTO history (track) VALUES (?)",
356 connection.execute("UPDATE history SET last_play = ? "
357 " WHERE track = ?", (date, track_id,))
359 self.close_database_connection(connection)
361 def purge_history(self, duration=__HIST_DURATION__):
362 """Remove old entries in history
363 :param duration int: Purge history record older than duration in hours
364 (defaults to __HIST_DURATION__)"""
365 connection = self.get_database_connection()
366 connection.execute("DELETE FROM history WHERE last_play"
367 " < datetime('now', '-%i hours')" % duration)
369 self.close_database_connection(connection)
371 def fetch_artists_history(self, duration=__HIST_DURATION__):
372 date = datetime.utcnow() - timedelta(hours=duration)
373 connection = self.get_database_connection()
374 connection.row_factory = sqlite3.Row
375 rows = connection.execute("""
376 SELECT artists.name AS name,
379 JOIN tracks ON history.track = tracks.id
380 LEFT OUTER JOIN artists ON tracks.artist = artists.id
381 WHERE history.last_play > ?
382 ORDER BY history.last_play DESC""", (date.isoformat(' '),))
385 if hist and hist[-1] == Album(**row): # remove consecutive dupes
387 hist.append(Album(**row))
391 def fetch_history(self, duration=__HIST_DURATION__):
392 """Fetches tracks history, more recent first
393 :param int duration: How long ago to fetch history from
395 date = datetime.utcnow() - timedelta(hours=duration)
396 connection = self.get_database_connection()
397 connection.row_factory = sqlite3.Row
398 rows = connection.execute("""
399 SELECT tracks.title, tracks.file, artists.name AS artist,
400 albumartists.name AS albumartist,
401 artists.mbid as musicbrainz_artistid,
402 albums.name AS album,
403 albums.mbid AS musicbrainz_albumid,
404 tracks.mbid as musicbrainz_trackid
406 JOIN tracks ON history.track = tracks.id
407 LEFT OUTER JOIN artists ON tracks.artist = artists.id
408 LEFT OUTER JOIN albumartists ON tracks.albumartist = albumartists.id
409 LEFT OUTER JOIN albums ON tracks.album = albums.id
410 WHERE history.last_play > ?
411 ORDER BY history.last_play DESC""", (date.isoformat(' '),))
414 hist.append(Track(**row))
418 def get_bl_track(self, track, with_connection=None, add=True):
419 """Add a track to blocklist
420 :param sima.lib.track.Track track: Track object to add to blocklist
421 :param sqlite3.Connection with_connection: sqlite3.Connection to reuse, else create a new one
422 :param bool add: Default is to add a new record, set to False to fetch associated record"""
424 connection = with_connection
426 connection = self.get_database_connection()
427 track_id = self.get_track(track, with_connection=connection, add=True)
428 rows = connection.execute(
429 "SELECT * FROM blocklist WHERE track = ?", (track_id,))
430 if not rows.fetchone():
433 connection.execute('INSERT INTO blocklist (track) VALUES (?)',
436 rows = connection.execute(
437 "SELECT * FROM blocklist WHERE track = ?", (track_id,))
438 return rows.fetchone()[0]
440 def get_bl_album(self, album, with_connection=None, add=True):
441 """Add an album to blocklist
442 :param sima.lib.meta.Album: Album object to add to blocklist
443 :param sqlite3.Connection with_connection: sqlite3.Connection to reuse, ele create a new one
444 :param bool add: Default is to add a new record, set to False to fetch associated record"""
446 connection = with_connection
448 connection = self.get_database_connection()
449 album_id = self.get_album(album, with_connection=connection, add=True)
450 rows = connection.execute(
451 "SELECT * FROM blocklist WHERE album = ?", (album_id,))
452 if not rows.fetchone():
455 connection.execute('INSERT INTO blocklist (album) VALUES (?)',
458 rows = connection.execute(
459 "SELECT * FROM blocklist WHERE album = ?", (album_id,))
460 return rows.fetchone()
462 def get_bl_artist(self, artist, with_connection=None, add=True):
463 """Add an artist to blocklist
464 :param sima.lib.meta.Artist: Artist object to add to blocklist
465 :param sqlite3.Connection with_connection: sqlite3.Connection to reuse, ele create a new one
466 :param bool add: Default is to add a new record, set to False to fetch associated record"""
468 connection = with_connection
470 connection = self.get_database_connection()
471 artist_id = self.get_artist(artist, with_connection=connection, add=True)
472 rows = connection.execute(
473 "SELECT * FROM blocklist WHERE artist = ?", (artist_id,))
474 if not rows.fetchone():
477 connection.execute('INSERT INTO blocklist (artist) VALUES (?)',
480 rows = connection.execute(
481 "SELECT * FROM blocklist WHERE artist = ?", (artist_id,))
482 return rows.fetchone()
488 'albumartist': 'Devolt',
490 'date': '2011-12-01',
491 'file': 'music/Devolt/2011-Grey/03-Devolt - Crazy.mp3',
492 'musicbrainz_albumartistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
493 'musicbrainz_albumid': 'ea2ef2cf-59e1-443a-817e-9066e3e0be4b',
494 'musicbrainz_artistid': 'd8e7e3e2-49ab-4f7c-b148-fc946d521f99',
495 'musicbrainz_trackid': 'fabf8fc9-2ae5-49c9-8214-a839c958d872',
496 'duration': '220.000',
498 db = SimaDB('/dev/shm/test.sqlite')
500 db.add_history(Track(**DEVOLT))
501 DEVOLT['file'] = 'foo'
502 print(db.get_bl_track(Track(**DEVOLT)))
503 db.add_history(Track(**DEVOLT))
506 # vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8