"""SQlite database library
"""
+#: DB Version
__DB_VERSION__ = 4
-__HIST_DURATION__ = int(30 * 24) # in hours
+#: Default history duration for both request and purge in hours
+__HIST_DURATION__ = int(30 * 24)
import sqlite3
from sima.lib.meta import Artist, Album
from sima.lib.track import Track
+from sima.utils.utils import MPDSimaException
-class SimaDBError(Exception):
+class SimaDBError(MPDSimaException):
"""
Exceptions.
"""
connection = self.get_database_connection()
rows = connection.execute(
"SELECT name FROM sqlite_master WHERE type='table'")
- for r in rows.fetchall():
- connection.execute(f'DROP TABLE IF EXISTS {r[0]}')
+ for row in rows.fetchall():
+ connection.execute(f'DROP TABLE IF EXISTS {row[0]}')
connection.close()
def _remove_blocklist_id(self, blid, with_connection=None):
- """Remove id"""
+ """Remove a blocklist id"""
if with_connection:
connection = with_connection
else:
return connection.execute(
"SELECT id FROM albums WHERE mbid = ?",
(album.mbid,))
- else:
- return connection.execute(
- "SELECT id FROM albums WHERE name = ? AND mbid IS NULL",
- (album.name,))
+ return connection.execute(
+ "SELECT id FROM albums WHERE name = ? AND mbid IS NULL",
+ (album.name,))
def get_album(self, album, with_connection=None, add=True):
"""get album information from the database.
return connection.execute(
"SELECT id FROM albumartists WHERE mbid = ?",
(artist.mbid,))
- else:
- return connection.execute(
- "SELECT id FROM albumartists WHERE name = ? AND mbid IS NULL",
- (artist.name,))
+ return connection.execute(
+ "SELECT id FROM albumartists WHERE name = ? AND mbid IS NULL",
+ (artist.name,))
def get_albumartist(self, artist, with_connection=None, add=True):
"""get albumartist information from the database.
return connection.execute(
"SELECT id FROM artists WHERE mbid = ?",
(artist.mbid,))
- else:
- return connection.execute(
- "SELECT id FROM artists WHERE name = ? AND mbid IS NULL", (artist.name,))
+ return connection.execute(
+ "SELECT id FROM artists WHERE name = ? AND mbid IS NULL", (artist.name,))
def get_artist(self, artist, with_connection=None, add=True):
"""get artist information from the database.
def get_track(self, track, with_connection=None, add=True):
"""Get a track id from Tracks table, add if not existing,
+
:param sima.lib.track.Track track: track to use
:param bool add: add non existing track to database"""
if not track.file:
- raise SimaDBError('Got a track with no file attribute: %r' % track)
+ raise SimaDBError(f'Got a track with no file attribute: {track}')
if with_connection:
connection = with_connection
else:
def _add_tracks_genres(self, track, connection):
if not track.genres:
- return None
+ return
rows = connection.execute(
"SELECT id FROM tracks WHERE file = ?", (track.file,))
trk_id = rows.fetchone()[0]
def add_history(self, track, date=None):
"""Record last play date of track (ie. not a real play history).
- :param track sima.lib.track.Track: track to add to history
- :param date datetime.datetime: UTC datetime object (use "datetime.now(timezone.utc)" is not set)"""
+
+ :param sima.lib.track.Track track: track to add to history
+ :param datetime.datetime date: UTC datetime object (use "datetime.now(timezone.utc)" is not set)"""
if not date:
date = datetime.now(timezone.utc)
connection = self.get_database_connection()
def purge_history(self, duration=__HIST_DURATION__):
"""Remove old entries in history
- :param duration int: Purge history record older than duration in hours
- (defaults to __HIST_DURATION__)"""
+
+ :param int duration: Purge history record older than duration in hours"""
connection = self.get_database_connection()
connection.execute("DELETE FROM history WHERE last_play"
" < datetime('now', '-%i hours')" % duration)
def fetch_albums_history(self, needle=None, duration=__HIST_DURATION__):
"""
:param sima.lib.meta.Artist needle: When specified, returns albums history for this artist.
+ :param int duration: How long ago to fetch history from (in hours)
"""
date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
SELECT albums.name AS name,
albums.mbid as mbid,
artists.name as artist,
- artists.mbid as artist_mbib
+ artists.mbid as artist_mbib,
+ albumartists.name as albumartist,
+ albumartists.mbid as albumartist_mbib
FROM history
JOIN tracks ON history.track = tracks.id
LEFT OUTER JOIN albums ON tracks.album = albums.id
LEFT OUTER JOIN artists ON tracks.artist = artists.id
+ LEFT OUTER JOIN albumartists ON tracks.albumartist = albumartists.id
WHERE history.last_play > ? AND albums.name NOT NULL AND artists.name NOT NULL
ORDER BY history.last_play DESC""", (date.isoformat(' '),))
- hist = list()
+ hist = []
for row in rows:
vals = dict(row)
- artist = Artist(name=vals.pop('artist'),
- mbid=vals.pop('artist_mbib'))
- if needle:
- if needle != artist:
+ if needle: # Here use artist instead of albumartist
+ if needle != Artist(name=vals.get('artist'),
+ mbid=vals.get('artist_mbib')):
continue
- album = Album(**vals, artist=artist)
+ # Use albumartist / MBIDs if possible to build album artist
+ if not vals.get('albumartist'):
+ vals['albumartist'] = vals.get('artist')
+ if not vals.get('albumartist_mbib'):
+ vals['albumartist_mbib'] = vals.get('artist_mbib')
+ artist = Artist(name=vals.get('albumartist'),
+ mbid=vals.pop('albumartist_mbib'))
+ album = Album(**vals, Artist=artist)
if hist and hist[-1] == album:
# remove consecutive dupes
continue
def fetch_artists_history(self, needle=None, duration=__HIST_DURATION__):
"""Returns a list of Artist objects
- :param sima.lib.meta.Artist|sima.lib.meta.MetaContainer needle: When specified, returns history for this artist, it's actually testing the artist presence in history.
+
:param sima.lib.meta.MetaContainer needle: When specified, returns history for these artists only
+ :param int duration: How long ago to fetch history from (in hours)
+ :type needle: sima.lib.meta.Artist or sima.lib.meta.MetaContainer
"""
date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
WHERE history.last_play > ? AND artists.name NOT NULL
ORDER BY history.last_play DESC""", (date.isoformat(' '),))
last = deque(maxlen=1)
- hist = list()
+ hist = []
for row in rows:
artist = Artist(**row)
if last and last[0] == artist: # remove consecutive dupes
hist.append(artist) # No need to go further
break
continue
- elif needle and getattr(needle, '__contains__'):
+ if needle and getattr(needle, '__contains__'):
if artist in needle:
hist.append(artist) # No need to go further
continue
return hist
def fetch_genres_history(self, duration=__HIST_DURATION__, limit=20):
+ """Returns genre history
+
+ :param int duration: How long ago to fetch history from (in hours)
+ :param int limit: number of genre to fetch
+ """
date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
rows = connection.execute("""
WHERE history.last_play > ? AND genres.name NOT NULL
ORDER BY history.last_play DESC
""", (date.isoformat(' '),))
- genres = list()
+ genres = []
for row in rows:
genres.append(row)
if len({g[0] for g in genres}) >= limit:
def fetch_history(self, artist=None, duration=__HIST_DURATION__):
"""Fetches tracks history, more recent first
+
:param sima.lib.meta.Artist artist: limit history to this artist
- :param int duration: How long ago to fetch history from
+ :param int duration: How long ago to fetch history from (in hours)
"""
date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
else:
rows = connection.execute(sql+'ORDER BY history.last_play DESC',
(date.isoformat(' '),))
- hist = list()
+ hist = []
for row in rows:
hist.append(Track(**row))
connection.close()
def get_bl_track(self, track, with_connection=None, add=True):
"""Add a track to blocklist
+
:param sima.lib.track.Track track: Track object to add to blocklist
:param sqlite3.Connection with_connection: sqlite3.Connection to reuse, else create a new one
:param bool add: Default is to add a new record, set to False to fetch associated record"""
connection.commit()
rows = connection.execute(
"SELECT id FROM blocklist WHERE track = ?", (track_id,))
- bl = rows.fetchone()[0]
+ blt = rows.fetchone()[0]
if not with_connection:
connection.close()
- return bl
+ return blt
def get_bl_album(self, album, with_connection=None, add=True):
"""Add an album to blocklist
+
:param sima.lib.meta.Album: Album object to add to blocklist
:param sqlite3.Connection with_connection: sqlite3.Connection to reuse, else create a new one
:param bool add: Default is to add a new record, set to False to fetch associated record"""
connection.commit()
rows = connection.execute(
"SELECT id FROM blocklist WHERE album = ?", (album_id,))
- bl = rows.fetchone()[0]
+ blitem = rows.fetchone()[0]
if not with_connection:
connection.close()
- return bl
+ return blitem
def get_bl_artist(self, artist, with_connection=None, add=True):
"""Add an artist to blocklist
+
:param sima.lib.meta.Artist: Artist object to add to blocklist
:param sqlite3.Connection with_connection: sqlite3.Connection to reuse, else create a new one
:param bool add: Default is to add a new record, set to False to fetch associated record"""
connection.commit()
rows = connection.execute(
"SELECT id FROM blocklist WHERE artist = ?", (artist_id,))
- bl = rows.fetchone()[0]
+ blitem = rows.fetchone()[0]
if not with_connection:
connection.close()
- return bl
+ return blitem
def view_bl(self):
connection = self.get_database_connection()
home / dev / repositories / mpd-sima.git / blobdiff