import os import yaml import shutil from argparse import ArgumentParser as argp, SUPPRESS, RawTextHelpFormatter as raw # Defaults CONTEXT_FILE = '.context.yaml' CACHE_FILE = '.index.yaml' class OrvaniTag: INDEX = {} # our search index CACHEFILE = '' # search index cache ('.index.yaml') DB = dict() # holds the folder context data FILENAME = '' # folder context data filename ('.context.yaml') FILE = '' # holds the folder context file path CTX = { # default contexts 'contexts' : '@', 'projects' : '+', 'tags' : '#' } def __init__(self, path: str=CONTEXT_FILE, cache: str=CACHE_FILE, contexts=CTX): """ Initialization path (str): override default filename for folder context cache (str): override default filename for cached index contexts (dict): override the default contexts """ self.FILENAME = os.path.basename(path) self.CACHEFILE = cache self.INDEX = self.CTX.copy() for ctx in self.INDEX: self.INDEX[ctx] = {} self.INDEX['files'] = {} self.__cd('') def __cd(self, path: str): """ Change folder context to other directory. Checks if a folder context file exists and loads context DB. path (str): path of the new folder context """ self.FILE = os.path.join(path, self.FILENAME) if os.path.isfile(self.FILE): try: with open(self.FILE, 'r') as data: self.DB = yaml.safe_load(data) except Exception: print('File does not exist') else: self.DB = dict() def __save(self): """ Save current folder context to file """ try: with open(self.FILE, 'w+') as data: yaml.dump(self.DB, data) except Exception: print('File write error') def __context(self, argv) -> dict: """ Creates a context entry for an existing file argv (list): sets file contexts according to CTX schema """ context = dict() for ctx in self.CTX.keys(): context.update( { ctx : []}) for arg in argv: for ctx in self.CTX.keys(): if str(arg)[0] in self.CTX.values(): if str(arg).startswith(self.CTX[ctx]): context[ctx].append(str(arg)[1:]) else: if arg != ' ': context.update( { 'note' : str(arg)}) else: context.update( { 'note' : 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 """ if os.path.isfile(file): if os.path.dirname(file): self.__cd(os.path.dirname(file)) file = os.path.basename(file) context = self.__context(argv) if file in self.DB: for ctx in self.CTX.keys(): self.DB[file][ctx] = list(set( self.DB[file][ctx] + context[ctx])) if 'note' in context.keys(): self.DB[file]['note'] = context['note'] else: self.DB.update( { file : context } ) self.__save() def remove(self, file: str, *argv: str): """ Remove context of existing filename and saves changed folder context file file (str): path to existing file argv (list): context parameters to be removed """ if os.path.isfile(file): if os.path.dirname(file): self.__cd(os.path.dirname(file)) file = os.path.basename(file) context = self.__context(argv) if file in self.DB: for ctx in self.CTX.keys(): for item in context[ctx]: if item in self.DB[file][ctx]: self.DB[file][ctx].remove(item) self.__save() def delete(self, file: str, fop:bool=False): """ Deletes all entries of an existing file and optionally deletes the file file (str): path to existing file fop (bool): if True removes the referenced file """ if os.path.dirname(file): self.__cd(os.path.dirname(file)) if fop: os.remove(file) file = os.path.basename(file) if file in self.DB: del self.DB[file] if not self.DB: os.remove(self.FILE) self.__save() def move(self, source: str, dest: str, retain: bool=False, fop: bool=True): """ Copies or moves existing file with contexts to a new destination source (str): existing source file dest (str): the new file destination retain (bool): if True then copies, if False then moves fop (bool): if True also does the file operation, if False only moves context data """ if os.path.isfile(source) and os.path.isdir(os.path.dirname(dest)): sourcefolder = os.path.dirname(source) sourcefile = os.path.basename(source) destfolder = os.path.dirname(dest) destfile = os.path.basename(dest) if not destfile: destfile = sourcefile sourcedata = {} self.__cd(sourcefolder) if self.DB[sourcefile]: sourcedata = self.DB[sourcefile] if not retain: self.purge(source) self.__cd(destfolder) self.DB.update( { destfile : sourcedata } ) self.__save() if fop: if retain: shutil.copy2(source, dest) else: shutil.move(source, dest, copy_function=shutil.copy2) def purge(self, path: str='', recursive=True): """ Removes context data from folder if files do not exist path (str): folder path """ if path == '': path = './' if not os.path.dirname(path): path += '/' if os.path.dirname(path): self.__cd(os.path.dirname(path)) DB = self.DB.copy() for file in self.DB.keys(): if not os.path.isfile(file): if file in self.DB: del DB[file] self.DB = DB self.__save() if not self.DB: os.remove(self.FILE) if recursive: for (root,dirs,files) in os.walk(path): if dirs: for dir in dirs: rpath = os.path.join(root, dir) self.purge(rpath, False) def list(self, search: str, path: str='./', filter=None, exact=True, recursive: bool=True, cache: bool=False): """ Returns an index object with query results and optionally saves a cached index search (str): the search query path (str): the root context for the search and index filter (str): limit the search to a context exact (bool): exact search match if True, otherwise partial match recursive (bool): if True also searches all subfolder contexts cache (bool): if True loads index from cache file if exists, writes it if it doesn't exist """ if path == '': path = './' if not os.path.dirname(path): path += '/' cachefile = os.path.join(path, self.CACHEFILE) if cache and os.path.isfile(cachefile): try: with open(cachefile, 'r') as cache: self.INDEX = yaml.safe_load(cache) except Exception: print('File does not exist') else: self.__indexfolder(path) if recursive: for (root,dirs,files) in os.walk(path): if dirs: for dir in dirs: rpath = os.path.join(root, dir) self.__indexfolder(rpath) if cache: try: with open(cachefile, 'w+') as data: yaml.dump(self.INDEX, data) except Exception: print('File write error') result = { 'index' : self.INDEX, 'found' : [] } for ctx in self.CTX.keys(): for context in self.INDEX[ctx]: if filter and not filter == ctx: continue if exact: if search == context: result['found'] = self.INDEX[ctx][context] else: if search in context: result['found'] = self.INDEX[ctx][context] return result def __indexfolder(self, sub: str): """ Indexes a folder context sub (str): path of folder """ if os.path.dirname(sub): self.__cd(sub) if self.DB: for file in self.DB.keys(): path = os.path.join(sub, file) self.INDEX['files'].update( { path : self.DB[file]} ) for ctx in self.CTX.keys(): for item in self.DB[file][ctx]: if item: filelist = [path] if ctx in self.INDEX: if item in self.INDEX[ctx]: filelist += self.INDEX[ctx][item] self.INDEX.update( { ctx : { item : filelist } } ) def main(): command_help = """{add,a} FILE [CONTEXT ...] {remove,rem,r} FILE [CONTEXT ...] {copy,cp} SOURCE DEST {move,mv} SOURCE DEST {del,rm} FILE {clear,c} FILE {clean} PATH {list,ls,dir,l} QUERY {listproj,lsprj} QUERY {listcon,lsc} QUERY {listtag,lst} QUERY """ p = argp(description="OrvaniTagger", formatter_class=raw) p.add_argument("--not-recursive", "-nr", action="store_true", help="Include subdirectories in search") p.add_argument("--no-file_operation", "-nf", action="store_true", help="Copy/Move/Delete file, not just context") p.add_argument("--path", "-p", action="store", type=str, help="List/search custom path", default='') p.add_argument("--exact", "-e", action="store_true", help="Match exact search term") p.add_argument("--cache", action="store_true", help="Cache index search (useful with deep directory trees)") p.add_argument("--context_filter", "-c", action="store", type=str, help="Filter context", choices=['context', 'project', 'tag'], default=None) p.add_argument("command", nargs="?", default=None, metavar='Command', help=command_help, type=str, choices=[ 'add', 'a', 'remove', 'rem', 'r', 'list', 'ls', 'dir', 'l', 'copy', 'cp', 'move', 'mv', 'del', 'rm', 'clear', 'c', 'listproj', 'lsprj', 'listcon', 'lsc', 'listtag', 'lst', 'clean', 'purge' ]) 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() orvani = OrvaniTag() if args.command in ['add', 'a']: if args.file and args.context: orvani.add(args.file, *args.context) elif args.command in ['remove', 'rem', 'r']: if args.file and args.context: orvani.remove(args.file, *args.context) elif args.command in ['list', 'ls', 'dir', 'l', 'listproj', 'lsprj', 'lp', 'listcon', 'lsc', 'lc', 'listtag', 'lst', 'lt']: query = args.file if args.command in ['listproj', 'lsprj', 'lp'] or args.context_filter == 'project': args.context_filter = 'projects' if args.command in ['listcon', 'lsc', 'lc'] or args.context_filter == 'context': args.context_filter = 'contexts' if args.command in ['listtag', 'lst', 'lt'] or args.context_filter == 'tag': args.context_filter = 'tags' if query: print(args.exact) r = orvani.list(query, args.path, args.context_filter, args.exact, args.not_recursive, args.cache) print(r) elif args.command in ['copy', 'cp']: if args.file and args.context: orvani.move(args.file, args.context[0], True, args.no_file_operation) elif args.command in ['move', 'mv']: if args.file and args.context: orvani.move(args.file, args.context[0], False, args.no_file_operation) elif args.command in ['del', 'rm']: if args.file: orvani.delete(args.file, args.no_file_operation) elif args.command in ['clear', 'c']: if args.file: orvani.delete(args.file, False) elif args.command in ['clean', 'purge']: if not args.file: args.file = '' orvani.purge(args.file, args.not_recursive) if __name__ == '__main__': main()