mirror of
https://github.com/krateng/maloja.git
synced 2023-08-10 21:12:55 +03:00
Refactoring (#83)
* Merge isinstance calls * Inline variable that is immediately returned * Replace set() with comprehension * Replace assignment with augmented assignment * Remove unnecessary else after guard condition * Convert for loop into list comprehension * Replace unused for index with underscore * Merge nested if conditions * Convert for loop into list comprehension * Convert for loop into set comprehension * Remove unnecessary else after guard condition * Replace if statements with if expressions * Simplify sequence comparison * Replace multiple comparisons with in operator * Merge isinstance calls * Merge nested if conditions * Add guard clause * Merge duplicate blocks in conditional * Replace unneeded comprehension with generator * Inline variable that is immediately returned * Remove unused imports * Replace unneeded comprehension with generator * Remove unused imports * Remove unused import * Inline variable that is immediately returned * Swap if/else branches and remove unnecessary else * Use str.join() instead of for loop * Multiple refactors - Remove redundant pass statement - Hoist repeated code outside conditional statement - Swap if/else to remove empty if body * Inline variable that is immediately returned * Simplify generator expression * Replace if statement with if expression * Multiple refactoring - Replace range(0, x) with range(x) - Swap if/else branches - Remove unnecessary else after guard condition * Use str.join() instead of for loop * Hoist repeated code outside conditional statement * Use str.join() instead of for loop * Inline variables that are immediately returned * Merge dictionary assignment with declaration * Use items() to directly unpack dictionary values * Extract dup code from methods into a new one
This commit is contained in:
parent
c31770a34c
commit
d1b598a32b
@ -54,26 +54,25 @@ class Audioscrobbler(APIHandler):
|
|||||||
def submit_scrobble(self,pathnodes,keys):
|
def submit_scrobble(self,pathnodes,keys):
|
||||||
if keys.get("sk") is None or keys.get("sk") not in self.mobile_sessions:
|
if keys.get("sk") is None or keys.get("sk") not in self.mobile_sessions:
|
||||||
raise InvalidSessionKey()
|
raise InvalidSessionKey()
|
||||||
|
if "track" in keys and "artist" in keys:
|
||||||
|
artiststr,titlestr = keys["artist"], keys["track"]
|
||||||
|
#(artists,title) = cla.fullclean(artiststr,titlestr)
|
||||||
|
try:
|
||||||
|
timestamp = int(keys["timestamp"])
|
||||||
|
except:
|
||||||
|
timestamp = None
|
||||||
|
#database.createScrobble(artists,title,timestamp)
|
||||||
|
self.scrobble(artiststr,titlestr,time=timestamp)
|
||||||
else:
|
else:
|
||||||
if "track" in keys and "artist" in keys:
|
for num in range(50):
|
||||||
artiststr,titlestr = keys["artist"], keys["track"]
|
if "track[" + str(num) + "]" in keys:
|
||||||
#(artists,title) = cla.fullclean(artiststr,titlestr)
|
artiststr,titlestr = keys["artist[" + str(num) + "]"], keys["track[" + str(num) + "]"]
|
||||||
try:
|
#(artists,title) = cla.fullclean(artiststr,titlestr)
|
||||||
timestamp = int(keys["timestamp"])
|
timestamp = int(keys["timestamp[" + str(num) + "]"])
|
||||||
except:
|
#database.createScrobble(artists,title,timestamp)
|
||||||
timestamp = None
|
self.scrobble(artiststr,titlestr,time=timestamp)
|
||||||
#database.createScrobble(artists,title,timestamp)
|
|
||||||
self.scrobble(artiststr,titlestr,time=timestamp)
|
return 200,{"scrobbles":{"@attr":{"ignored":0}}}
|
||||||
return 200,{"scrobbles":{"@attr":{"ignored":0}}}
|
|
||||||
else:
|
|
||||||
for num in range(50):
|
|
||||||
if "track[" + str(num) + "]" in keys:
|
|
||||||
artiststr,titlestr = keys["artist[" + str(num) + "]"], keys["track[" + str(num) + "]"]
|
|
||||||
#(artists,title) = cla.fullclean(artiststr,titlestr)
|
|
||||||
timestamp = int(keys["timestamp[" + str(num) + "]"])
|
|
||||||
#database.createScrobble(artists,title,timestamp)
|
|
||||||
self.scrobble(artiststr,titlestr,time=timestamp)
|
|
||||||
return 200,{"scrobbles":{"@attr":{"ignored":0}}}
|
|
||||||
|
|
||||||
|
|
||||||
import hashlib
|
import hashlib
|
||||||
@ -85,8 +84,11 @@ def md5(input):
|
|||||||
return m.hexdigest()
|
return m.hexdigest()
|
||||||
|
|
||||||
def generate_key(ls):
|
def generate_key(ls):
|
||||||
key = ""
|
key = "".join(
|
||||||
for i in range(64):
|
str(
|
||||||
key += str(random.choice(list(range(10)) + list("abcdefghijklmnopqrstuvwxyz") + list("ABCDEFGHIJKLMNOPQRSTUVWXYZ")))
|
random.choice(
|
||||||
|
list(range(10)) + list("abcdefghijklmnopqrstuvwxyz") +
|
||||||
|
list("ABCDEFGHIJKLMNOPQRSTUVWXYZ"))) for _ in range(64))
|
||||||
|
|
||||||
ls.append(key)
|
ls.append(key)
|
||||||
return key
|
return key
|
||||||
|
@ -38,8 +38,7 @@ class AudioscrobblerLegacy(APIHandler):
|
|||||||
timestamp = keys.get("t")
|
timestamp = keys.get("t")
|
||||||
apikey = keys.get("api_key")
|
apikey = keys.get("api_key")
|
||||||
host = keys.get("Host")
|
host = keys.get("Host")
|
||||||
protocol = request.urlparts.scheme
|
protocol = 'http' if (keys.get("u") == 'nossl') else request.urlparts.scheme
|
||||||
if (keys.get("u") == 'nossl'): protocol = 'http' #user override
|
|
||||||
|
|
||||||
if auth is not None:
|
if auth is not None:
|
||||||
for key in database.allAPIkeys():
|
for key in database.allAPIkeys():
|
||||||
@ -68,22 +67,20 @@ class AudioscrobblerLegacy(APIHandler):
|
|||||||
def submit_scrobble(self,pathnodes,keys):
|
def submit_scrobble(self,pathnodes,keys):
|
||||||
if keys.get("s") is None or keys.get("s") not in self.mobile_sessions:
|
if keys.get("s") is None or keys.get("s") not in self.mobile_sessions:
|
||||||
raise InvalidSessionKey()
|
raise InvalidSessionKey()
|
||||||
else:
|
for count in range(50):
|
||||||
for count in range(0,50):
|
artist_key = f"a[{count}]"
|
||||||
artist_key = f"a[{count}]"
|
track_key = f"t[{count}]"
|
||||||
track_key = f"t[{count}]"
|
time_key = f"i[{count}]"
|
||||||
time_key = f"i[{count}]"
|
if artist_key not in keys or track_key not in keys:
|
||||||
if artist_key in keys and track_key in keys:
|
return 200,"OK\n"
|
||||||
artiststr,titlestr = keys[artist_key], keys[track_key]
|
artiststr,titlestr = keys[artist_key], keys[track_key]
|
||||||
try:
|
try:
|
||||||
timestamp = int(keys[time_key])
|
timestamp = int(keys[time_key])
|
||||||
except:
|
except:
|
||||||
timestamp = None
|
timestamp = None
|
||||||
#database.createScrobble(artists,title,timestamp)
|
#database.createScrobble(artists,title,timestamp)
|
||||||
self.scrobble(artiststr,titlestr,time=timestamp)
|
self.scrobble(artiststr,titlestr,time=timestamp)
|
||||||
else:
|
return 200,"OK\n"
|
||||||
return 200,"OK\n"
|
|
||||||
return 200,"OK\n"
|
|
||||||
|
|
||||||
|
|
||||||
import hashlib
|
import hashlib
|
||||||
@ -95,9 +92,12 @@ def md5(input):
|
|||||||
return m.hexdigest()
|
return m.hexdigest()
|
||||||
|
|
||||||
def generate_key(ls):
|
def generate_key(ls):
|
||||||
key = ""
|
key = "".join(
|
||||||
for i in range(64):
|
str(
|
||||||
key += str(random.choice(list(range(10)) + list("abcdefghijklmnopqrstuvwxyz") + list("ABCDEFGHIJKLMNOPQRSTUVWXYZ")))
|
random.choice(
|
||||||
|
list(range(10)) + list("abcdefghijklmnopqrstuvwxyz") +
|
||||||
|
list("ABCDEFGHIJKLMNOPQRSTUVWXYZ"))) for _ in range(64))
|
||||||
|
|
||||||
ls.append(key)
|
ls.append(key)
|
||||||
return key
|
return key
|
||||||
|
|
||||||
|
@ -72,8 +72,7 @@ def info_external(**keys):
|
|||||||
response.set_header("Access-Control-Allow-Origin","*")
|
response.set_header("Access-Control-Allow-Origin","*")
|
||||||
response.set_header("Content-Type","application/json")
|
response.set_header("Content-Type","application/json")
|
||||||
|
|
||||||
result = info()
|
return info()
|
||||||
return result
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@ -178,8 +177,7 @@ def artistInfo_external(**keys):
|
|||||||
k_filter, _, _, _, _ = uri_to_internal(keys,forceArtist=True)
|
k_filter, _, _, _, _ = uri_to_internal(keys,forceArtist=True)
|
||||||
ckeys = {**k_filter}
|
ckeys = {**k_filter}
|
||||||
|
|
||||||
results = artistInfo(**ckeys)
|
return artistInfo(**ckeys)
|
||||||
return results
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@ -192,15 +190,12 @@ def trackInfo_external(artist:Multi[str],**keys):
|
|||||||
k_filter, _, _, _, _ = uri_to_internal(keys,forceTrack=True)
|
k_filter, _, _, _, _ = uri_to_internal(keys,forceTrack=True)
|
||||||
ckeys = {**k_filter}
|
ckeys = {**k_filter}
|
||||||
|
|
||||||
results = trackInfo(**ckeys)
|
return trackInfo(**ckeys)
|
||||||
return results
|
|
||||||
|
|
||||||
|
|
||||||
@api.get("compare")
|
@api.get("compare")
|
||||||
def compare_external(**keys):
|
def compare_external(**keys):
|
||||||
|
return compare(keys["remote"])
|
||||||
results = compare(keys["remote"])
|
|
||||||
return results
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@ -297,8 +292,10 @@ def search(**keys):
|
|||||||
# add links
|
# add links
|
||||||
artists_result = []
|
artists_result = []
|
||||||
for a in artists:
|
for a in artists:
|
||||||
result = {"name":a}
|
result = {
|
||||||
result["link"] = "/artist?" + compose_querystring(internal_to_uri({"artist":a}))
|
'name': a,
|
||||||
|
'link': "/artist?" + compose_querystring(internal_to_uri({"artist": a})),
|
||||||
|
}
|
||||||
result["image"] = "/image?" + compose_querystring(internal_to_uri({"artist":a}))
|
result["image"] = "/image?" + compose_querystring(internal_to_uri({"artist":a}))
|
||||||
artists_result.append(result)
|
artists_result.append(result)
|
||||||
|
|
||||||
|
@ -35,12 +35,12 @@ class CleanerAgent:
|
|||||||
reqartists, allartists = self.rules_addartists[title.lower()]
|
reqartists, allartists = self.rules_addartists[title.lower()]
|
||||||
reqartists = reqartists.split("␟")
|
reqartists = reqartists.split("␟")
|
||||||
allartists = allartists.split("␟")
|
allartists = allartists.split("␟")
|
||||||
if set(reqartists).issubset(set(a.lower() for a in artists)):
|
if set(reqartists).issubset({a.lower() for a in artists}):
|
||||||
artists += allartists
|
artists += allartists
|
||||||
elif title.lower() in self.rules_fixartists:
|
elif title.lower() in self.rules_fixartists:
|
||||||
allartists = self.rules_fixartists[title.lower()]
|
allartists = self.rules_fixartists[title.lower()]
|
||||||
allartists = allartists.split("␟")
|
allartists = allartists.split("␟")
|
||||||
if len(set(a.lower() for a in allartists) & set(a.lower() for a in artists)) > 0:
|
if len({a.lower() for a in allartists} & {a.lower() for a in artists}) > 0:
|
||||||
artists = allartists
|
artists = allartists
|
||||||
artists = list(set(artists))
|
artists = list(set(artists))
|
||||||
artists.sort()
|
artists.sort()
|
||||||
@ -50,10 +50,9 @@ class CleanerAgent:
|
|||||||
def removespecial(self,s):
|
def removespecial(self,s):
|
||||||
if isinstance(s,list):
|
if isinstance(s,list):
|
||||||
return [self.removespecial(se) for se in s]
|
return [self.removespecial(se) for se in s]
|
||||||
else:
|
s = s.replace("\t","").replace("␟","").replace("\n","")
|
||||||
s = s.replace("\t","").replace("␟","").replace("\n","")
|
s = re.sub(" +"," ",s)
|
||||||
s = re.sub(" +"," ",s)
|
return s
|
||||||
return s
|
|
||||||
|
|
||||||
|
|
||||||
# if an artist appears in any created rule, we can assume that artist is meant to exist and be spelled like that
|
# if an artist appears in any created rule, we can assume that artist is meant to exist and be spelled like that
|
||||||
@ -206,9 +205,7 @@ class CollectorAgent:
|
|||||||
|
|
||||||
# get all credited artists for the artists given
|
# get all credited artists for the artists given
|
||||||
def getCreditedList(self,artists):
|
def getCreditedList(self,artists):
|
||||||
updatedArtists = []
|
updatedArtists = [self.getCredited(artist) for artist in artists]
|
||||||
for artist in artists:
|
|
||||||
updatedArtists.append(self.getCredited(artist))
|
|
||||||
return list(set(updatedArtists))
|
return list(set(updatedArtists))
|
||||||
|
|
||||||
# get artists who the given artist is given credit for
|
# get artists who the given artist is given credit for
|
||||||
@ -218,7 +215,7 @@ class CollectorAgent:
|
|||||||
# this function is there to check for artists that we should include in the
|
# this function is there to check for artists that we should include in the
|
||||||
# database even though they never have any scrobble.
|
# database even though they never have any scrobble.
|
||||||
def getAllArtists(self):
|
def getAllArtists(self):
|
||||||
return list(set([self.rules_countas[a] for a in self.rules_countas]))
|
return list({self.rules_countas[a] for a in self.rules_countas})
|
||||||
# artists that count can be nonexisting (counting HyunA as 4Minute even
|
# artists that count can be nonexisting (counting HyunA as 4Minute even
|
||||||
# though 4Minute has never been listened to)
|
# though 4Minute has never been listened to)
|
||||||
# but artists that are counted as someone else are only relevant if they
|
# but artists that are counted as someone else are only relevant if they
|
||||||
@ -239,6 +236,6 @@ def flatten(lis):
|
|||||||
if isinstance(l, str):
|
if isinstance(l, str):
|
||||||
newlist.append(l)
|
newlist.append(l)
|
||||||
else:
|
else:
|
||||||
newlist = newlist + l
|
newlist += l
|
||||||
|
|
||||||
return list(set(newlist))
|
return list(set(newlist))
|
||||||
|
@ -149,10 +149,9 @@ def createScrobble(artists,title,time,album=None,duration=None,volatile=False):
|
|||||||
i = getTrackID(artists,title)
|
i = getTrackID(artists,title)
|
||||||
|
|
||||||
# idempotence
|
# idempotence
|
||||||
if time in SCROBBLESDICT:
|
if time in SCROBBLESDICT and i == SCROBBLESDICT[time].track:
|
||||||
if i == SCROBBLESDICT[time].track:
|
dblock.release()
|
||||||
dblock.release()
|
return get_track_dict(TRACKS[i])
|
||||||
return get_track_dict(TRACKS[i])
|
|
||||||
# timestamp as unique identifier
|
# timestamp as unique identifier
|
||||||
while (time in SCROBBLESDICT):
|
while (time in SCROBBLESDICT):
|
||||||
time += 1
|
time += 1
|
||||||
@ -192,35 +191,31 @@ def getArtistID(name):
|
|||||||
if obj_normalized in ARTISTS_NORMALIZED_SET:
|
if obj_normalized in ARTISTS_NORMALIZED_SET:
|
||||||
return ARTISTS_NORMALIZED.index(obj_normalized)
|
return ARTISTS_NORMALIZED.index(obj_normalized)
|
||||||
|
|
||||||
else:
|
i = len(ARTISTS)
|
||||||
i = len(ARTISTS)
|
ARTISTS.append(obj)
|
||||||
ARTISTS.append(obj)
|
ARTISTS_NORMALIZED_SET.add(obj_normalized)
|
||||||
ARTISTS_NORMALIZED_SET.add(obj_normalized)
|
ARTISTS_NORMALIZED.append(obj_normalized)
|
||||||
ARTISTS_NORMALIZED.append(obj_normalized)
|
|
||||||
|
|
||||||
# with a new artist added, we might also get new artists that they are credited as
|
# with a new artist added, we might also get new artists that they are credited as
|
||||||
cr = coa.getCredited(name)
|
cr = coa.getCredited(name)
|
||||||
getArtistID(cr)
|
getArtistID(cr)
|
||||||
|
|
||||||
coa.updateIDs(ARTISTS)
|
coa.updateIDs(ARTISTS)
|
||||||
|
|
||||||
return i
|
return i
|
||||||
|
|
||||||
def getTrackID(artists,title):
|
def getTrackID(artists,title):
|
||||||
artistset = set()
|
artistset = {getArtistID(name=a) for a in artists}
|
||||||
for a in artists:
|
|
||||||
artistset.add(getArtistID(name=a))
|
|
||||||
obj = Track(artists=frozenset(artistset),title=title)
|
obj = Track(artists=frozenset(artistset),title=title)
|
||||||
obj_normalized = Track(artists=frozenset(artistset),title=normalize_name(title))
|
obj_normalized = Track(artists=frozenset(artistset),title=normalize_name(title))
|
||||||
|
|
||||||
if obj_normalized in TRACKS_NORMALIZED_SET:
|
if obj_normalized in TRACKS_NORMALIZED_SET:
|
||||||
return TRACKS_NORMALIZED.index(obj_normalized)
|
return TRACKS_NORMALIZED.index(obj_normalized)
|
||||||
else:
|
i = len(TRACKS)
|
||||||
i = len(TRACKS)
|
TRACKS.append(obj)
|
||||||
TRACKS.append(obj)
|
TRACKS_NORMALIZED_SET.add(obj_normalized)
|
||||||
TRACKS_NORMALIZED_SET.add(obj_normalized)
|
TRACKS_NORMALIZED.append(obj_normalized)
|
||||||
TRACKS_NORMALIZED.append(obj_normalized)
|
return i
|
||||||
return i
|
|
||||||
|
|
||||||
import unicodedata
|
import unicodedata
|
||||||
|
|
||||||
@ -330,11 +325,7 @@ def get_scrobbles_num(**keys):
|
|||||||
|
|
||||||
def get_tracks(artist=None):
|
def get_tracks(artist=None):
|
||||||
|
|
||||||
if artist is not None:
|
artistid = ARTISTS.index(artist) if artist is not None else None
|
||||||
artistid = ARTISTS.index(artist)
|
|
||||||
else:
|
|
||||||
artistid = None
|
|
||||||
|
|
||||||
# Option 1
|
# Option 1
|
||||||
return [get_track_dict(t) for t in TRACKS if (artistid in t.artists) or (artistid==None)]
|
return [get_track_dict(t) for t in TRACKS if (artistid in t.artists) or (artistid==None)]
|
||||||
|
|
||||||
@ -639,7 +630,7 @@ def check_issues():
|
|||||||
duplicates.append((a,ar))
|
duplicates.append((a,ar))
|
||||||
|
|
||||||
st = st.replace("&","").replace("and","").replace("with","").strip()
|
st = st.replace("&","").replace("and","").replace("with","").strip()
|
||||||
if st != "" and st != a:
|
if st not in ["", a]:
|
||||||
if len(st) < 5 and len(lis) == 1:
|
if len(st) < 5 and len(lis) == 1:
|
||||||
#check if we havent just randomly found the string in another word
|
#check if we havent just randomly found the string in another word
|
||||||
#if (" " + st + " ") in lis[0] or (lis[0].endswith(" " + st)) or (lis[0].startswith(st + " ")):
|
#if (" " + st + " ") in lis[0] or (lis[0].endswith(" " + st)) or (lis[0].startswith(st + " ")):
|
||||||
@ -694,14 +685,9 @@ def get_predefined_rulesets():
|
|||||||
if f.endswith(".tsv"):
|
if f.endswith(".tsv"):
|
||||||
|
|
||||||
rawf = f.replace(".tsv","")
|
rawf = f.replace(".tsv","")
|
||||||
valid = True
|
valid = all(char in validchars for char in rawf)
|
||||||
for char in rawf:
|
|
||||||
if char not in validchars:
|
|
||||||
valid = False
|
|
||||||
break # don't even show up invalid filenames
|
|
||||||
|
|
||||||
if not valid: continue
|
if not valid: continue
|
||||||
if not "_" in rawf: continue
|
if "_" not in rawf: continue
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with open(data_dir['rules']("predefined",f)) as tsvfile:
|
with open(data_dir['rules']("predefined",f)) as tsvfile:
|
||||||
@ -711,21 +697,14 @@ def get_predefined_rulesets():
|
|||||||
if "# NAME: " in line1:
|
if "# NAME: " in line1:
|
||||||
name = line1.replace("# NAME: ","")
|
name = line1.replace("# NAME: ","")
|
||||||
else: name = rawf.split("_")[1]
|
else: name = rawf.split("_")[1]
|
||||||
if "# DESC: " in line2:
|
desc = line2.replace("# DESC: ","") if "# DESC: " in line2 else ""
|
||||||
desc = line2.replace("# DESC: ","")
|
|
||||||
else: desc = ""
|
|
||||||
|
|
||||||
author = rawf.split("_")[0]
|
author = rawf.split("_")[0]
|
||||||
except:
|
except:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
ruleset = {"file":rawf}
|
ruleset = {"file":rawf}
|
||||||
rulesets.append(ruleset)
|
rulesets.append(ruleset)
|
||||||
if os.path.exists(data_dir['rules'](f)):
|
ruleset["active"] = bool(os.path.exists(data_dir['rules'](f)))
|
||||||
ruleset["active"] = True
|
|
||||||
else:
|
|
||||||
ruleset["active"] = False
|
|
||||||
|
|
||||||
ruleset["name"] = name
|
ruleset["name"] = name
|
||||||
ruleset["author"] = author
|
ruleset["author"] = author
|
||||||
ruleset["desc"] = desc
|
ruleset["desc"] = desc
|
||||||
@ -805,7 +784,7 @@ def build_db():
|
|||||||
STAMPS.sort()
|
STAMPS.sort()
|
||||||
|
|
||||||
# inform malojatime module about earliest scrobble
|
# inform malojatime module about earliest scrobble
|
||||||
if len(STAMPS) > 0: register_scrobbletime(STAMPS[0])
|
if STAMPS: register_scrobbletime(STAMPS[0])
|
||||||
|
|
||||||
# NOT NEEDED BECAUSE WE DO THAT ON ADDING EVERY ARTIST ANYWAY
|
# NOT NEEDED BECAUSE WE DO THAT ON ADDING EVERY ARTIST ANYWAY
|
||||||
# get extra artists with no real scrobbles from countas rules
|
# get extra artists with no real scrobbles from countas rules
|
||||||
@ -1155,20 +1134,13 @@ def db_aggregate_full(by=None,since=None,to=None,within=None,timerange=None,arti
|
|||||||
|
|
||||||
# Search for strings
|
# Search for strings
|
||||||
def db_search(query,type=None):
|
def db_search(query,type=None):
|
||||||
|
results = []
|
||||||
if type=="ARTIST":
|
if type=="ARTIST":
|
||||||
results = []
|
results = [a for a in ARTISTS if simplestr(query) in simplestr(a)]
|
||||||
for a in ARTISTS:
|
|
||||||
#if query.lower() in a.lower():
|
|
||||||
if simplestr(query) in simplestr(a):
|
|
||||||
results.append(a)
|
|
||||||
|
|
||||||
if type=="TRACK":
|
if type=="TRACK":
|
||||||
results = []
|
results = [
|
||||||
for t in TRACKS:
|
get_track_dict(t) for t in TRACKS if simplestr(query) in simplestr(t[1])
|
||||||
#if query.lower() in t[1].lower():
|
]
|
||||||
if simplestr(query) in simplestr(t[1]):
|
|
||||||
results.append(get_track_dict(t))
|
|
||||||
|
|
||||||
return results
|
return results
|
||||||
|
|
||||||
|
|
||||||
@ -1227,7 +1199,7 @@ def scrobbles_in_range(start,end,reverse=False):
|
|||||||
# for performance testing
|
# for performance testing
|
||||||
def generateStuff(num=0,pertrack=0,mult=0):
|
def generateStuff(num=0,pertrack=0,mult=0):
|
||||||
import random
|
import random
|
||||||
for i in range(num):
|
for _ in range(num):
|
||||||
track = random.choice(TRACKS)
|
track = random.choice(TRACKS)
|
||||||
t = get_track_dict(track)
|
t = get_track_dict(track)
|
||||||
time = random.randint(STAMPS[0],STAMPS[-1])
|
time = random.randint(STAMPS[0],STAMPS[-1])
|
||||||
@ -1235,7 +1207,7 @@ def generateStuff(num=0,pertrack=0,mult=0):
|
|||||||
|
|
||||||
for track in TRACKS:
|
for track in TRACKS:
|
||||||
t = get_track_dict(track)
|
t = get_track_dict(track)
|
||||||
for i in range(pertrack):
|
for _ in range(pertrack):
|
||||||
time = random.randint(STAMPS[0],STAMPS[-1])
|
time = random.randint(STAMPS[0],STAMPS[-1])
|
||||||
createScrobble(t["artists"],t["title"],time,volatile=True)
|
createScrobble(t["artists"],t["title"],time,volatile=True)
|
||||||
|
|
||||||
|
@ -42,12 +42,11 @@ def find_representative(sequence,attribute_id,attribute_count):
|
|||||||
|
|
||||||
|
|
||||||
def combine_dicts(dictlist):
|
def combine_dicts(dictlist):
|
||||||
res = {k:d[k] for d in dictlist for k in d}
|
return {k:d[k] for d in dictlist for k in d}
|
||||||
return res
|
|
||||||
|
|
||||||
|
|
||||||
def compare_key_in_dicts(key,d1,d2):
|
def compare_key_in_dicts(key,d1,d2):
|
||||||
return d1[key] == d2[key]
|
return d1[key] == d2[key]
|
||||||
|
|
||||||
def alltrue(seq):
|
def alltrue(seq):
|
||||||
return all(s for s in seq)
|
return all(seq)
|
||||||
|
@ -77,7 +77,7 @@ class MRangeDescriptor:
|
|||||||
class MTime(MRangeDescriptor):
|
class MTime(MRangeDescriptor):
|
||||||
def __init__(self,*ls):
|
def __init__(self,*ls):
|
||||||
# in case we want to call with non-unpacked arguments
|
# in case we want to call with non-unpacked arguments
|
||||||
if isinstance(ls[0],tuple) or isinstance(ls[0],list):
|
if isinstance(ls[0], (tuple, list)):
|
||||||
ls = ls[0]
|
ls = ls[0]
|
||||||
|
|
||||||
self.tup = tuple(ls)
|
self.tup = tuple(ls)
|
||||||
@ -104,9 +104,7 @@ class MTime(MRangeDescriptor):
|
|||||||
if tod.year == self.year:
|
if tod.year == self.year:
|
||||||
if tod.month > self.month: return False
|
if tod.month > self.month: return False
|
||||||
if self.precision == 2: return True
|
if self.precision == 2: return True
|
||||||
if tod.month == self.month:
|
if tod.month == self.month and tod.day > self.day: return False
|
||||||
if tod.day > self.day: return False
|
|
||||||
|
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
@ -144,21 +142,21 @@ class MTime(MRangeDescriptor):
|
|||||||
|
|
||||||
# describes only the parts that are different than another range object
|
# describes only the parts that are different than another range object
|
||||||
def contextual_desc(self,other):
|
def contextual_desc(self,other):
|
||||||
if isinstance(other,MTime):
|
if not isinstance(other, MTime):
|
||||||
relevant = self.desc().split(" ")
|
return self.desc()
|
||||||
if self.year == other.year:
|
relevant = self.desc().split(" ")
|
||||||
|
if self.year == other.year:
|
||||||
|
relevant.pop()
|
||||||
|
if self.precision > 1 and other.precision > 1 and self.month == other.month:
|
||||||
relevant.pop()
|
relevant.pop()
|
||||||
if self.precision > 1 and other.precision > 1 and self.month == other.month:
|
if self.precision > 2 and other.precision > 2 and self.day == other.day:
|
||||||
relevant.pop()
|
relevant.pop()
|
||||||
if self.precision > 2 and other.precision > 2 and self.day == other.day:
|
return " ".join(relevant)
|
||||||
relevant.pop()
|
|
||||||
return " ".join(relevant)
|
|
||||||
return self.desc()
|
|
||||||
|
|
||||||
# gets object with one higher precision that starts this one
|
# gets object with one higher precision that starts this one
|
||||||
def start(self):
|
def start(self):
|
||||||
if self.precision == 1: return MTime(self.tup + (1,))
|
if self.precision in [1, 2]: return MTime(self.tup + (1,))
|
||||||
elif self.precision == 2: return MTime(self.tup + (1,))
|
|
||||||
# gets object with one higher precision that ends this one
|
# gets object with one higher precision that ends this one
|
||||||
def end(self):
|
def end(self):
|
||||||
if self.precision == 1: return MTime(self.tup + (12,))
|
if self.precision == 1: return MTime(self.tup + (12,))
|
||||||
@ -251,8 +249,7 @@ class MTimeWeek(MRangeDescriptor):
|
|||||||
return self.desc()
|
return self.desc()
|
||||||
|
|
||||||
def contextual_desc(self,other):
|
def contextual_desc(self,other):
|
||||||
if isinstance(other,MTimeWeek):
|
if isinstance(other, MTimeWeek) and other.year == self.year: return "Week " + str(self.week)
|
||||||
if other.year == self.year: return "Week " + str(self.week)
|
|
||||||
return self.desc()
|
return self.desc()
|
||||||
|
|
||||||
def start(self):
|
def start(self):
|
||||||
|
@ -132,7 +132,7 @@ def compose_querystring(*dicts,exclude=[]):
|
|||||||
for k in keys:
|
for k in keys:
|
||||||
if k in exclude: continue
|
if k in exclude: continue
|
||||||
values = keys.getall(k)
|
values = keys.getall(k)
|
||||||
st += "&".join([urllib.parse.urlencode({k:v},safe="/") for v in values])
|
st += "&".join(urllib.parse.urlencode({k:v},safe="/") for v in values)
|
||||||
st += "&"
|
st += "&"
|
||||||
return st[:-1] if st.endswith("&") else st # remove last ampersand
|
return st[:-1] if st.endswith("&") else st # remove last ampersand
|
||||||
|
|
||||||
|
@ -11,16 +11,14 @@ from . import tasks
|
|||||||
def getInstance():
|
def getInstance():
|
||||||
try:
|
try:
|
||||||
output = subprocess.check_output(["pidof","Maloja"])
|
output = subprocess.check_output(["pidof","Maloja"])
|
||||||
pid = int(output)
|
return int(output)
|
||||||
return pid
|
|
||||||
except:
|
except:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def getInstanceSupervisor():
|
def getInstanceSupervisor():
|
||||||
try:
|
try:
|
||||||
output = subprocess.check_output(["pidof","maloja_supervisor"])
|
output = subprocess.check_output(["pidof","maloja_supervisor"])
|
||||||
pid = int(output)
|
return int(output)
|
||||||
return pid
|
|
||||||
except:
|
except:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@ -60,12 +58,12 @@ def stop():
|
|||||||
if pid is not None:
|
if pid is not None:
|
||||||
os.kill(pid,signal.SIGTERM)
|
os.kill(pid,signal.SIGTERM)
|
||||||
|
|
||||||
if pid is not None or pid_sv is not None:
|
if pid is None and pid_sv is None:
|
||||||
print("Maloja stopped!")
|
|
||||||
return True
|
|
||||||
else:
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
print("Maloja stopped!")
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def direct():
|
def direct():
|
||||||
|
@ -26,10 +26,7 @@ def copy_initial_local_files():
|
|||||||
charset = list(range(10)) + list("abcdefghijklmnopqrstuvwxyz") + list("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
|
charset = list(range(10)) + list("abcdefghijklmnopqrstuvwxyz") + list("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
|
||||||
def randomstring(length=32):
|
def randomstring(length=32):
|
||||||
import random
|
import random
|
||||||
key = ""
|
return "".join(str(random.choice(charset)) for _ in range(length))
|
||||||
for i in range(length):
|
|
||||||
key += str(random.choice(charset))
|
|
||||||
return key
|
|
||||||
|
|
||||||
def setup():
|
def setup():
|
||||||
|
|
||||||
@ -50,18 +47,13 @@ def setup():
|
|||||||
|
|
||||||
|
|
||||||
# OWN API KEY
|
# OWN API KEY
|
||||||
if os.path.exists(data_dir['clients']("authenticated_machines.tsv")):
|
if not os.path.exists(data_dir['clients']("authenticated_machines.tsv")):
|
||||||
pass
|
|
||||||
else:
|
|
||||||
answer = ask("Do you want to set up a key to enable scrobbling? Your scrobble extension needs that key so that only you can scrobble tracks to your database.",default=True,skip=SKIP)
|
answer = ask("Do you want to set up a key to enable scrobbling? Your scrobble extension needs that key so that only you can scrobble tracks to your database.",default=True,skip=SKIP)
|
||||||
if answer:
|
if answer:
|
||||||
key = randomstring(64)
|
key = randomstring(64)
|
||||||
print("Your API Key: " + col["yellow"](key))
|
print("Your API Key: " + col["yellow"](key))
|
||||||
with open(data_dir['clients']("authenticated_machines.tsv"),"w") as keyfile:
|
with open(data_dir['clients']("authenticated_machines.tsv"),"w") as keyfile:
|
||||||
keyfile.write(key + "\t" + "Default Generated Key")
|
keyfile.write(key + "\t" + "Default Generated Key")
|
||||||
else:
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
# PASSWORD
|
# PASSWORD
|
||||||
defaultpassword = settings.get_settings("DEFAULT_PASSWORD")
|
defaultpassword = settings.get_settings("DEFAULT_PASSWORD")
|
||||||
@ -81,14 +73,11 @@ def setup():
|
|||||||
if newpw is None:
|
if newpw is None:
|
||||||
newpw = defaultpassword
|
newpw = defaultpassword
|
||||||
print("Generated password:",newpw)
|
print("Generated password:",newpw)
|
||||||
auth.defaultuser.setpw(newpw)
|
|
||||||
else:
|
else:
|
||||||
# docker installation (or settings file, but don't do that)
|
# docker installation (or settings file, but don't do that)
|
||||||
# we still 'ask' the user to set one, but for docker this will be skipped
|
# we still 'ask' the user to set one, but for docker this will be skipped
|
||||||
newpw = prompt("Please set a password for web backend access. Leave this empty to use the default password.",skip=SKIP,default=defaultpassword,secret=True)
|
newpw = prompt("Please set a password for web backend access. Leave this empty to use the default password.",skip=SKIP,default=defaultpassword,secret=True)
|
||||||
auth.defaultuser.setpw(newpw)
|
auth.defaultuser.setpw(newpw)
|
||||||
|
|
||||||
|
|
||||||
if settings.get_settings("NAME") is None:
|
if settings.get_settings("NAME") is None:
|
||||||
name = prompt("Please enter your name. This will be displayed e.g. when comparing your charts to another user. Leave this empty if you would not like to specify a name right now.",default="Generic Maloja User",skip=SKIP)
|
name = prompt("Please enter your name. This will be displayed e.g. when comparing your charts to another user. Leave this empty if you would not like to specify a name right now.",default="Generic Maloja User",skip=SKIP)
|
||||||
settings.update_settings(data_dir['settings']("settings.ini"),{"NAME":name},create_new=True)
|
settings.update_settings(data_dir['settings']("settings.ini"),{"NAME":name},create_new=True)
|
||||||
|
@ -21,8 +21,11 @@ def update():
|
|||||||
|
|
||||||
def start():
|
def start():
|
||||||
try:
|
try:
|
||||||
p = subprocess.Popen(["python3","-m","maloja.server"],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
|
return subprocess.Popen(
|
||||||
return p
|
["python3", "-m", "maloja.server"],
|
||||||
|
stdout=subprocess.DEVNULL,
|
||||||
|
stderr=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
except e:
|
except e:
|
||||||
log("Error starting Maloja: " + str(e),module="supervisor")
|
log("Error starting Maloja: " + str(e),module="supervisor")
|
||||||
|
|
||||||
|
@ -39,8 +39,8 @@ def backup(folder,level="full"):
|
|||||||
archivefile = os.path.join(folder,filename)
|
archivefile = os.path.join(folder,filename)
|
||||||
assert not os.path.exists(archivefile)
|
assert not os.path.exists(archivefile)
|
||||||
with tarfile.open(name=archivefile,mode="x:gz") as archive:
|
with tarfile.open(name=archivefile,mode="x:gz") as archive:
|
||||||
for cat in real_files:
|
for cat, value in real_files.items():
|
||||||
for f in real_files[cat]:
|
for f in value:
|
||||||
p = PurePath(f)
|
p = PurePath(f)
|
||||||
r = p.relative_to(data_dir[cat]())
|
r = p.relative_to(data_dir[cat]())
|
||||||
archive.add(f,arcname=os.path.join(cat,r))
|
archive.add(f,arcname=os.path.join(cat,r))
|
||||||
|
@ -94,8 +94,7 @@ def clean_html(inp):
|
|||||||
@webserver.route("")
|
@webserver.route("")
|
||||||
@webserver.route("/")
|
@webserver.route("/")
|
||||||
def mainpage():
|
def mainpage():
|
||||||
response = static_html("start")
|
return static_html("start")
|
||||||
return response
|
|
||||||
|
|
||||||
@webserver.error(400)
|
@webserver.error(400)
|
||||||
@webserver.error(403)
|
@webserver.error(403)
|
||||||
@ -112,8 +111,12 @@ def customerror(error):
|
|||||||
adminmode = request.cookies.get("adminmode") == "true" and auth.check(request)
|
adminmode = request.cookies.get("adminmode") == "true" and auth.check(request)
|
||||||
|
|
||||||
template = jinja_environment.get_template('error.jinja')
|
template = jinja_environment.get_template('error.jinja')
|
||||||
res = template.render(errorcode=errorcode,errordesc=errordesc,traceback=traceback,adminmode=adminmode)
|
return template.render(
|
||||||
return res
|
errorcode=errorcode,
|
||||||
|
errordesc=errordesc,
|
||||||
|
traceback=traceback,
|
||||||
|
adminmode=adminmode,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
13
maloja/thirdparty/__init__.py
vendored
13
maloja/thirdparty/__init__.py
vendored
@ -177,17 +177,14 @@ class MetadataInterface(GenericInterface,abstract=True):
|
|||||||
# default function to parse response by descending down nodes
|
# default function to parse response by descending down nodes
|
||||||
# override if more complicated
|
# override if more complicated
|
||||||
def metadata_parse_response_artist(self,data):
|
def metadata_parse_response_artist(self,data):
|
||||||
res = data
|
return self._parse_response("response_parse_tree_artist", data)
|
||||||
for node in self.metadata["response_parse_tree_artist"]:
|
|
||||||
try:
|
|
||||||
res = res[node]
|
|
||||||
except:
|
|
||||||
return None
|
|
||||||
return res
|
|
||||||
|
|
||||||
def metadata_parse_response_track(self,data):
|
def metadata_parse_response_track(self,data):
|
||||||
|
return self._parse_response("response_parse_tree_track", data)
|
||||||
|
|
||||||
|
def _parse_response(self, resp, data):
|
||||||
res = data
|
res = data
|
||||||
for node in self.metadata["response_parse_tree_track"]:
|
for node in self.metadata[resp]:
|
||||||
try:
|
try:
|
||||||
res = res[node]
|
res = res[node]
|
||||||
except:
|
except:
|
||||||
|
3
maloja/thirdparty/musicbrainz.py
vendored
3
maloja/thirdparty/musicbrainz.py
vendored
@ -1,5 +1,4 @@
|
|||||||
from . import MetadataInterface, utf, b64
|
from . import MetadataInterface
|
||||||
import hashlib
|
|
||||||
import urllib.parse, urllib.request
|
import urllib.parse, urllib.request
|
||||||
import json
|
import json
|
||||||
import time
|
import time
|
||||||
|
1
maloja/thirdparty/spotify.py
vendored
1
maloja/thirdparty/spotify.py
vendored
@ -1,5 +1,4 @@
|
|||||||
from . import MetadataInterface, utf, b64
|
from . import MetadataInterface, utf, b64
|
||||||
import hashlib
|
|
||||||
import urllib.parse, urllib.request
|
import urllib.parse, urllib.request
|
||||||
import json
|
import json
|
||||||
from threading import Timer
|
from threading import Timer
|
||||||
|
@ -159,11 +159,10 @@ def getTrackImage(artists,title,fast=False):
|
|||||||
#result = cachedTracks[(frozenset(artists),title)]
|
#result = cachedTracks[(frozenset(artists),title)]
|
||||||
result = track_cache.get((frozenset(artists),title)) #track_from_cache(artists,title)
|
result = track_cache.get((frozenset(artists),title)) #track_from_cache(artists,title)
|
||||||
if result is not None: return result
|
if result is not None: return result
|
||||||
else:
|
for a in artists:
|
||||||
for a in artists:
|
res = getArtistImage(artist=a,fast=True)
|
||||||
res = getArtistImage(artist=a,fast=True)
|
if res != "": return res
|
||||||
if res != "": return res
|
return ""
|
||||||
return ""
|
|
||||||
except:
|
except:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@ -173,7 +172,9 @@ def getTrackImage(artists,title,fast=False):
|
|||||||
|
|
||||||
|
|
||||||
# fast request only retuns cached and local results, generates redirect link for rest
|
# fast request only retuns cached and local results, generates redirect link for rest
|
||||||
if fast: return "/image?title=" + urllib.parse.quote(title) + "&" + "&".join(["artist=" + urllib.parse.quote(a) for a in artists])
|
if fast:
|
||||||
|
return ("/image?title=" + urllib.parse.quote(title) + "&" + "&".join(
|
||||||
|
"artist=" + urllib.parse.quote(a) for a in artists))
|
||||||
|
|
||||||
# non-fast lookup (essentially only the resolver lookup)
|
# non-fast lookup (essentially only the resolver lookup)
|
||||||
result = thirdparty.get_image_track_all((artists,title))
|
result = thirdparty.get_image_track_all((artists,title))
|
||||||
@ -184,11 +185,10 @@ def getTrackImage(artists,title,fast=False):
|
|||||||
|
|
||||||
# return either result or redirect to artist
|
# return either result or redirect to artist
|
||||||
if result is not None: return result
|
if result is not None: return result
|
||||||
else:
|
for a in artists:
|
||||||
for a in artists:
|
res = getArtistImage(artist=a,fast=False)
|
||||||
res = getArtistImage(artist=a,fast=False)
|
if res != "": return res
|
||||||
if res != "": return res
|
return ""
|
||||||
return ""
|
|
||||||
|
|
||||||
|
|
||||||
def getArtistImage(artist,fast=False):
|
def getArtistImage(artist,fast=False):
|
||||||
|
@ -12,7 +12,7 @@ def serialize(obj):
|
|||||||
try:
|
try:
|
||||||
return json.dumps(obj)
|
return json.dumps(obj)
|
||||||
except:
|
except:
|
||||||
if isinstance(obj,list) or isinstance(obj,tuple):
|
if isinstance(obj, (list, tuple)):
|
||||||
return "[" + ",".join(serialize(o) for o in obj) + "]"
|
return "[" + ",".join(serialize(o) for o in obj) + "]"
|
||||||
elif isinstance(obj,dict):
|
elif isinstance(obj,dict):
|
||||||
return "{" + ",".join(serialize(o) + ":" + serialize(obj[o]) for o in obj) + "}"
|
return "{" + ",".join(serialize(o) + ":" + serialize(obj[o]) for o in obj) + "}"
|
||||||
|
Loading…
x
Reference in New Issue
Block a user