refactor: use pathlib for all path handling and normalize index paths
This commit is contained in:
randogoth
parent
f18d0300b6
commit
a9a44ccb7e
3 changed files with 112 additions and 141 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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']
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue