From 85d93e3dae7af6c3107898abcdddcf732c8e9477 Mon Sep 17 00:00:00 2001 From: randogoth Date: Mon, 12 Jun 2023 23:26:39 +0300 Subject: [PATCH] cleanup --- build/lib/orvanitag/cmd.py | 2 +- build/lib/orvanitag/orvani.py | 469 +++++++++++++++++----------------- orvanitag/cmd.py | 2 +- orvanitag/orvani.py | 469 +++++++++++++++++----------------- 4 files changed, 472 insertions(+), 470 deletions(-) diff --git a/build/lib/orvanitag/cmd.py b/build/lib/orvanitag/cmd.py index 76a1687..3a7063a 100644 --- a/build/lib/orvanitag/cmd.py +++ b/build/lib/orvanitag/cmd.py @@ -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) diff --git a/build/lib/orvanitag/orvani.py b/build/lib/orvanitag/orvani.py index 0aeaaed..40e20e2 100644 --- a/build/lib/orvanitag/orvani.py +++ b/build/lib/orvanitag/orvani.py @@ -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] diff --git a/orvanitag/cmd.py b/orvanitag/cmd.py index 76a1687..3a7063a 100644 --- a/orvanitag/cmd.py +++ b/orvanitag/cmd.py @@ -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) diff --git a/orvanitag/orvani.py b/orvanitag/orvani.py index 0aeaaed..40e20e2 100644 --- a/orvanitag/orvani.py +++ b/orvanitag/orvani.py @@ -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]