cleanup
This commit is contained in:
randogoth
parent
a6724c11ef
commit
85d93e3dae
4 changed files with 472 additions and 470 deletions
|
|
@ -22,7 +22,7 @@ def mktable(query: str, index: dict, filter=None, trunc_len: int=25):
|
|||
if len(index['found']) > 0: # is there a search result?
|
||||
|
||||
# Create table
|
||||
table = Table(title=f"Results for `{query}`", box=box.HORIZONTALS, show_lines=True)
|
||||
table = Table(box=box.HORIZONTALS, show_lines=True)
|
||||
|
||||
# File Path Column Header
|
||||
table.add_column("Path", style="white", no_wrap=False, overflow='fold', max_width=40)
|
||||
|
|
|
|||
|
|
@ -178,6 +178,236 @@ class OrvaniTag:
|
|||
return context
|
||||
|
||||
|
||||
def __add(self, path: str, context: dict):
|
||||
"""Programmatically add context"""
|
||||
ctx_str = self.__collect_contexts(context)
|
||||
if ctx_str:
|
||||
argv = self.__get_argv(' '.join(ctx_str))
|
||||
self.add(path, *argv)
|
||||
|
||||
|
||||
def __makeindex(self, path:str, recursive:bool=True, cache: bool=False, delete_cache: bool=False):
|
||||
"""
|
||||
Creates global context index for directory for searching, optionally also a cache file
|
||||
Loads plugin readers to also read file specific metadata
|
||||
|
||||
path (dict): root directory
|
||||
recursive (bool): traverse subdirectories
|
||||
|
||||
"""
|
||||
cachefile = Path(path) / self.CACHEFILE
|
||||
loaded_from_cache = False
|
||||
if delete_cache:
|
||||
cachefile.unlink(missing_ok=True)
|
||||
if cache and cachefile.is_file(): # load index from cache
|
||||
self.INDEX = yaml.safe_load(cachefile.read_text())
|
||||
loaded_from_cache = True
|
||||
|
||||
else: # index the directory manually
|
||||
self.__indexfolder(path) # index current directory
|
||||
|
||||
# Dive into the subdirectories
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
# index each subdirectory
|
||||
if dirs:
|
||||
for dir in dirs:
|
||||
rpath = os.path.join(root, dir)
|
||||
self.__indexfolder(rpath) # index subdirectory
|
||||
|
||||
# If we just created an index, write it to cache
|
||||
if cache and not loaded_from_cache:
|
||||
with open(cachefile, 'w+') as data:
|
||||
yaml.dump(self.INDEX, data)
|
||||
|
||||
|
||||
def __indexfolder(self, sub: str):
|
||||
"""
|
||||
Load a folder's context and add it to the global context index
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
target = Path(sub)
|
||||
if target.is_dir():
|
||||
self.__cd(sub)
|
||||
if self.DB:
|
||||
for file in self.DB.keys():
|
||||
path = os.path.join(sub, file)
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if context_key in self.DB[file]:
|
||||
self.__updateindex(path, self.DB[file], context_key)
|
||||
|
||||
|
||||
def __updateindex(self, path: str, context: dict, context_key: str):
|
||||
"""
|
||||
Adds entries of a single context type of a file to the global context index.
|
||||
The context entry is associated with the file in the 'items' index
|
||||
The file is associated with its context entry
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
# Update files index
|
||||
for key, value in context.items():
|
||||
if path in self.INDEX['items']: # check if there is already an entry
|
||||
# if yes, merge the entries
|
||||
if key in self.INDEX['items'][path] and isinstance(self.INDEX['items'][path][key], list) and isinstance(value, list):
|
||||
self.INDEX['items'][path][key] += value
|
||||
self.INDEX['items'][path][key] = list(set(self.INDEX['items'][path][key]))
|
||||
# if not, just add them
|
||||
else:
|
||||
self.INDEX['items'][path][key] = value
|
||||
else: # if not, just add the entry
|
||||
self.INDEX['items'][path] = { key : value }
|
||||
|
||||
# sort the file index alphabetically
|
||||
self.INDEX['items'] = {key: value for key, value in sorted(self.INDEX['items'].items())}
|
||||
|
||||
# Update context index
|
||||
# Add each file path to a context entry
|
||||
|
||||
if type(context[context_key]) == list: # is it a context entry
|
||||
for item in context[context_key]:
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
else: # or is it a note
|
||||
item = context[context_key]
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
|
||||
|
||||
def __search(self, search, exact:bool=False, either:bool=False):
|
||||
"""
|
||||
Searches the index and adds search result to it
|
||||
|
||||
search (list): search query list
|
||||
exact (bool): exact match or partial match
|
||||
either (bool): combine found items by OR instead of AND
|
||||
|
||||
"""
|
||||
result = {
|
||||
'index' : self.INDEX,
|
||||
'found' : []
|
||||
}
|
||||
if search[0]: # is there a search query?
|
||||
|
||||
query = self.__context(search) # turn it into a context dict
|
||||
return_dict = self.__context([' ']) # create an empty context dict
|
||||
universal = {i for i in self.SCHEMA if self.SCHEMA[i]==None}.pop() # get the schema key with the None value, so we know which one to use for the note
|
||||
|
||||
if universal in query: # does the query have an entry without context prefix?
|
||||
query[universal] = [query[universal]] # turn it into a list for better processing
|
||||
|
||||
for context_key in self.SCHEMA.keys(): # turn all values in the empty dict into list types
|
||||
return_dict[context_key] = []
|
||||
|
||||
# Iterate Search Query Terms
|
||||
for param in query.keys():
|
||||
needles = query[param]
|
||||
|
||||
# Add the query without content prefix to all other contexts
|
||||
# It is a wildcard to search all contexts at once
|
||||
if universal in query.keys():
|
||||
needles += query[universal]
|
||||
|
||||
# Run the search and add every found file path to the return dict
|
||||
# in its respective context (or all of them for the wildcard search)
|
||||
if needles:
|
||||
haystack = self.INDEX[param].keys()
|
||||
for hay in haystack:
|
||||
for needle in needles:
|
||||
if exact:
|
||||
if needle == hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
else:
|
||||
if needle in hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
|
||||
if len(search) > 1 and not either:
|
||||
# we do an AND search by default, but only if there was more than 1 query item
|
||||
result['found'] = list(set(self.__intersect(return_dict)))
|
||||
|
||||
else:
|
||||
# for 1 query searches and OR searches we simply add all of the matches
|
||||
min = []
|
||||
for sublist in return_dict.values():
|
||||
min += sublist
|
||||
result['found'] = list(set(min))
|
||||
|
||||
else:
|
||||
# return the whole index if there was no query item
|
||||
result['found'] = self.INDEX['items'].keys()
|
||||
|
||||
# Sort the result directory tree from shallow to deep based on the occurrence of /
|
||||
result['found'] = sorted(result['found'], key=lambda x: x.count('/'))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def __intersect(self, result: dict):
|
||||
"""
|
||||
Returns an intersection of two search queries (AND)
|
||||
|
||||
result (dict): dict with search result lists to be intersected
|
||||
"""
|
||||
|
||||
mresult = {key: value for key, value in result.items() if value} # remove empty items
|
||||
|
||||
if len(mresult) < 2: # if there is not enough to intersect, just return it
|
||||
return []
|
||||
|
||||
common_elements = set(mresult[next(iter(mresult))]) # elements common in all
|
||||
|
||||
for key in mresult.keys():
|
||||
# intersect common elements
|
||||
common_elements = common_elements.intersection(mresult[key])
|
||||
|
||||
return list(common_elements)
|
||||
|
||||
|
||||
def __collect_contexts(self, context: dict):
|
||||
"""
|
||||
Reuturns a list of prefixed context entries from a context dict
|
||||
"""
|
||||
elements = []
|
||||
context_keys = list(CONTEXTS.keys())
|
||||
for context_key in context_keys:
|
||||
if not context_key in context:
|
||||
context[context_key] = ''
|
||||
if context[context_key]:
|
||||
if CONTEXTS[context_key]:
|
||||
prefix = CONTEXTS[context_key]
|
||||
else:
|
||||
prefix = ''
|
||||
if not type(context[context_key]) == str:
|
||||
elements.append(' '.join([prefix + tag for tag in context[context_key]]))
|
||||
else:
|
||||
elements.append(context[context_key])
|
||||
return elements
|
||||
|
||||
|
||||
def __get_argv(self, context_string: str):
|
||||
"""
|
||||
Make a list of args from a context string to pass as an *argv parameter
|
||||
"""
|
||||
|
||||
tags = re.findall(self.REGEX, context_string)
|
||||
note = re.sub(self.REGEX, '', context_string).lstrip()
|
||||
argv = ' '.join(tags).split(' ') + [note]
|
||||
argv = [x for x in list(set(argv)) if x != ''] # make unique and remove empty items
|
||||
return argv
|
||||
|
||||
|
||||
def add(self, file: str, *argv: str):
|
||||
"""
|
||||
Add context to existing filename and saves local folder context file.
|
||||
|
|
@ -215,13 +445,6 @@ class OrvaniTag:
|
|||
return [1, file]
|
||||
|
||||
|
||||
def __add(self, path: str, context: dict):
|
||||
"""Programmatically add context"""
|
||||
ctx_str = self.__collect_contexts(context)
|
||||
if ctx_str:
|
||||
argv = self.get_argv(' '.join(ctx_str))
|
||||
self.add(path, *argv)
|
||||
|
||||
def remove(self, file: str, *argv: str):
|
||||
"""
|
||||
Remove context of existing filename and saves changed folder context file
|
||||
|
|
@ -397,44 +620,6 @@ class OrvaniTag:
|
|||
return result
|
||||
|
||||
|
||||
def __makeindex(self, path:str, recursive:bool=True, cache: bool=False, delete_cache: bool=False):
|
||||
"""
|
||||
Creates global context index for directory for searching, optionally also a cache file
|
||||
Loads plugin readers to also read file specific metadata
|
||||
|
||||
path (dict): root directory
|
||||
recursive (bool): traverse subdirectories
|
||||
|
||||
"""
|
||||
cachefile = Path(path) / self.CACHEFILE
|
||||
loaded_from_cache = False
|
||||
if delete_cache:
|
||||
cachefile.unlink(missing_ok=True)
|
||||
if cache and cachefile.is_file(): # load index from cache
|
||||
self.INDEX = yaml.safe_load(cachefile.read_text())
|
||||
loaded_from_cache = True
|
||||
|
||||
else: # index the directory manually
|
||||
self.__indexfolder(path) # index current directory
|
||||
|
||||
# Dive into the subdirectories
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
# index each subdirectory
|
||||
if dirs:
|
||||
for dir in dirs:
|
||||
rpath = os.path.join(root, dir)
|
||||
self.__indexfolder(rpath) # index subdirectory
|
||||
|
||||
# If we just created an index, write it to cache
|
||||
if cache and not loaded_from_cache:
|
||||
with open(cachefile, 'w+') as data:
|
||||
yaml.dump(self.INDEX, data)
|
||||
|
||||
|
||||
def scan(self, path: str, recursive: bool=False):
|
||||
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
|
@ -445,162 +630,11 @@ class OrvaniTag:
|
|||
for file in files:
|
||||
fpath = os.path.join(root, file)
|
||||
self.__cd(root, True)
|
||||
self.__metadata(fpath)
|
||||
self.read_metadata(fpath)
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
|
||||
def __indexfolder(self, sub: str):
|
||||
"""
|
||||
Load a folder's context and add it to the global context index
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
target = Path(sub)
|
||||
if target.is_dir():
|
||||
self.__cd(sub)
|
||||
if self.DB:
|
||||
for file in self.DB.keys():
|
||||
path = os.path.join(sub, file)
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if context_key in self.DB[file]:
|
||||
self.__updateindex(path, self.DB[file], context_key)
|
||||
|
||||
|
||||
def __updateindex(self, path: str, context: dict, context_key: str):
|
||||
"""
|
||||
Adds entries of a single context type of a file to the global context index.
|
||||
The context entry is associated with the file in the 'items' index
|
||||
The file is associated with its context entry
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
# Update files index
|
||||
for key, value in context.items():
|
||||
if path in self.INDEX['items']: # check if there is already an entry
|
||||
# if yes, merge the entries
|
||||
if key in self.INDEX['items'][path] and isinstance(self.INDEX['items'][path][key], list) and isinstance(value, list):
|
||||
self.INDEX['items'][path][key] += value
|
||||
self.INDEX['items'][path][key] = list(set(self.INDEX['items'][path][key]))
|
||||
# if not, just add them
|
||||
else:
|
||||
self.INDEX['items'][path][key] = value
|
||||
else: # if not, just add the entry
|
||||
self.INDEX['items'][path] = { key : value }
|
||||
|
||||
# sort the file index alphabetically
|
||||
self.INDEX['items'] = {key: value for key, value in sorted(self.INDEX['items'].items())}
|
||||
|
||||
# Update context index
|
||||
# Add each file path to a context entry
|
||||
|
||||
if type(context[context_key]) == list: # is it a context entry
|
||||
for item in context[context_key]:
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
else: # or is it a note
|
||||
item = context[context_key]
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
|
||||
|
||||
def __search(self, search, exact:bool=False, either:bool=False):
|
||||
"""
|
||||
Searches the index and adds search result to it
|
||||
|
||||
search (list): search query list
|
||||
exact (bool): exact match or partial match
|
||||
either (bool): combine found items by OR instead of AND
|
||||
|
||||
"""
|
||||
result = {
|
||||
'index' : self.INDEX,
|
||||
'found' : []
|
||||
}
|
||||
if search[0]: # is there a search query?
|
||||
|
||||
query = self.__context(search) # turn it into a context dict
|
||||
return_dict = self.__context([' ']) # create an empty context dict
|
||||
universal = {i for i in self.SCHEMA if self.SCHEMA[i]==None}.pop() # get the schema key with the None value, so we know which one to use for the note
|
||||
|
||||
if universal in query: # does the query have an entry without context prefix?
|
||||
query[universal] = [query[universal]] # turn it into a list for better processing
|
||||
|
||||
for context_key in self.SCHEMA.keys(): # turn all values in the empty dict into list types
|
||||
return_dict[context_key] = []
|
||||
|
||||
# Iterate Search Query Terms
|
||||
for param in query.keys():
|
||||
needles = query[param]
|
||||
|
||||
# Add the query without content prefix to all other contexts
|
||||
# It is a wildcard to search all contexts at once
|
||||
if universal in query.keys():
|
||||
needles += query[universal]
|
||||
|
||||
# Run the search and add every found file path to the return dict
|
||||
# in its respective context (or all of them for the wildcard search)
|
||||
if needles:
|
||||
haystack = self.INDEX[param].keys()
|
||||
for hay in haystack:
|
||||
for needle in needles:
|
||||
if exact:
|
||||
if needle == hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
else:
|
||||
if needle in hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
|
||||
if len(search) > 1 and not either:
|
||||
# we do an AND search by default, but only if there was more than 1 query item
|
||||
result['found'] = list(set(self.__intersect(return_dict)))
|
||||
|
||||
else:
|
||||
# for 1 query searches and OR searches we simply add all of the matches
|
||||
min = []
|
||||
for sublist in return_dict.values():
|
||||
min += sublist
|
||||
result['found'] = list(set(min))
|
||||
|
||||
else:
|
||||
# return the whole index if there was no query item
|
||||
result['found'] = self.INDEX['items'].keys()
|
||||
|
||||
# Sort the result directory tree from shallow to deep based on the occurrence of /
|
||||
result['found'] = sorted(result['found'], key=lambda x: x.count('/'))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def __intersect(self, result: dict):
|
||||
"""
|
||||
Returns an intersection of two search queries (AND)
|
||||
|
||||
result (dict): dict with search result lists to be intersected
|
||||
"""
|
||||
|
||||
mresult = {key: value for key, value in result.items() if value} # remove empty items
|
||||
|
||||
if len(mresult) < 2: # if there is not enough to intersect, just return it
|
||||
return []
|
||||
|
||||
common_elements = set(mresult[next(iter(mresult))]) # elements common in all
|
||||
|
||||
for key in mresult.keys():
|
||||
# intersect common elements
|
||||
common_elements = common_elements.intersection(mresult[key])
|
||||
|
||||
return list(common_elements)
|
||||
|
||||
|
||||
def register_plugin(self, function):
|
||||
"""
|
||||
|
|
@ -610,19 +644,7 @@ class OrvaniTag:
|
|||
return function
|
||||
|
||||
|
||||
def get_argv(self, context_string: str):
|
||||
"""
|
||||
Make a list of args from a context string to pass as an *argv parameter
|
||||
"""
|
||||
|
||||
tags = re.findall(self.REGEX, context_string)
|
||||
note = re.sub(self.REGEX, '', context_string).lstrip()
|
||||
argv = ' '.join(tags).split(' ') + [note]
|
||||
argv = [x for x in list(set(argv)) if x != ''] # make unique and remove empty items
|
||||
return argv
|
||||
|
||||
|
||||
def __metadata(self, filename: str):
|
||||
def read_metadata(self, filename: str):
|
||||
"""
|
||||
Processes a file with all reader plugins to extract its metadata
|
||||
"""
|
||||
|
|
@ -665,27 +687,6 @@ class OrvaniTag:
|
|||
print(f'Plugin Error ({plugin_name} reading {filename}): `{e}`')
|
||||
|
||||
|
||||
def __collect_contexts(self, context: dict):
|
||||
"""
|
||||
Reuturns a list of prefixed context entries from a context dict
|
||||
"""
|
||||
elements = []
|
||||
context_keys = list(CONTEXTS.keys())
|
||||
for context_key in context_keys:
|
||||
if not context_key in context:
|
||||
context[context_key] = ''
|
||||
if context[context_key]:
|
||||
if CONTEXTS[context_key]:
|
||||
prefix = CONTEXTS[context_key]
|
||||
else:
|
||||
prefix = ''
|
||||
if not type(context[context_key]) == str:
|
||||
elements.append(' '.join([prefix + tag for tag in context[context_key]]))
|
||||
else:
|
||||
elements.append(context[context_key])
|
||||
return elements
|
||||
|
||||
|
||||
def write_all(self, path: str, method: str, recursive: bool=False):
|
||||
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
|
@ -696,13 +697,13 @@ class OrvaniTag:
|
|||
for file in files:
|
||||
fpath = os.path.join(root, file)
|
||||
self.__cd(root, True)
|
||||
self.write(fpath, method)
|
||||
self.write_metadata(fpath, method)
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
|
||||
def write(self, filename: str, method: str):
|
||||
def write_metadata(self, filename: str, method: str):
|
||||
"""
|
||||
Writes a file's context entries as metadata to the file using a writer plugin
|
||||
|
||||
|
|
@ -820,7 +821,7 @@ class OrvaniTag:
|
|||
if new_file.exists(): # Check if file exists
|
||||
string = line_elements[1].strip()
|
||||
if string:
|
||||
new_file_context = self.__context(self.get_argv(string))
|
||||
new_file_context = self.__context(self.__get_argv(string))
|
||||
# Check if file is already known and don't import anything we already have
|
||||
if path in known.keys():
|
||||
known_file_context = known[path]
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ def mktable(query: str, index: dict, filter=None, trunc_len: int=25):
|
|||
if len(index['found']) > 0: # is there a search result?
|
||||
|
||||
# Create table
|
||||
table = Table(title=f"Results for `{query}`", box=box.HORIZONTALS, show_lines=True)
|
||||
table = Table(box=box.HORIZONTALS, show_lines=True)
|
||||
|
||||
# File Path Column Header
|
||||
table.add_column("Path", style="white", no_wrap=False, overflow='fold', max_width=40)
|
||||
|
|
|
|||
|
|
@ -178,6 +178,236 @@ class OrvaniTag:
|
|||
return context
|
||||
|
||||
|
||||
def __add(self, path: str, context: dict):
|
||||
"""Programmatically add context"""
|
||||
ctx_str = self.__collect_contexts(context)
|
||||
if ctx_str:
|
||||
argv = self.__get_argv(' '.join(ctx_str))
|
||||
self.add(path, *argv)
|
||||
|
||||
|
||||
def __makeindex(self, path:str, recursive:bool=True, cache: bool=False, delete_cache: bool=False):
|
||||
"""
|
||||
Creates global context index for directory for searching, optionally also a cache file
|
||||
Loads plugin readers to also read file specific metadata
|
||||
|
||||
path (dict): root directory
|
||||
recursive (bool): traverse subdirectories
|
||||
|
||||
"""
|
||||
cachefile = Path(path) / self.CACHEFILE
|
||||
loaded_from_cache = False
|
||||
if delete_cache:
|
||||
cachefile.unlink(missing_ok=True)
|
||||
if cache and cachefile.is_file(): # load index from cache
|
||||
self.INDEX = yaml.safe_load(cachefile.read_text())
|
||||
loaded_from_cache = True
|
||||
|
||||
else: # index the directory manually
|
||||
self.__indexfolder(path) # index current directory
|
||||
|
||||
# Dive into the subdirectories
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
# index each subdirectory
|
||||
if dirs:
|
||||
for dir in dirs:
|
||||
rpath = os.path.join(root, dir)
|
||||
self.__indexfolder(rpath) # index subdirectory
|
||||
|
||||
# If we just created an index, write it to cache
|
||||
if cache and not loaded_from_cache:
|
||||
with open(cachefile, 'w+') as data:
|
||||
yaml.dump(self.INDEX, data)
|
||||
|
||||
|
||||
def __indexfolder(self, sub: str):
|
||||
"""
|
||||
Load a folder's context and add it to the global context index
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
target = Path(sub)
|
||||
if target.is_dir():
|
||||
self.__cd(sub)
|
||||
if self.DB:
|
||||
for file in self.DB.keys():
|
||||
path = os.path.join(sub, file)
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if context_key in self.DB[file]:
|
||||
self.__updateindex(path, self.DB[file], context_key)
|
||||
|
||||
|
||||
def __updateindex(self, path: str, context: dict, context_key: str):
|
||||
"""
|
||||
Adds entries of a single context type of a file to the global context index.
|
||||
The context entry is associated with the file in the 'items' index
|
||||
The file is associated with its context entry
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
# Update files index
|
||||
for key, value in context.items():
|
||||
if path in self.INDEX['items']: # check if there is already an entry
|
||||
# if yes, merge the entries
|
||||
if key in self.INDEX['items'][path] and isinstance(self.INDEX['items'][path][key], list) and isinstance(value, list):
|
||||
self.INDEX['items'][path][key] += value
|
||||
self.INDEX['items'][path][key] = list(set(self.INDEX['items'][path][key]))
|
||||
# if not, just add them
|
||||
else:
|
||||
self.INDEX['items'][path][key] = value
|
||||
else: # if not, just add the entry
|
||||
self.INDEX['items'][path] = { key : value }
|
||||
|
||||
# sort the file index alphabetically
|
||||
self.INDEX['items'] = {key: value for key, value in sorted(self.INDEX['items'].items())}
|
||||
|
||||
# Update context index
|
||||
# Add each file path to a context entry
|
||||
|
||||
if type(context[context_key]) == list: # is it a context entry
|
||||
for item in context[context_key]:
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
else: # or is it a note
|
||||
item = context[context_key]
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
|
||||
|
||||
def __search(self, search, exact:bool=False, either:bool=False):
|
||||
"""
|
||||
Searches the index and adds search result to it
|
||||
|
||||
search (list): search query list
|
||||
exact (bool): exact match or partial match
|
||||
either (bool): combine found items by OR instead of AND
|
||||
|
||||
"""
|
||||
result = {
|
||||
'index' : self.INDEX,
|
||||
'found' : []
|
||||
}
|
||||
if search[0]: # is there a search query?
|
||||
|
||||
query = self.__context(search) # turn it into a context dict
|
||||
return_dict = self.__context([' ']) # create an empty context dict
|
||||
universal = {i for i in self.SCHEMA if self.SCHEMA[i]==None}.pop() # get the schema key with the None value, so we know which one to use for the note
|
||||
|
||||
if universal in query: # does the query have an entry without context prefix?
|
||||
query[universal] = [query[universal]] # turn it into a list for better processing
|
||||
|
||||
for context_key in self.SCHEMA.keys(): # turn all values in the empty dict into list types
|
||||
return_dict[context_key] = []
|
||||
|
||||
# Iterate Search Query Terms
|
||||
for param in query.keys():
|
||||
needles = query[param]
|
||||
|
||||
# Add the query without content prefix to all other contexts
|
||||
# It is a wildcard to search all contexts at once
|
||||
if universal in query.keys():
|
||||
needles += query[universal]
|
||||
|
||||
# Run the search and add every found file path to the return dict
|
||||
# in its respective context (or all of them for the wildcard search)
|
||||
if needles:
|
||||
haystack = self.INDEX[param].keys()
|
||||
for hay in haystack:
|
||||
for needle in needles:
|
||||
if exact:
|
||||
if needle == hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
else:
|
||||
if needle in hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
|
||||
if len(search) > 1 and not either:
|
||||
# we do an AND search by default, but only if there was more than 1 query item
|
||||
result['found'] = list(set(self.__intersect(return_dict)))
|
||||
|
||||
else:
|
||||
# for 1 query searches and OR searches we simply add all of the matches
|
||||
min = []
|
||||
for sublist in return_dict.values():
|
||||
min += sublist
|
||||
result['found'] = list(set(min))
|
||||
|
||||
else:
|
||||
# return the whole index if there was no query item
|
||||
result['found'] = self.INDEX['items'].keys()
|
||||
|
||||
# Sort the result directory tree from shallow to deep based on the occurrence of /
|
||||
result['found'] = sorted(result['found'], key=lambda x: x.count('/'))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def __intersect(self, result: dict):
|
||||
"""
|
||||
Returns an intersection of two search queries (AND)
|
||||
|
||||
result (dict): dict with search result lists to be intersected
|
||||
"""
|
||||
|
||||
mresult = {key: value for key, value in result.items() if value} # remove empty items
|
||||
|
||||
if len(mresult) < 2: # if there is not enough to intersect, just return it
|
||||
return []
|
||||
|
||||
common_elements = set(mresult[next(iter(mresult))]) # elements common in all
|
||||
|
||||
for key in mresult.keys():
|
||||
# intersect common elements
|
||||
common_elements = common_elements.intersection(mresult[key])
|
||||
|
||||
return list(common_elements)
|
||||
|
||||
|
||||
def __collect_contexts(self, context: dict):
|
||||
"""
|
||||
Reuturns a list of prefixed context entries from a context dict
|
||||
"""
|
||||
elements = []
|
||||
context_keys = list(CONTEXTS.keys())
|
||||
for context_key in context_keys:
|
||||
if not context_key in context:
|
||||
context[context_key] = ''
|
||||
if context[context_key]:
|
||||
if CONTEXTS[context_key]:
|
||||
prefix = CONTEXTS[context_key]
|
||||
else:
|
||||
prefix = ''
|
||||
if not type(context[context_key]) == str:
|
||||
elements.append(' '.join([prefix + tag for tag in context[context_key]]))
|
||||
else:
|
||||
elements.append(context[context_key])
|
||||
return elements
|
||||
|
||||
|
||||
def __get_argv(self, context_string: str):
|
||||
"""
|
||||
Make a list of args from a context string to pass as an *argv parameter
|
||||
"""
|
||||
|
||||
tags = re.findall(self.REGEX, context_string)
|
||||
note = re.sub(self.REGEX, '', context_string).lstrip()
|
||||
argv = ' '.join(tags).split(' ') + [note]
|
||||
argv = [x for x in list(set(argv)) if x != ''] # make unique and remove empty items
|
||||
return argv
|
||||
|
||||
|
||||
def add(self, file: str, *argv: str):
|
||||
"""
|
||||
Add context to existing filename and saves local folder context file.
|
||||
|
|
@ -215,13 +445,6 @@ class OrvaniTag:
|
|||
return [1, file]
|
||||
|
||||
|
||||
def __add(self, path: str, context: dict):
|
||||
"""Programmatically add context"""
|
||||
ctx_str = self.__collect_contexts(context)
|
||||
if ctx_str:
|
||||
argv = self.get_argv(' '.join(ctx_str))
|
||||
self.add(path, *argv)
|
||||
|
||||
def remove(self, file: str, *argv: str):
|
||||
"""
|
||||
Remove context of existing filename and saves changed folder context file
|
||||
|
|
@ -397,44 +620,6 @@ class OrvaniTag:
|
|||
return result
|
||||
|
||||
|
||||
def __makeindex(self, path:str, recursive:bool=True, cache: bool=False, delete_cache: bool=False):
|
||||
"""
|
||||
Creates global context index for directory for searching, optionally also a cache file
|
||||
Loads plugin readers to also read file specific metadata
|
||||
|
||||
path (dict): root directory
|
||||
recursive (bool): traverse subdirectories
|
||||
|
||||
"""
|
||||
cachefile = Path(path) / self.CACHEFILE
|
||||
loaded_from_cache = False
|
||||
if delete_cache:
|
||||
cachefile.unlink(missing_ok=True)
|
||||
if cache and cachefile.is_file(): # load index from cache
|
||||
self.INDEX = yaml.safe_load(cachefile.read_text())
|
||||
loaded_from_cache = True
|
||||
|
||||
else: # index the directory manually
|
||||
self.__indexfolder(path) # index current directory
|
||||
|
||||
# Dive into the subdirectories
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
# index each subdirectory
|
||||
if dirs:
|
||||
for dir in dirs:
|
||||
rpath = os.path.join(root, dir)
|
||||
self.__indexfolder(rpath) # index subdirectory
|
||||
|
||||
# If we just created an index, write it to cache
|
||||
if cache and not loaded_from_cache:
|
||||
with open(cachefile, 'w+') as data:
|
||||
yaml.dump(self.INDEX, data)
|
||||
|
||||
|
||||
def scan(self, path: str, recursive: bool=False):
|
||||
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
|
@ -445,162 +630,11 @@ class OrvaniTag:
|
|||
for file in files:
|
||||
fpath = os.path.join(root, file)
|
||||
self.__cd(root, True)
|
||||
self.__metadata(fpath)
|
||||
self.read_metadata(fpath)
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
|
||||
def __indexfolder(self, sub: str):
|
||||
"""
|
||||
Load a folder's context and add it to the global context index
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
target = Path(sub)
|
||||
if target.is_dir():
|
||||
self.__cd(sub)
|
||||
if self.DB:
|
||||
for file in self.DB.keys():
|
||||
path = os.path.join(sub, file)
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if context_key in self.DB[file]:
|
||||
self.__updateindex(path, self.DB[file], context_key)
|
||||
|
||||
|
||||
def __updateindex(self, path: str, context: dict, context_key: str):
|
||||
"""
|
||||
Adds entries of a single context type of a file to the global context index.
|
||||
The context entry is associated with the file in the 'items' index
|
||||
The file is associated with its context entry
|
||||
|
||||
sub (str): path of folder
|
||||
"""
|
||||
# Update files index
|
||||
for key, value in context.items():
|
||||
if path in self.INDEX['items']: # check if there is already an entry
|
||||
# if yes, merge the entries
|
||||
if key in self.INDEX['items'][path] and isinstance(self.INDEX['items'][path][key], list) and isinstance(value, list):
|
||||
self.INDEX['items'][path][key] += value
|
||||
self.INDEX['items'][path][key] = list(set(self.INDEX['items'][path][key]))
|
||||
# if not, just add them
|
||||
else:
|
||||
self.INDEX['items'][path][key] = value
|
||||
else: # if not, just add the entry
|
||||
self.INDEX['items'][path] = { key : value }
|
||||
|
||||
# sort the file index alphabetically
|
||||
self.INDEX['items'] = {key: value for key, value in sorted(self.INDEX['items'].items())}
|
||||
|
||||
# Update context index
|
||||
# Add each file path to a context entry
|
||||
|
||||
if type(context[context_key]) == list: # is it a context entry
|
||||
for item in context[context_key]:
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
else: # or is it a note
|
||||
item = context[context_key]
|
||||
filelist = [path]
|
||||
if context_key in self.INDEX:
|
||||
if item in self.INDEX[context_key]:
|
||||
filelist += self.INDEX[context_key][item]
|
||||
self.INDEX[context_key][item] = list(set(filelist))
|
||||
|
||||
|
||||
def __search(self, search, exact:bool=False, either:bool=False):
|
||||
"""
|
||||
Searches the index and adds search result to it
|
||||
|
||||
search (list): search query list
|
||||
exact (bool): exact match or partial match
|
||||
either (bool): combine found items by OR instead of AND
|
||||
|
||||
"""
|
||||
result = {
|
||||
'index' : self.INDEX,
|
||||
'found' : []
|
||||
}
|
||||
if search[0]: # is there a search query?
|
||||
|
||||
query = self.__context(search) # turn it into a context dict
|
||||
return_dict = self.__context([' ']) # create an empty context dict
|
||||
universal = {i for i in self.SCHEMA if self.SCHEMA[i]==None}.pop() # get the schema key with the None value, so we know which one to use for the note
|
||||
|
||||
if universal in query: # does the query have an entry without context prefix?
|
||||
query[universal] = [query[universal]] # turn it into a list for better processing
|
||||
|
||||
for context_key in self.SCHEMA.keys(): # turn all values in the empty dict into list types
|
||||
return_dict[context_key] = []
|
||||
|
||||
# Iterate Search Query Terms
|
||||
for param in query.keys():
|
||||
needles = query[param]
|
||||
|
||||
# Add the query without content prefix to all other contexts
|
||||
# It is a wildcard to search all contexts at once
|
||||
if universal in query.keys():
|
||||
needles += query[universal]
|
||||
|
||||
# Run the search and add every found file path to the return dict
|
||||
# in its respective context (or all of them for the wildcard search)
|
||||
if needles:
|
||||
haystack = self.INDEX[param].keys()
|
||||
for hay in haystack:
|
||||
for needle in needles:
|
||||
if exact:
|
||||
if needle == hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
else:
|
||||
if needle in hay:
|
||||
if self.INDEX[param][hay] not in return_dict[param]:
|
||||
return_dict[param] += self.INDEX[param][hay]
|
||||
|
||||
if len(search) > 1 and not either:
|
||||
# we do an AND search by default, but only if there was more than 1 query item
|
||||
result['found'] = list(set(self.__intersect(return_dict)))
|
||||
|
||||
else:
|
||||
# for 1 query searches and OR searches we simply add all of the matches
|
||||
min = []
|
||||
for sublist in return_dict.values():
|
||||
min += sublist
|
||||
result['found'] = list(set(min))
|
||||
|
||||
else:
|
||||
# return the whole index if there was no query item
|
||||
result['found'] = self.INDEX['items'].keys()
|
||||
|
||||
# Sort the result directory tree from shallow to deep based on the occurrence of /
|
||||
result['found'] = sorted(result['found'], key=lambda x: x.count('/'))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def __intersect(self, result: dict):
|
||||
"""
|
||||
Returns an intersection of two search queries (AND)
|
||||
|
||||
result (dict): dict with search result lists to be intersected
|
||||
"""
|
||||
|
||||
mresult = {key: value for key, value in result.items() if value} # remove empty items
|
||||
|
||||
if len(mresult) < 2: # if there is not enough to intersect, just return it
|
||||
return []
|
||||
|
||||
common_elements = set(mresult[next(iter(mresult))]) # elements common in all
|
||||
|
||||
for key in mresult.keys():
|
||||
# intersect common elements
|
||||
common_elements = common_elements.intersection(mresult[key])
|
||||
|
||||
return list(common_elements)
|
||||
|
||||
|
||||
def register_plugin(self, function):
|
||||
"""
|
||||
|
|
@ -610,19 +644,7 @@ class OrvaniTag:
|
|||
return function
|
||||
|
||||
|
||||
def get_argv(self, context_string: str):
|
||||
"""
|
||||
Make a list of args from a context string to pass as an *argv parameter
|
||||
"""
|
||||
|
||||
tags = re.findall(self.REGEX, context_string)
|
||||
note = re.sub(self.REGEX, '', context_string).lstrip()
|
||||
argv = ' '.join(tags).split(' ') + [note]
|
||||
argv = [x for x in list(set(argv)) if x != ''] # make unique and remove empty items
|
||||
return argv
|
||||
|
||||
|
||||
def __metadata(self, filename: str):
|
||||
def read_metadata(self, filename: str):
|
||||
"""
|
||||
Processes a file with all reader plugins to extract its metadata
|
||||
"""
|
||||
|
|
@ -665,27 +687,6 @@ class OrvaniTag:
|
|||
print(f'Plugin Error ({plugin_name} reading {filename}): `{e}`')
|
||||
|
||||
|
||||
def __collect_contexts(self, context: dict):
|
||||
"""
|
||||
Reuturns a list of prefixed context entries from a context dict
|
||||
"""
|
||||
elements = []
|
||||
context_keys = list(CONTEXTS.keys())
|
||||
for context_key in context_keys:
|
||||
if not context_key in context:
|
||||
context[context_key] = ''
|
||||
if context[context_key]:
|
||||
if CONTEXTS[context_key]:
|
||||
prefix = CONTEXTS[context_key]
|
||||
else:
|
||||
prefix = ''
|
||||
if not type(context[context_key]) == str:
|
||||
elements.append(' '.join([prefix + tag for tag in context[context_key]]))
|
||||
else:
|
||||
elements.append(context[context_key])
|
||||
return elements
|
||||
|
||||
|
||||
def write_all(self, path: str, method: str, recursive: bool=False):
|
||||
|
||||
for (root,dirs,files) in os.walk(path, topdown=True):
|
||||
|
|
@ -696,13 +697,13 @@ class OrvaniTag:
|
|||
for file in files:
|
||||
fpath = os.path.join(root, file)
|
||||
self.__cd(root, True)
|
||||
self.write(fpath, method)
|
||||
self.write_metadata(fpath, method)
|
||||
|
||||
if not recursive: # check if to include subdirectories
|
||||
break
|
||||
|
||||
|
||||
def write(self, filename: str, method: str):
|
||||
def write_metadata(self, filename: str, method: str):
|
||||
"""
|
||||
Writes a file's context entries as metadata to the file using a writer plugin
|
||||
|
||||
|
|
@ -820,7 +821,7 @@ class OrvaniTag:
|
|||
if new_file.exists(): # Check if file exists
|
||||
string = line_elements[1].strip()
|
||||
if string:
|
||||
new_file_context = self.__context(self.get_argv(string))
|
||||
new_file_context = self.__context(self.__get_argv(string))
|
||||
# Check if file is already known and don't import anything we already have
|
||||
if path in known.keys():
|
||||
known_file_context = known[path]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue