diff --git a/orvanitag/cmd.py b/orvanitag/cmd.py index 49f7b21..19a49a0 100644 --- a/orvanitag/cmd.py +++ b/orvanitag/cmd.py @@ -1,4 +1,3 @@ -import os import sys import math from argparse import ArgumentParser as argp, SUPPRESS, RawTextHelpFormatter as raw @@ -7,7 +6,7 @@ from rich import box from rich.console import Console from rich.markup import escape from rich.table import Table -from orvanitag.orvani import OrvaniTag, CONTEXT_FILE, CACHE_FILE, CONTEXTS +from orvanitag.orvani import OrvaniTag, CONTEXT_FILE, CACHE_FILE, CONTEXTS, walk from orvanitag.orvani import OrvaniError, NotFound, NoEntry, UnsupportedType, PluginError # Checked in order, so the base class comes last @@ -56,8 +55,8 @@ def mktable(query: str, index: dict, filter=None, trunc_len: int=25): # 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('./', '') + name = (item.name[:long_file_pre] + '…' + item.name[long_file_post:]) if len(item.name) > trunc_len else item.name + path = (item.parent / name).as_posix() fc = index['index']['items'][file] elements = [] @@ -219,12 +218,8 @@ def main(): 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) + for folder, files in walk(args.file): + purge(folder) elif args.command in ['clean', 'c']: count = orvani.clean(args.file or '', args.recursive) diff --git a/orvanitag/orvani.py b/orvanitag/orvani.py index df27def..5d184f7 100644 --- a/orvanitag/orvani.py +++ b/orvanitag/orvani.py @@ -28,6 +28,22 @@ DEFAULT_PLUGINS = [ 'pdf' ] +def walk(path, recursive: bool=True): + """ + Yields each folder from path downwards with its file names. + Uses os.walk, since Path.walk needs Python 3.12. + """ + for (root,dirs,files) in os.walk(Path(path), topdown=True): + yield Path(root), files + if not recursive: # check if to include subdirectories + break + + +def key(path) -> str: + """The index and export form of a path: normalized, with forward slashes on every system""" + return Path(path).as_posix() + + class OrvaniError(Exception): """An error caused by the user's input, reported without a traceback""" @@ -61,7 +77,7 @@ class OrvaniTag: self.PLUGINS = list(plugins) # List of plugin names self.MODULES = None # Imported plugin modules, see __plugins() - self.FILENAME = os.path.basename(path) # folder context data filename ('.otag') + self.FILENAME = Path(path).name # folder context data filename ('.otag') self.CACHEFILE = cache # search index cache ('.otag.cache.json') self.SCHEMA = contexts # Default context schema self.DB = dict() # holds the folder context data @@ -83,7 +99,7 @@ class OrvaniTag: path (str): path of the new folder context """ - self.FILE = Path(str(path)) / self.FILENAME + self.FILE = Path(path) / self.FILENAME self.DB = self.__read(self.FILE) if self.FILE.exists() else dict() @@ -92,7 +108,7 @@ class OrvaniTag: Returns the configuration of a folder: the nearest .otag.toml in it or any parent folder. A nearer configuration replaces the ones above it entirely. """ - folder = Path(str(folder) or '.').resolve() + folder = Path(folder).resolve() for directory in [folder, *folder.parents]: cfile = directory / CONFIG_FILE if cfile.is_file(): @@ -202,18 +218,13 @@ class OrvaniTag: cachefile.unlink(missing_ok=True) # Only folders with a context file contribute to the index - folders = [] - for (root,dirs,files) in os.walk(path, topdown=True): - if self.FILENAME in files: - folders.append(root) - if not recursive: # check if to include subdirectories - break + folders = [folder for folder, files in walk(path, recursive) if self.FILENAME in files] if cache: sources = {} for folder in folders: - stat = os.stat(os.path.join(folder, self.FILENAME)) - sources[folder] = [stat.st_mtime_ns, stat.st_size] + stat = (folder / self.FILENAME).stat() + sources[key(folder)] = [stat.st_mtime_ns, stat.st_size] if cachefile.is_file(): # load index from cache cached = json.loads(cachefile.read_text(encoding='utf-8')) # A cache built from other or since changed context files is rebuilt @@ -234,15 +245,15 @@ class OrvaniTag: cachefile.write_text(json.dumps({'recursive': recursive, 'sources': sources, 'index': self.INDEX}), encoding='utf-8') - def __indexfolder(self, sub: str): + def __indexfolder(self, folder: Path): """ Load a folder's context and add it to the global context index - sub (str): path of folder + folder (Path): path of folder """ - self.__cd(sub) + self.__cd(folder) for file, context in self.DB.items(): - self.__updateindex(os.path.join(sub, file), context) + self.__updateindex(key(folder / file), context) def __updateindex(self, path: str, context: dict): @@ -285,8 +296,8 @@ class OrvaniTag: # return the whole index if there was no query item found = self.INDEX['items'].keys() - # Sort the result directory tree from shallow to deep based on the occurrence of / - result['found'] = sorted(found, key=lambda x: (x.count('/'), x)) + # Sort the result directory tree from shallow to deep + result['found'] = sorted(found, key=lambda x: (len(Path(x).parts), x)) return result @@ -444,55 +455,35 @@ class OrvaniTag: Returns the moved context """ - # Let's use trusty old os lib over pathlib for this one + src = Path(source) + target = Path(dest) # An existing folder as destination of a file operation receives the file under its own name - if fop and os.path.isdir(dest): - dest = os.path.join(dest, '') + if fop and target.is_dir(): + target = target / src.name - # Check if source file and destination folder exist ('' is the current directory) - if os.path.exists(source) and os.path.isdir(os.path.dirname(dest) or '.'): - - 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) - - self.__cd(sourcefolder) # load local context - sourcedata = self.DB.get(sourcefile) # load its context data - if sourcedata is None and not fop: # without an entry there is nothing to move - raise NoEntry(f"No entry for {source} found.") - if sourcedata is not None: - 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 sourcedata or {} - - if not os.path.exists(source) and not os.path.isdir(os.path.dirname(dest) or '.'): + # Check if source file and destination folder exist + if not src.exists() and not target.parent.is_dir(): raise NotFound(f"Source file {source} and destination {dest} don't exist.") - if not os.path.exists(source): + if not src.exists(): raise NotFound(f"Source file {source} doesn't exist.") - raise NotFound(f"Destination {dest} doesn't exist.") + if not target.parent.is_dir(): + raise NotFound(f"Destination {dest} doesn't exist.") + + self.__cd(src.parent) # load local context + sourcedata = self.DB.get(src.name) # load its context data + if sourcedata is None and not fop: # without an entry there is nothing to move + raise NoEntry(f"No entry for {source} found.") + if sourcedata is not None: + self.__cd(target.parent) # load local context of destination folder + self.DB.update( { target.name : 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? + shutil.copy2(src, target) + if not retain: + src.unlink() + return sourcedata or {} def clean(self, path: str='.', recursive=True): @@ -503,32 +494,24 @@ class OrvaniTag: Returns the number of removed entries """ - if path == '': - path = '.' target = Path(path) - count = 0 if target.is_file(): - target = Path(target.parent) + target = target.parent if not target.is_dir(): raise NotFound(f"Folder {path} doesn't exist.") - 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.exists(): # 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) - count += self.clean(rpath, False) + + count = 0 + for folder, files in walk(target, recursive): + if self.FILENAME not in files: + continue + self.__cd(folder) # load folder context + # delete the entries of files that do not exist anymore + missing = [file for file in self.DB if not (folder / file).exists()] + for file in missing: + del self.DB[file] + if missing: + self.__save() # deletes the context file once it is empty + count += len(missing) return count @@ -544,38 +527,28 @@ class OrvaniTag: 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 index(self, path): - if path == '': # fix empty path argument - path = './' self.__makeindex(path) # index the directories return self.INDEX def context(self, filepath: str): - if filepath in self.INDEX['items']: - context = self.INDEX['items'][filepath] + if key(filepath) in self.INDEX['items']: + context = self.INDEX['items'][key(filepath)] return self.__collect_contexts(context) else: return [] def scan(self, path: str, recursive: bool=False): - # os.walk('') yields nothing, so an empty path means the current folder - for (root,dirs,files) in os.walk(path or '.', topdown=True): - + for folder, files in walk(path, recursive): # process every file with a reader plugin if self.ENABLE_PLUGINS: - if files: - for file in files: - self.read_metadata(os.path.join(root, file)) - - if not recursive: # check if to include subdirectories - break + for file in files: + self.read_metadata(folder / file) def register_plugin(self, function): @@ -647,19 +620,14 @@ class OrvaniTag: def write_all(self, path: str, method: str, recursive: bool=False): - for (root,dirs,files) in os.walk(path, topdown=True): - + for folder, files in walk(path, recursive): # process every file with a writer plugin if self.ENABLE_PLUGINS: - if files: - for file in files: - try: - self.write_metadata(os.path.join(root, file), method) - except OrvaniError: # skip untagged files and other file types - pass - - if not recursive: # check if to include subdirectories - break + for file in files: + try: + self.write_metadata(folder / file, method) + except OrvaniError: # skip untagged files and other file types + pass def write_metadata(self, filename: str, method: str): @@ -708,16 +676,14 @@ class OrvaniTag: 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 o.context - else: - raise PluginError(f"Plugin {plugin_name} can't write.") - raise NoEntry(f"No entry for {filename} found.") + # If there is a file with matching entries in the local context + if key(o.file) not in items: + raise NoEntry(f"No entry for {filename} found.") + o.context = items[key(o.file)] + if not hasattr(plugin, 'writer'): # If the plugin has a writer method + raise PluginError(f"Plugin {plugin_name} can't write.") + plugin.writer(o) # Write the context as metadata + return o.context else: configured = f' Its types are set in {CONFIG_FILE}.' if plugin_name in config else '' raise UnsupportedType(f"Plugin {plugin_name} can only write to file types: {', '.join(suffixes)}.{configured}") @@ -779,7 +745,7 @@ class OrvaniTag: if len(import_line) > 5: line_elements = import_line.split(' --- ') if len(line_elements) > 1: - path = line_elements[0].strip() + path = key(line_elements[0].strip()) # also matches older `./` paths string = line_elements[1].strip() if string and Path(path).exists(): # Check if file exists context = self.__context(self.__get_argv(string)) diff --git a/tests/test_orvani.py b/tests/test_orvani.py index 28bbee3..bcb8577 100644 --- a/tests/test_orvani.py +++ b/tests/test_orvani.py @@ -28,7 +28,7 @@ def test_and_search_within_one_category(tree): o = OrvaniTag() o.add('a.mp3', '#lofi', '#snare') o.add('b.mp3', '#lofi') - assert found(o, '#lofi', '#snare') == ['./a.mp3'] + assert found(o, '#lofi', '#snare') == ['a.mp3'] def test_and_search_across_categories(tree): @@ -36,7 +36,7 @@ def test_and_search_across_categories(tree): o.add('a.mp3', '#lofi', '@drums') o.add('b.mp3', '#lofi') o.add('c.mp3', '@drums') - assert found(o, '#lofi', '@drums') == ['./a.mp3'] + assert found(o, '#lofi', '@drums') == ['a.mp3'] def test_or_search(tree): @@ -44,7 +44,7 @@ def test_or_search(tree): o.add('a.mp3', '#lofi') o.add('b.mp3', '#snare') o.add('c.mp3', '#other') - assert found(o, '#lofi', '#snare', either=True) == ['./a.mp3', './b.mp3'] + assert found(o, '#lofi', '#snare', either=True) == ['a.mp3', 'b.mp3'] def test_unprefixed_term_matches_any_category(tree): @@ -52,14 +52,14 @@ def test_unprefixed_term_matches_any_category(tree): o.add('a.mp3', '+lofi') o.add('b.mp3', '#lofi') o.add('c.mp3', '#other') - assert found(o, 'lofi') == ['./a.mp3', './b.mp3'] + assert found(o, 'lofi') == ['a.mp3', 'b.mp3'] def test_empty_query_lists_everything(tree): o = OrvaniTag() o.add('a.mp3', '#lofi') o.add('sub/d.txt', '#doc') - assert found(o, None) == ['./a.mp3', './sub/d.txt'] + assert found(o, None) == ['a.mp3', 'sub/d.txt'] def test_multi_word_note(tree): @@ -92,7 +92,7 @@ def test_index_is_rebuilt_per_list_call(tree): o = OrvaniTag() o.add('a.mp3', '#lofi') o.add('sub/d.txt', '#lofi') - assert found(o, '#lofi') == ['./a.mp3', './sub/d.txt'] + assert found(o, '#lofi') == ['a.mp3', 'sub/d.txt'] assert found(o, '#lofi', path='sub') == ['sub/d.txt'] @@ -176,8 +176,8 @@ def test_file_names_with_spaces_and_separator_in_note(tree): o = OrvaniTag() o.add('My Track.flac', '#boombap', 'a', '---', 'b') assert lines(tree) == ['My Track.flac --- #boombap a --- b'] - assert found(o, '#boombap') == ['./My Track.flac'] - assert o.index('')['items']['./My Track.flac']['note'] == 'a --- b' + assert found(o, '#boombap') == ['My Track.flac'] + assert o.index('')['items']['My Track.flac']['note'] == 'a --- b' def test_malformed_line_raises_and_keeps_file(tree): @@ -216,13 +216,13 @@ def test_remove_saves_once(tree, monkeypatch): def test_cache_is_rebuilt_when_context_changes(tree): o = OrvaniTag() o.add('a.mp3', '#lofi') - assert o.list('', True, False, True, True, False, '#lofi')['found'] == ['./a.mp3'] + assert o.list('', True, False, True, True, False, '#lofi')['found'] == ['a.mp3'] assert (tree / orvani.CACHE_FILE).exists() o.add('b.mp3', '#lofi') stat = (tree / orvani.CONTEXT_FILE).stat() os.utime(tree / orvani.CONTEXT_FILE, ns=(stat.st_atime_ns, stat.st_mtime_ns + 10**9)) - assert OrvaniTag().list('', True, False, True, True, False, '#lofi')['found'] == ['./a.mp3', './b.mp3'] + assert OrvaniTag().list('', True, False, True, True, False, '#lofi')['found'] == ['a.mp3', 'b.mp3'] def test_import_untagged_files(tree): @@ -230,6 +230,7 @@ def test_import_untagged_files(tree): o.add('a.mp3', '#lofi', 'a note') o.add('sub/d.txt', '@docs') o.export('context.txt') + assert (tree / 'context.txt').read_text() == 'a.mp3 --- #lofi a note\nsub/d.txt --- @docs\n' (tree / orvani.CONTEXT_FILE).unlink() (tree / 'sub' / orvani.CONTEXT_FILE).unlink() @@ -284,3 +285,12 @@ def test_errors_of_remove_delete_export_and_import(tree): o.delete('a.mp3') with pytest.raises(NoEntry): o.export('context.txt') + + +def test_clean_with_a_file_path_cleans_its_folder(tree): + o = OrvaniTag() + o.add('a.mp3', '#lofi') + o.add('b.mp3', '#gone') + (tree / 'b.mp3').unlink() + assert o.clean('a.mp3', False) == 1 + assert lines(tree) == ['a.mp3 --- #lofi']