274 lines
9.1 KiB
Python
274 lines
9.1 KiB
Python
import os
|
|
import yaml
|
|
import shutil
|
|
|
|
# Defaults
|
|
|
|
CONTEXT_FILE = '.context.yaml'
|
|
CACHE_FILE = '.index.yaml'
|
|
|
|
class OrvaniTag:
|
|
|
|
INDEX = {} # our search index
|
|
CACHEFILE = '' # search index cache ('.index.yaml')
|
|
DB = dict() # holds the folder context data
|
|
FILENAME = '' # folder context data filename ('.context.yaml')
|
|
FILE = '' # holds the folder context file path
|
|
CTX = { # default contexts
|
|
'contexts' : '@',
|
|
'projects' : '+',
|
|
'tags' : '#'
|
|
}
|
|
|
|
def __init__(self, path: str=CONTEXT_FILE, cache: str=CACHE_FILE, contexts=CTX):
|
|
"""
|
|
Initialization
|
|
|
|
path (str): override default filename for folder context
|
|
cache (str): override default filename for cached index
|
|
contexts (dict): override the default contexts
|
|
|
|
"""
|
|
self.FILENAME = os.path.basename(path)
|
|
self.CACHEFILE = cache
|
|
self.INDEX = self.CTX.copy()
|
|
for ctx in self.INDEX:
|
|
self.INDEX[ctx] = {}
|
|
self.INDEX['files'] = {}
|
|
self.__cd('')
|
|
|
|
def __cd(self, path: str):
|
|
"""
|
|
Change folder context to other directory.
|
|
Checks if a folder context file exists and loads context DB.
|
|
|
|
path (str): path of the new folder context
|
|
|
|
"""
|
|
self.FILE = os.path.join(path, self.FILENAME)
|
|
if os.path.isfile(self.FILE):
|
|
try:
|
|
with open(self.FILE, 'r') as data:
|
|
self.DB = yaml.safe_load(data)
|
|
except Exception:
|
|
print('File does not exist')
|
|
else:
|
|
self.DB = dict()
|
|
|
|
def __save(self):
|
|
"""
|
|
Save current folder context to file
|
|
|
|
"""
|
|
try:
|
|
with open(self.FILE, 'w+') as data:
|
|
yaml.dump(self.DB, data)
|
|
except Exception:
|
|
print('File write error')
|
|
|
|
def __context(self, argv) -> dict:
|
|
"""
|
|
Creates a context entry for an existing file
|
|
|
|
argv (list): sets file contexts according to CTX schema
|
|
|
|
"""
|
|
context = dict()
|
|
for ctx in self.CTX.keys():
|
|
context.update( { ctx : []})
|
|
for arg in argv:
|
|
for ctx in self.CTX.keys():
|
|
if str(arg)[0] in self.CTX.values():
|
|
if str(arg).startswith(self.CTX[ctx]):
|
|
context[ctx].append(str(arg)[1:])
|
|
else:
|
|
context.update( { 'note' : str(arg)})
|
|
return context
|
|
|
|
def add(self, file: str, *argv: str):
|
|
"""
|
|
Add context to existing filename and saves local folder context file.
|
|
|
|
file (str): path to existing file
|
|
argv (list): context parameters passed to be added
|
|
|
|
"""
|
|
if os.path.isfile(file):
|
|
if os.path.dirname(file):
|
|
self.__cd(os.path.dirname(file))
|
|
file = os.path.basename(file)
|
|
context = self.__context(argv)
|
|
if file in self.DB:
|
|
for ctx in self.CTX.keys():
|
|
self.DB[file][ctx] = list(set( self.DB[file][ctx] + context[ctx]))
|
|
if 'note' in context.keys():
|
|
self.DB[file]['note'] = context['note']
|
|
else:
|
|
self.DB.update( { file : context } )
|
|
self.__save()
|
|
|
|
def remove(self, file: str, *argv: str):
|
|
"""
|
|
Remove context of existing filename and saves changed folder context file
|
|
|
|
file (str): path to existing file
|
|
argv (list): context parameters to be removed
|
|
|
|
"""
|
|
if os.path.isfile(file):
|
|
if os.path.dirname(file):
|
|
self.__cd(os.path.dirname(file))
|
|
file = os.path.basename(file)
|
|
context = self.__context(argv)
|
|
if file in self.DB:
|
|
for ctx in self.CTX.keys():
|
|
for item in context[ctx]:
|
|
if item in self.DB[file][ctx]:
|
|
self.DB[file][ctx].remove(item)
|
|
self.__save()
|
|
|
|
def delete(self, file: str, fop:bool=False):
|
|
"""
|
|
Deletes all entries of an existing file and optionally deletes the file
|
|
|
|
file (str): path to existing file
|
|
fop (bool): if True removes the referenced file
|
|
|
|
"""
|
|
if os.path.dirname(file):
|
|
self.__cd(os.path.dirname(file))
|
|
if fop:
|
|
os.remove(file)
|
|
file = os.path.basename(file)
|
|
if file in self.DB:
|
|
del self.DB[file]
|
|
self.__save()
|
|
|
|
def move(self, source: str, dest: str, retain: bool=False, fop: bool=True):
|
|
"""
|
|
Copies or moves existing file with contexts to a new destination
|
|
|
|
source (str): existing source file
|
|
dest (str): the new file destination
|
|
retain (bool): if True then copies, if False then moves
|
|
fop (bool): if True also does the file operation, if False only moves context data
|
|
|
|
"""
|
|
if os.path.isfile(source) and os.path.isdir(os.path.dirname(dest)):
|
|
sourcefolder = os.path.dirname(source)
|
|
sourcefile = os.path.basename(source)
|
|
destfolder = os.path.dirname(dest)
|
|
destfile = os.path.basename(dest)
|
|
if not destfile:
|
|
destfile = sourcefile
|
|
sourcedata = {}
|
|
self.__cd(sourcefolder)
|
|
if self.DB[sourcefile]:
|
|
sourcedata = self.DB[sourcefile]
|
|
if not retain:
|
|
self.purge(source)
|
|
self.__cd(destfolder)
|
|
self.DB.update( { destfile : sourcedata } )
|
|
self.__save()
|
|
if fop:
|
|
if retain:
|
|
shutil.copy2(source, dest)
|
|
else:
|
|
shutil.move(source, dest, copy_function=shutil.copy2)
|
|
|
|
def purge(self, path: str=''):
|
|
"""
|
|
Removes context data from folder if files do not exist
|
|
|
|
path (str): folder path
|
|
|
|
"""
|
|
if os.path.dirname(path):
|
|
self.__cd(os.path.dirname(path))
|
|
for file in self.DB.keys():
|
|
if not os.path.isfile(file):
|
|
if file in self.DB:
|
|
del self.DB[file]
|
|
self.__save()
|
|
if not self.DB:
|
|
os.remove(self.FILE)
|
|
|
|
def list(self, search: str, path: str='./', filter=None, exact=True, recursive: bool=True, cache: bool=False):
|
|
"""
|
|
Returns an index object with query results and optionally saves a cached index
|
|
|
|
search (str): the search query
|
|
path (str): the root context for the search and index
|
|
filter (str): limit the search to a context
|
|
exact (bool): exact search match if True, otherwise partial match
|
|
recursive (bool): if True also searches all subfolder contexts
|
|
cache (bool): if True loads index from cache file if exists, writes it if it doesn't exist
|
|
|
|
"""
|
|
if path == '':
|
|
path = './'
|
|
if not os.path.dirname(path):
|
|
path += '/'
|
|
cachefile = os.path.join(path, self.CACHEFILE)
|
|
if cache and os.path.isfile(cachefile):
|
|
try:
|
|
with open(cachefile, 'r') as cache:
|
|
self.INDEX = yaml.safe_load(cache)
|
|
except Exception:
|
|
print('File does not exist')
|
|
else:
|
|
self.__indexfolder(path)
|
|
if recursive:
|
|
for (root,dirs,files) in os.walk(path):
|
|
if dirs:
|
|
for dir in dirs:
|
|
rpath = os.path.join(root, dir)
|
|
self.__indexfolder(rpath)
|
|
if cache:
|
|
try:
|
|
with open(cachefile, 'w+') as data:
|
|
yaml.dump(self.INDEX, data)
|
|
except Exception:
|
|
print('File write error')
|
|
result = {
|
|
'index' : self.INDEX,
|
|
'found' : []
|
|
}
|
|
for ctx in self.CTX.keys():
|
|
for context in self.INDEX[ctx]:
|
|
if filter and not filter == ctx:
|
|
continue
|
|
if exact:
|
|
if search == context:
|
|
result['found'] = self.INDEX[ctx][context]
|
|
else:
|
|
if search in context:
|
|
result['found'] = self.INDEX[ctx][context]
|
|
return result
|
|
|
|
def __indexfolder(self, sub: str):
|
|
"""
|
|
Indexes a folder context
|
|
|
|
sub (str): path of folder
|
|
|
|
"""
|
|
if os.path.dirname(sub):
|
|
self.__cd(sub)
|
|
if self.DB:
|
|
for file in self.DB.keys():
|
|
path = os.path.join(sub, file)
|
|
self.INDEX['files'].update( { path : self.DB[file]} )
|
|
for ctx in self.CTX.keys():
|
|
for item in self.DB[file][ctx]:
|
|
if item:
|
|
filelist = [path]
|
|
if ctx in self.INDEX:
|
|
if item in self.INDEX[ctx]:
|
|
filelist += self.INDEX[ctx][item]
|
|
self.INDEX.update( { ctx : { item : filelist } } )
|
|
|
|
|
|
o = OrvaniTag()
|
|
r = o.list('markdown', cache=True)
|
|
print(r)
|