diff --git a/.context.yaml b/.context.yaml index 438432d..44362ef 100644 --- a/.context.yaml +++ b/.context.yaml @@ -1,24 +1 @@ -items: - COMMANDS.md: - contexts: [] - projects: [] - tags: - - great - README.md: - contexts: [] - projects: [] - tags: - - fool - TODO.md: - contexts: - - bbb - - python - note: Chawera - projects: - - proooo - - fool - - bokertov - tags: - - rusty - - xxx - - ttt +items: {} diff --git a/COMMANDS.md b/COMMANDS.md deleted file mode 100644 index efbe93b..0000000 --- a/COMMANDS.md +++ /dev/null @@ -1,17 +0,0 @@ ---- -#// @erfwegf #super #great +coding This is awesome ---- -# Tagging - -+ add file.txt @context +project #tag This is a note -+ del file.txt @context - -# File Operations - -+ copy/cp file.txt dir/file.txt -+ move/mv file.txt dir/file.txt -+ del/rm file.txt - -# DB Operations - -+ dir/list/ls/l TERM -e -p \ No newline at end of file diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..cf1a457 --- /dev/null +++ b/LICENSE @@ -0,0 +1,7 @@ +Copyright 2023 randogoth + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. \ No newline at end of file diff --git a/orvani/__init__.py b/build/lib/orvanitag/__init__.py similarity index 100% rename from orvani/__init__.py rename to build/lib/orvanitag/__init__.py diff --git a/orvani/cmd.py b/build/lib/orvanitag/cmd.py similarity index 88% rename from orvani/cmd.py rename to build/lib/orvanitag/cmd.py index 69dfab4..76a1687 100644 --- a/orvani/cmd.py +++ b/build/lib/orvanitag/cmd.py @@ -6,7 +6,7 @@ from pathlib import Path from rich import box from rich.console import Console from rich.table import Table -from orvani import OrvaniTag, CONTEXT_FILE, CACHE_FILE, CONTEXTS +from orvanitag.orvani import OrvaniTag, CONTEXT_FILE, CACHE_FILE, CONTEXTS def mktable(query: str, index: dict, filter=None, trunc_len: int=25): """Creates a rich table from `otag ls` search result @@ -31,7 +31,7 @@ def mktable(query: str, index: dict, filter=None, trunc_len: int=25): ctx = list(CONTEXTS.keys()) styles = ["cyan", "yellow", "magenta", "green"] for i, c in enumerate(ctx): - table.add_column(ctx[i].title(), style=styles[i], max_width=12, overflow='fold') + table.add_column(ctx[i].title(), style=styles[i], overflow='fold') # Create Rows Based on Search Results for file in index['found']: @@ -46,7 +46,8 @@ def mktable(query: str, index: dict, filter=None, trunc_len: int=25): long_file_pre = math.floor(trunc_len/2-1) long_file_post = len(item.name) - math.ceil(trunc_len/2-1) path = f'{item.parent}/' + ((item.name[:long_file_pre] + '…' + item.name[long_file_post:]) if len(item.name) > trunc_len else item.name) - + path = path.replace('./', '') + fc = index['index']['items'][file] elements = [] elements.append(path) @@ -82,9 +83,9 @@ def nice_context(context: dict): if context['projects']: string += '[cyan]' + ' '.join(['+' + tag for tag in context['projects']]) + '[/cyan] ' if context['contexts']: - string += '[yellow]' + ' '.join(['+' + tag for tag in context['contexts']]) + '[/yellow] ' + string += '[yellow]' + ' '.join(['@' + tag for tag in context['contexts']]) + '[/yellow] ' if context['tags']: - string += '[magenta]' + ' '.join(['+' + tag for tag in context['tags']]) + '[/magenta] ' + string += '[magenta]' + ' '.join(['#' + tag for tag in context['tags']]) + '[/magenta] ' if 'note' in context: string += '[green]\'' + context['note'] + '\'[/green] ' return string @@ -113,12 +114,11 @@ def main(): """ # Command Line Argument Setup p = argp(description="OrvaniTagger (otag)", formatter_class=raw) - p.add_argument("--not-recursive", "-nr", action="store_false", help="(list,clean,purge,export) Include subdirectories in search") - p.add_argument("--no-file-operation", "-nf", action="store_false", help="(copy,move,del) Skip file operation, just apply to context") - p.add_argument("--no-plugins", "-np", action="store_false", help="(list,export) Don't use plug-ins to search file metadata") + p.add_argument("--recursive", "-r", action="store_false", help="(list,clean,purge,export) Include subdirectories in search") + p.add_argument("--file-operation", "-f", action="store_true", help="(copy,move,del) Do file operation, not just context") p.add_argument("--path", "-p", action="store", type=str, help="(list,export) Use custom path", default='') p.add_argument("--exact", "-e", action="store_true", help="(list) Match exact search term") - p.add_argument("--files", "-f", action="store_true", help="(list) Only show files, ignore folders") + p.add_argument("--files", action="store_true", help="(list) Only show files, ignore folders") p.add_argument("--dirs", "-d", action="store_true", help="(list) Only show directories, ignore files") p.add_argument("--either", "-or", action="store_true", help="(list) Combine search terms by OR instead of AND") p.add_argument("--cache", "-c", action="store_true", help="(list) Cache index search (useful with deep directory trees)") @@ -133,10 +133,11 @@ def main(): 'move', 'mv', 'del', 'rm', 'clean', 'c', - 'purge', 'p' + 'purge', 'p', 'write', 'w', 'import', 'i', - 'export', 'e' + 'export', 'e', + 'scan', 's' ]) p.add_argument("file", nargs="?", type=str, help=SUPPRESS) p.add_argument("context", nargs="*", type=str, help=SUPPRESS) @@ -145,7 +146,6 @@ def main(): # Load Orvani orvani = OrvaniTag() - orvani.ENABLE_PLUGINS = args.no_plugins # Parse Arguments and Run otag Methods try: @@ -158,10 +158,13 @@ def main(): console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.") elif args.command in ['remove', 'r']: - if args.file and args.context: + if args.file: fb = orvani.remove(args.file, *args.context) if fb[0] == 0: - console.print(f':wastebasket: Entry {nice_context(fb[2])}successfully removed from file [strong]{fb[1]}[/strong].') + if fb[2] == None: + console.print(f':wastebasket: All entries successfully removed from file [strong]{fb[1]}[/strong].') + else: + console.print(f':wastebasket: Entry {nice_context(fb[2])}successfully removed from file [strong]{fb[1]}[/strong].') elif fb[0] == 1: console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.") elif fb[0] == 2: @@ -170,7 +173,7 @@ def main(): elif args.command in ['list', 'ls', 'l']: query = [args.file] + args.context if query: - r = orvani.list(args.path, args.exact, args.either, args.not_recursive, args.cache, args.delete_cache, *query) + r = orvani.list(args.path, args.exact, args.either, args.recursive, args.cache, args.delete_cache, *query) filter = None if args.files: filter = 'files' @@ -180,7 +183,7 @@ def main(): elif args.command in ['copy', 'cp']: if args.file and args.context: - fb = orvani.move(args.file, args.context[0], True, args.no_file_operation) + fb = orvani.move(args.file, args.context[0], True, args.file_operation) if fb[0] == 0: console.print(f':dancers: File [strong]{fb[1]}[/strong] with entry {nice_context(fb[3])}successfully copied to [strong]{fb[2]}[/strong]') elif fb[0] == 1: @@ -193,7 +196,7 @@ def main(): elif args.command in ['move', 'mv']: if args.file and args.context: - fb = orvani.move(args.file, args.context[0], False, args.no_file_operation) + fb = orvani.move(args.file, args.context[0], False, args.file_operation) if fb[0] == 0: console.print(f':dancers: File [strong]{fb[1]}[/strong] with entry {nice_context(fb[3])}successfully moved to [strong]{fb[2]}[/strong]') elif fb[0] == 1: @@ -206,7 +209,7 @@ def main(): elif args.command in ['del', 'rm']: if args.file: - fb = orvani.delete(args.file, args.no_file_operation) + fb = orvani.delete(args.file, args.file_operation) if fb[0] == 0: status = f':wastebasket: Entry for [strong]{fb[1]}[/strong] successfully removed.' if fb[3]: @@ -248,10 +251,10 @@ def main(): elif args.command in ['clean', 'c']: if not args.file: args.file = '' - fb = orvani.clean(args.file, args.not_recursive) + fb = orvani.clean(args.file, args.recursive) if fb[0] == 0: sub = '' - if args.not_recursive: + if args.recursive: sub = ' and subdirectories' console.print(f':broom: {fb[2]} entries cleaned from [strong]{fb[1]}[/strong]{sub}') @@ -285,8 +288,16 @@ def main(): console.print(f":mag: No entries found in file [strong]{fb[1]}[/strong].") elif fb[0] == 4: console.print(f":thinking_face: File [strong]{fb[1]}[/strong] has wrong format. Cannot import.") + + elif args.command in ['scan', 's']: + if not args.file: + args.file = '' + orvani.scan(args.file, args.recursive) + console.print(f':mag: Finished scanning subdirectories.') + except Exception as e: - console.print(f":scream: An Exception occurred: [italic]{e}[/italic]") + raise Exception + #console.print(f":scream: An Exception occurred: [italic]{e}[/italic]") if __name__ == '__main__': main() diff --git a/orvani/orvani.py b/build/lib/orvanitag/orvani.py similarity index 54% rename from orvani/orvani.py rename to build/lib/orvanitag/orvani.py index 82f670f..0aeaaed 100644 --- a/orvani/orvani.py +++ b/build/lib/orvanitag/orvani.py @@ -7,6 +7,10 @@ from importlib import import_module from types import SimpleNamespace from pprint import pprint +class VerboseSafeDumper(yaml.SafeDumper): + def ignore_aliases(self, data): + return True + # Defaults CONTEXT_FILE = '.context.yaml' @@ -17,6 +21,7 @@ CONTEXTS = { 'tags' : '#', 'note' : None } + DEFAULT_PLUGINS = [ 'frontmatter', 'hashdoubleslash', @@ -25,31 +30,30 @@ DEFAULT_PLUGINS = [ 'pdf' ] +""" +ERRORS: +0 - Success +1 - File Not Found +2 - Entry not found +3 - Path is wrong type +4 - Wrong File Type +5 - Plugin has no method +""" + class OrvaniTag: INDEX = {} # our search index CACHEFILE = '' # search index cache ('.index.yaml') DB = dict() # holds the folder context data - CONFIG = dict() - CONF_FILES = [] - ROOT_CONFIG = dict() + CONFIG = dict() # Currently used configuration + CONF_FILES = [] # List to keep track of config files in the directory tree FILENAME = '' # folder context data filename ('.context.yaml') FILE = '' # holds the folder context file path - CTX = {} - ENABLE_PLUGINS = True - PLUGINS = [] + SCHEMA = {} # Default context schema + ENABLE_PLUGINS = True # Toggle use of plugins + PLUGINS = [] # List of loaded plugins REGEX = r'\s*([@+]\w+|#\w+)' - """ - ERRORS: - 0 - Success - 1 - File Not Found - 2 - Entry not found - 3 - Path is wrong type - 4 - Wrong File Type - 5 - Plugin has no method - """ - def __init__(self, path: str=CONTEXT_FILE, cache: str=CACHE_FILE, contexts=CONTEXTS, plugins:list=DEFAULT_PLUGINS): """ Initialization @@ -57,109 +61,123 @@ class OrvaniTag: path (str): override default filename for folder context cache (str): override default filename for cached index contexts (dict): override the default contexts - """ + self.PLUGINS = plugins self.FILENAME = os.path.basename(path) self.CACHEFILE = cache - self.CTX = contexts - self.INDEX = self.CTX.copy() + self.SCHEMA = contexts + self.INDEX = self.SCHEMA.copy() for context_key in self.INDEX: self.INDEX[context_key] = {} self.INDEX['items'] = {} self.__cd('') - def __cd(self, path: str): + def __cd(self, path: str, config: bool=False): """ 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 - """ + # Helper function to load the custom configuration from a folder context + # If none is found, we check if there is one further up the directory tree def __config(path): - if self.CONF_FILES: + if self.CONF_FILES: # do we have known configurations up the tree? for conf_file in self.CONF_FILES: - if path == '.': - path = './' - if '.' + os.path.commonpath([conf_file, str(path)]) == conf_file: + if path == '.': path = './' + # has current directory has a common tree branch with a known configuration? + common = '.' + os.path.commonpath([conf_file, str(path)]) + if common == '.': common = './' + if common == conf_file: cfile = Path(conf_file) / self.FILENAME if cfile.exists(): conf = yaml.safe_load(cfile.read_text()) if 'config' in conf: self.CONFIG = conf['config'] continue - else: + else: # if there is no config up the tree, then use defaults self.CONFIG = dict() self.FILE = Path(str(path)) / self.FILENAME - if self.FILE.exists(): + if self.FILE.exists(): # is there a folder ocntext file? y = yaml.safe_load(self.FILE.read_text()) + if 'items' in y: - self.DB = y['items'] - if 'config' in y: + self.DB = y['items'] # load items to current context DB + + if 'config' in y: # load optional local configuration self.CONFIG = y['config'] - self.CONF_FILES.append(str(path)) + self.CONF_FILES.append(str(path)) # and save it in our known files list self.CONF_FILES = [x for x in self.CONF_FILES if x != ''] self.CONF_FILES = sorted(list(set(self.CONF_FILES)), key=lambda x: x.count('/'), reverse=True) else: - __config(path) + if config: + __config(path) + else: - self.DB = dict() - __config(path) + self.DB = dict() + else: self.DB = dict() - __config(path) + if config: + __config(path) + def __save(self): """ Save current folder context to file - """ file = Path(self.FILE) - yf = { 'items' : self.DB } - if file.is_file(): - tmp = yaml.safe_load(file.read_text()) + yf = { 'items' : self.DB } # load current context into dict + if file.is_file(): # does current folder have a context file? + tmp = yaml.safe_load(file.read_text()) # if yes, load it... if 'config' in tmp: - yf['config'] = tmp['config'] + yf['config'] = tmp['config'] # ...and save its optional configuration try: - with file.open('w+') as data: - yaml.safe_dump(yf, data) + with file.open('w+') as data: # ...then write it all to the file + yaml.dump(yf, data, Dumper=VerboseSafeDumper) return self.FILE except: return 0 - - def __context(self, argv) -> dict: - """ - Creates a context entry for an existing file - argv (list): sets file contexts according to CTX schema - + def __context(self, argv: list) -> dict: + """ + Creates a context entry for an existing file according to the schema + + argv (list): sets file contexts according to context schema """ context = dict() - for context_key in self.CTX.keys(): - if self.CTX[context_key] != None: + # Create context dict according to schema + for context_key in self.SCHEMA.keys(): + if self.SCHEMA[context_key] != None: context.update( { context_key : []}) - for arg in [x for x in argv if len(x) > 0]: - arg.replace(',', '') - for context_key in self.CTX.keys(): - if str(arg)[0] in self.CTX.values(): + + # Parse the argument list + for arg in [x for x in argv if len(x) > 0]: # remove empty list elements + arg.replace(',', '') # remove commas + + # Add each argument list element to context dict based on the + # prefix symbols defined in the schema + for context_key in self.SCHEMA.keys(): + if str(arg)[0] in self.SCHEMA.values(): # Schema with prefix - if not self.CTX[context_key] == None: - if str(arg).startswith(self.CTX[context_key]): + if not self.SCHEMA[context_key] == None: + if str(arg).startswith(self.SCHEMA[context_key]): context[context_key].append(str(arg)[1:]) else: - # Otherwise it's a note - if self.CTX[context_key] == None: + # No prefix, so it must be a note + if self.SCHEMA[context_key] == None: if arg != ' ': context.update( { context_key : str(arg)}) else: context.update( { context_key : None}) return context + def add(self, file: str, *argv: str): """ Add context to existing filename and saves local folder context file. @@ -169,24 +187,41 @@ class OrvaniTag: """ target = Path(file) - if target.exists(): - self.__cd(target.parent) - context = self.__context(argv) - if target.name in self.DB: - for context_key in self.CTX.keys(): - if self.CTX[context_key] != None: + + if target.exists(): # Only existing files and folders + self.__cd(target.parent) # Load folder context + context = self.__context(argv) # Create context dict from argument list + + if target.name in self.DB: # Does the file exist in our local context DB? + + # Add entries from argument context to context DB + for context_key in self.SCHEMA.keys(): + if self.SCHEMA[context_key] != None: + # Add to context and remove duplicate entries self.DB[target.name][context_key] = list(set( self.DB[target.name][context_key] + context[context_key])) else: + # Add the note if context_key in context: self.DB[target.name][context_key] = context[context_key] - else: + + else: # If file or folder does not exist in context DB yet, add it self.DB.update( { target.name : context } ) + + # Save to file saved = self.__save() if saved and context: return [0, file, context] else: 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 @@ -196,18 +231,29 @@ class OrvaniTag: """ target = Path(file) - if target.exists(): - self.__cd(target.parent) - context = self.__context(argv) - if target.name in self.DB: + all = False + if target.exists(): # Only existing files and folders + self.__cd(target.parent) # Load folder context + if len(argv) == 0: + all = True + context = self.__context(argv) # Create context dict from argument list + + if target.name in self.DB: # Does the file exist in our local context DB? + saved = False - for context_key in self.CTX.keys(): - # Only works for context list items - if self.CTX[context_key] != None: - for item in context[context_key]: - if item in self.DB[target.name][context_key]: - self.DB[target.name][context_key].remove(item) - saved = self.__save() + if all: + del self.DB[target.name] + saved = self.__save() + return [0, file, None] + else: + for context_key in self.SCHEMA.keys(): + # Only works for context list items + if self.SCHEMA[context_key] != None: + for item in context[context_key]: + if item in self.DB[target.name][context_key]: + self.DB[target.name][context_key].remove(item) + saved = self.__save() + if saved and self.DB[target.name]: return [0, file, context] else: @@ -217,6 +263,7 @@ class OrvaniTag: else: return [1, file] + def delete(self, file: str, fop:bool=False): """ Deletes all entries of an existing file and optionally deletes the file @@ -226,21 +273,22 @@ class OrvaniTag: """ target = Path(file) - if target.exists(): - self.__cd(target.parent) + if target.exists(): # Does the target file exist? + self.__cd(target.parent) # Load its local context else: return [1, file] - if target.name in self.DB: - del self.DB[target.name] + if target.name in self.DB: # Do we have a record in the DB? + del self.DB[target.name] # Delete it else: return [2, file] saved = self.__save() - if not self.DB: + if not self.DB: # If local context is empty now, we can delete the file Path(self.FILE).unlink(missing_ok=True) - if fop: + if fop: # delete the referenced file too? target.unlink(missing_ok=True) return [0, file, self.FILE if not self.DB else None, fop] + def move(self, source: str, dest: str, retain: bool=False, fop: bool=True): """ Copies or moves existing file with contexts to a new destination @@ -251,23 +299,39 @@ class OrvaniTag: 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)): + # Let's use trusty old os lib over pathlib for this one + # Check if source file and destination folder exist + if os.path.exists(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] - self.__cd(destfolder) - self.DB.update( { destfile : sourcedata } ) - self.__save() - if not retain: - self.delete(source) + + # If no destination file name was defined, but fop is True + # use the same as the source if fop: + if not destfile: + destfile = sourcefile + else: + if not destfile: + destfile = destfolder + destfolder = os.path.dirname(destfolder) + if not sourcefile: + sourcefile = sourcefolder + sourcefolder = os.path.dirname(sourcefolder) + + sourcedata = {} + self.__cd(sourcefolder) # load local context + if self.DB[sourcefile]: # check if we have an entry for this file + sourcedata = self.DB[sourcefile] # load its context data + if not os.path.isdir(destfile): + self.__cd(destfolder) # load local context of destination folder + self.DB.update( { destfile : sourcedata } ) # update its DB with the context data of our source file + self.__save() # and save it + if not retain: # do we copy or move the data? + self.delete(source) # delete for move + if fop: # do we also physically move or copy the reference file? if retain: shutil.copy2(source, dest) else: @@ -276,8 +340,8 @@ class OrvaniTag: return [0, source, dest, sourcedata] else: return [1, source, dest, os.path.isfile(source), os.path.isdir(os.path.dirname(dest))] - - + + def clean(self, path: str='.', recursive=True): """ Removes context data from folder if files do not exist @@ -292,19 +356,19 @@ class OrvaniTag: if target.is_file(): target = Path(target.parent) if target.is_dir(): - self.__cd(path) - DB = self.DB.copy() - for file in self.DB.keys(): + self.__cd(path) # load folder context + DB = self.DB.copy() # make a copy of it + for file in self.DB.keys(): # check for each entry if the file actually exists f = target / file - if not f.is_file(): + if not f.is_file(): # if not, delete its entry if file in self.DB: del DB[file] count += 1 - self.DB = DB - self.__save() - if not self.DB: + self.DB = DB # make our local working copy of the DB the context DB + self.__save() # save it + if not self.DB: # if it is empty we can delete the context file Path(self.FILE).unlink(missing_ok=True) - if recursive: + if recursive: # do it all recursively? for (root,dirs,files) in os.walk(path): if dirs: for dir in dirs: @@ -313,6 +377,7 @@ class OrvaniTag: count += fb[2] return [0, str(target), count] + def list(self, path: str='./', exact=True, either=False, recursive: bool=True, cache: bool=False, delete_cache: bool=False, *search: str): """ Returns an index object with query results and optionally saves a cached index @@ -325,52 +390,72 @@ class OrvaniTag: search (str): the search query """ - if path == '': + if path == '': # fix empty path argument path = './' - self.__makeindex(path, recursive, cache, delete_cache) - result = self.__search(search, exact, either) + self.__makeindex(path, recursive, cache, delete_cache) # index the directories + result = self.__search(search, exact, either) # apply the search query return result def __makeindex(self, path:str, recursive:bool=True, cache: bool=False, delete_cache: bool=False): """ - Creates index for searching, optionally also a cache file + 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(): + if cache and cachefile.is_file(): # load index from cache self.INDEX = yaml.safe_load(cachefile.read_text()) - else: - self.__indexfolder(path) + 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 self.ENABLE_PLUGINS: - if files: - for file in files: - fpath = os.path.join(root, file) - self.__cd(root) - self.__metadata(fpath) + + 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) - if not recursive: - break - - if cache: + 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): + + # process every file with a reader plugin + if self.ENABLE_PLUGINS: + if files: + for file in files: + fpath = os.path.join(root, file) + self.__cd(root, True) + self.__metadata(fpath) + + if not recursive: # check if to include subdirectories + break + + def __indexfolder(self, sub: str): """ - Indexes a folder context + 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(): @@ -378,40 +463,53 @@ class OrvaniTag: if self.DB: for file in self.DB.keys(): path = os.path.join(sub, file) - for context_key in self.CTX.keys(): + 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']: + 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: + 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 - if type(context[context_key]) == list: + # 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: + 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)) + self.INDEX[context_key][item] = list(set(filelist)) + def __search(self, search, exact:bool=False, either:bool=False): """ @@ -419,24 +517,36 @@ class OrvaniTag: 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]: - query = self.__context(search) - return_dict = self.__context([' ']) - universal = {i for i in self.CTX if self.CTX[i]==None}.pop() - if universal in query: - query[universal] = [query[universal]] - for context_key in self.CTX.keys(): + 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: @@ -449,72 +559,116 @@ class OrvaniTag: 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 + 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} - if len(mresult) < 2: + + 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))]) + + 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): + """ + Add a plugin to the plugin list + """ self.PLUGINS.append(function) 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 != ''] + argv = [x for x in list(set(argv)) if x != ''] # make unique and remove empty items return argv + def __metadata(self, filename: str): + """ + Processes a file with all reader plugins to extract its metadata + """ + + # Utilities for reader plugins obj = { 'file' : Path(), 'config' : self.CONFIG, - 'schema' : self.CTX, + 'schema' : self.SCHEMA, 'context' : self.__context, 'string' : self.get_argv } o = SimpleNamespace(**obj) + for plugin_name in self.PLUGINS: + # Check if plugin is (de)activated by a local folder configuration if 'plugins' in self.CONFIG: if plugin_name not in self.CONFIG['plugins']: continue - plugin = import_module('plugins.' + plugin_name) - suffixes = plugin.TYPES - if plugin_name in self.CONFIG: + # Load the plugin from plugins folder + plugin = import_module('orvanitag.plugins.' + plugin_name) + + # Load a list of file types the plugin is able to read + suffixes = plugin.TYPES # The plugin's default file types + if plugin_name in self.CONFIG: # check for local folder config overrides if 'types' in self.CONFIG[plugin_name]: suffixes = self.CONFIG[plugin_name]['types'] + o.file = Path(filename) - if o.file.suffix in suffixes: - if hasattr(plugin, 'reader'): - context = plugin.reader(o) - if context: - for context_key in context.keys(): - self.__updateindex(filename, context, context_key) + if o.file.suffix in suffixes: # Does the file match a permitted file type + if hasattr(plugin, 'reader'): # Does the plugin have a reader method? + + # Process the file using the reader plugin + try: + context = plugin.reader(o) + if context: + self.__add(filename, context) + print(f'Found Metadata for `{filename}`.') + except Exception as e: + 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: @@ -531,24 +685,56 @@ class OrvaniTag: 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): + + # process every file with a writer plugin + if self.ENABLE_PLUGINS: + if files: + for file in files: + fpath = os.path.join(root, file) + self.__cd(root, True) + self.write(fpath, method) + + if not recursive: # check if to include subdirectories + break + + def write(self, filename: str, method: str): + """ + Writes a file's context entries as metadata to the file using a writer plugin + + filename (str): A valid file path + method (str): The name of a plugin able to process the file type + """ + + # Utilities for writer plugins obj = { 'file' : Path(), 'config' : self.CONFIG, - 'schema' : self.CTX, + 'schema' : self.SCHEMA, 'context' : {}, 'collect' : self.__collect_contexts, 'string' : self.get_argv } o = SimpleNamespace(**obj) + for plugin_name in self.PLUGINS: + + # Check if plugin is (de)activated by a local folder configuration if 'plugins' in self.CONFIG: if plugin_name not in self.CONFIG['plugins']: continue - plugin = import_module('plugins.' + plugin_name) - if plugin.NAME == method: - suffixes = plugin.TYPES - if plugin_name in self.CONFIG: + + plugin = import_module('orvanitag.plugins.' + plugin_name) + + if plugin.NAME == method: # select the reference plugin + + # Load a list of file types the plugin is able to read + suffixes = plugin.TYPES # Load the plugin's defaults + if plugin_name in self.CONFIG: # Check for local folder config overrides if 'types' in self.CONFIG[plugin_name]: suffixes = self.CONFIG[plugin_name]['types'] for suffix in suffixes: @@ -556,26 +742,44 @@ class OrvaniTag: suffixes.remove(suffix) if not suffixes: suffixes = plugin.TYPES + o.file = Path(filename) if o.file.suffix in suffixes: + + # Get all local context entries (including file metadata) self.__makeindex(o.file.parent, recursive=False) items = self.INDEX['items'] + for found in items.keys(): + # If there is a file with matching entries in the local context if str(o.file).replace('./', '') == found.replace('./', ''): o.context = items[found] - if hasattr(plugin, 'writer'): - plugin.writer(o) + if hasattr(plugin, 'writer'): # If the plugin has a writer method + plugin.writer(o) # Write the context as metadata return [0, filename, o.context] else: return [5, plugin_name] else: return [4, suffixes, plugin_name, plugin_name in self.CONFIG] - + + def export(self, filename:str): + """ + Exports the global context index to a text file in a format that can be reimported + + filename (str): optional filename + """ + + # If no filename give, use the default one if filename == '' or filename == None: filename = 'context.txt' + file = Path(filename) + + # Run `otag ls` to create the global context index index = self.list(file.parent, False, False, True, False, False, '') + + # Write the index as lines to file items = index['index']['items'] content = "" if len(items): @@ -586,14 +790,24 @@ class OrvaniTag: return [0, filename] else: return [2] - + + def load(self, filename:str): + """ + Imports lines from a text file created by `otag export` to the global context index, + filtering out the already known entries + + filename (str): optional filename + """ + + # If no filename was given, try the default one if filename == '' or filename == None: filename = 'context.txt' file = Path(filename) if file.exists(): imported = file.read_text().split('\n') if len(imported): + # Make an index of all context items that are already known known = self.list(file.parent, False, False, True, False, False, '')['index']['items'] count = 0 for import_line in imported: @@ -603,10 +817,11 @@ class OrvaniTag: context = {} path = line_elements[0].strip() new_file = Path(path) - if new_file.exists(): + if new_file.exists(): # Check if file exists string = line_elements[1].strip() if 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] context = new_file_context.copy() @@ -620,12 +835,7 @@ class OrvaniTag: if new_context in known_file_context.keys(): if new_file_context[new_context] == known_file_context[new_context]: del context[new_context] - ctx_str = self.__collect_contexts(context) - if ctx_str: - count += 1 - argv = self.get_argv(' '.join(ctx_str)) - print(argv) - self.add(path, *argv) + self.__add(path, context) else: return [4, filename] return [0, filename, count] diff --git a/orvani/plugins/__init__.py b/build/lib/orvanitag/plugins/__init__.py similarity index 100% rename from orvani/plugins/__init__.py rename to build/lib/orvanitag/plugins/__init__.py diff --git a/orvani/plugins/frontmatter.py b/build/lib/orvanitag/plugins/frontmatter.py similarity index 100% rename from orvani/plugins/frontmatter.py rename to build/lib/orvanitag/plugins/frontmatter.py diff --git a/orvani/plugins/hashdoubleslash.py b/build/lib/orvanitag/plugins/hashdoubleslash.py similarity index 100% rename from orvani/plugins/hashdoubleslash.py rename to build/lib/orvanitag/plugins/hashdoubleslash.py diff --git a/orvani/plugins/id3.py b/build/lib/orvanitag/plugins/id3.py similarity index 100% rename from orvani/plugins/id3.py rename to build/lib/orvanitag/plugins/id3.py diff --git a/orvani/plugins/office.py b/build/lib/orvanitag/plugins/office.py similarity index 100% rename from orvani/plugins/office.py rename to build/lib/orvanitag/plugins/office.py diff --git a/orvani/plugins/pdf.py b/build/lib/orvanitag/plugins/pdf.py similarity index 100% rename from orvani/plugins/pdf.py rename to build/lib/orvanitag/plugins/pdf.py diff --git a/context.txt b/context.txt deleted file mode 100644 index 7cfe933..0000000 --- a/context.txt +++ /dev/null @@ -1,12 +0,0 @@ -./COMMANDS.md --- +coding @erfwegf #great #super This is awesome -./README.md --- @frrreee #writeme #fool -./TODO.md --- +bokertov +proooo +fool @bbb @python #ttt #xxx #frrr #rusty Chawera -./test/chirpy.wav --- +parakeetio +sensorwatch @Chripo @watch #funk McFlu McFlu This is a title This is a title -./test/folder with spaces/file with spaces.txt --- #spaces -./test/office.odt --- +oopsie +orvani @this #whoa #that and a note -./test/oldword.doc --- and a note -./test/sheet.xlsx --- +orvani @calc #tessssting -./test/sub/statement_14793553_USD_2022-11-01_2023-01-31.pdf --- +yeeha @pdfcontext #yay #frrrrrrrr And a comment too -./test/table.ods --- @calc #tessssting -./test/tableold.xls --- @calc #tessssting -./test/word.docx --- @concon #tagz #choo what is that smell? diff --git a/context.yaml b/context.yaml deleted file mode 100644 index b893c3b..0000000 --- a/context.yaml +++ /dev/null @@ -1,25 +0,0 @@ -items: - README.md: - contexts: [] - projects: [] - tags: - - writeme - TODO.md: - contexts: - - python - note: Chawera - projects: - - bokertov - tags: - - rusty - requirements.txt: - contexts: [] - note: pip reqs - projects: [] - tags: [] - test: - contexts: - - folder - projects: [] - tags: - - spaces diff --git a/orvanitag.egg-info/PKG-INFO b/orvanitag.egg-info/PKG-INFO new file mode 100644 index 0000000..dbfadbd --- /dev/null +++ b/orvanitag.egg-info/PKG-INFO @@ -0,0 +1,52 @@ +Metadata-Version: 2.1 +Name: orvanitag +Version: 1.0.0 +Summary: Platform independent, file-based, portable tagging system to organize directories and files. +Home-page: https://github.com/randogoth/orvanitag +Author: randogoth +Author-email: randogoth@posteo.org +License: UNKNOWN +Platform: UNKNOWN +Classifier: Programming Language :: Python :: 3 +Classifier: License :: OSI Approved :: MIT License +Classifier: Operating System :: OS Independent +Requires-Python: >=3.6 +Description-Content-Type: text/markdown +License-File: LICENSE + +--- +contexts: +- frrreee +projects: [] +tags: +- fool +- writeme +--- + +# Orvani Tagger + +(From עוֹרְבָנִי, the Hebrew name for th Eurasian Jay) + +Platform independent, folder-based, and portable tagging system for organizing and searching files. The command line interface and organizing principles are heavily inspired by the todo.txt standard. + +Every file can be assigned projects, contexts, and tags and have a descriptive note. + +Exactly like tasks in todo.txt, files can be categorized by freely defined contexts, defined by the +`@` prefix or projects, defined by the `+` prefix. In addition, the `#` prefix defines hashtags that can be assigned. + +So a as an example an audio-file could be assigned to project `+album` and have a context of `@drums` and tagged with `#boombap`, `#lofi`, `#snare`. It's completely up to you how you use the different means of categorization. + +## Command Line Tool + +### Assigning Contextual Tags + +Let's +``` +$ orvani add [contextual tags ...] +``` + +Removing contextual tags from a file +``` +$ orvani remove [contextual tags ...] +``` + diff --git a/orvanitag.egg-info/SOURCES.txt b/orvanitag.egg-info/SOURCES.txt new file mode 100644 index 0000000..827e9cb --- /dev/null +++ b/orvanitag.egg-info/SOURCES.txt @@ -0,0 +1,18 @@ +LICENSE +README.md +setup.py +orvanitag/__init__.py +orvanitag/cmd.py +orvanitag/orvani.py +orvanitag.egg-info/PKG-INFO +orvanitag.egg-info/SOURCES.txt +orvanitag.egg-info/dependency_links.txt +orvanitag.egg-info/entry_points.txt +orvanitag.egg-info/requires.txt +orvanitag.egg-info/top_level.txt +orvanitag/plugins/__init__.py +orvanitag/plugins/frontmatter.py +orvanitag/plugins/hashdoubleslash.py +orvanitag/plugins/id3.py +orvanitag/plugins/office.py +orvanitag/plugins/pdf.py \ No newline at end of file diff --git a/orvanitag.egg-info/dependency_links.txt b/orvanitag.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/orvanitag.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/orvanitag.egg-info/entry_points.txt b/orvanitag.egg-info/entry_points.txt new file mode 100644 index 0000000..2537197 --- /dev/null +++ b/orvanitag.egg-info/entry_points.txt @@ -0,0 +1,3 @@ +[console_scripts] +otag = orvanitag.cmd:main + diff --git a/orvanitag.egg-info/requires.txt b/orvanitag.egg-info/requires.txt new file mode 100644 index 0000000..f229360 --- /dev/null +++ b/orvanitag.egg-info/requires.txt @@ -0,0 +1 @@ +requests diff --git a/orvanitag.egg-info/top_level.txt b/orvanitag.egg-info/top_level.txt new file mode 100644 index 0000000..fc5b9db --- /dev/null +++ b/orvanitag.egg-info/top_level.txt @@ -0,0 +1 @@ +orvanitag diff --git a/orvanitag/__init__.py b/orvanitag/__init__.py new file mode 100644 index 0000000..9409f5b --- /dev/null +++ b/orvanitag/__init__.py @@ -0,0 +1,2 @@ +from .orvani import OrvaniTag +from .cmd import main \ No newline at end of file diff --git a/orvanitag/cmd.py b/orvanitag/cmd.py new file mode 100644 index 0000000..76a1687 --- /dev/null +++ b/orvanitag/cmd.py @@ -0,0 +1,303 @@ +import os +import sys +import math +from argparse import ArgumentParser as argp, SUPPRESS, RawTextHelpFormatter as raw +from pathlib import Path +from rich import box +from rich.console import Console +from rich.table import Table +from orvanitag.orvani import OrvaniTag, CONTEXT_FILE, CACHE_FILE, CONTEXTS + +def mktable(query: str, index: dict, filter=None, trunc_len: int=25): + """Creates a rich table from `otag ls` search result + + query (str): The user query + index (dict): The context index returned from `otag ls` + filter (bool): Show only 'files' or 'dirs' + trunc_len (int): Max. filename length beyond which overflow method kicks in + + """ + console = Console() + + 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) + + # File Path Column Header + table.add_column("Path", style="white", no_wrap=False, overflow='fold', max_width=40) + + # Create Column Headers based on Context Schema + ctx = list(CONTEXTS.keys()) + styles = ["cyan", "yellow", "magenta", "green"] + for i, c in enumerate(ctx): + table.add_column(ctx[i].title(), style=styles[i], overflow='fold') + + # Create Rows Based on Search Results + for file in index['found']: + item = Path(file) + + if filter == 'files' and not item.is_file(): # Only show directories (--dirs flag) + continue + if filter == 'dirs' and not item.is_dir(): # Only show files (--files flag) + continue + + # Our own overflow method for filenames longer than 25 characters (file…name.ext) + long_file_pre = math.floor(trunc_len/2-1) + long_file_post = len(item.name) - math.ceil(trunc_len/2-1) + path = f'{item.parent}/' + ((item.name[:long_file_pre] + '…' + item.name[long_file_post:]) if len(item.name) > trunc_len else item.name) + path = path.replace('./', '') + + fc = index['index']['items'][file] + elements = [] + elements.append(path) + + # Fetch contextual data for each file to add to a table row + for c in ctx: + if not c in fc: + fc[c] = None + if fc[c]: + if CONTEXTS[c]: + prefix = CONTEXTS[c] + else: + prefix = '' + if not type(fc[c]) == str: # Add respective context prefix + elements.append(' '.join([prefix + tag for tag in fc[c]])) + else: # Except for notes + elements.append(fc[c]) + else: + elements.append(None) + + table.add_row(*elements) + console.print(table) + else: + console.print(f":thinking_face: [red]No results found for [white]{query}") + +def nice_context(context: dict): + """ + Rich colors for displaying contextual data + + context (dict): context dict + """ + string = '' + if context['projects']: + string += '[cyan]' + ' '.join(['+' + tag for tag in context['projects']]) + '[/cyan] ' + if context['contexts']: + string += '[yellow]' + ' '.join(['@' + tag for tag in context['contexts']]) + '[/yellow] ' + if context['tags']: + string += '[magenta]' + ' '.join(['#' + tag for tag in context['tags']]) + '[/magenta] ' + if 'note' in context: + string += '[green]\'' + context['note'] + '\'[/green] ' + return string + +def main(): + """otag CLI""" + + # Rich Console + console = Console() + + # Detailed Help Text for the --help flag + command_help = """{add,a} FILE CONTEXT # add context to an existing file +{remove,r} FILE CONTEXT # remove context from an existing file +{list,ls,l} QUERY # list all files with queried context in current directory tree + +{copy,cp} SOURCE DEST # copy file and its context to another directory +{move,mv} SOURCE DEST # move file and its context to another directory +{del,rm} FILE # delete file and its context + +{clean,c} PATH # remove all context for nonexisting files +{purge,p} PATH # remove all context for all files in current directory tree + +{write,w} PLUGIN FILE # write all context of a file to the file as metadata +{import,i} FILE # import context for files from a text file +{export,e} FILE # export context for files to a text file +""" + # Command Line Argument Setup + p = argp(description="OrvaniTagger (otag)", formatter_class=raw) + p.add_argument("--recursive", "-r", action="store_false", help="(list,clean,purge,export) Include subdirectories in search") + p.add_argument("--file-operation", "-f", action="store_true", help="(copy,move,del) Do file operation, not just context") + p.add_argument("--path", "-p", action="store", type=str, help="(list,export) Use custom path", default='') + p.add_argument("--exact", "-e", action="store_true", help="(list) Match exact search term") + p.add_argument("--files", action="store_true", help="(list) Only show files, ignore folders") + p.add_argument("--dirs", "-d", action="store_true", help="(list) Only show directories, ignore files") + p.add_argument("--either", "-or", action="store_true", help="(list) Combine search terms by OR instead of AND") + p.add_argument("--cache", "-c", action="store_true", help="(list) Cache index search (useful with deep directory trees)") + p.add_argument("--delete-cache", "-dc", action="store_true", help="(list) Delete cache file") + p.add_argument("command", nargs="?", default=None, metavar='Command', + help=command_help, type=str, + choices=[ + 'add', 'a', + 'remove', 'r', + 'list', 'ls', 'l', + 'copy', 'cp', + 'move', 'mv', + 'del', 'rm', + 'clean', 'c', + 'purge', 'p', + 'write', 'w', + 'import', 'i', + 'export', 'e', + 'scan', 's' + ]) + p.add_argument("file", nargs="?", type=str, help=SUPPRESS) + p.add_argument("context", nargs="*", type=str, help=SUPPRESS) + p.print_usage = p.print_help + args = p.parse_args() + + # Load Orvani + orvani = OrvaniTag() + + # Parse Arguments and Run otag Methods + try: + if args.command in ['add', 'a']: + if args.file and args.context: + fb = orvani.add(args.file, *args.context) + if fb[0] == 0: + console.print(f':label: Entry {nice_context(fb[2])}successfully added to file [strong]{fb[1]}[/strong].') + elif fb[0] == 1: + console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.") + + elif args.command in ['remove', 'r']: + if args.file: + fb = orvani.remove(args.file, *args.context) + if fb[0] == 0: + if fb[2] == None: + console.print(f':wastebasket: All entries successfully removed from file [strong]{fb[1]}[/strong].') + else: + console.print(f':wastebasket: Entry {nice_context(fb[2])}successfully removed from file [strong]{fb[1]}[/strong].') + elif fb[0] == 1: + console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.") + elif fb[0] == 2: + console.print(f":mag: No match found for [white][strong]{fb[1]}[/strong][/white] and {nice_context(fb[2])}") + + elif args.command in ['list', 'ls', 'l']: + query = [args.file] + args.context + if query: + r = orvani.list(args.path, args.exact, args.either, args.recursive, args.cache, args.delete_cache, *query) + filter = None + if args.files: + filter = 'files' + elif args.dirs: + filter = 'dirs' + mktable(' '.join(sys.argv[1:]), r, filter) + + elif args.command in ['copy', 'cp']: + if args.file and args.context: + fb = orvani.move(args.file, args.context[0], True, args.file_operation) + if fb[0] == 0: + console.print(f':dancers: File [strong]{fb[1]}[/strong] with entry {nice_context(fb[3])}successfully copied to [strong]{fb[2]}[/strong]') + elif fb[0] == 1: + if fb[3] == False and fb[4] == True: + console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] doesn't exist.") + elif fb[3] == True and fb[4] == False: + console.print(f":thinking_face: Destination [strong]{fb[2]}[/strong] doesn't exist.") + elif fb[3] == False and fb[4] == False: + console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] and destination [strong]{fb[2]}[/strong] don't exist.") + + elif args.command in ['move', 'mv']: + if args.file and args.context: + fb = orvani.move(args.file, args.context[0], False, args.file_operation) + if fb[0] == 0: + console.print(f':dancers: File [strong]{fb[1]}[/strong] with entry {nice_context(fb[3])}successfully moved to [strong]{fb[2]}[/strong]') + elif fb[0] == 1: + if fb[3] == False and fb[4] == True: + console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] doesn't exist.") + elif fb[3] == True and fb[4] == False: + console.print(f":thinking_face: Destination [strong]{fb[2]}[/strong] doesn't exist.") + elif fb[3] == False and fb[4] == False: + console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] and destination [strong]{fb[2]}[/strong] don't exist.") + + elif args.command in ['del', 'rm']: + if args.file: + fb = orvani.delete(args.file, args.file_operation) + if fb[0] == 0: + status = f':wastebasket: Entry for [strong]{fb[1]}[/strong] successfully removed.' + if fb[3]: + status += f'\n:wastebasket: File [strong]{fb[1]}[/strong] successfully deleted.' + if not fb[2] == None: + status += f'\n:wastebasket: File [strong]{fb[2]}[/strong] automatically deleted.' + console.print(status) + elif fb[0] == 1: + console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.") + elif fb[0] == 2: + console.print(f":mag: No entry for [strong]{fb[1]}[/strong] found.") + + + elif args.command in ['purge', 'p']: + + def purge(path): + target = Path(path) + if target.is_file(): + target = Path(target.parent) + if target.is_dir(): + context_file = target / CONTEXT_FILE + if context_file.exists(): + context_file.unlink(missing_ok=True) + console.print(f":wastebasket: File [strong]{context_file}[/strong] successfully purged.") + cache_file = target / CACHE_FILE + if cache_file.exists(): + cache_file.unlink(missing_ok=True) + console.print(f":wastebasket: File [strong]{cache_file}[/strong] successfully purged.") + + if not args.file: + args.file = '.' + purge(args.file) + for (root,dirs,files) in os.walk(args.file): + if dirs: + for dir in dirs: + pp = os.path.join(root, dir) + purge(pp) + + elif args.command in ['clean', 'c']: + if not args.file: + args.file = '' + fb = orvani.clean(args.file, args.recursive) + if fb[0] == 0: + sub = '' + if args.recursive: + sub = ' and subdirectories' + console.print(f':broom: {fb[2]} entries cleaned from [strong]{fb[1]}[/strong]{sub}') + + elif args.command in ['write', 'w']: + if args.file and args.context: + fb = orvani.write(args.context[0], args.file) + if fb[0] == 0: + console.print(f':floppy_disk: Metadata {nice_context(fb[2])}successfully written to file [strong]{fb[1]}[/strong].') + if fb[0] == 4: + console.print(f':card_file_box: Plugin [strong]{fb[2]}[/strong] can only write to files types: [strong]{"[/strong], [strong]".join(fb[1])}[/strong].') + if fb[3]: + console.print(f':information: Plugin configuration found in folder context file.') + if fb[0] == 5: + console.print(f':disappointed: Plugin [strong]{fb[1]}[/strong] can\'t write.') + + elif args.command in ['export', 'e']: + fb = orvani.export(args.file) + if fb[0] == 0: + console.print(f':floppy_disk: All entries successfully written to file [strong]{fb[1]}[/strong].') + elif fb[0] == 2: + console.print(f":mag: No entries found.") + + elif args.command in ['import', 'i']: + + fb = orvani.load(args.file) + if fb[0] == 0: + console.print(f':inbox_tray: [strong]{fb[2]}[/strong] entries successfully imported from file [strong]{fb[1]}[/strong].') + elif fb[0] == 1: + console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.") + elif fb[0] == 2: + console.print(f":mag: No entries found in file [strong]{fb[1]}[/strong].") + elif fb[0] == 4: + console.print(f":thinking_face: File [strong]{fb[1]}[/strong] has wrong format. Cannot import.") + + elif args.command in ['scan', 's']: + if not args.file: + args.file = '' + orvani.scan(args.file, args.recursive) + console.print(f':mag: Finished scanning subdirectories.') + + except Exception as e: + raise Exception + #console.print(f":scream: An Exception occurred: [italic]{e}[/italic]") + +if __name__ == '__main__': + main() diff --git a/orvanitag/orvani.py b/orvanitag/orvani.py new file mode 100644 index 0000000..0aeaaed --- /dev/null +++ b/orvanitag/orvani.py @@ -0,0 +1,845 @@ +import os +import re +import yaml +import shutil +from pathlib import Path +from importlib import import_module +from types import SimpleNamespace +from pprint import pprint + +class VerboseSafeDumper(yaml.SafeDumper): + def ignore_aliases(self, data): + return True + +# Defaults + +CONTEXT_FILE = '.context.yaml' +CACHE_FILE = '.index.yaml' +CONTEXTS = { + 'projects' : '+', + 'contexts' : '@', + 'tags' : '#', + 'note' : None +} + +DEFAULT_PLUGINS = [ + 'frontmatter', + 'hashdoubleslash', + 'id3', + 'office', + 'pdf' +] + +""" +ERRORS: +0 - Success +1 - File Not Found +2 - Entry not found +3 - Path is wrong type +4 - Wrong File Type +5 - Plugin has no method +""" + +class OrvaniTag: + + INDEX = {} # our search index + CACHEFILE = '' # search index cache ('.index.yaml') + DB = dict() # holds the folder context data + CONFIG = dict() # Currently used configuration + CONF_FILES = [] # List to keep track of config files in the directory tree + FILENAME = '' # folder context data filename ('.context.yaml') + FILE = '' # holds the folder context file path + SCHEMA = {} # Default context schema + ENABLE_PLUGINS = True # Toggle use of plugins + PLUGINS = [] # List of loaded plugins + REGEX = r'\s*([@+]\w+|#\w+)' + + def __init__(self, path: str=CONTEXT_FILE, cache: str=CACHE_FILE, contexts=CONTEXTS, plugins:list=DEFAULT_PLUGINS): + """ + Initialization + + path (str): override default filename for folder context + cache (str): override default filename for cached index + contexts (dict): override the default contexts + """ + + self.PLUGINS = plugins + self.FILENAME = os.path.basename(path) + self.CACHEFILE = cache + self.SCHEMA = contexts + self.INDEX = self.SCHEMA.copy() + for context_key in self.INDEX: + self.INDEX[context_key] = {} + self.INDEX['items'] = {} + self.__cd('') + + + def __cd(self, path: str, config: bool=False): + """ + 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 + """ + + # Helper function to load the custom configuration from a folder context + # If none is found, we check if there is one further up the directory tree + def __config(path): + if self.CONF_FILES: # do we have known configurations up the tree? + for conf_file in self.CONF_FILES: + if path == '.': path = './' + # has current directory has a common tree branch with a known configuration? + common = '.' + os.path.commonpath([conf_file, str(path)]) + if common == '.': common = './' + if common == conf_file: + cfile = Path(conf_file) / self.FILENAME + if cfile.exists(): + conf = yaml.safe_load(cfile.read_text()) + if 'config' in conf: + self.CONFIG = conf['config'] + continue + else: # if there is no config up the tree, then use defaults + self.CONFIG = dict() + + self.FILE = Path(str(path)) / self.FILENAME + if self.FILE.exists(): # is there a folder ocntext file? + y = yaml.safe_load(self.FILE.read_text()) + + if 'items' in y: + self.DB = y['items'] # load items to current context DB + + if 'config' in y: # load optional local configuration + self.CONFIG = y['config'] + self.CONF_FILES.append(str(path)) # and save it in our known files list + self.CONF_FILES = [x for x in self.CONF_FILES if x != ''] + self.CONF_FILES = sorted(list(set(self.CONF_FILES)), key=lambda x: x.count('/'), reverse=True) + else: + if config: + __config(path) + + else: + self.DB = dict() + + else: + self.DB = dict() + if config: + __config(path) + + + def __save(self): + """ + Save current folder context to file + """ + file = Path(self.FILE) + yf = { 'items' : self.DB } # load current context into dict + if file.is_file(): # does current folder have a context file? + tmp = yaml.safe_load(file.read_text()) # if yes, load it... + if 'config' in tmp: + yf['config'] = tmp['config'] # ...and save its optional configuration + try: + with file.open('w+') as data: # ...then write it all to the file + yaml.dump(yf, data, Dumper=VerboseSafeDumper) + return self.FILE + except: + return 0 + + + def __context(self, argv: list) -> dict: + """ + Creates a context entry for an existing file according to the schema + + argv (list): sets file contexts according to context schema + """ + context = dict() + # Create context dict according to schema + for context_key in self.SCHEMA.keys(): + if self.SCHEMA[context_key] != None: + context.update( { context_key : []}) + + # Parse the argument list + for arg in [x for x in argv if len(x) > 0]: # remove empty list elements + arg.replace(',', '') # remove commas + + # Add each argument list element to context dict based on the + # prefix symbols defined in the schema + for context_key in self.SCHEMA.keys(): + if str(arg)[0] in self.SCHEMA.values(): + # Schema with prefix + if not self.SCHEMA[context_key] == None: + if str(arg).startswith(self.SCHEMA[context_key]): + context[context_key].append(str(arg)[1:]) + else: + # No prefix, so it must be a note + if self.SCHEMA[context_key] == None: + if arg != ' ': + context.update( { context_key : str(arg)}) + else: + context.update( { context_key : None}) + 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 + + """ + target = Path(file) + + if target.exists(): # Only existing files and folders + self.__cd(target.parent) # Load folder context + context = self.__context(argv) # Create context dict from argument list + + if target.name in self.DB: # Does the file exist in our local context DB? + + # Add entries from argument context to context DB + for context_key in self.SCHEMA.keys(): + if self.SCHEMA[context_key] != None: + # Add to context and remove duplicate entries + self.DB[target.name][context_key] = list(set( self.DB[target.name][context_key] + context[context_key])) + else: + # Add the note + if context_key in context: + self.DB[target.name][context_key] = context[context_key] + + else: # If file or folder does not exist in context DB yet, add it + self.DB.update( { target.name : context } ) + + # Save to file + saved = self.__save() + if saved and context: + return [0, file, context] + else: + 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 + + file (str): path to existing file + argv (list): context parameters to be removed + + """ + target = Path(file) + all = False + if target.exists(): # Only existing files and folders + self.__cd(target.parent) # Load folder context + if len(argv) == 0: + all = True + context = self.__context(argv) # Create context dict from argument list + + if target.name in self.DB: # Does the file exist in our local context DB? + + saved = False + if all: + del self.DB[target.name] + saved = self.__save() + return [0, file, None] + else: + for context_key in self.SCHEMA.keys(): + # Only works for context list items + if self.SCHEMA[context_key] != None: + for item in context[context_key]: + if item in self.DB[target.name][context_key]: + self.DB[target.name][context_key].remove(item) + saved = self.__save() + + if saved and self.DB[target.name]: + return [0, file, context] + else: + return [2, file, context] + else: + return [2, file, context] + else: + return [1, file] + + + 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 + + """ + target = Path(file) + if target.exists(): # Does the target file exist? + self.__cd(target.parent) # Load its local context + else: + return [1, file] + if target.name in self.DB: # Do we have a record in the DB? + del self.DB[target.name] # Delete it + else: + return [2, file] + saved = self.__save() + if not self.DB: # If local context is empty now, we can delete the file + Path(self.FILE).unlink(missing_ok=True) + if fop: # delete the referenced file too? + target.unlink(missing_ok=True) + return [0, file, self.FILE if not self.DB else None, fop] + + + 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 + + """ + # Let's use trusty old os lib over pathlib for this one + # Check if source file and destination folder exist + if os.path.exists(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 no destination file name was defined, but fop is True + # use the same as the source + if fop: + if not destfile: + destfile = sourcefile + else: + if not destfile: + destfile = destfolder + destfolder = os.path.dirname(destfolder) + if not sourcefile: + sourcefile = sourcefolder + sourcefolder = os.path.dirname(sourcefolder) + + sourcedata = {} + self.__cd(sourcefolder) # load local context + if self.DB[sourcefile]: # check if we have an entry for this file + sourcedata = self.DB[sourcefile] # load its context data + if not os.path.isdir(destfile): + self.__cd(destfolder) # load local context of destination folder + self.DB.update( { destfile : sourcedata } ) # update its DB with the context data of our source file + self.__save() # and save it + if not retain: # do we copy or move the data? + self.delete(source) # delete for move + if fop: # do we also physically move or copy the reference file? + if retain: + shutil.copy2(source, dest) + else: + shutil.copy2(source, dest) + os.remove(source) + return [0, source, dest, sourcedata] + else: + return [1, source, dest, os.path.isfile(source), os.path.isdir(os.path.dirname(dest))] + + + def clean(self, path: str='.', recursive=True): + """ + Removes context data from folder if files do not exist + + path (str): folder path + + """ + if path == '': + path = '.' + target = Path(path) + count = 0 + if target.is_file(): + target = Path(target.parent) + if target.is_dir(): + self.__cd(path) # load folder context + DB = self.DB.copy() # make a copy of it + for file in self.DB.keys(): # check for each entry if the file actually exists + f = target / file + if not f.is_file(): # if not, delete its entry + if file in self.DB: + del DB[file] + count += 1 + self.DB = DB # make our local working copy of the DB the context DB + self.__save() # save it + if not self.DB: # if it is empty we can delete the context file + Path(self.FILE).unlink(missing_ok=True) + if recursive: # do it all recursively? + for (root,dirs,files) in os.walk(path): + if dirs: + for dir in dirs: + rpath = os.path.join(root, dir) + fb = self.clean(rpath, False) + count += fb[2] + return [0, str(target), count] + + + def list(self, path: str='./', exact=True, either=False, recursive: bool=True, cache: bool=False, delete_cache: bool=False, *search: str): + """ + Returns an index object with query results and optionally saves a cached index + + 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 + search (str): the search query + + """ + if path == '': # fix empty path argument + path = './' + self.__makeindex(path, recursive, cache, delete_cache) # index the directories + result = self.__search(search, exact, either) # apply the search query + 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): + + # process every file with a reader plugin + if self.ENABLE_PLUGINS: + if files: + for file in files: + fpath = os.path.join(root, file) + self.__cd(root, True) + self.__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): + """ + Add a plugin to the plugin list + """ + self.PLUGINS.append(function) + 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): + """ + Processes a file with all reader plugins to extract its metadata + """ + + # Utilities for reader plugins + obj = { + 'file' : Path(), + 'config' : self.CONFIG, + 'schema' : self.SCHEMA, + 'context' : self.__context, + 'string' : self.get_argv + } + o = SimpleNamespace(**obj) + + for plugin_name in self.PLUGINS: + # Check if plugin is (de)activated by a local folder configuration + if 'plugins' in self.CONFIG: + if plugin_name not in self.CONFIG['plugins']: + continue + # Load the plugin from plugins folder + plugin = import_module('orvanitag.plugins.' + plugin_name) + + # Load a list of file types the plugin is able to read + suffixes = plugin.TYPES # The plugin's default file types + if plugin_name in self.CONFIG: # check for local folder config overrides + if 'types' in self.CONFIG[plugin_name]: + suffixes = self.CONFIG[plugin_name]['types'] + + o.file = Path(filename) + if o.file.suffix in suffixes: # Does the file match a permitted file type + if hasattr(plugin, 'reader'): # Does the plugin have a reader method? + + # Process the file using the reader plugin + try: + context = plugin.reader(o) + if context: + self.__add(filename, context) + print(f'Found Metadata for `{filename}`.') + except Exception as e: + 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): + + # process every file with a writer plugin + if self.ENABLE_PLUGINS: + if files: + for file in files: + fpath = os.path.join(root, file) + self.__cd(root, True) + self.write(fpath, method) + + if not recursive: # check if to include subdirectories + break + + + def write(self, filename: str, method: str): + """ + Writes a file's context entries as metadata to the file using a writer plugin + + filename (str): A valid file path + method (str): The name of a plugin able to process the file type + """ + + # Utilities for writer plugins + obj = { + 'file' : Path(), + 'config' : self.CONFIG, + 'schema' : self.SCHEMA, + 'context' : {}, + 'collect' : self.__collect_contexts, + 'string' : self.get_argv + } + o = SimpleNamespace(**obj) + + for plugin_name in self.PLUGINS: + + # Check if plugin is (de)activated by a local folder configuration + if 'plugins' in self.CONFIG: + if plugin_name not in self.CONFIG['plugins']: + continue + + plugin = import_module('orvanitag.plugins.' + plugin_name) + + if plugin.NAME == method: # select the reference plugin + + # Load a list of file types the plugin is able to read + suffixes = plugin.TYPES # Load the plugin's defaults + if plugin_name in self.CONFIG: # Check for local folder config overrides + if 'types' in self.CONFIG[plugin_name]: + suffixes = self.CONFIG[plugin_name]['types'] + for suffix in suffixes: + if not suffix in plugin.TYPES: + suffixes.remove(suffix) + if not suffixes: + suffixes = plugin.TYPES + + o.file = Path(filename) + if o.file.suffix in suffixes: + + # Get all local context entries (including file metadata) + self.__makeindex(o.file.parent, recursive=False) + items = self.INDEX['items'] + + for found in items.keys(): + # If there is a file with matching entries in the local context + if str(o.file).replace('./', '') == found.replace('./', ''): + o.context = items[found] + if hasattr(plugin, 'writer'): # If the plugin has a writer method + plugin.writer(o) # Write the context as metadata + return [0, filename, o.context] + else: + return [5, plugin_name] + else: + return [4, suffixes, plugin_name, plugin_name in self.CONFIG] + + + def export(self, filename:str): + """ + Exports the global context index to a text file in a format that can be reimported + + filename (str): optional filename + """ + + # If no filename give, use the default one + if filename == '' or filename == None: + filename = 'context.txt' + + file = Path(filename) + + # Run `otag ls` to create the global context index + index = self.list(file.parent, False, False, True, False, False, '') + + # Write the index as lines to file + items = index['index']['items'] + content = "" + if len(items): + for path in items: + string = ' '.join([x for x in self.__collect_contexts(items[path]) if x != None]) + content += f'{path} --- {string}\n' + file.write_text(content) + return [0, filename] + else: + return [2] + + + def load(self, filename:str): + """ + Imports lines from a text file created by `otag export` to the global context index, + filtering out the already known entries + + filename (str): optional filename + """ + + # If no filename was given, try the default one + if filename == '' or filename == None: + filename = 'context.txt' + file = Path(filename) + if file.exists(): + imported = file.read_text().split('\n') + if len(imported): + # Make an index of all context items that are already known + known = self.list(file.parent, False, False, True, False, False, '')['index']['items'] + count = 0 + for import_line in imported: + if len(import_line) > 5: + line_elements = import_line.split(' --- ') + if len(line_elements) > 1: + context = {} + path = line_elements[0].strip() + new_file = Path(path) + if new_file.exists(): # Check if file exists + string = line_elements[1].strip() + if 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] + context = new_file_context.copy() + for new_context in new_file_context: + if type(new_file_context[new_context]) == list: + for context_item in new_file_context[new_context]: + if new_context in known_file_context.keys(): + if context_item in known_file_context[new_context]: + context[new_context].remove(context_item) + else: + if new_context in known_file_context.keys(): + if new_file_context[new_context] == known_file_context[new_context]: + del context[new_context] + self.__add(path, context) + else: + return [4, filename] + return [0, filename, count] + else: + return [2, filename] + else: + return [1, filename] \ No newline at end of file diff --git a/test/test.txt b/orvanitag/plugins/__init__.py similarity index 100% rename from test/test.txt rename to orvanitag/plugins/__init__.py diff --git a/orvanitag/plugins/frontmatter.py b/orvanitag/plugins/frontmatter.py new file mode 100644 index 0000000..ea8e13b --- /dev/null +++ b/orvanitag/plugins/frontmatter.py @@ -0,0 +1,22 @@ +import frontmatter +from pathlib import Path + +NAME = 'frontmatter' +TYPES = ['.md', '.txt', '.markdown', '.rst'] + +def reader(o): + with open(o.file, 'r') as data: + y = frontmatter.load(data) + if y.metadata: + context = y.metadata.copy() + for meta in y.metadata.keys(): + if not meta in o.schema: + del context[meta] + return context + +def writer(o): + y = frontmatter.load(o.file) + for key in o.context: + y[key] = o.context[key] + with open(o.file, 'w') as data: + data.write(frontmatter.dumps(y)) \ No newline at end of file diff --git a/orvanitag/plugins/hashdoubleslash.py b/orvanitag/plugins/hashdoubleslash.py new file mode 100644 index 0000000..28e8130 --- /dev/null +++ b/orvanitag/plugins/hashdoubleslash.py @@ -0,0 +1,63 @@ +import re +from pathlib import Path + +NAME = 'hash//' +REGEX = r'\s*([@+]\w+|#\w+(?'], + '.java' : MLC_C, + '.js' : MLC_C, + '.kt' : MLC_C, + '.md' : ['---', '---'], + '.php' : MLC_C, + '.py' : MLC_Python, + '.rb' : MLC_Perl, + '.rs' : MLC_C, + '.ts' : MLC_C, + '.txt' : ['---', '---'], +} +TYPES = MAP.keys() + +def reader(o): + with o.file.open() as data: + for line_num, line in enumerate(data, start=1): + if line_num > 5: + break + if line.strip().startswith("#// "): + argv = o.string(line.strip()[3:]) + context = o.context(argv) + return context + +def writer(o): + string = PREFIX + ' '.join(o.collect(o.context)) + '\n' + with o.file.open() as data: + content = data.readlines() + num = 0 + line = '' + if len(content): + for num, line in enumerate(content): + if line.strip().startswith("#// ") or num > 3: + break + if not num > 3 and line.strip().startswith("#// "): + content[num] = string + with o.file.open('w') as file: + file.writelines(content) + else: + frontmatter = MAP[o.file.suffix][0] + '\n' + string + MAP[o.file.suffix][1] + '\n' + with o.file.open('r+') as file: + content = file.read() + file.seek(0, 0) + file.write(frontmatter + content) \ No newline at end of file diff --git a/orvanitag/plugins/id3.py b/orvanitag/plugins/id3.py new file mode 100644 index 0000000..f43e60e --- /dev/null +++ b/orvanitag/plugins/id3.py @@ -0,0 +1,42 @@ +import re +from pathlib import Path +import taglib + +NAME = 'id3' +TYPES = ['.mp3', '.wav', '.flac', '.ogg'] +REGEX = r'\s*([@+]\w+|#\w+)' + +def reader(o): + universal = {i for i in o.schema if o.schema[i]==None}.pop() + with taglib.File(o.file, save_on_exit=False) as song: + metadata = song.tags + id3map = {} + if 'id3' in o.config: + if 'map' in o.config['id3']: + id3map = o.config['id3']['map'].copy() + for context_key in id3map: + if context_key not in o.schema.keys(): + del id3map[context_key] + else: + ctag = str(id3map[context_key]).upper() + if ctag in metadata: + id3map[context_key] = metadata[ctag] + if 'COMMENT' in metadata: + comment = metadata['COMMENT'] + argv = o.string(comment[0].strip()) + context = o.context(argv) + if not universal in context: + context[universal] = [] + else: + context[universal] = [context[universal]] + if id3map.keys(): + for key in id3map.keys(): + if key in context.keys(): + context[key] += id3map[key] + context[universal] = '; '.join(context[universal]) + return context + +def writer(o): + string = ' '.join(o.collect(o.context)) + with taglib.File(o.file, save_on_exit=True) as song: + song.tags['COMMENT'] = [string] \ No newline at end of file diff --git a/orvanitag/plugins/office.py b/orvanitag/plugins/office.py new file mode 100644 index 0000000..3f1d255 --- /dev/null +++ b/orvanitag/plugins/office.py @@ -0,0 +1,105 @@ +import os +import re +import mimetypes +import shutil +from pathlib import Path +from olefile import OleFileIO + +NAME = 'office' +TYPES = ['.doc', '.xls', '.ppt', '.docx', '.xlsx', '.pptx', '.odt', '.ods', '.odp'] +XML_REGEX = r"(?P
)(?P.*?)(?P<\/dc:description>)"
+REGEX = r'\s*([@+]\w+|#\w+)'
+
+# Office Document processing inspired by
+# https://github.com/kappapiana/anonymize/blob/master/anonymize.py
+
+def unzip_file(orig_file, tmp_dir):
+    type = mimetypes.guess_type(orig_file)
+    filetype = {
+        "application/vnd.oasis.opendocument.text" : "odt",
+        "application/vnd.oasis.opendocument.spreadsheet" : "ods",
+        "application/vnd.oasis.opendocument.presentation" : "odp",
+        "application/vnd.openxmlformats-officedocument.wordprocessingml.document" : "docx",
+        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" : "xlsx",
+        "application/vnd.openxmlformats-officedocument.presentationml.presentation" : "pptx",
+    }
+    if type[0] in filetype:
+        shutil.unpack_archive(orig_file, tmp_dir, 'zip')
+        return filetype[type[0]]
+    else:
+        return type[0]
+
+def get_context_string(meta_file):
+    ctx = ''
+    meta_content = meta_file.read_text()
+    csearch = re.search(XML_REGEX, meta_content)
+    if csearch:
+        return csearch.group('body')
+    
+
+def reader(o):
+    tmp_folder = Path('unzip__' + o.file.stem)
+    file_type = unzip_file(o.file, tmp_folder)
+    tags = note = []
+
+    # OpenOffice / LibreOffice / Open Document Format
+    if file_type in ['odt', 'ods', 'odp']:
+        meta_file = tmp_folder / 'meta.xml'
+        argv = o.string(get_context_string(meta_file))
+        context = o.context(argv)
+        shutil.rmtree(tmp_folder)
+        return context
+
+    # Microsoft Office Format
+    elif file_type in ['docx', 'xlsx', 'pptx']:
+        meta_file = tmp_folder / 'docProps' / 'core.xml'
+        argv = o.string(get_context_string(meta_file))
+        context = o.context(argv)
+        shutil.rmtree(tmp_folder)
+        return context
+
+    # Legacy Microsoft Office Format
+    elif file_type in ['application/msword', 'application/vnd.ms-excel', 'application/vnd.ms-powerpoint']:
+        ole = OleFileIO(o.file)
+        meta = ole.get_metadata()
+        ole.close()
+        if meta.comments:
+            argv = o.string(meta.comments.decode('UTF-8'))
+            context = o.context(argv)
+            return context
+    
+
+def writer(o):
+    string = ' '.join(o.collect(o.context))
+    tmp_folder = Path('unzip__' + o.file.stem)
+    file_type = unzip_file(o.file, tmp_folder)
+    regex = re.compile(XML_REGEX)
+    string = f"\\g
{string}\\g"
+
+    # OpenOffice / LibreOffice / Open Document Format
+    if file_type in ['odt', 'ods', 'odp']:
+        meta_file = tmp_folder / 'meta.xml'
+        with meta_file.open('r+') as file:
+            content = regex.sub(string, file.read())
+            file.seek(0, 0)
+            file.write(content)
+        shutil.make_archive(o.file, "zip", tmp_folder)
+        output_file_zip = str(o.file) + ".zip"
+        os.rename(output_file_zip, o.file)
+        shutil.rmtree(tmp_folder)
+
+    # Microsoft Office Format
+    elif file_type in ['docx', 'xlsx', 'pptx']:
+        meta_file = tmp_folder / 'docProps' / 'core.xml'
+        with meta_file.open('r+') as file:
+            content = regex.sub(string, file.read())
+            file.seek(0, 0)
+            file.write(content)
+        shutil.make_archive(o.file, "zip", tmp_folder)
+        output_file_zip = str(o.file) + ".zip"
+        os.rename(output_file_zip, o.file)
+        shutil.rmtree(tmp_folder) 
+
+    # Legacy Microsoft Office Format
+    elif file_type in ['application/msword', 'application/vnd.ms-excel', 'application/vnd.ms-powerpoint']:
+        print('unsupported file format')
diff --git a/orvanitag/plugins/pdf.py b/orvanitag/plugins/pdf.py
new file mode 100644
index 0000000..1b30884
--- /dev/null
+++ b/orvanitag/plugins/pdf.py
@@ -0,0 +1,24 @@
+import re
+import shutil
+from pathlib import Path
+from pdfrw import PdfReader, PdfWriter
+
+NAME = 'pdf'
+TYPES = ['.pdf']
+REGEX = r'\s*([@+]\w+|#\w+)'
+
+def reader(o):
+    pdf = PdfReader(o.file)
+    if pdf.Info.Keywords:
+        tags = re.findall(REGEX, pdf.Info.Keywords.strip())
+        note = re.sub(REGEX, '', pdf.Info.Keywords.strip()).replace('(', '').replace(',', '').replace(')', '').strip()
+        argv = ' '.join(tags).split(' ') + [note]
+        argv = [x for x in argv if x != '']
+        context = o.context(argv)
+        return context
+    
+def writer(o):
+    string = ' '.join(o.collect(o.context))
+    pdf = PdfReader(o.file)
+    pdf.Info.Keywords = string
+    PdfWriter(o.file, trailer=pdf).write()
diff --git a/setup.py b/setup.py
new file mode 100644
index 0000000..7fd8bd4
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,29 @@
+"""
+#// +orvanitag @orga #tagging #python
+"""
+import setuptools
+
+with open("README.md", "r") as fh:
+    long_description = fh.read()
+
+setuptools.setup(
+    name="orvanitag",
+    version="1.0.0",
+    author="randogoth",
+    author_email="randogoth@posteo.org",
+    description="Platform independent, file-based, portable tagging system to organize directories and files.",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    url="https://github.com/randogoth/orvanitag",
+    packages=setuptools.find_packages(),
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: OS Independent",
+    ],
+    entry_points= {
+        'console_scripts': ['otag=orvanitag.cmd:main']
+    },
+    install_requires=['requests'],
+    python_requires='>=3.6',
+)
\ No newline at end of file
diff --git a/test/.context.yaml b/test/.context.yaml
deleted file mode 100644
index 5a75a24..0000000
--- a/test/.context.yaml
+++ /dev/null
@@ -1,19 +0,0 @@
-items:
-  chirpy.wav:
-    contexts:
-    - Chripo
-    projects:
-    - sensorwatch
-    tags: []
-  office.odt:
-    contexts: []
-    projects:
-    - oopsie
-    tags:
-    - whoa
-  word.docx:
-    contexts: []
-    note: what is that smell?
-    projects: []
-    tags:
-    - choo
diff --git a/test/chirpy.wav b/test/chirpy.wav
deleted file mode 100644
index e2a7baa..0000000
Binary files a/test/chirpy.wav and /dev/null differ
diff --git a/test/folder with spaces/file with spaces.txt b/test/folder with spaces/file with spaces.txt
deleted file mode 100644
index 56aacb0..0000000
--- a/test/folder with spaces/file with spaces.txt	
+++ /dev/null
@@ -1,3 +0,0 @@
----
-#// #spaces
----
diff --git a/test/office.odt b/test/office.odt
deleted file mode 100644
index bcf4fb8..0000000
Binary files a/test/office.odt and /dev/null differ
diff --git a/test/oldword.doc b/test/oldword.doc
deleted file mode 100644
index 17fc207..0000000
Binary files a/test/oldword.doc and /dev/null differ
diff --git a/test/sheet.xlsx b/test/sheet.xlsx
deleted file mode 100644
index cfa167b..0000000
Binary files a/test/sheet.xlsx and /dev/null differ
diff --git a/test/sub/.context.yaml b/test/sub/.context.yaml
deleted file mode 100644
index 65e0f78..0000000
--- a/test/sub/.context.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-items:
-  statement_14793553_USD_2022-11-01_2023-01-31.pdf:
-    contexts: []
-    projects: []
-    tags:
-    - yay
diff --git a/test/sub/statement_14793553_USD_2022-11-01_2023-01-31.pdf b/test/sub/statement_14793553_USD_2022-11-01_2023-01-31.pdf
deleted file mode 100644
index 82cb63e..0000000
Binary files a/test/sub/statement_14793553_USD_2022-11-01_2023-01-31.pdf and /dev/null differ
diff --git a/test/table.ods b/test/table.ods
deleted file mode 100644
index bee9848..0000000
Binary files a/test/table.ods and /dev/null differ
diff --git a/test/tableold.xls b/test/tableold.xls
deleted file mode 100644
index 65cd308..0000000
Binary files a/test/tableold.xls and /dev/null differ
diff --git a/test/word.docx b/test/word.docx
deleted file mode 100644
index e22a651..0000000
Binary files a/test/word.docx and /dev/null differ