import os import sys import yaml import shutil from argparse import ArgumentParser as argp, SUPPRESS, RawTextHelpFormatter as raw from rich.console import Console from rich.table import Table import frontmatter # Defaults CONTEXT_FILE = '.context.yaml' CACHE_FILE = '.index.yaml' class OrvaniTag: META_IN = [] META_OUT = [] INDEX = {} # our search index CACHEFILE = '' # search index cache ('.index.yaml') DB = dict() # holds the folder context data CONFIG = dict() 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: y = yaml.safe_load(data) self.DB = y['files'] if 'config' in y: self.CONFIG = y['config'] except Exception: print('File does not exist') else: self.DB = dict() self.CONFIG = dict() def __save(self): """ Save current folder context to file """ try: with open(self.FILE, 'w+') as data: y = { 'files' : self.DB } yaml.dump(y, 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, path: str='./', filter=None, exact=True, recursive: bool=True, 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 == '': path = './' if not os.path.dirname(path): path += '/' self.__makeindex(path, recursive, cache) result = self.__search(search, exact) return result def __intersect(self, result: dict): """ Returns an intersection of two search queries 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: return [] common_elements = set(mresult[next(iter(mresult))]) for key in mresult.keys(): common_elements = common_elements.intersection(mresult[key]) return list(common_elements) def __makeindex(self, path:str, recursive:bool=True, cache: bool=False): """ Creates index for searching, optionally also a cache file path (dict): root directory recursive (bool): traverse subdirectories """ 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') 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]: filelist = [path] if ctx in self.INDEX: if item in self.INDEX[ctx]: filelist += self.INDEX[ctx][item] self.INDEX[ctx][item] = filelist def __search(self, search, filter: str=None, exact:bool=False): """ Searches the index and adds search result to it search (list): search query list exact (bool): exact match or partial match """ result = { 'index' : self.INDEX, 'found' : [] } if search[0]: query = self.__context(search) ret = self.__context([' ']) ret['note'] = [] for ctx in self.CTX.keys(): for context in self.INDEX[ctx]: if filter and not filter == ctx: continue for q in query[ctx]: if exact: if q == context: if self.INDEX[ctx][context] not in ret[ctx]: ret[ctx] += self.INDEX[ctx][context] else: if q in context: if self.INDEX[ctx][context] not in ret[ctx]: ret[ctx] += self.INDEX[ctx][context] if exact: if 'note' in query and query['note'] == context: if self.INDEX[ctx][context] not in ret['note']: ret['note'] += self.INDEX[ctx][context] else: if 'note' in query and query['note'] in context: if self.INDEX[ctx][context] not in ret['note']: ret['note'] += self.INDEX[ctx][context] ret[ctx] = list(set(ret[ctx])) ret['note'] = list(set(ret['note'])) if len(search) > 1: result['found'] = sorted(list(set(self.__intersect(ret))), key=lambda x: x.count('/')) else: min = [] for sublist in ret.values(): min += sublist result['found'] = min else: result['found'] = sorted(self.INDEX['files'].keys(), key=lambda x: x.count('/')) return result def metareader(self, function): self.META_IN.append(function) return function def metawriter(self, function): self.META_OUT.append(function) return function def read_meta(self): for callback in self.META_IN: entry = callback(self.CONFIG, self.DB) print(entry) def mktable(query: str, index: dict): console = Console() if len(index['found']) > 0: table = Table(title=f"Results for `{query}`") table.add_column("Path", justify="right", style="white", no_wrap=True) table.add_column("Note", style="#cccccc") table.add_column("Projects", justify="center", style="cyan") table.add_column("Contexts", justify="center", style="magenta") table.add_column("Tags", justify="center", style="green") note = tags = projects = contexts = tags = '' for file in index['found']: path = file fc = index['index']['files'][file] if 'note' in fc: note = fc['note'] else: note = '' tags = ' '.join(['#' + tag for tag in fc['tags']]) projects = ' '.join(['+' + p for p in fc['projects']]) contexts = ' '.join(['@' + c for c in fc['contexts']]) table.add_row(path, note, projects, contexts, tags) console.print(table) else: console.print(f"No results found for `{query}`", style="red") 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_false", help="Include subdirectories in search") p.add_argument("--no-file_operation", "-nf", action="store_false", 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', 'read' ]) 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] + args.context 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: r = orvani.list(args.path, args.context_filter, args.exact, args.not_recursive, args.cache, *query) mktable(' '.join(sys.argv[1:]), 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) elif args.command in ['read']: orvani.read_meta() if __name__ == '__main__': main()