randogoth/orvanitag
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: add Entry type that owns the tag syntax and drop the configurable schema

Browse source
This commit is contained in:
randogoth 2026-10-24 21:55:16 +03:00
parent 6fbd54c754
commit 921f06b4fa
4 changed files with 187 additions and 174 deletions
Download patch file Download diff file Expand all files Collapse all files

21
orvanitag/cmd.py
View file

@ -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():

90
orvanitag/entry.py Normal file
View file

@ -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,
)

215
orvanitag/orvani.py
View file

@ -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.")

35
tests/test_entry.py Normal file
View file

@ -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'])