# -*- coding: utf-8 -*-
#
-# Copyright (c) 2009-2013 Jack Kaliko <jack@azylum.org>
+# Copyright (c) 2009-2013, 2019-2020 kaliko <kaliko@azylum.org>
# Copyright (c) 2009, Eric Casteleijn <thisfred@gmail.com>
# Copyright (c) 2008 Rick van Hattem
#
from datetime import (datetime, timedelta)
from os.path import dirname, isdir
from os import (access, W_OK, F_OK)
-from shutil import copyfile
class SimaDBError(Exception):
"""
Exceptions.
"""
- pass
class SimaDBAccessError(SimaDBError):
"""Error on accessing DB file"""
- pass
class SimaDBNoFile(SimaDBError):
"""No DB file present"""
- pass
-class SimaDB(object):
+class SimaDB:
"SQLite management"
def __init__(self, db_path=None):
# Controls directory access
if not isdir(dirname(db_path)):
raise SimaDBAccessError('Not a regular directory: "%s"' %
- dirname(db_path))
+ dirname(db_path))
if not access(dirname(db_path), W_OK):
raise SimaDBAccessError('No write access to "%s"' % dirname(db_path))
# Is a file but no write access
def get_database_connection(self):
"""get database reference"""
connection = sqlite3.connect(
- self._db_path, timeout=5.0, isolation_level="immediate")
+ self._db_path, timeout=5.0, isolation_level=None)
#connection.text_factory = str
return connection
def get_artist(self, artist_name, mbid=None,
- with_connection=None, add_not=False):
+ with_connection=None, add_not=False):
"""get artist information from the database.
if not in database insert new entry."""
if with_connection:
self.close_database_connection(connection)
def get_album(self, track, mbid=None,
- with_connection=None, add_not=False):
+ with_connection=None, add_not=False):
"""
get album information from the database.
if not in database insert new entry.
yield artist
def get_bl_artist(self, artist_name,
- with_connection=None, add_not=None):
+ with_connection=None, add_not=None):
"""get blacklisted artist information from the database."""
if with_connection:
connection = with_connection
else:
connection = self.get_database_connection()
- art = self.get_artist(artist_name,
- with_connection=connection, add_not=add_not)
+ art = self.get_artist(artist_name, with_connection=connection,
+ add_not=add_not)
if not art:
return False
art_id = art[0]
rows = connection.execute("SELECT * FROM black_list WHERE artist = ?",
- (art_id,))
+ (art_id,))
for row in rows:
if not with_connection:
self.close_database_connection(connection)
self.close_database_connection(connection)
return False
connection.execute("INSERT INTO black_list (artist) VALUES (?)",
- (art_id,))
+ (art_id,))
connection.execute("UPDATE black_list SET updated = DATETIME('now')"
- " WHERE artist = ?", (art_id,))
+ " WHERE artist = ?", (art_id,))
connection.commit()
rows = connection.execute("SELECT * FROM black_list WHERE artist = ?",
- (art_id,))
+ (art_id,))
for row in rows:
if not with_connection:
self.close_database_connection(connection)
self.close_database_connection(connection)
return False
- def get_bl_album(self, track,
- with_connection=None, add_not=None):
+ def get_bl_album(self, track, with_connection=None, add_not=None):
"""get blacklisted album information from the database."""
if with_connection:
connection = with_connection
else:
connection = self.get_database_connection()
- album = self.get_album(track,
- with_connection=connection, add_not=add_not)
+ album = self.get_album(track, with_connection=connection,
+ add_not=add_not)
if not album:
return False
alb_id = album[0]
rows = connection.execute("SELECT * FROM black_list WHERE album = ?",
- (alb_id,))
+ (alb_id,))
for row in rows:
if not with_connection:
self.close_database_connection(connection)
self.close_database_connection(connection)
return False
connection.execute("INSERT INTO black_list (album) VALUES (?)",
- (alb_id,))
+ (alb_id,))
connection.execute("UPDATE black_list SET updated = DATETIME('now')"
- " WHERE album = ?", (alb_id,))
+ " WHERE album = ?", (alb_id,))
connection.commit()
rows = connection.execute("SELECT * FROM black_list WHERE album = ?",
- (alb_id,))
+ (alb_id,))
for row in rows:
if not with_connection:
self.close_database_connection(connection)
connection = with_connection
else:
connection = self.get_database_connection()
- track = self.get_track(track,
- with_connection=connection, add_not=add_not)
+ track = self.get_track(track, with_connection=connection,
+ add_not=add_not)
if not track:
return False
track_id = track[0]
rows = connection.execute("SELECT * FROM black_list WHERE track = ?",
- (track_id,))
+ (track_id,))
for row in rows:
if not with_connection:
self.close_database_connection(connection)
self.close_database_connection(connection)
return False
connection.execute("INSERT INTO black_list (track) VALUES (?)",
- (track_id,))
+ (track_id,))
connection.execute("UPDATE black_list SET updated = DATETIME('now')"
- " WHERE track = ?", (track_id,))
+ " WHERE track = ?", (track_id,))
connection.commit()
rows = connection.execute("SELECT * FROM black_list WHERE track = ?",
- (track_id,))
+ (track_id,))
for row in rows:
if not with_connection:
self.close_database_connection(connection)
def get_artists_history(self, artists, duration=__HIST_DURATION__):
"""
+ :param list artists: list of object that can evaluate equality with
+ artist name, iterable of str or Artist object
"""
date = datetime.utcnow() - timedelta(hours=duration)
connection = self.get_database_connection()
"""Retrieve complete black list."""
connection = self.get_database_connection()
rows = connection.execute('SELECT black_list.rowid, artists.name'
- ' FROM artists INNER JOIN black_list'
- ' ON artists.id = black_list.artist')
+ ' FROM artists INNER JOIN black_list'
+ ' ON artists.id = black_list.artist')
yield ('Row ID', 'Actual black listed element', 'Extra information',)
yield ('',)
yield ('Row ID', 'Artist',)
else:
connection = self.get_database_connection()
connection.execute("UPDATE artists SET mbid = ? WHERE id = ?",
- (mbid, artist_id))
+ (mbid, artist_id))
connection.commit()
if not with_connection:
self.close_database_connection(connection)
"""Remove bl row id"""
connection = self.get_database_connection()
connection.execute('DELETE FROM black_list'
- ' WHERE black_list.rowid = ?', (rowid,))
+ ' WHERE black_list.rowid = ?', (rowid,))
connection.commit()
self.close_database_connection(connection)
connection.execute("INSERT INTO history (track) VALUES (?)",
(track_id,))
connection.execute("UPDATE history SET last_play = DATETIME('now') "
- " WHERE track = ?", (track_id,))
+ " WHERE track = ?", (track_id,))
connection.commit()
self.close_database_connection(connection)
connection = with_connection
else:
connection = self.get_database_connection()
- artists_ids = set([row[0] for row in connection.execute(
- "SELECT id FROM artists")])
- artist_2_artist_ids = set([row[0] for row in connection.execute(
- "SELECT artist FROM black_list")] +
- [row[0] for row in connection.execute(
- "SELECT artist FROM albums")] +
- [row[0] for row in connection.execute(
- "SELECT artist FROM tracks")])
+ artists_ids = {row[0] for row in connection.execute(
+ "SELECT id FROM artists")}
+ artist_2_artist_ids = {row[0] for row in connection.execute(
+ "SELECT artist FROM black_list")} | {
+ row[0] for row in connection.execute(
+ "SELECT artist FROM albums")} | {
+ row[0] for row in connection.execute(
+ "SELECT artist FROM tracks")}
orphans = [(orphan,) for orphan in artists_ids - artist_2_artist_ids]
connection.executemany('DELETE FROM artists WHERE id = (?);', orphans)
if not with_connection:
connection = with_connection
else:
connection = self.get_database_connection()
- orphan_black_ids = set([row[0] for row in connection.execute(
+ orphan_black_ids = {row[0] for row in connection.execute(
"""SELECT albums.id FROM albums
LEFT JOIN black_list ON albums.id = black_list.album
- WHERE ( black_list.album IS NULL )""")])
- orphan_tracks_ids = set([row[0] for row in connection.execute(
+ WHERE ( black_list.album IS NULL )""")}
+ orphan_tracks_ids = {row[0] for row in connection.execute(
"""SELECT albums.id FROM albums
LEFT JOIN tracks ON albums.id = tracks.album
- WHERE tracks.album IS NULL""")])
+ WHERE tracks.album IS NULL""")}
orphans = [(orphan,) for orphan in orphan_black_ids & orphan_tracks_ids]
connection.executemany('DELETE FROM albums WHERE id = (?);', orphans)
if not with_connection:
connection = with_connection
else:
connection = self.get_database_connection()
- hist_orphan_ids = set([row[0] for row in connection.execute(
+ hist_orphan_ids = {row[0] for row in connection.execute(
"""SELECT tracks.id FROM tracks
LEFT JOIN history ON tracks.id = history.track
- WHERE history.track IS NULL""")])
- black_list_orphan_ids = set([row[0] for row in connection.execute(
+ WHERE history.track IS NULL""")}
+ black_list_orphan_ids = {row[0] for row in connection.execute(
"""SELECT tracks.id FROM tracks
LEFT JOIN black_list ON tracks.id = black_list.track
- WHERE black_list.track IS NULL""")])
+ WHERE black_list.track IS NULL""")}
orphans = [(orphan,) for orphan in hist_orphan_ids & black_list_orphan_ids]
connection.executemany('DELETE FROM tracks WHERE id = (?);', orphans)
if not with_connection:
"""Remove old entries in history"""
connection = self.get_database_connection()
connection.execute("DELETE FROM history WHERE last_play"
- " < datetime('now', '-%i hours')" % duration)
+ " < datetime('now', '-%i hours')" % duration)
connection.commit()
self.close_database_connection(connection)
"""Add db version"""
connection = self.get_database_connection()
connection.execute('INSERT INTO db_info (version, name) VALUES (?, ?)',
- (__DB_VERSION__, 'Sima DB'))
+ (__DB_VERSION__, 'Sima DB'))
connection.commit()
self.close_database_connection(connection)
home / dev / repositories / mpd-sima.git / blobdiff