import os import sys import yaml import shutil from pathlib import Path 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' : '#', 'note' : None } 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 = Path(path) / self.FILENAME if self.FILE.exists(): y = yaml.safe_load(self.FILE.read_text()) self.DB = y['files'] if 'config' in y: self.CONFIG = y['config'] else: self.DB = dict() self.CONFIG = dict() def __save(self): """ Save current folder context to file """ y = { 'files' : self.DB } if self.CONFIG: y['config'] = self.CONFIG if Path(self.FILE).is_file(): with Path(self.FILE).open('w') as data: yaml.safe_dump(y, data) 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(): if self.CTX[ctx] != None: context.update( { ctx : []}) for arg in argv: for ctx in self.CTX.keys(): if str(arg)[0] in self.CTX.values(): # Schema with prefix if not self.CTX[ctx] == None: if str(arg).startswith(self.CTX[ctx]): context[ctx].append(str(arg)[1:]) else: # Otherwise it's a note if self.CTX[ctx] == None: if arg != ' ': context.update( { ctx : str(arg)}) else: context.update( { ctx : 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(): self.__cd(target.parent.name) context = self.__context(argv) if file in self.DB: for ctx in self.CTX.keys(): if self.CTX[ctx] != None: self.DB[file][ctx] = list(set( self.DB[file][ctx] + context[ctx])) else: self.DB[file][ctx] = context[ctx] 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 """ target = Path(file) if target.exists(): self.__cd(target.parent.name) context = self.__context(argv) if file in self.DB: for ctx in self.CTX.keys(): # Only works for context list items if self.CTX[ctx] != None: 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 """ target = Path(file) if target.exists(): self.__cd(target.parent.name) if fop: target.unlink(missing_ok=True) if file in self.DB: del self.DB[file] if not self.DB: Path(self.FILE).unlink(missing_ok=True) 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 = '.' target = Path(path) if target.exists(): self.__cd(target.parent.name) DB = self.DB.copy() for file in self.DB.keys(): if not Path(file).is_file(): if file in self.DB: del DB[file] self.DB = DB self.__save() if not self.DB: Path(self.FILE).unlink(missing_ok=True) 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 ['purge']: if not args.file: args.file = '.' target = Path(args.file) context_file = target / CONTEXT_FILE cache_file = target / CACHE_FILE context_file.unlink(missing_ok=True) cache_file.unlink(missing_ok=True) if args.not_recursive: for (root,dirs,files) in os.walk(args.file): if dirs: for dir in dirs: rpath = os.path.join(root, dir) Path(CONTEXT_FILE).unlink(missing_ok=True) Path(CACHE_FILE).unlink(missing_ok=True) elif args.command in ['clean']: 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()