refactor: replace return-code lists with exceptions and plain return values
This commit is contained in:
parent
675e069a51
commit
f18d0300b6
7 changed files with 260 additions and 263 deletions
138
orvanitag/cmd.py
138
orvanitag/cmd.py
|
|
@ -8,6 +8,16 @@ 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 OrvaniError, NotFound, NoEntry, UnsupportedType, PluginError
|
||||
|
||||
# Checked in order, so the base class comes last
|
||||
ERROR_ICONS = {
|
||||
NotFound: ':thinking_face:',
|
||||
NoEntry: ':mag:',
|
||||
UnsupportedType: ':card_file_box: ',
|
||||
PluginError: ':disappointed: ',
|
||||
OrvaniError: ':thinking_face:',
|
||||
}
|
||||
|
||||
def mktable(query: str, index: dict, filter=None, trunc_len: int=25):
|
||||
"""Creates a rich table from `otag ls` search result
|
||||
|
|
@ -152,24 +162,16 @@ def main():
|
|||
|
||||
if args.command in ['add', 'a']:
|
||||
if args.file and args.context:
|
||||
fb = orvani.add(args.file, *args.context)
|
||||
if fb[0] == 0:
|
||||
console.print(f':label: Entry {nice_context(fb[2])}successfully added to file [strong]{fb[1]}[/strong].')
|
||||
elif fb[0] == 1:
|
||||
console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.")
|
||||
context = orvani.add(args.file, *args.context)
|
||||
console.print(f':label: Entry {nice_context(context)}successfully added to file [strong]{args.file}[/strong].')
|
||||
|
||||
elif args.command in ['remove', 'r']:
|
||||
if args.file:
|
||||
fb = orvani.remove(args.file, *args.context)
|
||||
if fb[0] == 0:
|
||||
if fb[2] == None:
|
||||
console.print(f':wastebasket: All entries successfully removed from file [strong]{fb[1]}[/strong].')
|
||||
else:
|
||||
console.print(f':wastebasket: Entry {nice_context(fb[2])}successfully removed from file [strong]{fb[1]}[/strong].')
|
||||
elif fb[0] == 1:
|
||||
console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.")
|
||||
elif fb[0] == 2:
|
||||
console.print(f":mag: No match found for [white][strong]{fb[1]}[/strong][/white] and {nice_context(fb[2])}")
|
||||
context = orvani.remove(args.file, *args.context)
|
||||
if context is None:
|
||||
console.print(f':wastebasket: All entries successfully removed from file [strong]{args.file}[/strong].')
|
||||
else:
|
||||
console.print(f':wastebasket: Entry {nice_context(context)}successfully removed from file [strong]{args.file}[/strong].')
|
||||
|
||||
elif args.command in ['list', 'ls', 'l']:
|
||||
query = [args.file] + args.context
|
||||
|
|
@ -182,50 +184,21 @@ def main():
|
|||
filter = 'dirs'
|
||||
mktable(' '.join(sys.argv[1:]), r, filter)
|
||||
|
||||
elif args.command in ['copy', 'cp']:
|
||||
elif args.command in ['copy', 'cp', 'move', 'mv']:
|
||||
if args.file and args.context:
|
||||
fb = orvani.move(args.file, args.context[0], True, args.file_operation)
|
||||
if fb[0] == 0:
|
||||
console.print(f':dancers: File [strong]{fb[1]}[/strong] with entry {nice_context(fb[3])}successfully copied to [strong]{fb[2]}[/strong]')
|
||||
elif fb[0] == 1:
|
||||
if fb[3] == False and fb[4] == True:
|
||||
console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] doesn't exist.")
|
||||
elif fb[3] == True and fb[4] == False:
|
||||
console.print(f":thinking_face: Destination [strong]{fb[2]}[/strong] doesn't exist.")
|
||||
elif fb[3] == False and fb[4] == False:
|
||||
console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] and destination [strong]{fb[2]}[/strong] don't exist.")
|
||||
elif fb[0] == 2:
|
||||
console.print(f":mag: No entry for [strong]{fb[1]}[/strong] found.")
|
||||
|
||||
elif args.command in ['move', 'mv']:
|
||||
if args.file and args.context:
|
||||
fb = orvani.move(args.file, args.context[0], False, args.file_operation)
|
||||
if fb[0] == 0:
|
||||
console.print(f':dancers: File [strong]{fb[1]}[/strong] with entry {nice_context(fb[3])}successfully moved to [strong]{fb[2]}[/strong]')
|
||||
elif fb[0] == 1:
|
||||
if fb[3] == False and fb[4] == True:
|
||||
console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] doesn't exist.")
|
||||
elif fb[3] == True and fb[4] == False:
|
||||
console.print(f":thinking_face: Destination [strong]{fb[2]}[/strong] doesn't exist.")
|
||||
elif fb[3] == False and fb[4] == False:
|
||||
console.print(f":thinking_face: Source file [strong]{fb[1]}[/strong] and destination [strong]{fb[2]}[/strong] don't exist.")
|
||||
elif fb[0] == 2:
|
||||
console.print(f":mag: No entry for [strong]{fb[1]}[/strong] found.")
|
||||
copy = args.command in ['copy', 'cp']
|
||||
context = orvani.move(args.file, args.context[0], copy, args.file_operation)
|
||||
console.print(f':dancers: File [strong]{args.file}[/strong] with entry {nice_context(context)}successfully {"copied" if copy else "moved"} to [strong]{args.context[0]}[/strong]')
|
||||
|
||||
elif args.command in ['del', 'rm']:
|
||||
if args.file:
|
||||
fb = orvani.delete(args.file, args.file_operation)
|
||||
if fb[0] == 0:
|
||||
status = f':wastebasket: Entry for [strong]{fb[1]}[/strong] successfully removed.'
|
||||
if fb[3]:
|
||||
status += f'\n:wastebasket: File [strong]{fb[1]}[/strong] successfully deleted.'
|
||||
if not fb[2] == None:
|
||||
status += f'\n:wastebasket: File [strong]{fb[2]}[/strong] automatically deleted.'
|
||||
console.print(status)
|
||||
elif fb[0] == 1:
|
||||
console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.")
|
||||
elif fb[0] == 2:
|
||||
console.print(f":mag: No entry for [strong]{fb[1]}[/strong] found.")
|
||||
context_file = orvani.delete(args.file, args.file_operation)
|
||||
status = f':wastebasket: Entry for [strong]{args.file}[/strong] successfully removed.'
|
||||
if args.file_operation:
|
||||
status += f'\n:wastebasket: File [strong]{args.file}[/strong] successfully deleted.'
|
||||
if context_file:
|
||||
status += f'\n:wastebasket: File [strong]{context_file}[/strong] automatically deleted.'
|
||||
console.print(status)
|
||||
|
||||
|
||||
elif args.command in ['purge', 'p']:
|
||||
|
|
@ -254,56 +227,31 @@ def main():
|
|||
purge(pp)
|
||||
|
||||
elif args.command in ['clean', 'c']:
|
||||
if not args.file:
|
||||
args.file = ''
|
||||
fb = orvani.clean(args.file, args.recursive)
|
||||
if fb[0] == 0:
|
||||
sub = ''
|
||||
if args.recursive:
|
||||
sub = ' and subdirectories'
|
||||
console.print(f':broom: {fb[2]} entries cleaned from [strong]{fb[1]}[/strong]{sub}')
|
||||
count = orvani.clean(args.file or '', args.recursive)
|
||||
sub = ' and subdirectories' if args.recursive else ''
|
||||
console.print(f':broom: {count} entries cleaned from [strong]{args.file or "."}[/strong]{sub}')
|
||||
|
||||
elif args.command in ['write', 'w']:
|
||||
if args.file and args.context:
|
||||
fb = orvani.write_metadata(args.context[0], args.file)
|
||||
if fb[0] == 0:
|
||||
console.print(f':floppy_disk: Metadata {nice_context(fb[2])}successfully written to file [strong]{fb[1]}[/strong].')
|
||||
if fb[0] == 4:
|
||||
console.print(f':card_file_box: Plugin [strong]{fb[2]}[/strong] can only write to files types: [strong]{"[/strong], [strong]".join(fb[1])}[/strong].')
|
||||
if fb[3]:
|
||||
console.print(f':information: Plugin configuration found in folder context file.')
|
||||
if fb[0] == 5:
|
||||
console.print(f':disappointed: Plugin [strong]{fb[1]}[/strong] can\'t write.')
|
||||
if fb[0] == 2:
|
||||
console.print(f":mag: No entry for [strong]{fb[1]}[/strong] found.")
|
||||
if fb[0] == 6:
|
||||
console.print(f":thinking_face: Plugin [strong]{fb[1]}[/strong] not found or its dependencies are not installed.")
|
||||
context = orvani.write_metadata(args.context[0], args.file)
|
||||
console.print(f':floppy_disk: Metadata {nice_context(context)}successfully written to file [strong]{args.context[0]}[/strong].')
|
||||
|
||||
elif args.command in ['export', 'e']:
|
||||
fb = orvani.export(args.file)
|
||||
if fb[0] == 0:
|
||||
console.print(f':floppy_disk: All entries successfully written to file [strong]{fb[1]}[/strong].')
|
||||
elif fb[0] == 2:
|
||||
console.print(f":mag: No entries found.")
|
||||
filename = orvani.export(args.file)
|
||||
console.print(f':floppy_disk: All entries successfully written to file [strong]{filename}[/strong].')
|
||||
|
||||
elif args.command in ['import', 'i']:
|
||||
|
||||
fb = orvani.load(args.file)
|
||||
if fb[0] == 0:
|
||||
console.print(f':inbox_tray: [strong]{fb[2]}[/strong] entries successfully imported from file [strong]{fb[1]}[/strong].')
|
||||
elif fb[0] == 1:
|
||||
console.print(f":thinking_face: File [strong]{fb[1]}[/strong] doesn't exist.")
|
||||
elif fb[0] == 2:
|
||||
console.print(f":mag: No entries found in file [strong]{fb[1]}[/strong].")
|
||||
elif fb[0] == 4:
|
||||
console.print(f":thinking_face: File [strong]{fb[1]}[/strong] has wrong format. Cannot import.")
|
||||
count = orvani.load(args.file)
|
||||
console.print(f':inbox_tray: [strong]{count}[/strong] entries successfully imported from file [strong]{args.file or "context.txt"}[/strong].')
|
||||
|
||||
elif args.command in ['scan', 's']:
|
||||
if not args.file:
|
||||
args.file = ''
|
||||
orvani.scan(args.file, args.recursive)
|
||||
orvani.scan(args.file or '', args.recursive)
|
||||
console.print(f':mag: Finished scanning subdirectories.')
|
||||
|
||||
|
||||
except OrvaniError as e:
|
||||
icon = next(icon for kind, icon in ERROR_ICONS.items() if isinstance(e, kind))
|
||||
console.print(f'{icon} {escape(str(e))}')
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
console.print(f":scream: An Exception occurred: [italic]{escape(str(e))}[/italic]")
|
||||
sys.exit(1)
|
||||
|
|
|
|||
|
|
@ -28,15 +28,23 @@ DEFAULT_PLUGINS = [
|
|||
'pdf'
|
||||
]
|
||||
|
||||
"""
|
||||
ERRORS:
|
||||
0 - Success
|
||||
1 - File Not Found
|
||||
2 - Entry not found
|
||||
3 - Path is wrong type
|
||||
4 - Wrong File Type
|
||||
5 - Plugin has no method
|
||||
"""
|
||||
class OrvaniError(Exception):
|
||||
"""An error caused by the user's input, reported without a traceback"""
|
||||
|
||||
class NotFound(OrvaniError):
|
||||
"""A file or folder does not exist"""
|
||||
|
||||
class NoEntry(OrvaniError):
|
||||
"""There are no matching tags"""
|
||||
|
||||
class UnsupportedType(OrvaniError):
|
||||
"""A plugin cannot handle the file type"""
|
||||
|
||||
class PluginError(OrvaniError):
|
||||
"""A plugin is unknown, not installed or cannot write"""
|
||||
|
||||
class FormatError(OrvaniError, ValueError):
|
||||
"""A context or import file has a malformed line"""
|
||||
|
||||
class OrvaniTag:
|
||||
|
||||
|
|
@ -104,7 +112,7 @@ class OrvaniTag:
|
|||
name, separator, string = line.partition(SEPARATOR)
|
||||
if not separator or not name:
|
||||
# Raising instead of skipping, since the next save would silently drop the line
|
||||
raise ValueError(f'{file}:{num}: expected "name{SEPARATOR}contexts", got "{line}"')
|
||||
raise FormatError(f'{file}:{num}: expected "name{SEPARATOR}contexts", got "{line}"')
|
||||
name = name[:-1] if name.endswith('/') else name
|
||||
db[name] = self.__context(string.split())
|
||||
return db
|
||||
|
|
@ -340,32 +348,29 @@ class OrvaniTag:
|
|||
|
||||
"""
|
||||
target = Path(file)
|
||||
if not target.exists(): # Only existing files and folders
|
||||
raise NotFound(f"File {file} doesn't exist.")
|
||||
|
||||
if target.exists(): # Only existing files and folders
|
||||
self.__cd(target.parent) # Load folder context
|
||||
context = self.__context(argv) # Create context dict from argument list
|
||||
self.__cd(target.parent) # Load folder context
|
||||
context = self.__context(argv) # Create context dict from argument list
|
||||
|
||||
if target.name in self.DB: # Does the file exist in our local context DB?
|
||||
if target.name in self.DB: # Does the file exist in our local context DB?
|
||||
|
||||
# Add entries from argument context to context DB
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if self.SCHEMA[context_key] != None:
|
||||
# Add to context and remove duplicate entries
|
||||
self.DB[target.name][context_key] = list(dict.fromkeys( self.DB[target.name][context_key] + context[context_key]))
|
||||
else:
|
||||
# Add the note
|
||||
if context_key in context:
|
||||
self.DB[target.name][context_key] = context[context_key]
|
||||
|
||||
else: # If file or folder does not exist in context DB yet, add it
|
||||
self.DB.update( { target.name : context } )
|
||||
|
||||
# Save to file
|
||||
saved = self.__save()
|
||||
if saved and context:
|
||||
return [0, file, context]
|
||||
else:
|
||||
return [1, file]
|
||||
# Add entries from argument context to context DB
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if self.SCHEMA[context_key] != None:
|
||||
# Add to context and remove duplicate entries
|
||||
self.DB[target.name][context_key] = list(dict.fromkeys( self.DB[target.name][context_key] + context[context_key]))
|
||||
else:
|
||||
# Add the note
|
||||
if context_key in context:
|
||||
self.DB[target.name][context_key] = context[context_key]
|
||||
|
||||
else: # If file or folder does not exist in context DB yet, add it
|
||||
self.DB.update( { target.name : context } )
|
||||
|
||||
self.__save()
|
||||
return context
|
||||
|
||||
|
||||
def remove(self, file: str, *argv: str):
|
||||
|
|
@ -375,41 +380,35 @@ class OrvaniTag:
|
|||
file (str): path to existing file
|
||||
argv (list): context parameters to be removed
|
||||
|
||||
Returns the removed context, or None if the whole entry was removed
|
||||
"""
|
||||
target = Path(file)
|
||||
all = False
|
||||
if target.exists(): # Only existing files and folders
|
||||
self.__cd(target.parent) # Load folder context
|
||||
if len(argv) == 0:
|
||||
all = True
|
||||
context = self.__context(argv) # Create context dict from argument list
|
||||
if not target.exists(): # Only existing files and folders
|
||||
raise NotFound(f"File {file} doesn't exist.")
|
||||
|
||||
if target.name in self.DB: # Does the file exist in our local context DB?
|
||||
self.__cd(target.parent) # Load folder context
|
||||
context = self.__context(argv) # Create context dict from argument list
|
||||
no_match = NoEntry(f"No match found for {file} and {' '.join(self.__collect_contexts(dict(context)))}.")
|
||||
if target.name not in self.DB: # Does the file exist in our local context DB?
|
||||
raise no_match
|
||||
|
||||
saved = False
|
||||
if all:
|
||||
del self.DB[target.name]
|
||||
saved = self.__save()
|
||||
return [0, file, None]
|
||||
else:
|
||||
for context_key in self.SCHEMA.keys():
|
||||
# Only works for context list items
|
||||
if self.SCHEMA[context_key] != None:
|
||||
for item in context[context_key]:
|
||||
if item in self.DB[target.name][context_key]:
|
||||
self.DB[target.name][context_key].remove(item)
|
||||
saved = True
|
||||
if saved:
|
||||
saved = self.__save()
|
||||
if not argv:
|
||||
del self.DB[target.name]
|
||||
self.__save()
|
||||
return None
|
||||
|
||||
if saved and self.DB[target.name]:
|
||||
return [0, file, context]
|
||||
else:
|
||||
return [2, file, context]
|
||||
else:
|
||||
return [2, file, context]
|
||||
else:
|
||||
return [1, file]
|
||||
removed = False
|
||||
for context_key in self.SCHEMA.keys():
|
||||
# Only works for context list items
|
||||
if self.SCHEMA[context_key] != None:
|
||||
for item in context[context_key]:
|
||||
if item in self.DB[target.name][context_key]:
|
||||
self.DB[target.name][context_key].remove(item)
|
||||
removed = True
|
||||
if not removed:
|
||||
raise no_match
|
||||
self.__save()
|
||||
return context
|
||||
|
||||
|
||||
def delete(self, file: str, fop:bool=False):
|
||||
|
|
@ -419,22 +418,19 @@ class OrvaniTag:
|
|||
file (str): path to existing file
|
||||
fop (bool): if True removes the referenced file
|
||||
|
||||
Returns the folder context file if it was deleted because it became empty
|
||||
"""
|
||||
target = Path(file)
|
||||
if target.exists(): # Does the target file exist?
|
||||
self.__cd(target.parent) # Load its local context
|
||||
else:
|
||||
return [1, file]
|
||||
if target.name in self.DB: # Do we have a record in the DB?
|
||||
del self.DB[target.name] # Delete it
|
||||
else:
|
||||
return [2, file]
|
||||
saved = self.__save()
|
||||
if not self.DB: # If local context is empty now, we can delete the file
|
||||
Path(self.FILE).unlink(missing_ok=True)
|
||||
if not target.exists(): # Does the target file exist?
|
||||
raise NotFound(f"File {file} doesn't exist.")
|
||||
self.__cd(target.parent) # Load its local context
|
||||
if target.name not in self.DB: # Do we have a record in the DB?
|
||||
raise NoEntry(f"No entry for {file} found.")
|
||||
del self.DB[target.name] # Delete it
|
||||
self.__save() # deletes the context file once it is empty
|
||||
if fop: # delete the referenced file too?
|
||||
target.unlink(missing_ok=True)
|
||||
return [0, file, self.FILE if not self.DB else None, fop]
|
||||
return None if self.FILE.exists() else self.FILE
|
||||
|
||||
|
||||
def move(self, source: str, dest: str, retain: bool=False, fop: bool=True):
|
||||
|
|
@ -446,6 +442,7 @@ class OrvaniTag:
|
|||
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
|
||||
|
||||
Returns the moved context
|
||||
"""
|
||||
# Let's use trusty old os lib over pathlib for this one
|
||||
# An existing folder as destination of a file operation receives the file under its own name
|
||||
|
|
@ -476,7 +473,7 @@ class OrvaniTag:
|
|||
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
|
||||
return [2, source]
|
||||
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
|
||||
|
|
@ -489,9 +486,13 @@ class OrvaniTag:
|
|||
else:
|
||||
shutil.copy2(source, dest)
|
||||
os.remove(source)
|
||||
return [0, source, dest, sourcedata or {}]
|
||||
else:
|
||||
return [1, source, dest, os.path.isfile(source), os.path.isdir(os.path.dirname(dest) or '.')]
|
||||
return sourcedata or {}
|
||||
|
||||
if not os.path.exists(source) and not os.path.isdir(os.path.dirname(dest) or '.'):
|
||||
raise NotFound(f"Source file {source} and destination {dest} don't exist.")
|
||||
if not os.path.exists(source):
|
||||
raise NotFound(f"Source file {source} doesn't exist.")
|
||||
raise NotFound(f"Destination {dest} doesn't exist.")
|
||||
|
||||
|
||||
def clean(self, path: str='.', recursive=True):
|
||||
|
|
@ -500,6 +501,7 @@ class OrvaniTag:
|
|||
|
||||
path (str): folder path
|
||||
|
||||
Returns the number of removed entries
|
||||
"""
|
||||
if path == '':
|
||||
path = '.'
|
||||
|
|
@ -507,27 +509,27 @@ class OrvaniTag:
|
|||
count = 0
|
||||
if target.is_file():
|
||||
target = Path(target.parent)
|
||||
if target.is_dir():
|
||||
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)
|
||||
fb = self.clean(rpath, False)
|
||||
count += fb[2]
|
||||
return [0, str(target), count]
|
||||
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)
|
||||
return count
|
||||
|
||||
|
||||
def list(self, path: str='./', exact=True, either=False, recursive: bool=True, cache: bool=False, delete_cache: bool=False, *search: str):
|
||||
|
|
@ -651,7 +653,10 @@ class OrvaniTag:
|
|||
if self.ENABLE_PLUGINS:
|
||||
if files:
|
||||
for file in files:
|
||||
self.write_metadata(os.path.join(root, file), method)
|
||||
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
|
||||
|
|
@ -663,8 +668,10 @@ class OrvaniTag:
|
|||
|
||||
filename (str): A valid file path
|
||||
method (str): The name of a plugin able to process the file type
|
||||
|
||||
Returns the written context
|
||||
"""
|
||||
|
||||
|
||||
config = self.config(Path(filename).parent)
|
||||
|
||||
# Utilities for writer plugins
|
||||
|
|
@ -707,13 +714,14 @@ class OrvaniTag:
|
|||
o.context = items[found]
|
||||
if hasattr(plugin, 'writer'): # If the plugin has a writer method
|
||||
plugin.writer(o) # Write the context as metadata
|
||||
return [0, filename, o.context]
|
||||
return o.context
|
||||
else:
|
||||
return [5, plugin_name]
|
||||
return [2, filename]
|
||||
raise PluginError(f"Plugin {plugin_name} can't write.")
|
||||
raise NoEntry(f"No entry for {filename} found.")
|
||||
else:
|
||||
return [4, suffixes, plugin_name, plugin_name in config]
|
||||
return [6, method]
|
||||
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}")
|
||||
raise PluginError(f"Plugin {method} not found or its dependencies are not installed.")
|
||||
|
||||
|
||||
def export(self, filename:str):
|
||||
|
|
@ -721,6 +729,8 @@ class OrvaniTag:
|
|||
Exports the global context index to a text file in a format that can be reimported
|
||||
|
||||
filename (str): optional filename
|
||||
|
||||
Returns the filename
|
||||
"""
|
||||
|
||||
# If no filename give, use the default one
|
||||
|
|
@ -740,9 +750,9 @@ class OrvaniTag:
|
|||
string = ' '.join([x for x in self.__collect_contexts(items[path]) if x != None])
|
||||
content += f'{path} --- {string}\n'
|
||||
file.write_text(content)
|
||||
return [0, filename]
|
||||
return filename
|
||||
else:
|
||||
return [2]
|
||||
raise NoEntry("No entries found.")
|
||||
|
||||
|
||||
def load(self, filename:str):
|
||||
|
|
@ -751,40 +761,38 @@ class OrvaniTag:
|
|||
filtering out the already known entries
|
||||
|
||||
filename (str): optional filename
|
||||
|
||||
Returns the number of files that got new tags
|
||||
"""
|
||||
|
||||
# If no filename was given, try the default one
|
||||
if filename == '' or filename == None:
|
||||
filename = 'context.txt'
|
||||
file = Path(filename)
|
||||
if file.exists():
|
||||
imported = file.read_text().split('\n')
|
||||
if len(imported):
|
||||
# Make an index of all context items that are already known
|
||||
known = self.list(file.parent, False, False, True, False, False, '')['index']['items']
|
||||
count = 0
|
||||
for import_line in imported:
|
||||
if len(import_line) > 5:
|
||||
line_elements = import_line.split(' --- ')
|
||||
if len(line_elements) > 1:
|
||||
path = line_elements[0].strip()
|
||||
string = line_elements[1].strip()
|
||||
if string and Path(path).exists(): # Check if file exists
|
||||
context = self.__context(self.__get_argv(string))
|
||||
# Don't import anything we already have
|
||||
known_file_context = known.get(path, {})
|
||||
for new_context, value in list(context.items()):
|
||||
if type(value) == list:
|
||||
context[new_context] = [x for x in value if x not in (known_file_context.get(new_context) or [])]
|
||||
elif value == known_file_context.get(new_context):
|
||||
del context[new_context]
|
||||
if any(context.values()):
|
||||
self.__add(path, context)
|
||||
count += 1
|
||||
else:
|
||||
return [4, filename]
|
||||
return [0, filename, count]
|
||||
else:
|
||||
return [2, filename]
|
||||
else:
|
||||
return [1, filename]
|
||||
if not file.exists():
|
||||
raise NotFound(f"File {filename} doesn't exist.")
|
||||
|
||||
# Make an index of all context items that are already known
|
||||
known = self.list(file.parent, False, False, True, False, False, '')['index']['items']
|
||||
count = 0
|
||||
for import_line in file.read_text().split('\n'):
|
||||
if len(import_line) > 5:
|
||||
line_elements = import_line.split(' --- ')
|
||||
if len(line_elements) > 1:
|
||||
path = line_elements[0].strip()
|
||||
string = line_elements[1].strip()
|
||||
if string and Path(path).exists(): # Check if file exists
|
||||
context = self.__context(self.__get_argv(string))
|
||||
# Don't import anything we already have
|
||||
known_file_context = known.get(path, {})
|
||||
for new_context, value in list(context.items()):
|
||||
if type(value) == list:
|
||||
context[new_context] = [x for x in value if x not in (known_file_context.get(new_context) or [])]
|
||||
elif value == known_file_context.get(new_context):
|
||||
del context[new_context]
|
||||
if any(context.values()):
|
||||
self.__add(path, context)
|
||||
count += 1
|
||||
else:
|
||||
raise FormatError(f"File {filename} has wrong format. Cannot import.")
|
||||
return count
|
||||
|
|
@ -32,7 +32,7 @@ def test_python_keeps_shebang_coding_and_docstring(cwd):
|
|||
source = '#!/usr/bin/env python3\n# -*- coding: utf-8 -*-\n"""Real docstring."""\nprint(__doc__)\n'
|
||||
(cwd / 's.py').write_text(source)
|
||||
os.chmod(cwd / 's.py', 0o755)
|
||||
assert write('s.py', source, '+tools', '#demo', 'a note')[0] == 0
|
||||
write('s.py', source, '+tools', '#demo', 'a note')
|
||||
assert (cwd / 's.py').read_text() == (
|
||||
'#!/usr/bin/env python3\n# -*- coding: utf-8 -*-\n# otag: +tools #demo a note\n\n'
|
||||
'"""Real docstring."""\nprint(__doc__)\n'
|
||||
|
|
@ -45,13 +45,13 @@ def test_python_keeps_shebang_coding_and_docstring(cwd):
|
|||
|
||||
@pytest.mark.skipif(not shutil.which('perl'), reason='perl not installed')
|
||||
def test_perl_still_runs(cwd):
|
||||
assert write('s.pl', 'print "perl ok\\n";\n', '#demo')[0] == 0
|
||||
write('s.pl', 'print "perl ok\\n";\n', '#demo')
|
||||
run = subprocess.run(['perl', 's.pl'], capture_output=True, text=True)
|
||||
assert run.stdout == 'perl ok\n'
|
||||
|
||||
|
||||
def test_php_tag_goes_inside_the_php_block(cwd):
|
||||
assert write('p.php', '<?php\necho "ok";\n', '#demo')[0] == 0
|
||||
write('p.php', '<?php\necho "ok";\n', '#demo')
|
||||
assert (cwd / 'p.php').read_text() == '<?php\n// otag: #demo\n\necho "ok";\n'
|
||||
|
||||
|
||||
|
|
@ -62,19 +62,19 @@ def test_php_without_own_opening_line_is_refused(cwd):
|
|||
|
||||
|
||||
def test_go_tag_is_not_a_package_doc_comment(cwd):
|
||||
assert write('m.go', 'package main\n', '#demo')[0] == 0
|
||||
write('m.go', 'package main\n', '#demo')
|
||||
assert (cwd / 'm.go').read_text() == '// otag: #demo\n\npackage main\n'
|
||||
|
||||
|
||||
def test_html_keeps_doctype_first(cwd):
|
||||
assert write('i.html', '<!DOCTYPE html>\n<html></html>\n', '#demo', 'home page')[0] == 0
|
||||
write('i.html', '<!DOCTYPE html>\n<html></html>\n', '#demo', 'home page')
|
||||
assert (cwd / 'i.html').read_text() == '<!DOCTYPE html>\n<!-- otag: #demo home page -->\n\n<html></html>\n'
|
||||
assert read_back(cwd, 'i.html') == 'i.html --- #demo home page\n'
|
||||
|
||||
|
||||
def test_existing_tag_line_is_replaced_and_crlf_kept(cwd):
|
||||
source = '// otag: #old\r\n\r\nint main() {}\r\n'
|
||||
assert write('m.c', source, '#new')[0] == 0
|
||||
write('m.c', source, '#new')
|
||||
assert (cwd / 'm.c').read_bytes() == b'// otag: #new\r\n\r\nint main() {}\r\n'
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import pytest
|
||||
|
||||
import orvanitag.orvani as orvani
|
||||
from orvanitag.orvani import OrvaniTag
|
||||
from orvanitag.orvani import OrvaniTag, UnsupportedType
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -31,7 +31,7 @@ def test_write_changes_only_its_own_keys(cwd):
|
|||
(cwd / 'post.md').write_text(post)
|
||||
o = OrvaniTag()
|
||||
o.add('post.md', '#blog', '#lo-fi', '#new')
|
||||
assert o.write_metadata('post.md', 'frontmatter')[0] == 0
|
||||
o.write_metadata('post.md', 'frontmatter')
|
||||
assert (cwd / 'post.md').read_text() == (
|
||||
'---\ntitle: Post\n# my comment\ndate: 2024-01-05\ntags: [blog, lo-fi, new]\nlayout: post\n---\nBody text.\n'
|
||||
)
|
||||
|
|
@ -68,5 +68,6 @@ def test_rst_is_not_handled(cwd):
|
|||
(cwd / 'doc.rst').write_text('Title\n=====\n')
|
||||
o = OrvaniTag()
|
||||
o.add('doc.rst', '#x')
|
||||
assert o.write_metadata('doc.rst', 'frontmatter')[0] == 4
|
||||
with pytest.raises(UnsupportedType):
|
||||
o.write_metadata('doc.rst', 'frontmatter')
|
||||
assert (cwd / 'doc.rst').read_text() == 'Title\n=====\n'
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ def test_write_and_read_comment(cwd):
|
|||
make_wav(cwd / 'a.wav')
|
||||
o = OrvaniTag()
|
||||
o.add('a.wav', '+album', '@drums', '#lo-fi', 'first', 'take')
|
||||
assert o.write_metadata('a.wav', 'id3')[0] == 0
|
||||
o.write_metadata('a.wav', 'id3')
|
||||
with taglib.File(cwd / 'a.wav') as song:
|
||||
assert song.tags['COMMENT'] == ['+album @drums #lo-fi first take']
|
||||
(cwd / orvani.CONTEXT_FILE).unlink()
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import os
|
|||
import pytest
|
||||
|
||||
import orvanitag.orvani as orvani
|
||||
from orvanitag.orvani import OrvaniTag, DEFAULT_PLUGINS
|
||||
from orvanitag.orvani import OrvaniTag, DEFAULT_PLUGINS, FormatError, NoEntry, NotFound
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -77,7 +77,7 @@ def test_clean_keeps_tagged_directories(tree):
|
|||
o.add('sub', '#folder')
|
||||
o.add('a.mp3', '#lofi')
|
||||
(tree / 'a.mp3').unlink()
|
||||
assert o.clean('', True)[2] == 1
|
||||
assert o.clean('', True) == 1
|
||||
assert lines(tree) == ['sub/ --- #folder']
|
||||
|
||||
|
||||
|
|
@ -97,13 +97,14 @@ def test_index_is_rebuilt_per_list_call(tree):
|
|||
|
||||
|
||||
def test_move_untagged_file(tree):
|
||||
assert OrvaniTag().move('a.mp3', 'sub/a.mp3', False, False)[0] == 2
|
||||
with pytest.raises(NoEntry):
|
||||
OrvaniTag().move('a.mp3', 'sub/a.mp3', False, False)
|
||||
|
||||
|
||||
def test_move_tagged_file(tree):
|
||||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi')
|
||||
assert o.move('a.mp3', 'sub/', False, True)[0] == 0
|
||||
o.move('a.mp3', 'sub/', False, True)
|
||||
assert (tree / 'sub' / 'a.mp3').exists()
|
||||
assert lines(tree / 'sub') == ['a.mp3 --- #lofi']
|
||||
assert not (tree / orvani.CONTEXT_FILE).exists()
|
||||
|
|
@ -112,14 +113,14 @@ def test_move_tagged_file(tree):
|
|||
def test_copy_tags_within_current_directory(tree):
|
||||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi')
|
||||
assert o.move('a.mp3', 'b.mp3', True, False)[0] == 0
|
||||
o.move('a.mp3', 'b.mp3', True, False)
|
||||
assert lines(tree) == ['a.mp3 --- #lofi', 'b.mp3 --- #lofi']
|
||||
|
||||
|
||||
def test_rename_within_current_directory(tree):
|
||||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi')
|
||||
assert o.move('a.mp3', 'renamed.mp3', False, True)[0] == 0
|
||||
o.move('a.mp3', 'renamed.mp3', False, True)
|
||||
assert not (tree / 'a.mp3').exists()
|
||||
assert lines(tree) == ['renamed.mp3 --- #lofi']
|
||||
|
||||
|
|
@ -127,7 +128,7 @@ def test_rename_within_current_directory(tree):
|
|||
def test_move_into_directory_without_slash(tree):
|
||||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi')
|
||||
assert o.move('a.mp3', 'sub', False, True)[0] == 0
|
||||
o.move('a.mp3', 'sub', False, True)
|
||||
assert (tree / 'sub' / 'a.mp3').exists()
|
||||
assert lines(tree / 'sub') == ['a.mp3 --- #lofi']
|
||||
assert not (tree / orvani.CONTEXT_FILE).exists()
|
||||
|
|
@ -137,7 +138,7 @@ def test_copy_tags_from_subfolder_to_folder(tree):
|
|||
(tree / 'other').mkdir()
|
||||
o = OrvaniTag()
|
||||
o.add('sub/d.txt', '#doc')
|
||||
assert o.move('sub/d.txt', 'other/', True, False)[0] == 0
|
||||
o.move('sub/d.txt', 'other/', True, False)
|
||||
assert lines(tree) == ['other/ --- #doc']
|
||||
assert lines(tree / 'sub') == ['d.txt --- #doc']
|
||||
|
||||
|
|
@ -207,7 +208,7 @@ def test_remove_saves_once(tree, monkeypatch):
|
|||
saves = []
|
||||
save = o._OrvaniTag__save
|
||||
monkeypatch.setattr(o, '_OrvaniTag__save', lambda: saves.append(1) or save())
|
||||
assert o.remove('a.mp3', '#lofi', '#snare')[0] == 0
|
||||
o.remove('a.mp3', '#lofi', '#snare')
|
||||
assert len(saves) == 1
|
||||
assert lines(tree) == ['a.mp3 --- #keep']
|
||||
|
||||
|
|
@ -228,11 +229,11 @@ def test_import_untagged_files(tree):
|
|||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi', 'a note')
|
||||
o.add('sub/d.txt', '@docs')
|
||||
assert o.export('context.txt')[0] == 0
|
||||
o.export('context.txt')
|
||||
(tree / orvani.CONTEXT_FILE).unlink()
|
||||
(tree / 'sub' / orvani.CONTEXT_FILE).unlink()
|
||||
|
||||
assert OrvaniTag().load('context.txt') == [0, 'context.txt', 2]
|
||||
assert OrvaniTag().load('context.txt') == 2
|
||||
assert lines(tree) == ['a.mp3 --- #lofi a note']
|
||||
assert lines(tree / 'sub') == ['d.txt --- @docs']
|
||||
|
||||
|
|
@ -241,16 +242,45 @@ def test_import_skips_known_tags_and_empty_lines(tree):
|
|||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi')
|
||||
(tree / 'context.txt').write_text('./a.mp3 --- #lofi #snare\n./b.mp3 --- \n')
|
||||
assert OrvaniTag().load('context.txt') == [0, 'context.txt', 1]
|
||||
assert OrvaniTag().load('context.txt') == 1
|
||||
assert lines(tree) == ['a.mp3 --- #lofi #snare']
|
||||
|
||||
|
||||
def test_import_keeps_hyphenated_tags(tree):
|
||||
(tree / 'context.txt').write_text('./a.mp3 --- @home-office #lo-fi rock+roll take-2\n')
|
||||
assert OrvaniTag().load('context.txt')[0] == 0
|
||||
OrvaniTag().load('context.txt')
|
||||
assert lines(tree) == ['a.mp3 --- @home-office #lo-fi rock+roll take-2']
|
||||
|
||||
|
||||
def test_duplicate_and_bare_prefixes(tree):
|
||||
OrvaniTag().add('a.mp3', '#lofi', '#lofi', 'C', '+', 'D')
|
||||
assert lines(tree) == ['a.mp3 --- #lofi C + D']
|
||||
|
||||
|
||||
def test_move_reports_what_is_missing(tree):
|
||||
o = OrvaniTag()
|
||||
with pytest.raises(NotFound, match=r"^Source file x\.mp3 doesn't exist\.$"):
|
||||
o.move('x.mp3', 'sub/', False, True)
|
||||
with pytest.raises(NotFound, match=r"^Destination nope/a\.mp3 doesn't exist\.$"):
|
||||
o.move('a.mp3', 'nope/a.mp3', False, True)
|
||||
with pytest.raises(NotFound, match=r"^Source file x\.mp3 and destination nope/x don't exist\.$"):
|
||||
o.move('x.mp3', 'nope/x', False, True)
|
||||
|
||||
|
||||
def test_errors_of_remove_delete_export_and_import(tree):
|
||||
o = OrvaniTag()
|
||||
o.add('a.mp3', '#lofi')
|
||||
with pytest.raises(NoEntry, match='No match found for a.mp3 and #snare'):
|
||||
o.remove('a.mp3', '#snare')
|
||||
with pytest.raises(NoEntry):
|
||||
o.delete('b.mp3')
|
||||
with pytest.raises(NotFound):
|
||||
o.add('missing.mp3', '#x')
|
||||
with pytest.raises(NotFound):
|
||||
o.load('missing.txt')
|
||||
(tree / 'bad.txt').write_text('no separator on this line\n')
|
||||
with pytest.raises(FormatError):
|
||||
o.load('bad.txt')
|
||||
o.delete('a.mp3')
|
||||
with pytest.raises(NoEntry):
|
||||
o.export('context.txt')
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from pypdf import PdfReader, PdfWriter
|
|||
|
||||
import orvanitag.orvani as orvani
|
||||
from orvanitag.cmd import main
|
||||
from orvanitag.orvani import OrvaniTag
|
||||
from orvanitag.orvani import OrvaniTag, PluginError, UnsupportedType
|
||||
|
||||
DOCX_CORE = (
|
||||
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||||
|
|
@ -44,7 +44,7 @@ def make_odt(path):
|
|||
def roundtrip(cwd, name):
|
||||
o = OrvaniTag()
|
||||
o.add(name, '#lofi', '+proj', 'R&B', 'C:\\1')
|
||||
assert o.write_metadata(name, 'office')[0] == 0
|
||||
o.write_metadata(name, 'office')
|
||||
(cwd / orvani.CONTEXT_FILE).unlink()
|
||||
OrvaniTag().read_metadata(name)
|
||||
assert (cwd / orvani.CONTEXT_FILE).read_text() == f'{name} --- +proj #lofi R&B C:\\1\n'
|
||||
|
|
@ -76,7 +76,7 @@ def test_pdf_roundtrip(cwd):
|
|||
|
||||
o = OrvaniTag()
|
||||
o.add('doc.pdf', '#lofi', '@desk', 'a note')
|
||||
assert o.write_metadata('doc.pdf', 'pdf')[0] == 0
|
||||
o.write_metadata('doc.pdf', 'pdf')
|
||||
meta = PdfReader(cwd / 'doc.pdf').metadata
|
||||
assert meta['/Title'] == 'T'
|
||||
|
||||
|
|
@ -130,7 +130,8 @@ def test_write_uses_config_of_the_files_folder(cwd):
|
|||
writer.write(cwd / 'sub' / 'doc.pdf')
|
||||
|
||||
OrvaniTag().add('sub/doc.pdf', '#lofi')
|
||||
assert OrvaniTag().write_metadata('sub/doc.pdf', 'pdf') == [6, 'pdf']
|
||||
with pytest.raises(PluginError):
|
||||
OrvaniTag().write_metadata('sub/doc.pdf', 'pdf')
|
||||
|
||||
|
||||
def test_cmd_scan_defaults_to_current_folder(cwd, monkeypatch):
|
||||
|
|
@ -149,7 +150,8 @@ def test_write_ignores_configured_types_the_plugin_cannot_handle(cwd):
|
|||
(cwd / orvani.CONFIG_FILE).write_text('[pdf]\ntypes = [".xyz", ".abc", ".pdf"]\n')
|
||||
(cwd / 'doc.abc').write_text('not a pdf')
|
||||
OrvaniTag().add('doc.abc', '#lofi')
|
||||
assert OrvaniTag().write_metadata('doc.abc', 'pdf') == [4, ['.pdf'], 'pdf', True]
|
||||
with pytest.raises(UnsupportedType, match=r'types: \.pdf\. Its types are set in \.otag\.toml'):
|
||||
OrvaniTag().write_metadata('doc.abc', 'pdf')
|
||||
assert (cwd / 'doc.abc').read_text() == 'not a pdf'
|
||||
|
||||
|
||||
|
|
@ -160,3 +162,11 @@ def test_pdf_keeps_hyphenated_tags(cwd):
|
|||
writer.write(cwd / 'doc.pdf')
|
||||
OrvaniTag().read_metadata('doc.pdf')
|
||||
assert (cwd / orvani.CONTEXT_FILE).read_text() == 'doc.pdf --- +side-project #lo-fi\n'
|
||||
|
||||
|
||||
def test_cmd_user_errors_exit_with_1(cwd, monkeypatch, capsys):
|
||||
monkeypatch.setattr(sys, 'argv', ['otag', 'a', 'missing.mp3', '#lofi'])
|
||||
with pytest.raises(SystemExit) as exit:
|
||||
main()
|
||||
assert exit.value.code == 1
|
||||
assert "File missing.mp3 doesn't exist." in capsys.readouterr().out
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue