diff --git a/README.md b/README.md index 918ffad..86182f5 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,7 @@ $ otag write --clear id3 song.mp3 | `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 ` 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='comment, ebook, frontmatter, id3, office or pdf') + sub.add_argument('plugin', help='archive, comment, ebook, frontmatter, id3, 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 81d461a..e539f31 100644 --- a/orvanitag/orvani.py +++ b/orvanitag/orvani.py @@ -18,6 +18,7 @@ SEPARATOR = ' --- ' # between file name and contexts, in context files and ex DEFAULT_PLUGINS = [ 'frontmatter', 'comment', + 'archive', 'ebook', 'id3', 'office', diff --git a/orvanitag/plugins/_zip.py b/orvanitag/plugins/_zip.py index 9e5a901..9f5a4b6 100644 --- a/orvanitag/plugins/_zip.py +++ b/orvanitag/plugins/_zip.py @@ -1,5 +1,5 @@ """ -Shared by the plugins for zip based formats; not a plugin itself. +Shared by the plugins for zip based formats and archive comments; not a plugin itself. """ import os import shutil @@ -7,6 +7,8 @@ from pathlib import Path from zipfile import ZipFile from orvanitag.errors import UnsupportedFile +MARKER = 'otag:' # starts the archive comments otag writes + def rewrite(path, changes: dict = None, comment: bytes = None): """ @@ -35,3 +37,35 @@ def rewrite(path, changes: dict = None, comment: bytes = None): os.replace(tmp, path) finally: 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: + return unmark(archive.comment.decode('utf-8', 'replace')) + + +def write_comment(path, line: str): + """Stores an otag line as the zip file's archive comment; an empty line clears it""" + with ZipFile(path) as archive: + comment = mark(archive.comment.decode('utf-8', 'replace'), line, path) + if comment is not None: + rewrite(path, comment=comment.encode('utf-8')) diff --git a/orvanitag/plugins/archive.py b/orvanitag/plugins/archive.py new file mode 100644 index 0000000..b00e46b --- /dev/null +++ b/orvanitag/plugins/archive.py @@ -0,0 +1,122 @@ +""" +Tags in archives as an `otag:` comment, leaving the packed files untouched: the archive comment of ZIP, +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.entry import Entry +from orvanitag.errors import UnsupportedFile + +NAME = 'archive' +TYPES = ['.zip', '.tar', '.gz', '.tgz'] + +# gzip header flags, RFC 1952 +FHCRC, FEXTRA, FNAME, FCOMMENT = 2, 4, 8, 16 +BLOCK = 512 # tar block size + + +def until_zero(file) -> bytes: + """Reads a zero terminated gzip header field, including the zero""" + data = bytearray() + while not data.endswith(b'\0'): + byte = file.read(1) + if not byte: + raise UnsupportedFile(f'{file.name} has a truncated gzip header') + data += byte + return bytes(data) + + +def gzip_header(path) -> tuple[bytes, str, int]: + """The gzip header without comment, the comment and where the compressed data starts""" + with open(path, 'rb') as file: + head = bytearray(file.read(10)) + if head[:3] != b'\x1f\x8b\x08' or len(head) < 10: + raise UnsupportedFile(f'{path} is no gzip file') + flags = head[3] + # The header checksum is optional, so it is dropped instead of recomputed + head[3] = flags & ~(FCOMMENT | FHCRC) + if flags & FEXTRA: + size = file.read(2) + head += size + file.read(int.from_bytes(size, 'little')) + if flags & FNAME: + head += until_zero(file) + # RFC 1952 asks for Latin-1, but tags may use any script + comment = until_zero(file)[:-1].decode('utf-8', 'replace') if flags & FCOMMENT else '' + if flags & FHCRC: + file.read(2) + return bytes(head), comment, file.tell() + + +def pax_globals(path) -> tuple[dict, int]: + """The records of the pax global header a tar starts with, and where the rest of the archive begins""" + try: + tarfile.open(path, 'r:').close() + except tarfile.TarError as error: + raise UnsupportedFile(f'{path} is no uncompressed tar: {error}') from error + with open(path, 'rb') as file: + header = file.read(BLOCK) + if len(header) < BLOCK or header[156:157] != tarfile.XGLTYPE: + return {}, 0 + size = int(header[124:136].strip(b'\0 ') or b'0', 8) + data = file.read(size) + records = {} + while data: # records are ` =\n`, the length counting the whole record + length = data.split(b' ', 1)[0] + key, _, value = data[len(length) + 1:int(length) - 1].partition(b'=') + records[key.decode('utf-8')] = value.decode('utf-8') + data = data[int(length):] + 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': + text = read_comment(path) + elif suffix == '.tar': + text = unmark(pax_globals(path)[0].get('comment', '')) + else: + text = unmark(gzip_header(path)[1]) + if text.strip(): + return Entry.parse(text.split()) + + +def write(path, entry, config): + suffix = path.suffix.lower() + line = str(entry) + if suffix == '.zip': + write_comment(path, line) + + elif suffix == '.tar': + records, end = pax_globals(path) + comment = mark(records.pop('comment', ''), line, path) + if comment is None: + return + if comment: + records['comment'] = comment + replace_head(path, end, tarfile.TarInfo.create_pax_global_header(records) if records else b'') + + else: + head, comment, end = gzip_header(path) + comment = mark(comment, line, path) + if comment is None: + return + if comment: + head = head[:3] + bytes([head[3] | FCOMMENT]) + head[4:] + comment.encode('utf-8') + b'\0' + replace_head(path, end, head) diff --git a/orvanitag/plugins/ebook.py b/orvanitag/plugins/ebook.py index 099f5fc..7184ab3 100644 --- a/orvanitag/plugins/ebook.py +++ b/orvanitag/plugins/ebook.py @@ -8,13 +8,12 @@ import re import shutil from xml.sax.saxutils import escape, unescape from zipfile import ZipFile -from orvanitag.plugins._zip import rewrite +from orvanitag.plugins._zip import rewrite, read_comment, write_comment from orvanitag.entry import Entry from orvanitag.errors import UnsupportedFile NAME = 'ebook' TYPES = ['.epub', '.fb2', '.cbz'] -MARKER = 'otag:' # starts the CBZ archive comment QUOTES = {'"': '"'} # Element prefixes like `opf:` are matched, so the new element can use the same namespace prefix @@ -59,9 +58,7 @@ def read(path, config): found = CUSTOM.search(read_fb2(path)[0]) text = unescape(found.group(1)) if found else '' elif suffix == '.cbz': - with ZipFile(path) as archive: - comment = archive.comment.decode('utf-8', 'replace') - text = comment.removeprefix(MARKER) if comment.startswith(MARKER) else '' + text = read_comment(path) if text.strip(): return Entry.parse(text.split()) @@ -114,13 +111,4 @@ def write(path, entry, config): tmp.unlink(missing_ok=True) elif suffix == '.cbz': - with ZipFile(path) as archive: - comment = archive.comment - ours = comment.startswith(MARKER.encode()) - if not entry: - if ours: # another tool's comment is not otag's to clear - rewrite(path, comment=b'') - elif comment and not ours: - raise UnsupportedFile(f'the archive comment of {path} is used by another tool') - else: - rewrite(path, comment=f'{MARKER} {line}'.encode('utf-8')) + write_comment(path, line) diff --git a/tests/test_archive.py b/tests/test_archive.py new file mode 100644 index 0000000..fa32f4d --- /dev/null +++ b/tests/test_archive.py @@ -0,0 +1,139 @@ +import gzip +import io +import struct +import tarfile +import zipfile +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 make_tar(path, **kwargs): + with tarfile.open(path, 'w', **kwargs) as archive: + data = b'packed' + info = tarfile.TarInfo('a.txt') + info.size = len(data) + archive.addfile(info, io.BytesIO(data)) + + +def members(path): + with tarfile.open(path) as archive: + return {member.name: archive.extractfile(member).read() for member in archive.getmembers()} + + +def roundtrip(cwd, name, *tags): + """Writes the tags, reads them back into a fresh entry and clears them again""" + OrvaniTag().add(name, *tags) + OrvaniTag().write_metadata(name, 'archive') + 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 test_zip_comment(cwd): + with zipfile.ZipFile(cwd / 'a.zip', 'w') as z: + z.writestr('a.txt', 'packed') + roundtrip(cwd, 'a.zip', '+backup', '#old') + with zipfile.ZipFile(cwd / 'a.zip') as z: + assert z.comment == b'otag: +backup #old' + assert z.read('a.txt') == b'packed' + + +@pytest.mark.parametrize('name', ['a.gz', 'a.tar.gz', 'a.tgz']) +def test_gzip_comment_and_clear_restores_the_file(cwd, name): + with gzip.GzipFile(cwd / name, 'wb', mtime=0) as file: + file.write(b'packed') + before = (cwd / name).read_bytes() + + roundtrip(cwd, name, '#фантастика', 'a note') + assert b'otag: #\xd1\x84' in (cwd / name).read_bytes() + assert gzip.decompress((cwd / name).read_bytes()) == b'packed' + + OrvaniTag().write_metadata(name, 'archive', clear=True) + assert (cwd / name).read_bytes() == before + + +def test_gzip_keeps_extra_field_and_drops_header_crc(cwd): + header = b'\x1f\x8b\x08' + bytes([2 | 4 | 8]) + bytes(6) + struct.pack('