mirror of
https://github.com/krateng/maloja.git
synced 2025-04-16 00:40:32 +03:00
Doreah no longer included, now a dependency
This commit is contained in:
Krateng
parent
6f9ea32d08
commit
6086a2df23
@ -23,6 +23,7 @@ There are only two scrobblers (YouTube Music and Plex, both for Chromium), but a
|
||||
* [python3](https://www.python.org/) - [GitHub](https://github.com/python/cpython)
|
||||
* [bottle.py](https://bottlepy.org/) - [GitHub](https://github.com/bottlepy/bottle)
|
||||
* [waitress](https://docs.pylonsproject.org/projects/waitress/) - [GitHub](https://github.com/Pylons/waitress)
|
||||
* [doreah](https://pypi.org/project/doreah/) - [GitHub](https://github.com/krateng/doreah)
|
||||
|
||||
## How to install
|
||||
|
||||
|
||||
@ -1,38 +0,0 @@
|
||||
import os
|
||||
|
||||
|
||||
## decorator to set default arguments that are only evaluated at runtime
|
||||
def defaultarguments(defaultdict,**defaultargs):
|
||||
def decorator(func): #actual decorator function
|
||||
def newfunc(*args,**kwargs):
|
||||
realargs = {}
|
||||
|
||||
# evaluate default values at runtime
|
||||
for k in defaultargs:
|
||||
realargs[k] = defaultdict[defaultargs[k]]
|
||||
|
||||
# overwrite given arguments of function
|
||||
realargs.update(kwargs)
|
||||
|
||||
# execute function with correct arguments
|
||||
return func(*args,**realargs)
|
||||
|
||||
return newfunc
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
## opens file, creates it if necessary (including all folders)
|
||||
def gopen(filepath,mode):
|
||||
directory = os.path.dirname(filepath)
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
|
||||
return open(filepath,mode)
|
||||
|
||||
|
||||
|
||||
# reads module configuration from file
|
||||
def doreahconfig(module):
|
||||
from .settings import get_settings
|
||||
s = get_settings(files=[".doreah"],prefix=module + ".",cut_prefix=True)
|
||||
return s
|
||||
@ -1,109 +0,0 @@
|
||||
import datetime
|
||||
import inspect
|
||||
import os
|
||||
|
||||
from ._internal import defaultarguments, gopen, doreahconfig
|
||||
|
||||
_config = {}
|
||||
|
||||
_queue = []
|
||||
_locked = False
|
||||
|
||||
# set configuration
|
||||
# logfolder folder to store logfiles in
|
||||
# timeformat strftime format for log files
|
||||
# defaultmodule name for the main running script
|
||||
# verbosity higher means more (less important) messages are shown on console
|
||||
def config(logfolder="logs",timeformat="%Y/%m/%d %H:%M:%S",defaultmodule="main",verbosity=0):
|
||||
global _config
|
||||
_config["logfolder"] = logfolder
|
||||
_config["timeformat"] = timeformat
|
||||
_config["defaultmodule"] = defaultmodule
|
||||
_config["verbosity"] = verbosity
|
||||
|
||||
|
||||
# initial config on import, set everything to default
|
||||
config()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# Log entry
|
||||
# module allows discrimination between modules of a program. Will be prepended in console output and will determine the separate file for disk storage
|
||||
# defaults to actual name of the calling module or "main" for the main script
|
||||
# header determines the hierarchical position of the entry.
|
||||
# indent adds indent to the log entry
|
||||
# importance low means important. if higher than the configured verbosity, entry will not be shown on console
|
||||
def log(*msgs,module=None,header=None,indent=0,importance=0):
|
||||
|
||||
now = datetime.datetime.utcnow().strftime(_config["timeformat"])
|
||||
|
||||
# log() can be used to add empty line
|
||||
if len(msgs) == 0: msgs = ("",)
|
||||
|
||||
# make it easier to log data structures and such
|
||||
msgs = tuple([str(msg) for msg in msgs])
|
||||
|
||||
# header formating
|
||||
if header == 2:
|
||||
msgs = ("","","####") + msgs + ("####","")
|
||||
elif header == 1:
|
||||
msgs = ("","","","# # # # #","") + msgs + ("","# # # # #","","")
|
||||
|
||||
# indent
|
||||
prefix = "\t" * indent
|
||||
|
||||
# module name
|
||||
if module is None:
|
||||
try:
|
||||
module = inspect.getmodule(inspect.stack()[1][0]).__name__
|
||||
if module == "__main__": module = _config["defaultmodule"]
|
||||
except:
|
||||
module = "interpreter"
|
||||
|
||||
global _locked, _queue
|
||||
if _locked:
|
||||
for msg in msgs:
|
||||
_queue.append({"time":now,"prefix":prefix,"msg":msg,"module":module,"console":(importance <= _config["verbosity"])})
|
||||
else:
|
||||
# console output
|
||||
if (importance <= _config["verbosity"]):
|
||||
for msg in msgs:
|
||||
print("[" + module + "] " + prefix + msg)
|
||||
|
||||
# file output
|
||||
logfilename = _config["logfolder"] + "/" + module + ".log"
|
||||
#os.makedirs(os.path.dirname(logfilename), exist_ok=True)
|
||||
with gopen(logfilename,"a") as logfile:
|
||||
for msg in msgs:
|
||||
logfile.write(now + " " + prefix + msg + "\n")
|
||||
|
||||
|
||||
def flush():
|
||||
global _queue
|
||||
for entry in _queue:
|
||||
# console output
|
||||
if entry["console"]:
|
||||
print("[" + entry["module"] + "] " + entry["prefix"] + entry["msg"])
|
||||
|
||||
# file output
|
||||
logfilename = _config["logfolder"] + "/" + entry["module"] + ".log"
|
||||
#os.makedirs(os.path.dirname(logfilename), exist_ok=True)
|
||||
with gopen(logfilename,"a") as logfile:
|
||||
logfile.write(entry["time"] + " " + entry["prefix"] + entry["msg"] + "\n")
|
||||
|
||||
_queue = []
|
||||
|
||||
# Quicker way to add header
|
||||
def logh1(*args,**kwargs):
|
||||
return log(*args,**kwargs,header=1)
|
||||
def logh2(*args,**kwargs):
|
||||
return log(*args,**kwargs,header=2)
|
||||
|
||||
|
||||
|
||||
|
||||
# now check local configuration file
|
||||
_config.update(doreahconfig("logging"))
|
||||
@ -1,46 +0,0 @@
|
||||
import pickle
|
||||
import os
|
||||
|
||||
from ._internal import defaultarguments, gopen, doreahconfig
|
||||
|
||||
_config = {}
|
||||
|
||||
# set configuration
|
||||
# folder folder to store log files
|
||||
def config(folder="storage"):
|
||||
global _config
|
||||
_config["folder"] = folder
|
||||
|
||||
|
||||
# initial config on import, set everything to default
|
||||
config()
|
||||
|
||||
|
||||
@defaultarguments(_config,folder="folder")
|
||||
def save(data,name,folder):
|
||||
|
||||
filename = os.path.join(folder,name + ".gilly")
|
||||
|
||||
fl = gopen(filename,"wb")
|
||||
stream = pickle.dumps(data)
|
||||
fl.write(stream)
|
||||
fl.close()
|
||||
|
||||
@defaultarguments(_config,folder="folder")
|
||||
def load(name,folder):
|
||||
|
||||
filename = os.path.join(folder,name + ".gilly")
|
||||
|
||||
try:
|
||||
fl = gopen(filename,"rb")
|
||||
ob = pickle.loads(fl.read())
|
||||
except: ob = None
|
||||
finally:
|
||||
fl.close()
|
||||
|
||||
return ob
|
||||
|
||||
|
||||
|
||||
# now check local configuration file
|
||||
_config.update(doreahconfig("persistence"))
|
||||
@ -1,225 +0,0 @@
|
||||
import os
|
||||
import shutil
|
||||
|
||||
from ._internal import defaultarguments, doreahconfig
|
||||
|
||||
|
||||
_config = {}
|
||||
|
||||
# set configuration
|
||||
# defaultextension unused
|
||||
# files list of all files that will be used for configuration. high indicies overwrite low indicies
|
||||
# comment symbols that indicate comments. careful!
|
||||
# category symbols that indicate start and end of category name. only works at start of line
|
||||
# onlytext interpret everything as a string. if False, strings can be put into quotes to avoid confusion
|
||||
def config(defaultextension=".ini",files=["settings.ini","settings.conf","configuration.ini","configuration.conf"],
|
||||
comment=["#","//"],category=("[","]"),onlytext=False):
|
||||
|
||||
global _config
|
||||
_config["defaultextension"] = defaultextension
|
||||
_config["files"] = files
|
||||
_config["comment"] = comment
|
||||
_config["category"] = category
|
||||
_config["onlytext"] = onlytext
|
||||
|
||||
# initial config on import, set everything to default
|
||||
config()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def _interpret(text):
|
||||
if _config["onlytext"]: return text
|
||||
|
||||
if text.lower() in ["true","yes"]: return True
|
||||
if text.lower() in ["false","no"]: return False
|
||||
if text.lower() in ["none","nan","n/a",""]: return None
|
||||
if text.startswith("[") and text.endswith("]"):
|
||||
list = []
|
||||
buffer = ""
|
||||
string = None
|
||||
stringended = False
|
||||
for c in text[1:-1]:
|
||||
if stringended and c != ",": pass #after a string is done, skip to the next delimiter
|
||||
elif c == '"' and string is None:
|
||||
string = '"' # start new string
|
||||
buffer = '"'
|
||||
elif c == "'" and string is None:
|
||||
string = "'" # start new string
|
||||
buffer = "'"
|
||||
elif c == '"' and string is '"':
|
||||
string = None # terminate string
|
||||
stringended = True
|
||||
buffer += '"'
|
||||
elif c == "'" and string is "'":
|
||||
string = None # terminate string
|
||||
stringended = True
|
||||
buffer += "'"
|
||||
elif c == "," and string is None:
|
||||
list.append(buffer)
|
||||
buffer = ""
|
||||
stringended = False
|
||||
else: buffer += c
|
||||
|
||||
list.append(buffer.strip())
|
||||
return [_interpret(entry) for entry in list]
|
||||
|
||||
if text.startswith("'") and text.endswith("'"): return text[1:-1]
|
||||
if text.startswith('"') and text.endswith('"'): return text[1:-1]
|
||||
try:
|
||||
return int(text)
|
||||
except:
|
||||
pass
|
||||
try:
|
||||
return float(text)
|
||||
except:
|
||||
pass
|
||||
return text
|
||||
|
||||
|
||||
# get settings
|
||||
# keys list of requested keys. if present, will return list of according values, if not, will return dict of key-value pairs
|
||||
# files which files to parse. later files (higher indicies) will overload earlier ones
|
||||
# prefix only request keys with a certain prefix. key filter will still be applied if present
|
||||
# cut_prefix return keys without the prefix
|
||||
# category return only keys of specific category
|
||||
# raw do not interpret data type, only return strings
|
||||
@defaultarguments(_config,files="files")
|
||||
def get_settings(*keys,files,prefix="",cut_prefix=False,category=None,raw=False):
|
||||
|
||||
allsettings = {}
|
||||
|
||||
for f in files:
|
||||
if not os.path.exists(f): continue
|
||||
|
||||
# if category is specified, ignore all entries by default and switch when we encounter the right heading
|
||||
ignore = False if category is None else True
|
||||
|
||||
with open(f) as file:
|
||||
lines = file.readlines()
|
||||
|
||||
|
||||
for l in lines:
|
||||
# clean up line
|
||||
l = l.replace("\n","")
|
||||
for symbol in _config["comment"]:
|
||||
l = l.split(symbol)[0]
|
||||
l = l.strip()
|
||||
|
||||
# check if valid settings entry
|
||||
if l == "": continue
|
||||
if l.startswith(_config["category"][0]):
|
||||
# ignore category headers if we don't care
|
||||
if category is None: continue
|
||||
# otherwise, find out if this is the category we want
|
||||
else:
|
||||
if l.endswith(_config["category"][1]):
|
||||
cat = l[len(_config["category"][0]):-len(_config["category"][1])]
|
||||
ignore = not (cat == category) #if this is the right heading, set ignore to false
|
||||
continue
|
||||
|
||||
if ignore: continue
|
||||
|
||||
if "=" not in l: continue
|
||||
|
||||
# read
|
||||
(key,val) = l.split("=")
|
||||
key = key.strip()
|
||||
val = val.strip() if raw else _interpret(val.strip())
|
||||
|
||||
if key.startswith(prefix):
|
||||
# return keys without the common prefix
|
||||
if cut_prefix:
|
||||
allsettings[key[len(prefix):]] = val
|
||||
|
||||
# return full keys
|
||||
else:
|
||||
allsettings[key] = val
|
||||
|
||||
# no arguments means all settings
|
||||
if len(keys) == 0:
|
||||
return allsettings
|
||||
|
||||
# specific keys requested
|
||||
else:
|
||||
if len(keys) == 1: return allsettings.get(keys[0])
|
||||
else: return [allsettings.get(k) for k in keys]
|
||||
|
||||
|
||||
def update_settings(file,settings,create_new=False):
|
||||
|
||||
if not os.path.exists(file): open(file,"w").close()
|
||||
|
||||
with open(file,"r") as origfile:
|
||||
lines = origfile.readlines()
|
||||
|
||||
newlines = []
|
||||
|
||||
for origline in lines:
|
||||
l = origline
|
||||
# clean up line
|
||||
l = l.replace("\n","")
|
||||
for symbol in _config["comment"]:
|
||||
l = l.split(symbol)[0]
|
||||
l = l.strip()
|
||||
|
||||
# check if valid settings entry
|
||||
if l == "":
|
||||
newlines.append(origline)
|
||||
continue
|
||||
if l.startswith(_config["category"][0]):
|
||||
newlines.append(origline)
|
||||
continue
|
||||
if "=" not in l:
|
||||
newlines.append(origline)
|
||||
continue
|
||||
|
||||
# read
|
||||
(key,val) = l.split("=")
|
||||
key = key.strip()
|
||||
val = val.strip()
|
||||
|
||||
|
||||
if key not in settings:
|
||||
newlines.append(origline)
|
||||
continue
|
||||
|
||||
else:
|
||||
#print("Found key")
|
||||
newline = origline.split("=",1)
|
||||
#print({"linepart":newline[1],"keytoreplace":val,"new":settings[key]})
|
||||
newline[1] = newline[1].replace(val,str(settings[key]),1)
|
||||
newline = "=".join(newline)
|
||||
newlines.append(newline)
|
||||
|
||||
del settings[key]
|
||||
|
||||
if create_new:
|
||||
# settings that were not present in the file
|
||||
for key in settings:
|
||||
newlines.append(key + " = " + settings[key] + "\n")
|
||||
|
||||
with open(file,"w") as newfile:
|
||||
newfile.write("".join(newlines))
|
||||
|
||||
|
||||
|
||||
|
||||
# updates a user settings file to a newer format from a default file without overwriting user's settings
|
||||
def update(source="default_settings.ini",target="settings.ini"):
|
||||
|
||||
if not os.path.exists(target):
|
||||
shutil.copyfile(source,target)
|
||||
else:
|
||||
usersettings = get_settings(files=[target],raw=True)
|
||||
shutil.copyfile(source,target)
|
||||
update_settings(target,usersettings)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# now check local configuration file
|
||||
_config.update(doreahconfig("settings"))
|
||||
@ -1,52 +0,0 @@
|
||||
import time
|
||||
|
||||
from ._internal import defaultarguments, doreahconfig
|
||||
|
||||
_config = {}
|
||||
|
||||
|
||||
# set configuration
|
||||
# si 0 means seconds, 1 ms, 2 μs, 3 ns etc
|
||||
def config(si=0):
|
||||
global _config
|
||||
_config["si"] = si
|
||||
|
||||
|
||||
# initial config on import, set everything to default
|
||||
config()
|
||||
|
||||
|
||||
# Take clock. Returns time passed since last call of this function. if called with an identifier, will only
|
||||
# consider calls with that identifier. No identifier means any call is valid.
|
||||
# identifiers arbitrary strings to remember different timers. guaranteed to set all timers to exactly the same time for
|
||||
# all identifiers in one call. will return tuple of all identifiers, singular value if only one identifier
|
||||
def clock(*identifiers,lastcalls={None:None}):
|
||||
|
||||
if len(identifiers) == 0: identifiers = (None,)
|
||||
|
||||
now = time.time()
|
||||
# get last calls
|
||||
stamps = (lastcalls.get(i) for i in identifiers)
|
||||
results = tuple(None if lc is None else (now - lc) * (1000**_config["si"]) for lc in stamps)
|
||||
if len(results) == 1: results = results[0]
|
||||
|
||||
# set new stamps
|
||||
for i in identifiers:
|
||||
lastcalls[i] = now
|
||||
lastcalls[None] = now # always save last overall call so we can directly access it
|
||||
|
||||
return results
|
||||
|
||||
|
||||
|
||||
def clockp(name,*identifiers):
|
||||
time = clock(*identifiers)
|
||||
print(name + ": " + str(time))
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# now check local configuration file
|
||||
_config.update(doreahconfig("timing"))
|
||||
120
doreah/tsv.py
120
doreah/tsv.py
@ -1,120 +0,0 @@
|
||||
import os
|
||||
|
||||
from ._internal import defaultarguments, doreahconfig
|
||||
|
||||
_config = {}
|
||||
|
||||
# set configuration
|
||||
# defaultextension files with this extension will be regarded as valid files. can be overwritten per request.
|
||||
# comments whether files may include commenting (indicated by #)
|
||||
# multitab whether fields can be separated by multiple tabs (this makes empty fields impossible except when trailing)
|
||||
def config(defaultextension=".tsv",comments=True,multitab=True):
|
||||
global _config
|
||||
_config["defaultextension"] = defaultextension
|
||||
_config["comments"] = comments
|
||||
_config["multitab"] = multitab
|
||||
|
||||
|
||||
# initial config on import, set everything to default
|
||||
config()
|
||||
|
||||
|
||||
@defaultarguments(_config,comments="comments",multitab="multitab")
|
||||
def parse(filename,*args,comments,multitab):
|
||||
|
||||
if not os.path.exists(filename):
|
||||
filename = filename + _config["defaultextension"]
|
||||
|
||||
f = open(filename)
|
||||
|
||||
result = []
|
||||
for l in [l for l in f if (not l.startswith("#")) and (not l.strip()=="")]:
|
||||
l = l.replace("\n","")
|
||||
|
||||
# if the file allows comments, we need to replace the escape sequence and properly stop parsing for inline comments
|
||||
if comments:
|
||||
l = l.split("#")[0]
|
||||
l = l.replace(r"\num","#")
|
||||
l = l.replace(r"\hashtag","#")
|
||||
|
||||
# we either allow multiple tabs, or we don't (in which case empty fields are possible)
|
||||
if multitab:
|
||||
data = list(filter(None,l.split("\t")))
|
||||
else:
|
||||
data = list(l.split("\t"))
|
||||
|
||||
entry = [] * len(args)
|
||||
for i in range(len(args)):
|
||||
if args[i] in ["list","ls","array"]:
|
||||
try:
|
||||
entry.append(data[i].split("␟"))
|
||||
except:
|
||||
entry.append([])
|
||||
elif args[i] in ["string","str","text"]:
|
||||
try:
|
||||
entry.append(data[i])
|
||||
except:
|
||||
entry.append("")
|
||||
elif args[i] in ["int","integer","num","number"]:
|
||||
try:
|
||||
entry.append(int(data[i]))
|
||||
except:
|
||||
entry.append(0)
|
||||
elif args[i] in ["bool","boolean"]:
|
||||
try:
|
||||
entry.append((data[i].lower() in ["true","yes","1","y"]))
|
||||
except:
|
||||
entry.append(False)
|
||||
else:
|
||||
raise TypeError()
|
||||
|
||||
result.append(entry)
|
||||
|
||||
f.close()
|
||||
return result
|
||||
|
||||
@defaultarguments(_config,extension="defaultextension")
|
||||
def parse_all(path,*args,extension,**kwargs):
|
||||
|
||||
result = []
|
||||
for f in os.listdir(path + "/"):
|
||||
if (f.endswith(extension)): # use "" if all files are valid
|
||||
result += parse(path + "/" + f,*args,**kwargs)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
|
||||
def create(filename):
|
||||
|
||||
if not os.path.exists(filename):
|
||||
open(filename,"w").close()
|
||||
|
||||
@defaultarguments(_config,comments="comments")
|
||||
def add_entry(filename,a,comments):
|
||||
|
||||
create(filename)
|
||||
# remove all tabs and create tab-separated string
|
||||
line = "\t".join([str(e).replace("\t"," ") for e in a])
|
||||
|
||||
# replace comment symbol
|
||||
if comments: line = line.replace("#",r"\num")
|
||||
|
||||
with open(filename,"a") as f:
|
||||
f.write(line + "\n")
|
||||
|
||||
@defaultarguments(_config,comments="comments")
|
||||
def add_entries(filename,al,comments):
|
||||
|
||||
create(filename)
|
||||
|
||||
with open(filename,"a") as f:
|
||||
for a in al:
|
||||
line = "\t".join([str(e).replace("\t"," ") for e in a])
|
||||
if comments: line = line.replace("#",r"\num")
|
||||
f.write(line + "\n")
|
||||
|
||||
|
||||
|
||||
# now check local configuration file
|
||||
_config.update(doreahconfig("tsv"))
|
||||
Loading…
x
Reference in New Issue
Block a user