"""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 collections import deque
from datetime import (datetime, timedelta)
+from datetime import timezone
+
from sima.lib.meta import Artist, Album
from sima.lib.track import Track
connection.close()
def _remove_blocklist_id(self, blid, with_connection=None):
- """Remove id"""
+ """Remove a blocklist id"""
if with_connection:
connection = with_connection
else:
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:
VALUES (?, ?)""", (trk_id, gen_id))
def add_history(self, track, date=None):
- """Record last play date of track (ie. not a real exhautive play history).
- :param track sima.lib.track.Track: track to add to history"""
+ """Record last play date of track (ie. not a real play history).
+
+ :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()
+ date = datetime.now(timezone.utc)
connection = self.get_database_connection()
track_id = self.get_track(track, with_connection=connection)
rows = connection.execute("SELECT * FROM history WHERE track = ? ",
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.utcnow() - timedelta(hours=duration)
+ date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
connection.row_factory = sqlite3.Row
rows = connection.execute("""
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()
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.utcnow() - timedelta(hours=duration)
+ date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
connection.row_factory = sqlite3.Row
rows = connection.execute("""
return hist
def fetch_genres_history(self, duration=__HIST_DURATION__, limit=20):
- date = datetime.utcnow() - timedelta(hours=duration)
+ """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("""
SELECT genres.name, artists.name
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.utcnow() - timedelta(hours=duration)
+ date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
connection.row_factory = sqlite3.Row
sql = """
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"""
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"""
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"""
home / dev / repositories / mpd-sima.git / blobdiff