From 921f06b4face22f00e19dbcabe4fc56c5f89cc29 Mon Sep 17 00:00:00 2001 From: randogoth Date: Sat, 24 Oct 2026 21:55:16 +0300 Subject: [PATCH] refactor: add Entry type that owns the tag syntax and drop the configurable schema --- orvanitag/cmd.py | 21 ++--- orvanitag/entry.py | 90 +++++++++++++++++++ orvanitag/orvani.py | 215 +++++++++++--------------------------------- tests/test_entry.py | 35 ++++++++ 4 files changed, 187 insertions(+), 174 deletions(-) create mode 100644 orvanitag/entry.py create mode 100644 tests/test_entry.py diff --git a/orvanitag/cmd.py b/orvanitag/cmd.py index 19a49a0..9cb259e 100644 --- a/orvanitag/cmd.py +++ b/orvanitag/cmd.py @@ -7,6 +7,7 @@ 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, walk +from orvanitag.entry import Entry from orvanitag.orvani import OrvaniError, NotFound, NoEntry, UnsupportedType, PluginError # Checked in order, so the base class comes last @@ -83,21 +84,17 @@ def mktable(query: str, index: dict, filter=None, trunc_len: int=25): else: console.print(f":thinking_face: [red]No results found for [white]{query}") -def nice_context(context: dict): +def nice_context(entry: Entry): """ - Rich colors for displaying contextual data - - context (dict): context dict + Rich colors for displaying an entry """ string = '' - if context.get('projects'): - string += '[cyan]' + ' '.join(['+' + tag for tag in context['projects']]) + '[/cyan] ' - if context.get('contexts'): - string += '[yellow]' + ' '.join(['@' + tag for tag in context['contexts']]) + '[/yellow] ' - if context.get('tags'): - string += '[magenta]' + ' '.join(['#' + tag for tag in context['tags']]) + '[/magenta] ' - if context.get('note'): - string += '[green]\'' + context['note'] + '\'[/green] ' + for name, color in [('projects', 'cyan'), ('contexts', 'yellow'), ('tags', 'magenta')]: + values = getattr(entry, name) + if values: + string += f'[{color}]' + escape(' '.join(CONTEXTS[name] + value for value in values)) + f'[/{color}] ' + if entry.note: + string += f"[green]'{escape(entry.note)}'[/green] " return string def main(): diff --git a/orvanitag/entry.py b/orvanitag/entry.py new file mode 100644 index 0000000..7cbe911 --- /dev/null +++ b/orvanitag/entry.py @@ -0,0 +1,90 @@ +""" +The tags of one file in todo.txt style: `+project @context #tag` words plus a free text note. +""" +from dataclasses import dataclass, field + +# Categories and their prefixes; the note has none +CONTEXTS = { + 'projects' : '+', + 'contexts' : '@', + 'tags' : '#', + 'note' : None +} +PREFIXES = {prefix: name for name, prefix in CONTEXTS.items() if prefix} +LISTS = list(PREFIXES.values()) + + +@dataclass +class Entry: + projects: list = field(default_factory=list) + contexts: list = field(default_factory=list) + tags: list = field(default_factory=list) + note: str = None + + @classmethod + def parse(cls, words) -> 'Entry': + """ + Words with a prefix become projects, contexts and tags; all others form the note. + A bare prefix is a note word, trailing commas of tags are dropped. + """ + entry = cls() + note = [] + for word in [str(word) for word in words if word and str(word).strip()]: + tag = word[1:].replace(',', '') + if word[0] in PREFIXES and tag: + values = getattr(entry, PREFIXES[word[0]]) + if tag not in values: + values.append(tag) + else: + note.append(word) + entry.note = ' '.join(note) or None + return entry + + @classmethod + def from_dict(cls, data: dict) -> 'Entry': + """Reads the form used by the search index and plugins""" + entry = cls(note=data.get('note') or None) + for name in LISTS: + value = data.get(name) or [] + setattr(entry, name, [str(item) for item in value] if isinstance(value, list) else [str(value)]) + return entry + + def to_dict(self) -> dict: + data = {name: list(getattr(self, name)) for name in LISTS} + if self.note: + data['note'] = self.note + return data + + def __str__(self) -> str: + words = [CONTEXTS[name] + value for name in LISTS for value in getattr(self, name)] + if self.note: + words.append(self.note) + return ' '.join(words) + + def __bool__(self) -> bool: + return bool(self.note) or any(getattr(self, name) for name in LISTS) + + def merge(self, other: 'Entry'): + """Adds the other entry's projects, contexts and tags; its note replaces this one""" + for name in LISTS: + setattr(self, name, list(dict.fromkeys(getattr(self, name) + getattr(other, name)))) + if other.note: + self.note = other.note + + def remove(self, other: 'Entry') -> bool: + """Removes the other entry's projects, contexts and tags; returns whether any were found""" + removed = False + for name in LISTS: + values = getattr(self, name) + for value in getattr(other, name): + if value in values: + values.remove(value) + removed = True + return removed + + def without(self, other: 'Entry') -> 'Entry': + """The parts of this entry the other one does not have yet""" + return Entry( + *[[value for value in getattr(self, name) if value not in getattr(other, name)] for name in LISTS], + note=self.note if self.note != other.note else None, + ) diff --git a/orvanitag/orvani.py b/orvanitag/orvani.py index 0ec115c..9761a7b 100644 --- a/orvanitag/orvani.py +++ b/orvanitag/orvani.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from pathlib import Path from importlib import import_module from types import SimpleNamespace -from pprint import pprint +from orvanitag.entry import Entry, CONTEXTS, PREFIXES, LISTS # Defaults @@ -14,12 +14,6 @@ CONTEXT_FILE = '.otag' CONFIG_FILE = '.otag.toml' CACHE_FILE = '.otag.cache.json' SEPARATOR = ' --- ' # between file name and contexts, in context files and exports -CONTEXTS = { - 'projects' : '+', - 'contexts' : '@', - 'tags' : '#', - 'note' : None -} DEFAULT_PLUGINS = [ 'frontmatter', @@ -50,7 +44,7 @@ class Folder: """The tag entries of one folder and the context file they are stored in""" path: Path file: Path - entries: dict + entries: dict # file name -> Entry class OrvaniError(Exception): @@ -75,25 +69,23 @@ class OrvaniTag: ENABLE_PLUGINS = True # Toggle use of plugins - def __init__(self, path: str=CONTEXT_FILE, cache: str=CACHE_FILE, contexts=CONTEXTS, plugins:list=DEFAULT_PLUGINS): + def __init__(self, path: str=CONTEXT_FILE, cache: str=CACHE_FILE, plugins:list=DEFAULT_PLUGINS): """ Initialization path (str): override default filename for folder context cache (str): override default filename for cached index - contexts (dict): override the default contexts """ self.PLUGINS = list(plugins) # List of plugin names self.MODULES = None # Imported plugin modules, see __plugins() 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.INDEX = self.__emptyindex() # our search index def __emptyindex(self) -> dict: - index = {context_key: {} for context_key in self.SCHEMA} + index = {context_key: {} for context_key in CONTEXTS} index['items'] = {} return index @@ -135,7 +127,7 @@ class OrvaniTag: # Raising instead of skipping, since the next save would silently drop the 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()) + db[name] = Entry.parse(string.split()) return db @@ -146,7 +138,7 @@ class OrvaniTag: file = folder.file lines = [] for name in sorted(folder.entries): - contexts = ' '.join(self.__collect_contexts(folder.entries[name])) + contexts = str(folder.entries[name]) if contexts: # entries without any context are dropped slash = '/' if (file.parent / name).is_dir() else '' lines.append(f'{name}{slash}{SEPARATOR}{contexts}\n') @@ -166,47 +158,12 @@ class OrvaniTag: tmp.unlink(missing_ok=True) - def __context(self, argv: list) -> dict: - """ - Creates a context entry for an existing file according to the schema - - argv (list): sets file contexts according to context schema - """ - context = dict() - # Create context dict according to schema - for context_key in self.SCHEMA.keys(): - if self.SCHEMA[context_key] != None: - context.update( { context_key : []}) - - # Parse the argument list - note = [] - for arg in [str(x) for x in argv if x and str(x).strip()]: # remove empty list elements - - # Add each argument list element to context dict based on the - # prefix symbols defined in the schema - tag = arg[1:].replace(',', '') - if arg[0] in self.SCHEMA.values() and tag: - for context_key in self.SCHEMA.keys(): - if self.SCHEMA[context_key] != None and arg.startswith(self.SCHEMA[context_key]): - if tag not in context[context_key]: - context[context_key].append(tag) - else: - # No prefix (or a bare prefix), so it is part of the note - note.append(arg) - - if note: - for context_key in self.SCHEMA.keys(): - if self.SCHEMA[context_key] == None: - context.update( { context_key : ' '.join(note)}) - return context - - - def __add(self, path: str, context: dict): - """Programmatically add context""" - ctx_str = self.__collect_contexts(context) - if ctx_str: - argv = self.__get_argv(' '.join(ctx_str)) - self.add(path, *argv) + def __merge(self, path, entry: Entry): + """Adds an entry to the tags of an existing file and saves its folder""" + target = Path(path) + folder = self.__load(target.parent) + folder.entries.setdefault(target.name, Entry()).merge(entry) + self.__save(folder) def __makeindex(self, path:str, recursive:bool=True, cache: bool=False, delete_cache: bool=False): @@ -241,7 +198,7 @@ class OrvaniTag: self.__indexfolder(folder) # Paths are collected in sets while indexing and sorted once here - for context_key in self.SCHEMA.keys(): + for context_key in CONTEXTS: self.INDEX[context_key] = {item: sorted(paths) for item, paths in self.INDEX[context_key].items()} self.INDEX['items'] = dict(sorted(self.INDEX['items'].items())) @@ -255,26 +212,25 @@ class OrvaniTag: folder (Path): path of folder """ - for file, context in self.__load(folder).entries.items(): - self.__updateindex(key(folder / file), context) + for file, entry in self.__load(folder).entries.items(): + self.__updateindex(key(folder / file), entry) - def __updateindex(self, path: str, context: dict): + def __updateindex(self, path: str, entry: Entry): """ - Adds the entries of a file to the global context index. - The context entry is associated with the file in the 'items' index - The file is associated with each of its context entries + Adds the entry of a file to the global context index. + The entry is associated with the file in the 'items' index + The file is associated with each of its projects, contexts, tags and its note path (str): path of the file - context (dict): context entry of the file + entry (Entry): tags of the file """ - self.INDEX['items'][path] = dict(context) - for context_key in self.SCHEMA.keys(): - value = context.get(context_key) - if not value: - continue - for item in (value if isinstance(value, list) else [value]): + self.INDEX['items'][path] = entry.to_dict() + for context_key in LISTS: + for item in getattr(entry, context_key): self.INDEX[context_key].setdefault(item, set()).add(path) + if entry.note: + self.INDEX['note'].setdefault(entry.note, set()).add(path) def __search(self, search, exact:bool=False, either:bool=False): @@ -310,11 +266,10 @@ class OrvaniTag: Returns the paths matching a single search term. A term without context prefix is a wildcard that searches all contexts at once. """ - prefixes = {prefix: context_key for context_key, prefix in self.SCHEMA.items() if prefix} - if term[0] in prefixes and len(term) > 1: - context_keys, needle = [prefixes[term[0]]], term[1:] + if term[0] in PREFIXES and len(term) > 1: + context_keys, needle = [PREFIXES[term[0]]], term[1:] else: - context_keys, needle = self.SCHEMA.keys(), term + context_keys, needle = CONTEXTS.keys(), term found = set() for context_key in context_keys: @@ -324,35 +279,6 @@ class OrvaniTag: return found - def __collect_contexts(self, context: dict): - """ - Reuturns a list of prefixed context entries from a context dict - """ - elements = [] - context_keys = list(CONTEXTS.keys()) - for context_key in context_keys: - if not context_key in context: - context[context_key] = '' - if context[context_key]: - if CONTEXTS[context_key]: - prefix = CONTEXTS[context_key] - else: - prefix = '' - if not type(context[context_key]) == str: - elements.append(' '.join([prefix + tag for tag in context[context_key]])) - else: - elements.append(context[context_key]) - return elements - - - def __get_argv(self, context_string: str): - """ - Make a list of args from a context string to pass as an *argv parameter. - Words are classified by __context, the same way as command line arguments. - """ - return context_string.split() - - def add(self, file: str, *argv: str): """ Add context to existing filename and saves local folder context file. @@ -360,32 +286,15 @@ class OrvaniTag: file (str): path to existing file argv (list): context parameters passed to be added + Returns the added entry """ target = Path(file) if not target.exists(): # Only existing files and folders raise NotFound(f"File {file} doesn't exist.") - folder = self.__load(target.parent) # Load folder context - context = self.__context(argv) # Create context dict from argument list - entry = folder.entries.get(target.name) - - if entry: # 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 - entry[context_key] = list(dict.fromkeys( entry[context_key] + context[context_key])) - else: - # Add the note - if context_key in context: - entry[context_key] = context[context_key] - - else: # If file or folder does not exist in context DB yet, add it - folder.entries[target.name] = context - - self.__save(folder) - return context + entry = Entry.parse(argv) + self.__merge(target, entry) + return entry def remove(self, file: str, *argv: str): @@ -402,8 +311,8 @@ class OrvaniTag: raise NotFound(f"File {file} doesn't exist.") folder = self.__load(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)))}.") + context = Entry.parse(argv) + no_match = NoEntry(f"No match found for {file} and {context}.") entry = folder.entries.get(target.name) if entry is None: # Does the file exist in our local context DB? raise no_match @@ -413,15 +322,7 @@ class OrvaniTag: self.__save(folder) return None - 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 entry[context_key]: - entry[context_key].remove(item) - removed = True - if not removed: + if not entry.remove(context): # only projects, contexts and tags can be removed raise no_match self.__save(folder) return context @@ -487,7 +388,7 @@ class OrvaniTag: shutil.copy2(src, target) if not retain: src.unlink() - return sourcedata or {} + return sourcedata if sourcedata is not None else Entry() def clean(self, path: str='.', recursive=True): @@ -539,12 +440,9 @@ class OrvaniTag: self.__makeindex(path) # index the directories return self.INDEX - def context(self, filepath: str): - if key(filepath) in self.INDEX['items']: - context = self.INDEX['items'][key(filepath)] - return self.__collect_contexts(context) - else: - return [] + def context(self, filepath: str) -> Entry: + """The entry of a file in the last built index""" + return Entry.from_dict(self.INDEX['items'].get(key(filepath), {})) def scan(self, path: str, recursive: bool=False): @@ -590,9 +488,9 @@ class OrvaniTag: obj = { 'file' : Path(), 'config' : config, - 'schema' : self.SCHEMA, - 'context' : self.__context, - 'string' : self.__get_argv + 'schema' : CONTEXTS, + 'context' : lambda words: Entry.parse(words).to_dict(), + 'string' : str.split } o = SimpleNamespace(**obj) @@ -614,9 +512,9 @@ class OrvaniTag: # Process the file using the reader plugin try: - context = plugin.reader(o) - if context: - self.__add(filename, context) + entry = Entry.from_dict(plugin.reader(o) or {}) + if entry: + self.__merge(filename, entry) print(f'Found Metadata for `{filename}`.') except Exception as e: print(f'Plugin Error ({plugin_name} reading {filename}): `{e}`') @@ -650,10 +548,10 @@ class OrvaniTag: obj = { 'file' : Path(), 'config' : config, - 'schema' : self.SCHEMA, + 'schema' : CONTEXTS, 'context' : {}, - 'collect' : self.__collect_contexts, - 'string' : self.__get_argv + 'collect' : lambda context: str(Entry.from_dict(context)).split(), + 'string' : str.split } o = SimpleNamespace(**obj) @@ -687,7 +585,7 @@ class OrvaniTag: 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 + return Entry.from_dict(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}") @@ -717,8 +615,7 @@ class OrvaniTag: content = "" if len(items): for path in items: - string = ' '.join([x for x in self.__collect_contexts(items[path]) if x != None]) - content += f'{path} --- {string}\n' + content += f'{path}{SEPARATOR}{Entry.from_dict(items[path])}\n' file.write_text(content) return filename else: @@ -752,16 +649,10 @@ class OrvaniTag: 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)) # 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) + entry = Entry.parse(string.split()).without(Entry.from_dict(known.get(path, {}))) + if entry: + self.__merge(path, entry) count += 1 else: raise FormatError(f"File {filename} has wrong format. Cannot import.") diff --git a/tests/test_entry.py b/tests/test_entry.py new file mode 100644 index 0000000..453bc94 --- /dev/null +++ b/tests/test_entry.py @@ -0,0 +1,35 @@ +from orvanitag.entry import Entry + + +def test_parse_and_format(): + entry = Entry.parse(['#lofi,', '+album', 'first', '#lofi', '@home-office', 'C', '+', 'take']) + assert entry == Entry(['album'], ['home-office'], ['lofi'], 'first C + take') + assert str(entry) == '+album @home-office #lofi first C + take' + assert Entry.parse(str(entry).split()) == entry + + +def test_empty_entry_is_false(): + assert not Entry.parse([]) + assert not Entry.parse(['', ' ']) + assert Entry.parse(['note']) + + +def test_dict_round_trip_and_tolerance(): + entry = Entry(['album'], [], ['lofi'], 'a note') + assert entry.to_dict() == {'projects': ['album'], 'contexts': [], 'tags': ['lofi'], 'note': 'a note'} + assert Entry.from_dict(entry.to_dict()) == entry + assert Entry.from_dict({'tags': 'single', 'projects': [2023]}) == Entry(['2023'], [], ['single']) + + +def test_merge_remove_without(): + entry = Entry(tags=['lofi'], note='old') + entry.merge(Entry(tags=['lofi', 'snare'], note='new')) + assert entry == Entry(tags=['lofi', 'snare'], note='new') + entry.merge(Entry(tags=['x'])) + assert entry.note == 'new' + + assert entry.remove(Entry(tags=['snare', 'missing'])) + assert not entry.remove(Entry(tags=['missing'])) + assert entry.tags == ['lofi', 'x'] + + assert entry.without(Entry(tags=['lofi'], note='new')) == Entry(tags=['x'])