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
38 def __init__(self, db_path=None):
39 self._db_path = db_path
41 def get_database_connection(self):
42 """get database reference"""
43 connection = sqlite3.connect(
44 self._db_path, isolation_level=None)
47 def close_database_connection(self, connection):
48 """Close the database connection."""
54 connection = self.get_database_connection()
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 ?, ?
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:
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)
90 DELETE FROM tracks WHERE id = old.id;
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)
99 DELETE FROM artists WHERE id = old.artist;
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)
108 DELETE FROM albums WHERE id = old.album;
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)
117 DELETE FROM albumartists WHERE id = old.albumartist;
120 self.close_database_connection(connection)
122 def _get_album(self, album, connection):
124 return connection.execute(
125 "SELECT id FROM albums WHERE mbid = ?",
128 return connection.execute(
129 "SELECT id FROM albums WHERE name = ? AND mbid IS NULL",
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.
136 :param sima.lib.meta.Album album: album objet
137 :param sqlite3.Connection with_connection: SQLite connection
140 connection = with_connection
142 connection = self.get_database_connection()
143 rows = self._get_album(album, connection)
145 if not with_connection:
146 self.close_database_connection(connection)
149 if not with_connection:
150 self.close_database_connection(connection)
153 "INSERT INTO albums (name, mbid) VALUES (?, ?)",
154 (album.name, album.mbid))
156 rows = self._get_album(album, connection)
158 if not with_connection:
159 self.close_database_connection(connection)
161 print('damned: %s' % album.mbid)
162 if not with_connection:
163 self.close_database_connection(connection)
166 def _get_albumartist(self, artist, connection):
168 return connection.execute(
169 "SELECT id FROM albumartists WHERE mbid = ?",
172 return connection.execute(
173 "SELECT id FROM albumartists WHERE name = ? AND mbid IS NULL",
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.
180 :param sima.lib.meta.Artist artist: artist
181 :param sqlite3.Connection with_connection: SQLite connection
184 connection = with_connection
186 connection = self.get_database_connection()
187 rows = self._get_albumartist(artist, connection)
189 if not with_connection:
190 self.close_database_connection(connection)
193 if not with_connection:
194 self.close_database_connection(connection)
197 "INSERT INTO albumartists (name, mbid) VALUES (?, ?)",
198 (artist.name, artist.mbid))
200 rows = self._get_albumartist(artist, connection)
202 if not with_connection:
203 self.close_database_connection(connection)
205 if not with_connection:
206 self.close_database_connection(connection)
208 def _get_artist(self, artist, connection):
210 return connection.execute(
211 "SELECT id FROM artists WHERE mbid = ?",
214 return connection.execute(
215 "SELECT id FROM artists WHERE name = ? AND mbid IS NULL", (artist.name,))
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.
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_artist(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 artists (name, mbid) VALUES (?, ?)",
239 (artist.name, artist.mbid))
241 rows = self._get_artist(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_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."""
254 connection = with_connection
256 connection = self.get_database_connection()
257 rows = connection.execute(
258 "SELECT * FROM tracks WHERE file = ?", (track.file,))
260 if not with_connection:
261 self.close_database_connection(connection)
263 if not add: # Not adding non existing track
265 # Get an artist record or None
267 art = Artist(name=track.artist, mbid=track.musicbrainz_artistid)
268 art_id = self.get_artist(art, with_connection=connection)
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)
278 # Get an album record or None
280 alb = Album(name=track.album, mbid=track.musicbrainz_albumid)
281 alb_id = self.get_album(alb, with_connection=connection)
285 """INSERT INTO tracks (artist, albumartist, album, title, mbid, file)
286 VALUES (?, ?, ?, ?, ?, ?)""",
287 (art_id, albart_id, alb_id, track.title, track.musicbrainz_trackid,
290 rows = connection.execute(
291 "SELECT id FROM tracks WHERE file = ?", (track.file,))
293 if not with_connection:
294 self.close_database_connection(connection)
296 if not with_connection:
298 self.close_database_connection(connection)
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"""
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 = ? ",
310 if not rows.fetchone():
311 connection.execute("INSERT INTO history (track) VALUES (?)",
313 connection.execute("UPDATE history SET last_play = ? "
314 " WHERE track = ?", (date, track_id,))
316 self.close_database_connection(connection)
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)
326 self.close_database_connection(connection)
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
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(' '),))
348 hist.append(Track(**row))
354 db = SimaDB('/dev/shm/test.sqlite')
358 # vim: ai ts=4 sw=4 sts=4 expandtab fileencoding=utf8