diff --git a/README.md b/README.md index 86182f5..3406e29 100644 --- a/README.md +++ b/README.md @@ -106,12 +106,13 @@ $ otag write --clear id3 song.mp3 | Plugin | File types | Where the tags are stored | |---|---|---| | `id3` | `.mp3` `.wav` `.flac` `.ogg` | The comment tag | +| `image` | `.jpg` `.jpeg` `.png` `.gif` | An `otag:` comment of its own next to any other comments: a COM segment in JPEG (after the JFIF and EXIF segments; some apps only show a JPEG's last comment), an iTXt `Comment` chunk in PNG, a comment extension in GIF (turning a GIF87a into GIF89a). The image data stays unchanged | | `pdf` | `.pdf` | The keywords field | | `office` | `.docx` `.xlsx` `.pptx` `.odt` `.ods` `.odp`, reading also `.doc` `.xls` `.ppt` | The description field | | `ebook` | `.epub` `.fb2` `.cbz` | One otag line where other apps don't look: a `` in the EPUB package file, a `` in FB2, the archive comment of CBZ (never overwriting another tool's comment); the book's own metadata is left untouched | | `archive` | `.zip` `.tar` `.gz` `.tgz` (so also `.tar.gz`) | An `otag:` comment that unpacking ignores: the archive comment of ZIP, the comment field of the gzip header, a pax global header at the start of a tar. Another tool's comment, like the commit id stored by `git archive`, is never replaced. Writing a tar or gzip file copies it once, the packed data stays unchanged | | `frontmatter` | `.md` `.markdown` `.txt` | `projects`, `contexts`, `tags` and `note` keys of the YAML front matter; other keys, comments and formatting are left untouched | -| `comment` | Source code such as `.py` `.sh` `.toml` `.js` `.c` `.go` `.rs` `.php` `.html` | An `otag: +project @context #tag note` comment line within the first ten lines, written in the file's comment style below any shebang, encoding, `` or ``, `` or ` ArgumentParser: no_recursive(sub) sub = command('write', ['w'], 'write the entry of a file into the file itself', run_write) - sub.add_argument('plugin', help='archive, comment, ebook, frontmatter, id3, office or pdf') + sub.add_argument('plugin', help='archive, comment, ebook, frontmatter, id3, image, office or pdf') file_arg(sub) sub.add_argument('-c', '--clear', action='store_true', help="remove the plugin's tags from the file instead") diff --git a/orvanitag/orvani.py b/orvanitag/orvani.py index e539f31..923d418 100644 --- a/orvanitag/orvani.py +++ b/orvanitag/orvani.py @@ -21,6 +21,7 @@ DEFAULT_PLUGINS = [ 'archive', 'ebook', 'id3', + 'image', 'office', 'pdf' ] diff --git a/orvanitag/plugins/_embed.py b/orvanitag/plugins/_embed.py new file mode 100644 index 0000000..24fd73a --- /dev/null +++ b/orvanitag/plugins/_embed.py @@ -0,0 +1,41 @@ +""" +Shared by the plugins that store an `otag:` line in a file format's own comment field; not a plugin itself. +""" +import os +import shutil +from orvanitag.errors import UnsupportedFile + +MARKER = 'otag:' # starts the comments otag writes + + +def unmark(comment: str) -> str: + """The otag line of a comment, or '' if the comment is not otag's""" + return comment.removeprefix(MARKER) if comment.startswith(MARKER) else '' + + +def mark(comment: str, line: str, path) -> str | None: + """ + The new content of a format's only comment field for an otag line, or None to leave it as it is. + An empty line clears otag's comment; another tool's comment is never replaced. + """ + ours = comment.startswith(MARKER) + if not line: + return '' if ours else None + if comment and not ours: + raise UnsupportedFile(f'the comment of {path} is used by another tool') + return f'{MARKER} {line}' + + +def splice(path, start: int, end: int, data: bytes): + """Atomically replaces the bytes from start to end of a file by data, copying the rest as it is""" + tmp = path.with_name(f'.{path.name}.tmp') + try: + with open(path, 'rb') as source, open(tmp, 'wb') as target: + target.write(source.read(start)) + target.write(data) + source.seek(end) + shutil.copyfileobj(source, target) + shutil.copymode(path, tmp) + os.replace(tmp, path) + finally: + tmp.unlink(missing_ok=True) diff --git a/orvanitag/plugins/_zip.py b/orvanitag/plugins/_zip.py index 9f5a4b6..e002a69 100644 --- a/orvanitag/plugins/_zip.py +++ b/orvanitag/plugins/_zip.py @@ -1,13 +1,12 @@ """ -Shared by the plugins for zip based formats and archive comments; not a plugin itself. +Shared by the plugins for zip based formats; not a plugin itself. """ import os import shutil from pathlib import Path from zipfile import ZipFile from orvanitag.errors import UnsupportedFile - -MARKER = 'otag:' # starts the archive comments otag writes +from orvanitag.plugins._embed import mark, unmark def rewrite(path, changes: dict = None, comment: bytes = None): @@ -39,24 +38,6 @@ def rewrite(path, changes: dict = None, comment: bytes = None): tmp.unlink(missing_ok=True) -def unmark(comment: str) -> str: - """The otag line of an archive comment, or '' if the comment is not otag's""" - return comment.removeprefix(MARKER) if comment.startswith(MARKER) else '' - - -def mark(comment: str, line: str, path) -> str | None: - """ - The new archive comment for an otag line, or None to leave the comment as it is. - An empty line clears otag's comment; another tool's comment is never replaced. - """ - ours = comment.startswith(MARKER) - if not line: - return '' if ours else None - if comment and not ours: - raise UnsupportedFile(f'the archive comment of {path} is used by another tool') - return f'{MARKER} {line}' - - def read_comment(path) -> str: """The otag line of a zip file's archive comment""" with ZipFile(path) as archive: diff --git a/orvanitag/plugins/archive.py b/orvanitag/plugins/archive.py index b00e46b..3180a54 100644 --- a/orvanitag/plugins/archive.py +++ b/orvanitag/plugins/archive.py @@ -3,10 +3,9 @@ Tags in archives as an `otag:` comment, leaving the packed files untouched: the the comment field of the gzip header (so also of `.tar.gz`) and a pax global header at the start of a tar. Another tool's comment, like the commit id `git archive` stores, is never replaced. """ -import os -import shutil import tarfile -from orvanitag.plugins._zip import mark, unmark, read_comment, write_comment +from orvanitag.plugins._embed import mark, unmark, splice +from orvanitag.plugins._zip import read_comment, write_comment from orvanitag.entry import Entry from orvanitag.errors import UnsupportedFile @@ -71,20 +70,6 @@ def pax_globals(path) -> tuple[dict, int]: return records, BLOCK + -(-size // BLOCK) * BLOCK -def replace_head(path, end: int, head: bytes): - """Atomically replaces the first `end` bytes of a file by head, copying the rest as it is""" - tmp = path.with_name(f'.{path.name}.tmp') - try: - with open(path, 'rb') as source, open(tmp, 'wb') as target: - source.seek(end) - target.write(head) - shutil.copyfileobj(source, target) - shutil.copymode(path, tmp) - os.replace(tmp, path) - finally: - tmp.unlink(missing_ok=True) - - def read(path, config): suffix = path.suffix.lower() if suffix == '.zip': @@ -110,7 +95,7 @@ def write(path, entry, config): return if comment: records['comment'] = comment - replace_head(path, end, tarfile.TarInfo.create_pax_global_header(records) if records else b'') + splice(path, 0, end, tarfile.TarInfo.create_pax_global_header(records) if records else b'') else: head, comment, end = gzip_header(path) @@ -119,4 +104,4 @@ def write(path, entry, config): return if comment: head = head[:3] + bytes([head[3] | FCOMMENT]) + head[4:] + comment.encode('utf-8') + b'\0' - replace_head(path, end, head) + splice(path, 0, end, head) diff --git a/orvanitag/plugins/comment.py b/orvanitag/plugins/comment.py index 3a80150..cdfb8b5 100644 --- a/orvanitag/plugins/comment.py +++ b/orvanitag/plugins/comment.py @@ -33,6 +33,7 @@ STYLES = { '.rb' : HASH, '.rs' : SLASH, '.sh' : HASH, + '.svg' : HTML, '.toml' : HASH, '.ts' : SLASH, } @@ -59,7 +60,8 @@ def find(lines: list): def header(lines: list, suffix: str) -> int: """Returns how many leading lines have to stay above the tag line""" count = 0 - if lines and (lines[0].startswith('#!') or lines[0].lower().startswith(' tuple[str, int, int]: + """ + The otag comment of a JPEG with the byte range of its COM segment, or an empty range where one goes: + after the APP segments, since JFIF and EXIF have to come first + """ + with open(path, 'rb') as file: + if file.read(2) != b'\xff\xd8': + raise UnsupportedFile(f'{path} is no JPEG file') + insert = None + while True: + start = file.tell() + marker = file.read(2) + while marker[1:] == b'\xff': # fill bytes before a marker + start += 1 + marker = marker[1:] + file.read(1) + if len(marker) < 2 or marker[0] != 0xFF: + raise UnsupportedFile(f'{path} has a broken JPEG header') + if marker[1] in (0xDA, 0xD9): # the image data starts, or the file ends + break + data = file.read(int.from_bytes(file.read(2), 'big') - 2) + if marker[1] == 0xFE and data.startswith(OURS): + return data.decode('utf-8', 'replace'), start, file.tell() + if insert is None and not 0xE0 <= marker[1] <= 0xEF: + insert = start + insert = start if insert is None else insert + return '', insert, insert + + +def jpeg_segment(comment: str) -> bytes: + data = comment.encode('utf-8') + if len(data) + 2 > 0xFFFF: + raise UnsupportedFile('the tags are too long for a JPEG comment') + return b'\xff\xfe' + (len(data) + 2).to_bytes(2, 'big') + data + + +def png(path) -> tuple[str, int, int]: + """The otag comment of a PNG with the byte range of its chunk, or an empty range after the IHDR chunk""" + with open(path, 'rb') as file: + if file.read(8) != PNG: + raise UnsupportedFile(f'{path} is no PNG file') + insert = None + while len(head := file.read(8)) == 8: + start = file.tell() - 8 + size, kind = int.from_bytes(head[:4], 'big'), head[4:] + if kind == b'iTXt': + # keyword, compression flag and method, language, translated keyword, text + keyword, _, rest = file.read(size).partition(b'\0') + text = rest[2:].split(b'\0', 2)[-1] + file.seek(4, 1) + if keyword == b'Comment' and rest[:1] == b'\0' and text.startswith(OURS): + return text.decode('utf-8', 'replace'), start, file.tell() + else: + file.seek(size + 4, 1) + if kind == b'IHDR': + insert = file.tell() + if kind == b'IEND': + break + if insert is None: + raise UnsupportedFile(f'{path} has no PNG header') + return '', insert, insert + + +def png_chunk(comment: str) -> bytes: + data = b'iTXt' + b'Comment\0\0\0\0\0' + comment.encode('utf-8') + return (len(data) - 4).to_bytes(4, 'big') + data + zlib.crc32(data).to_bytes(4, 'big') + + +def gif(path) -> tuple[str, int, int]: + """ + The otag comment of a GIF with the byte range of its comment extension, or an empty range after the color table. + Only the extensions before the first image are searched, since otag puts its comment there. + """ + with open(path, 'rb') as file: + header = file.read(13) + if header[:6] not in (b'GIF87a', b'GIF89a'): + raise UnsupportedFile(f'{path} is no GIF file') + if header[10] & 0x80: # global color table + file.seek(3 << ((header[10] & 7) + 1), 1) + insert = file.tell() + while file.read(1) == b'\x21': + start = file.tell() - 1 + label, data = file.read(1), b'' + while (size := file.read(1)) not in (b'\0', b''): # data sub-blocks + data += file.read(size[0]) + if label == b'\xfe' and data.startswith(OURS): + return data.decode('utf-8', 'replace'), start, file.tell() + return '', insert, insert + + +def gif_extension(comment: str) -> bytes: + # The GIF specification asks for ASCII, but tags may use any script + data = comment.encode('utf-8') + return b'\x21\xfe' + b''.join(bytes([len(data[i:i + 255])]) + data[i:i + 255] for i in range(0, len(data), 255)) + b'\0' + + +FORMATS = { + '.jpg': (jpeg, jpeg_segment), + '.jpeg': (jpeg, jpeg_segment), + '.png': (png, png_chunk), + '.gif': (gif, gif_extension), +} + + +def read(path, config): + text = unmark(FORMATS[path.suffix.lower()][0](path)[0]) + if text.strip(): + return Entry.parse(text.split()) + + +def write(path, entry, config): + find, block = FORMATS[path.suffix.lower()] + comment, start, end = find(path) + line = str(entry) + if not line: + if comment: + splice(path, start, end, b'') + return + data = block(f'{MARKER} {line}') + if path.suffix.lower() == '.gif': + with open(path, 'rb') as file: + head = file.read(start) + if head.startswith(b'GIF87a'): # extensions came with version 89a + start, data = 0, b'GIF89a' + head[6:] + data + splice(path, start, end, data) diff --git a/tests/test_comment.py b/tests/test_comment.py index ddd4af4..68223a5 100644 --- a/tests/test_comment.py +++ b/tests/test_comment.py @@ -90,3 +90,15 @@ def test_marker_inside_code_is_ignored(cwd): (cwd / 'a.py').write_text('x = "otag: #nope"\n') OrvaniTag().scan('.', False) assert not (cwd / orvani.CONTEXT_FILE).exists() + + +def test_svg_keeps_xml_declaration_first(cwd): + write('a.svg', '\n\n', '#icon') + assert (cwd / 'a.svg').read_text() == '\n\n\n\n' + assert read_back(cwd, 'a.svg') == 'a.svg --- #icon\n' + + +def test_double_dash_is_refused_in_markup_comments(cwd): + with pytest.raises(UnsupportedFile): + write('a.svg', '\n', '#icon', 'before -- after') + assert (cwd / 'a.svg').read_text() == '\n' diff --git a/tests/test_image.py b/tests/test_image.py new file mode 100644 index 0000000..de1cf0a --- /dev/null +++ b/tests/test_image.py @@ -0,0 +1,132 @@ +import struct +import zlib + +import pytest + +import orvanitag.orvani as orvani +from orvanitag.entry import Entry +from orvanitag.orvani import OrvaniTag +from orvanitag.errors import UnsupportedFile + + +@pytest.fixture +def cwd(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + return tmp_path + + +def roundtrip(cwd, name, *tags): + """Writes the tags and reads them back into a fresh entry""" + OrvaniTag().add(name, *tags) + OrvaniTag().write_metadata(name, 'image') + expected = (cwd / orvani.CONTEXT_FILE).read_text() + (cwd / orvani.CONTEXT_FILE).unlink() + OrvaniTag().read_metadata(name) + assert (cwd / orvani.CONTEXT_FILE).read_text() == expected + + +def clear(name): + OrvaniTag().write_metadata(name, 'image', clear=True) + + +def segment(marker, data): + return bytes([0xFF, marker]) + (len(data) + 2).to_bytes(2, 'big') + data + + +SOI = b'\xff\xd8' +APPS = segment(0xE0, b'JFIF\0\1\1\0\0\1\0\1\0\0') + segment(0xE1, b'Exif\0\0') +REST = segment(0xDB, bytes(65)) + segment(0xDA, bytes(10)) + b'\x12\x34\xff\xd9' + + +def test_jpeg_comment_goes_after_the_app_segments(cwd): + (cwd / 'a.jpg').write_bytes(SOI + APPS + REST) + roundtrip(cwd, 'a.jpg', '+trip', '#beach', 'sunset') + assert (cwd / 'a.jpg').read_bytes() == SOI + APPS + segment(0xFE, b'otag: +trip #beach sunset') + REST + clear('a.jpg') + assert (cwd / 'a.jpg').read_bytes() == SOI + APPS + REST + + +def test_jpeg_keeps_other_comments(cwd): + foreign = segment(0xFE, b'CREATOR: gd-jpeg v1.0') + (cwd / 'a.JPEG').write_bytes(SOI + APPS + foreign + REST) + assert OrvaniTag().read_metadata('a.JPEG') == Entry() + roundtrip(cwd, 'a.JPEG', '#beach') + OrvaniTag().add('a.JPEG', '#more') + OrvaniTag().write_metadata('a.JPEG', 'image') + assert (cwd / 'a.JPEG').read_bytes() == SOI + APPS + segment(0xFE, b'otag: #beach #more') + foreign + REST + clear('a.JPEG') + assert (cwd / 'a.JPEG').read_bytes() == SOI + APPS + foreign + REST + + +def test_jpeg_comment_length_is_limited(cwd): + (cwd / 'a.jpg').write_bytes(SOI + APPS + REST) + OrvaniTag().add('a.jpg', 'x' * 0x10000) + with pytest.raises(UnsupportedFile, match='too long'): + OrvaniTag().write_metadata('a.jpg', 'image') + assert (cwd / 'a.jpg').read_bytes() == SOI + APPS + REST + + +def chunk(kind, data): + return len(data).to_bytes(4, 'big') + kind + data + zlib.crc32(kind + data).to_bytes(4, 'big') + + +def chunks(data): + """The chunks of a PNG, checking their CRCs""" + found, pos = [], 8 + while pos < len(data): + size, kind = int.from_bytes(data[pos:pos + 4], 'big'), data[pos + 4:pos + 8] + body = data[pos + 8:pos + 8 + size] + assert data[pos + 8 + size:pos + 12 + size] == zlib.crc32(kind + body).to_bytes(4, 'big') + found.append((kind, body)) + pos += 12 + size + return found + + +PNG = b'\x89PNG\r\n\x1a\n' +IHDR = chunk(b'IHDR', struct.pack('>IIBBBBB', 1, 1, 8, 0, 0, 0, 0)) +IMAGE = chunk(b'IDAT', zlib.compress(b'\0\0')) + chunk(b'IEND', b'') + + +def test_png_comment_chunk_and_others_kept(cwd): + foreign = chunk(b'iTXt', b'Comment\0\0\0\0\0Created with GIMP') + chunk(b'tEXt', b'Software\0x') + (cwd / 'a.png').write_bytes(PNG + IHDR + foreign + IMAGE) + assert OrvaniTag().read_metadata('a.png') == Entry() + + roundtrip(cwd, 'a.png', '#фото', 'a note') + data = (cwd / 'a.png').read_bytes() + assert chunks(data)[1] == (b'iTXt', b'Comment\0\0\0\0\0' + 'otag: #фото a note'.encode()) + assert data == PNG + IHDR + chunk(b'iTXt', b'Comment\0\0\0\0\0' + 'otag: #фото a note'.encode()) + foreign + IMAGE + + clear('a.png') + assert (cwd / 'a.png').read_bytes() == PNG + IHDR + foreign + IMAGE + + +GIF_SCREEN = b'\x01\x00\x01\x00\x80\x00\x00' + b'\xff\xff\xff\x00\x00\x00' +GIF_IMAGE = b',\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02\x44\x01\x00;' + + +def test_gif_comment_in_sub_blocks(cwd): + foreign = b'\x21\xfe\x07made by\x00' + (cwd / 'a.gif').write_bytes(b'GIF89a' + GIF_SCREEN + foreign + GIF_IMAGE) + note = 'n' * 300 + roundtrip(cwd, 'a.gif', '#anim', note) + line = f'otag: #anim {note}'.encode() + assert (cwd / 'a.gif').read_bytes() == (b'GIF89a' + GIF_SCREEN + b'\x21\xfe\xff' + line[:255] + + bytes([len(line) - 255]) + line[255:] + b'\0' + foreign + GIF_IMAGE) + clear('a.gif') + assert (cwd / 'a.gif').read_bytes() == b'GIF89a' + GIF_SCREEN + foreign + GIF_IMAGE + + +def test_gif87a_becomes_89a(cwd): + (cwd / 'a.gif').write_bytes(b'GIF87a' + GIF_SCREEN + GIF_IMAGE) + roundtrip(cwd, 'a.gif', '#old') + assert (cwd / 'a.gif').read_bytes() == b'GIF89a' + GIF_SCREEN + b'\x21\xfe' + bytes([10]) + b'otag: #old\0' + GIF_IMAGE + + +@pytest.mark.parametrize('name', ['a.jpg', 'a.png', 'a.gif']) +def test_other_files_are_refused_unchanged(cwd, name): + (cwd / name).write_bytes(b'plain text') + OrvaniTag().add(name, '#lofi') + with pytest.raises(UnsupportedFile): + OrvaniTag().write_metadata(name, 'image') + assert (cwd / name).read_bytes() == b'plain text'