From 93661d664a1cc60737b8993c472ada84cd77c8e0 Mon Sep 17 00:00:00 2001 From: randogoth Date: Sat, 24 Oct 2026 22:15:10 +0300 Subject: [PATCH] refactor: give plugins an Entry based read/write interface and stop printing from the library --- README.md | 2 +- orvanitag/cmd.py | 11 +- orvanitag/errors.py | 24 +++++ orvanitag/orvani.py | 176 +++++++++++-------------------- orvanitag/plugins/__init__.py | 14 +++ orvanitag/plugins/comment.py | 26 ++--- orvanitag/plugins/ebook.py | 44 ++++---- orvanitag/plugins/frontmatter.py | 42 ++++---- orvanitag/plugins/id3.py | 31 +++--- orvanitag/plugins/office.py | 24 +++-- orvanitag/plugins/pdf.py | 18 ++-- tests/test_comment.py | 3 +- tests/test_ebook.py | 3 +- tests/test_plugin_interface.py | 54 ++++++++++ 14 files changed, 263 insertions(+), 209 deletions(-) create mode 100644 orvanitag/errors.py create mode 100644 tests/test_plugin_interface.py diff --git a/README.md b/README.md index 268179c..34e0e72 100644 --- a/README.md +++ b/README.md @@ -151,4 +151,4 @@ types = [".mp3", ".flac"] tags = "genre" ``` -`plugins` limits which plugins run, `types` overrides the file types a plugin handles (`write` ignores types the plugin does not support), and `id3.map` additionally reads an audio tag (here `GENRE`) into a category. +`plugins` limits which plugins run, `types` narrows the file types a plugin handles (types the plugin does not support are ignored), and `id3.map` additionally reads an audio tag (here `GENRE`) into a category. diff --git a/orvanitag/cmd.py b/orvanitag/cmd.py index a9831da..96f4eba 100644 --- a/orvanitag/cmd.py +++ b/orvanitag/cmd.py @@ -8,13 +8,14 @@ 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 +from orvanitag.errors import OrvaniError, NotFound, NoEntry, UnsupportedType, UnsupportedFile, PluginError # Checked in order, so the base class comes last ERROR_ICONS = { NotFound: ':thinking_face:', NoEntry: ':mag:', UnsupportedType: ':card_file_box: ', + UnsupportedFile: ':card_file_box: ', PluginError: ':disappointed: ', OrvaniError: ':thinking_face:', } @@ -167,8 +168,12 @@ def run_import(orvani, args, console): def run_scan(orvani, args, console): - orvani.scan(args.path, args.recursive) - console.print(f':mag: Finished scanning subdirectories.') + found, errors = orvani.scan(args.path, args.recursive) + for path, entry in found.items(): + console.print(f':label: Found {nice_context(entry)}in [strong]{escape(path)}[/strong].') + for error in errors: + console.print(f':disappointed: {escape(error)}') + console.print(f':mag: Found embedded tags in {len(found)} files.') def parser() -> ArgumentParser: diff --git a/orvanitag/errors.py b/orvanitag/errors.py new file mode 100644 index 0000000..3300868 --- /dev/null +++ b/orvanitag/errors.py @@ -0,0 +1,24 @@ +""" +Errors caused by the user's input or files, which the command line reports without a traceback. +""" + +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 UnsupportedFile(OrvaniError): + """A plugin cannot store tags in this particular file, e.g. one without the expected structure""" diff --git a/orvanitag/orvani.py b/orvanitag/orvani.py index e21674a..e160877 100644 --- a/orvanitag/orvani.py +++ b/orvanitag/orvani.py @@ -5,8 +5,8 @@ import tomllib from dataclasses import dataclass from pathlib import Path from importlib import import_module -from types import SimpleNamespace from orvanitag.entry import Entry, CONTEXTS, PREFIXES, LISTS +from orvanitag.errors import OrvaniError, NotFound, NoEntry, UnsupportedType, UnsupportedFile, PluginError, FormatError # Defaults @@ -48,24 +48,6 @@ class Folder: entries: dict # file name -> Entry -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: ENABLE_PLUGINS = True # Toggle use of plugins @@ -456,14 +438,24 @@ class OrvaniTag: return Entry.from_dict(self.INDEX['items'].get(key(filepath), {})) def scan(self, path: str, recursive: bool=False): + """ + Adds the tags embedded in the files of a folder to their entries + Returns the found entries by path and the messages of failed plugin reads + """ + found, errors = {}, [] for folder, files in walk(path, recursive): - # process every file with a reader plugin - if self.ENABLE_PLUGINS: - for file in files: - self.read_metadata(folder / file) + for file in files: + try: + entry = self.read_metadata(folder / file) + except PluginError as error: + errors.append(str(error)) + continue + if entry: + found[key(folder / file)] = entry + return found, errors + - def register_plugin(self, function): """ Add a plugin to the plugin list @@ -488,47 +480,40 @@ class OrvaniTag: return self.MODULES - def read_metadata(self, filename: str): + def __enabled(self, config: dict): """ - Processes a file with all reader plugins to extract its metadata + Yields the plugins a folder configuration enables, with the file types they handle there. + A configured `types` list can only narrow the plugin's own types. """ - - config = self.config(Path(filename).parent) - - # Utilities for reader plugins - obj = { - 'file' : Path(), - 'config' : config, - 'schema' : CONTEXTS, - 'context' : lambda words: Entry.parse(words).to_dict(), - 'string' : str.split - } - o = SimpleNamespace(**obj) - for plugin_name, plugin in self.__plugins().items(): - # Check if plugin is (de)activated by a local folder configuration - if 'plugins' in config: - if plugin_name not in config['plugins']: + if 'plugins' in config and plugin_name not in config['plugins']: + continue + configured = config.get(plugin_name, {}).get('types', []) + yield plugin_name, plugin, [suffix for suffix in configured if suffix in plugin.TYPES] or plugin.TYPES + + + def read_metadata(self, filename: str) -> Entry: + """ + Reads the tags embedded in a file with every plugin that handles it and adds them to its entry + + Returns the found tags; raises PluginError naming the plugins that failed + """ + path = Path(filename) + config = self.config(path.parent) + found, errors = Entry(), [] + if self.ENABLE_PLUGINS: + for plugin_name, plugin, suffixes in self.__enabled(config): + if path.suffix.lower() not in suffixes or not hasattr(plugin, 'read'): continue - - # Load a list of file types the plugin is able to read - suffixes = plugin.TYPES # The plugin's default file types - if plugin_name in config: # check for local folder config overrides - if 'types' in config[plugin_name]: - suffixes = config[plugin_name]['types'] - - o.file = Path(filename) - if o.file.suffix in suffixes: # Does the file match a permitted file type - if hasattr(plugin, 'reader'): # Does the plugin have a reader method? - - # Process the file using the reader plugin - try: - 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}`') + try: + found.merge(plugin.read(path, config) or Entry()) + except Exception as error: # a broken file must not stop a scan + errors.append(f'{plugin_name} reading {path}: {error}') + if found: + self.__merge(path, found) + if errors: + raise PluginError('; '.join(errors)) + return found def write_all(self, path: str, method: str, recursive: bool=False): @@ -543,63 +528,30 @@ class OrvaniTag: pass - def write_metadata(self, filename: str, method: str): + def write_metadata(self, filename: str, method: str) -> Entry: """ - Writes a file's context entries as metadata to the file using a writer plugin + Writes a file's entry as metadata to the file using a writer plugin filename (str): A valid file path method (str): The name of a plugin able to process the file type - Returns the written context + Returns the written entry """ - - config = self.config(Path(filename).parent) - - # Utilities for writer plugins - obj = { - 'file' : Path(), - 'config' : config, - 'schema' : CONTEXTS, - 'context' : {}, - 'collect' : lambda context: str(Entry.from_dict(context)).split(), - 'string' : str.split - } - o = SimpleNamespace(**obj) - - for plugin_name, plugin in self.__plugins().items(): - - # Check if plugin is (de)activated by a local folder configuration - if 'plugins' in config: - if plugin_name not in config['plugins']: - continue - - if plugin.NAME == method: # select the reference plugin - - # Load a list of file types the plugin is able to read - suffixes = plugin.TYPES # Load the plugin's defaults - if plugin_name in config: # Check for local folder config overrides - if 'types' in config[plugin_name]: - # Only types the plugin can write, falling back to its defaults - suffixes = [suffix for suffix in config[plugin_name]['types'] if suffix in plugin.TYPES] or plugin.TYPES - - o.file = Path(filename) - if o.file.suffix in suffixes: - - # Get all local context entries (including file metadata) - self.__makeindex(o.file.parent, recursive=False) - items = self.INDEX['items'] - - # 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 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}") + path = Path(filename) + config = self.config(path.parent) + for plugin_name, plugin, suffixes in self.__enabled(config): + if plugin.NAME != method: # select the reference plugin + continue + if path.suffix.lower() not in suffixes: + configured = f' Its types are set in {CONFIG_FILE}.' if 'types' in config.get(plugin_name, {}) else '' + raise UnsupportedType(f"Plugin {plugin_name} can only write to file types: {', '.join(suffixes)}.{configured}") + if not hasattr(plugin, 'write'): + raise PluginError(f"Plugin {plugin_name} can't write.") + entry = self.__load(path.parent).entries.get(path.name) + if not entry: + raise NoEntry(f"No entry for {filename} found.") + plugin.write(path, entry, config) + return entry raise PluginError(f"Plugin {method} not found or its dependencies are not installed.") diff --git a/orvanitag/plugins/__init__.py b/orvanitag/plugins/__init__.py index e69de29..02bdeb4 100644 --- a/orvanitag/plugins/__init__.py +++ b/orvanitag/plugins/__init__.py @@ -0,0 +1,14 @@ +""" +Plugins read and write tags embedded in files. A plugin is a module with: + +NAME the name used by `otag write NAME FILE` +TYPES the lowercase file suffixes it handles, e.g. ['.pdf'] +read(path: Path, config: dict) -> Entry | None + returns the tags found in the file, or None +write(path: Path, entry: Entry, config: dict) + stores the entry in the file, optional for read-only plugins + +`config` is the file's folder configuration from `.otag.toml`. +Expected problems with a file raise orvanitag.errors.UnsupportedFile. +Modules starting with `_` are helpers, not plugins. +""" diff --git a/orvanitag/plugins/comment.py b/orvanitag/plugins/comment.py index ceb2f66..13e1c7f 100644 --- a/orvanitag/plugins/comment.py +++ b/orvanitag/plugins/comment.py @@ -4,6 +4,8 @@ Tags in a source file as a single comment line near the top, like `# otag: +proj import os import re import shutil +from orvanitag.entry import Entry +from orvanitag.errors import UnsupportedFile NAME = 'comment' MARKER = 'otag:' @@ -62,42 +64,42 @@ def header(lines: list, suffix: str) -> int: if suffix == '.php': # Outside of the PHP block the comment would be sent as output if not lines or lines[0].strip() != ' str: """The name of the EPUB's OPF package file""" found = ROOTFILE.search(archive.read('META-INF/container.xml').decode('utf-8')) if not found: - raise ValueError('EPUB without package file') + raise UnsupportedFile('EPUB without package file') return unescape(found.group(1)) @@ -44,24 +46,24 @@ def read_fb2(path): return data.decode(encoding), encoding, bom -def reader(o): - suffix = o.file.suffix.lower() +def read(path, config): + suffix = path.suffix.lower() text = '' if suffix == '.epub': - with ZipFile(o.file) as archive: + with ZipFile(path) as archive: opf = archive.read(package(archive)).decode('utf-8') meta = META.search(opf) content = CONTENT.search(meta.group(0)) if meta else None text = unescape(content.group(2), {'"': '"'}) if content else '' elif suffix == '.fb2': - found = CUSTOM.search(read_fb2(o.file)[0]) + found = CUSTOM.search(read_fb2(path)[0]) text = unescape(found.group(1)) if found else '' elif suffix == '.cbz': - with ZipFile(o.file) as archive: + with ZipFile(path) as archive: comment = archive.comment.decode('utf-8', 'replace') text = comment.removeprefix(MARKER) if comment.startswith(MARKER) else '' if text.strip(): - return o.context(o.string(text)) + return Entry.parse(text.split()) def epub_meta(opf: str, line: str) -> str: @@ -71,7 +73,7 @@ def epub_meta(opf: str, line: str) -> str: if not found: found = METADATA_END.search(opf) if not found: - raise ValueError('EPUB package file without metadata') + raise UnsupportedFile('EPUB package file without metadata') start = end = found.start() return opf[:start] + f'<{found.group(1) or ""}meta name="otag" content="{escape(line, QUOTES)}"/>' + opf[end:] @@ -83,33 +85,33 @@ def fb2_custom_info(text: str, line: str) -> str: return text[:found.start(1)] + escape(line) + text[found.end(1):] found = DESCRIPTION_END.search(text) if not found: - raise ValueError('FB2 file without description') + raise UnsupportedFile('FB2 file without description') prefix = found.group(1) or found.group(2) or '' return text[:found.start()] + f'<{prefix}custom-info info-type="otag">{escape(line)}' + text[found.start():] -def writer(o): - suffix = o.file.suffix.lower() - line = ' '.join(o.collect(o.context)) +def write(path, entry, config): + suffix = path.suffix.lower() + line = str(entry) if suffix == '.epub': - with ZipFile(o.file) as archive: + with ZipFile(path) as archive: name = package(archive) - rewrite(o.file, {name: lambda opf: epub_meta(opf, line)}) + rewrite(path, {name: lambda opf: epub_meta(opf, line)}) elif suffix == '.fb2': - text, encoding, bom = read_fb2(o.file) + text, encoding, bom = read_fb2(path) data = fb2_custom_info(text, line).encode(encoding, 'xmlcharrefreplace') - tmp = o.file.with_name(f'.{o.file.name}.tmp') + tmp = path.with_name(f'.{path.name}.tmp') try: tmp.write_bytes((codecs.BOM_UTF8 if bom else b'') + data) - shutil.copymode(o.file, tmp) - os.replace(tmp, o.file) + shutil.copymode(path, tmp) + os.replace(tmp, path) finally: tmp.unlink(missing_ok=True) elif suffix == '.cbz': - with ZipFile(o.file) as archive: + with ZipFile(path) as archive: comment = archive.comment if comment and not comment.startswith(MARKER.encode()): - raise ValueError(f'the archive comment of {o.file} is used by another tool') - rewrite(o.file, comment=f'{MARKER} {line}'.encode('utf-8')) + raise UnsupportedFile(f'the archive comment of {path} is used by another tool') + rewrite(path, comment=f'{MARKER} {line}'.encode('utf-8')) diff --git a/orvanitag/plugins/frontmatter.py b/orvanitag/plugins/frontmatter.py index c346dd3..6edef39 100644 --- a/orvanitag/plugins/frontmatter.py +++ b/orvanitag/plugins/frontmatter.py @@ -7,6 +7,7 @@ import os import re import shutil import yaml +from orvanitag.entry import Entry, LISTS NAME = 'frontmatter' TYPES = ['.md', '.markdown', '.txt'] @@ -41,8 +42,8 @@ def scalar(value: str) -> str: return json.dumps(value, ensure_ascii=False) # a JSON string is a valid YAML double-quoted scalar -def reader(o): - with o.file.open(encoding='utf-8') as data: +def read(path, config): + with path.open(encoding='utf-8') as data: lines = data.readlines() span = block(lines) if not span: @@ -52,26 +53,23 @@ def reader(o): if not isinstance(metadata, dict): return None - context = {} - for key, prefix in o.schema.items(): - if metadata.get(key): - context[key] = words(metadata[key]) if prefix else ' '.join(str(metadata[key]).split()) - if any(context.values()): - return context + entry = Entry(note=' '.join(str(metadata['note']).split()) if metadata.get('note') else None) + for name in LISTS: + if metadata.get(name): + setattr(entry, name, words(metadata[name])) + return entry or None -def writer(o): - with o.file.open(encoding='utf-8', newline='') as data: # keeps \r\n line endings +def write(path, entry, config): + with path.open(encoding='utf-8', newline='') as data: # keeps \r\n line endings text = data.read() lines = text.splitlines(keepends=True) newline = '\r\n' if '\r\n' in text else '\n' - entries = {} - for key, prefix in o.schema.items(): - value = o.context.get(key) - if value: # empty categories leave existing keys alone - rendered = f'[{", ".join(scalar(tag) for tag in value)}]' if prefix else scalar(value) - entries[key] = f'{key}: {rendered}{newline}' + # Empty categories leave existing keys alone + entries = {name: f'{name}: [{", ".join(scalar(tag) for tag in getattr(entry, name))}]{newline}' for name in LISTS if getattr(entry, name)} + if entry.note: + entries['note'] = f'note: {scalar(entry.note)}{newline}' if not entries: return @@ -80,24 +78,24 @@ def writer(o): lines[0:0] = [DELIMITER + newline, *entries.values(), DELIMITER + newline] else: start, end = span - for key, entry in entries.items(): + for key, line in entries.items(): found = next((num for num in range(start, end) if re.match(rf'{re.escape(key)}\s*:', lines[num])), None) if found is None: - lines.insert(end, entry) + lines.insert(end, line) end += 1 else: # The value continues on indented lines or top-level list items stop = found + 1 while stop < end and lines[stop].startswith((' ', '\t', '-')): stop += 1 - lines[found:stop] = [entry] + lines[found:stop] = [line] end -= stop - found - 1 - tmp = o.file.with_name(f'.{o.file.name}.tmp') + tmp = path.with_name(f'.{path.name}.tmp') try: with open(tmp, 'w', encoding='utf-8', newline='') as data: data.write(''.join(lines)) - shutil.copymode(o.file, tmp) - os.replace(tmp, o.file) + shutil.copymode(path, tmp) + os.replace(tmp, path) finally: tmp.unlink(missing_ok=True) diff --git a/orvanitag/plugins/id3.py b/orvanitag/plugins/id3.py index e80532d..fe2277b 100644 --- a/orvanitag/plugins/id3.py +++ b/orvanitag/plugins/id3.py @@ -1,30 +1,27 @@ import taglib +from orvanitag.entry import Entry, LISTS NAME = 'id3' TYPES = ['.mp3', '.wav', '.flac', '.ogg'] -def reader(o): - with taglib.File(o.file, save_on_exit=False) as song: +def read(path, config): + with taglib.File(path, save_on_exit=False) as song: metadata = song.tags - comment = metadata.get('COMMENT', [''])[0] - context = o.context(o.string(comment)) + entry = Entry.parse(metadata.get('COMMENT', [''])[0].split()) # Audio tags mapped to categories by the folder config, e.g. `[id3.map] tags = "genre"` - mapping = o.config.get('id3', {}).get('map', {}) - for context_key, tag in mapping.items(): + for name, tag in config.get('id3', {}).get('map', {}).items(): values = [value.strip() for value in metadata.get(str(tag).upper(), []) if value.strip()] - if context_key not in o.schema or not values: + if not values: continue - if o.schema[context_key] is None: # the note - context[context_key] = '; '.join(([context[context_key]] if context.get(context_key) else []) + values) - else: # spaces would split the value into a tag and note words - context[context_key] += ['-'.join(value.split()) for value in values] + if name == 'note': + entry.note = '; '.join(([entry.note] if entry.note else []) + values) + elif name in LISTS: # spaces would split the value into a tag and note words + entry.merge(Entry(**{name: ['-'.join(value.split()) for value in values]})) + return entry or None - if any(context.values()): - return context -def writer(o): - string = ' '.join(o.collect(o.context)) - with taglib.File(o.file, save_on_exit=True) as song: - song.tags['COMMENT'] = [string] \ No newline at end of file +def write(path, entry, config): + with taglib.File(path, save_on_exit=True) as song: + song.tags['COMMENT'] = [str(entry)] diff --git a/orvanitag/plugins/office.py b/orvanitag/plugins/office.py index c06603e..30b37bd 100644 --- a/orvanitag/plugins/office.py +++ b/orvanitag/plugins/office.py @@ -2,6 +2,8 @@ import re from xml.sax.saxutils import escape, unescape from zipfile import ZipFile from orvanitag.plugins._zip import rewrite +from orvanitag.entry import Entry +from orvanitag.errors import UnsupportedFile NAME = 'office' TYPES = ['.doc', '.xls', '.ppt', '.docx', '.xlsx', '.pptx', '.odt', '.ods', '.odp'] @@ -19,32 +21,32 @@ META_FILES = { LEGACY_TYPES = ['.doc', '.xls', '.ppt'] -def reader(o): - suffix = o.file.suffix.lower() +def read(path, config): + suffix = path.suffix.lower() # OpenDocument and Microsoft Office Open XML formats if suffix in META_FILES: - with ZipFile(o.file) as archive: + with ZipFile(path) as archive: meta = archive.read(META_FILES[suffix]).decode('utf-8') found = XML_REGEX.search(meta) if found and found.group('body'): - return o.context(o.string(unescape(found.group('body')))) + return Entry.parse(unescape(found.group('body')).split()) # Legacy Microsoft Office Format elif suffix in LEGACY_TYPES: # Imported here so the zip based formats work without the optional olefile dependency from olefile import OleFileIO - ole = OleFileIO(o.file) + ole = OleFileIO(path) meta = ole.get_metadata() ole.close() if meta.comments: - return o.context(o.string(meta.comments.decode('UTF-8'))) + return Entry.parse(meta.comments.decode('UTF-8').split()) -def writer(o): - suffix = o.file.suffix.lower() +def write(path, entry, config): + suffix = path.suffix.lower() if suffix not in META_FILES: - raise NotImplementedError(f'writing metadata to {suffix} files is not supported') + raise UnsupportedFile(f'writing metadata to {suffix} files is not supported') - string = escape(' '.join(o.collect(o.context))) - rewrite(o.file, {META_FILES[suffix]: lambda meta: XML_REGEX.sub(lambda m: m['pre'] + string + m['post'], meta)}) + string = escape(str(entry)) + rewrite(path, {META_FILES[suffix]: lambda meta: XML_REGEX.sub(lambda m: m['pre'] + string + m['post'], meta)}) diff --git a/orvanitag/plugins/pdf.py b/orvanitag/plugins/pdf.py index 8bd89b1..dd3415a 100644 --- a/orvanitag/plugins/pdf.py +++ b/orvanitag/plugins/pdf.py @@ -1,23 +1,25 @@ import os from pypdf import PdfReader, PdfWriter +from orvanitag.entry import Entry +from orvanitag.errors import UnsupportedFile NAME = 'pdf' TYPES = ['.pdf'] -def reader(o): - meta = PdfReader(o.file).metadata +def read(path, config): + meta = PdfReader(path).metadata keywords = meta.get('/Keywords') if meta else None if keywords and keywords.strip(): - return o.context(o.string(keywords.strip())) + return Entry.parse(keywords.strip().split()) -def writer(o): - string = ' '.join(o.collect(o.context)) - tmp = o.file.with_name(f'.{o.file.name}.tmp') +def write(path, entry, config): + string = str(entry) + tmp = path.with_name(f'.{path.name}.tmp') try: - with open(o.file, 'rb') as source: + with open(path, 'rb') as source: pdf = PdfWriter(clone_from=PdfReader(source)) pdf.add_metadata({'/Keywords': string}) pdf.write(tmp) - os.replace(tmp, o.file) + os.replace(tmp, path) finally: tmp.unlink(missing_ok=True) diff --git a/tests/test_comment.py b/tests/test_comment.py index 995e096..ddd4af4 100644 --- a/tests/test_comment.py +++ b/tests/test_comment.py @@ -7,6 +7,7 @@ import pytest import orvanitag.orvani as orvani from orvanitag.orvani import OrvaniTag +from orvanitag.errors import UnsupportedFile @pytest.fixture @@ -56,7 +57,7 @@ def test_php_tag_goes_inside_the_php_block(cwd): def test_php_without_own_opening_line_is_refused(cwd): - with pytest.raises(ValueError): + with pytest.raises(UnsupportedFile): write('p.php', '\n', '#demo') assert (cwd / 'p.php').read_text() == '\n' diff --git a/tests/test_ebook.py b/tests/test_ebook.py index f3d8c59..bae32cc 100644 --- a/tests/test_ebook.py +++ b/tests/test_ebook.py @@ -4,6 +4,7 @@ import pytest import orvanitag.orvani as orvani from orvanitag.orvani import OrvaniTag +from orvanitag.errors import UnsupportedFile CONTAINER = ( '' @@ -114,6 +115,6 @@ def test_cbz_foreign_comment_is_kept(cwd): make_cbz(cwd / 'comic.cbz', b'{"ComicBookInfo/1.0": {}}') before = (cwd / 'comic.cbz').read_bytes() OrvaniTag().add('comic.cbz', '#manga') - with pytest.raises(ValueError): + with pytest.raises(UnsupportedFile): OrvaniTag().write_metadata('comic.cbz', 'ebook') assert (cwd / 'comic.cbz').read_bytes() == before diff --git a/tests/test_plugin_interface.py b/tests/test_plugin_interface.py new file mode 100644 index 0000000..4a78b98 --- /dev/null +++ b/tests/test_plugin_interface.py @@ -0,0 +1,54 @@ +import sys + +import pytest +from pypdf import PdfWriter + +import orvanitag.orvani as orvani +from orvanitag.cmd import main +from orvanitag.entry import Entry +from orvanitag.orvani import OrvaniTag +from orvanitag.plugins import comment + + +@pytest.fixture +def cwd(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + return tmp_path + + +def test_plugins_take_and_return_entries(cwd): + (cwd / 'a.py').write_text('print(1)\n') + comment.write(cwd / 'a.py', Entry(tags=['demo'], note='a note'), {}) + assert comment.read(cwd / 'a.py', {}) == Entry(tags=['demo'], note='a note') + + +def test_scan_returns_found_tags_and_errors_without_printing(cwd, capsys): + writer = PdfWriter() + writer.add_blank_page(72, 72) + writer.add_metadata({'/Keywords': '#lofi'}) + writer.write(cwd / 'good.pdf') + (cwd / 'broken.epub').write_text('not a zip') + + found, errors = OrvaniTag().scan('.', False) + assert found == {'good.pdf': Entry(tags=['lofi'])} + assert len(errors) == 1 and 'broken.epub' in errors[0] + assert capsys.readouterr().out == '' + + +def test_cmd_shows_plugin_refusals_as_user_errors(cwd, monkeypatch, capsys): + (cwd / 'p.php').write_text('\n') + OrvaniTag().add('p.php', '#demo') + monkeypatch.setattr(sys, 'argv', ['otag', 'w', 'comment', 'p.php']) + with pytest.raises(SystemExit) as exit: + main() + out = capsys.readouterr().out + assert exit.value.code == 1 + assert 'PHP files need' in out and 'An Exception occurred' not in out + + +def test_configured_types_narrow_reading_too(cwd): + (cwd / orvani.CONFIG_FILE).write_text('[comment]\ntypes = [".py"]\n') + (cwd / 'a.py').write_text('# otag: #python\n') + (cwd / 'a.sh').write_text('# otag: #shell\n') + found, errors = OrvaniTag().scan('.', False) + assert found == {'a.py': Entry(tags=['python'])}