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
def get_database_connection(self):
"""get database reference"""
- connection = sqlite3.connect(
- self._db_path, isolation_level=None)
+ connection = sqlite3.connect(self._db_path, isolation_level=None)
return connection
def get_info(self):
connection.close()
return row[0]
if not add: # Not adding non existing track
- connection.close()
+ if not with_connection:
+ connection.close()
return None
# Get an artist record or None
if track.artist:
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 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)"""
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 = ? ",
"""
:param sima.lib.meta.Artist needle: When specified, returns albums history for this artist.
"""
- 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
: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
"""
- 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)
+ date = datetime.now(timezone.utc) - timedelta(hours=duration)
connection = self.get_database_connection()
rows = connection.execute("""
SELECT genres.name, artists.name
:param sima.lib.meta.Artist artist: limit history to this artist
:param int duration: How long ago to fetch history from
"""
- 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 = """
connection = with_connection
else:
connection = self.get_database_connection()
- track_id = self.get_track(track, with_connection=connection, add=True)
+ track_id = self.get_track(track, with_connection=connection, add=add)
rows = connection.execute(
"SELECT id FROM blocklist WHERE track = ?", (track_id,))
if not rows.fetchone():
if not add:
+ if not with_connection:
+ connection.close()
return None
connection.execute('INSERT INTO blocklist (track) VALUES (?)',
(track_id,))
connection = with_connection
else:
connection = self.get_database_connection()
- album_id = self.get_album(album, with_connection=connection, add=True)
+ album_id = self.get_album(album, with_connection=connection, add=add)
rows = connection.execute(
"SELECT id FROM blocklist WHERE album = ?", (album_id,))
if not rows.fetchone():
if not add:
+ if not with_connection:
+ connection.close()
return None
connection.execute('INSERT INTO blocklist (album) VALUES (?)',
(album_id,))
connection = with_connection
else:
connection = self.get_database_connection()
- artist_id = self.get_artist(artist, with_connection=connection, add=True)
+ artist_id = self.get_artist(artist, with_connection=connection, add=add)
rows = connection.execute(
"SELECT id FROM blocklist WHERE artist = ?", (artist_id,))
if not rows.fetchone():
connection.close()
return bl
+ def view_bl(self):
+ connection = self.get_database_connection()
+ connection.row_factory = sqlite3.Row
+ rows = connection.execute("""SELECT artists.name AS artist,
+ artists.mbid AS musicbrainz_artist,
+ albums.name AS album,
+ albums.mbid AS musicbrainz_album,
+ tracks.title AS title,
+ tracks.mbid AS musicbrainz_title,
+ tracks.file AS file,
+ blocklist.id
+ FROM blocklist
+ LEFT OUTER JOIN artists ON blocklist.artist = artists.id
+ LEFT OUTER JOIN albums ON blocklist.album = albums.id
+ LEFT OUTER JOIN tracks ON blocklist.track = tracks.id""")
+ res = [dict(row) for row in rows.fetchall()]
+ connection.close()
+ return res
+
def delete_bl(self, track=None, album=None, artist=None):
if not (track or album or artist):
return
home / dev / repositories / mpd-sima.git / blobdiff