Add support for custom archive formats

src/instawow/_sources/github.py

@@ -13,6 +13,7 @@
 
 from .. import pkg_models
 from .. import results as R
+from ..archive import find_addon_zip_tocs
 from ..catalogue.cataloguer import AddonKey, CatalogueEntry
 from ..common import ChangelogFormat, Defn, Flavour, FlavourVersionRange, SourceMetadata, Strategy
 from ..http import CACHE_INDEFINITELY, ClientSessionType
@@ -22,7 +23,6 @@
     TocReader,
     as_plain_text_data_url,
     extract_byte_range_offset,
-    find_addon_zip_tocs,
 )
 
 

src/instawow/archive.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import posixpath
+from collections.abc import Callable, Iterable, Iterator, Mapping, Set
+from contextlib import AbstractContextManager, contextmanager
+from pathlib import Path
+from typing import NamedTuple, Protocol
+from zipfile import ZipFile
+
+
+class ArchiveOpener(Protocol):  # pragma: no cover
+    def __call__(self, path: Path) -> AbstractContextManager[Archive]:
+        ...
+
+
+class Archive(NamedTuple):
+    top_level_folders: Set[str]
+    extract: Callable[[Path], None]
+
+
+def find_addon_zip_tocs(names: Iterable[str]):
+    "Find top-level folders in a list of zip member paths."
+    for name in names:
+        if name.count(posixpath.sep) == 1:
+            head, tail = posixpath.split(name)
+            if tail.startswith(head) and tail[-4:].lower() == '.toc':
+                yield (name, head)
+
+
+def make_zip_member_filter_fn(base_dirs: Set[str]):
+    "Filter out items which are not sub-paths of top-level folders in a zip."
+
+    def is_subpath(name: str):
+        head, sep, _ = name.partition(posixpath.sep)
+        return head in base_dirs if sep else False
+
+    return is_subpath
+
+
+@contextmanager
+def open_zip_archive(path: Path) -> Iterator[Archive]:
+    with ZipFile(path) as archive:
+        names = archive.namelist()
+        top_level_folders = {h for _, h in find_addon_zip_tocs(names)}
+
+        def extract(parent: Path) -> None:
+            should_extract = make_zip_member_filter_fn(top_level_folders)
+            archive.extractall(parent, members=(n for n in names if should_extract(n)))
+
+        yield Archive(top_level_folders, extract)
+
+
+ARCHIVE_OPENERS: Mapping[str, ArchiveOpener] = {
+    'application/zip': open_zip_archive,
+}

src/instawow/common.py

@@ -87,6 +87,7 @@ class SourceMetadata:
     strategies: frozenset[Strategy]
     changelog_format: ChangelogFormat
     addon_toc_key: str | None
+    archive_format: str = 'application/zip'
 
 
 @frozen

src/instawow/manager_ctx.py

@@ -24,13 +24,14 @@
 from ._sources.tukui import TukuiResolver
 from ._sources.wago import WagoResolver
 from ._sources.wowi import WowiResolver
+from .archive import ARCHIVE_OPENERS, ArchiveOpener
 from .catalogue.cataloguer import (
     CATALOGUE_VERSION,
     ComputedCatalogue,
 )
 from .config import Config
 from .http import make_generic_progress_ctx
-from .plugins import get_plugin_resolvers
+from .plugins import get_plugin_archive_openers, get_plugin_resolvers
 from .resolvers import Resolver
 from .utils import (
     WeakValueDefaultDictionary,
@@ -97,6 +98,7 @@ def _parse_catalogue(raw_catalogue: bytes):
 
 class ManagerCtx:
     __slots__ = [
+        'archive_openers',
         'config',
         'database',
         'resolvers',
@@ -139,7 +141,13 @@ def __init__(
         resolver_classes = chain(
             (r for g in get_plugin_resolvers() for r in g), builtin_resolver_classes
         )
-        self.resolvers = _Resolvers((r.metadata.id, r(self)) for r in resolver_classes)
+        self.resolvers: _Resolvers = _Resolvers((r.metadata.id, r(self)) for r in resolver_classes)
+
+        archive_openers = dict(get_plugin_archive_openers()) | dict(ARCHIVE_OPENERS)
+        self.archive_openers: Mapping[str, ArchiveOpener] = {
+            r.metadata.id: archive_openers[r.metadata.archive_format]
+            for r in self.resolvers.values()
+        }
 
     @classmethod
     def from_config(cls, config: Config) -> Self:

src/instawow/pkg_management.py

@@ -8,10 +8,10 @@
     Mapping,
     Sequence,
 )
-from contextlib import asynccontextmanager, contextmanager
+from contextlib import asynccontextmanager
 from functools import wraps
 from itertools import filterfalse, product, repeat, starmap
-from pathlib import Path, PurePath
+from pathlib import Path
 from shutil import move
 from tempfile import NamedTemporaryFile
 from typing import Concatenate, TypeVar
@@ -32,10 +32,8 @@
     bucketise,
     chain_dict,
     file_uri_to_path,
-    find_addon_zip_tocs,
     gather,
     is_file_uri,
-    make_zip_member_filter_fn,
     run_in_thread,
     shasum,
     time_op,
@@ -83,21 +81,6 @@ async def _open_temp_writer_async():
         await run_in_thread(fh.close)()
 
 
-@contextmanager
-def _open_pkg_archive(path: PurePath):
-    from zipfile import ZipFile
-
-    with ZipFile(path) as archive:
-
-        def extract(parent: Path) -> None:
-            should_extract = make_zip_member_filter_fn(base_dirs)
-            archive.extractall(parent, members=(n for n in names if should_extract(n)))
-
-        names = archive.namelist()
-        base_dirs = {h for _, h in find_addon_zip_tocs(names)}
-        yield (base_dirs, extract)
-
-
 @object.__new__
 class _DummyResolver:
     async def resolve(self, defns: Sequence[Defn]) -> dict[Defn, R.AnyResult[pkg_models.Pkg]]:
@@ -125,7 +108,9 @@ async def inner(self: _TPkgManager, *args: _P.args, **kwargs: _P.kwargs) -> _T:
 def _install_pkg(
     ctx: ManagerCtx, pkg: pkg_models.Pkg, archive: Path, replace_folders: bool
 ) -> R.PkgInstalled:
-    with _open_pkg_archive(archive) as (top_level_folders, extract):
+    open_archive = ctx.archive_openers[pkg.source]
+
+    with open_archive(archive) as (top_level_folders, extract):
         with ctx.database.connect() as connection:
             installed_conflicts = connection.execute(
                 sa.select(pkg_db.pkg)
@@ -162,7 +147,9 @@ def _install_pkg(
 def _update_pkg(
     ctx: ManagerCtx, old_pkg: pkg_models.Pkg, new_pkg: pkg_models.Pkg, archive: Path
 ) -> R.PkgUpdated:
-    with _open_pkg_archive(archive) as (top_level_folders, extract):
+    open_archive = ctx.archive_openers[new_pkg.source]
+
+    with open_archive(archive) as (top_level_folders, extract):
         with ctx.database.connect() as connection:
             installed_conflicts = connection.execute(
                 sa.select(pkg_db.pkg)

src/instawow/plugins.py

@@ -7,7 +7,7 @@
 import click
 import pluggy
 
-from . import resolvers
+from . import archive, resolvers
 
 _project_name = __spec__.parent
 _entry_point = f'{_project_name}.plugins'
@@ -19,6 +19,11 @@
 class InstawowPlugin(Protocol):  # pragma: no cover
     "The plug-in interface."
 
+    @hookspec
+    def instawow_add_archive_openers(self) -> Iterable[tuple[str, archive.ArchiveOpener]]:
+        "Additional archive openers to load."
+        ...
+
     @hookspec
     def instawow_add_commands(self) -> Iterable[click.Command]:
         "Additional commands to register with ``click``."
@@ -38,6 +43,11 @@ def _load_plugins():
     return plugin_manager.hook
 
 
+def get_plugin_archive_openers() -> Iterable[tuple[str, archive.ArchiveOpener]]:
+    plugin_hook = _load_plugins()
+    return plugin_hook.instawow_add_archive_openers()
+
+
 def get_plugin_commands() -> Iterable[Iterable[click.Command]]:
     plugin_hook = _load_plugins()
     return plugin_hook.instawow_add_commands()

src/instawow/results.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Awaitable, Collection
+from collections.abc import Awaitable, Collection, Set
 from typing import Any, Final, Protocol, TypeAlias, TypeVar
 
 from attrs import asdict
@@ -104,7 +104,7 @@ def message(self) -> str:
 
 
 class PkgConflictsWithUnreconciled(ManagerError):
-    def __init__(self, folders: set[str]) -> None:
+    def __init__(self, folders: Set[str]) -> None:
         super().__init__()
         self.folders = folders
 

src/instawow/utils.py

@@ -3,7 +3,6 @@
 import asyncio
 import importlib.resources
 import os
-import posixpath
 import string
 import sys
 import time
@@ -16,7 +15,6 @@
     Iterable,
     Iterator,
     Sequence,
-    Set,
 )
 from contextlib import contextmanager
 from functools import wraps
@@ -200,25 +198,6 @@ def shasum(*values: object) -> str:
     return sha256(''.join(map(str, filter(None, values))).encode()).hexdigest()[:32]
 
 
-def find_addon_zip_tocs(names: Iterable[str]) -> Iterator[tuple[str, str]]:
-    "Find top-level folders in a list of ZIP member paths."
-    for name in names:
-        if name.count(posixpath.sep) == 1:
-            head, tail = posixpath.split(name)
-            if tail.startswith(head) and tail[-4:].lower() == '.toc':
-                yield (name, head)
-
-
-def make_zip_member_filter_fn(base_dirs: Set[str]) -> Callable[[str], bool]:
-    "Filter out items which are not sub-paths of top-level folders in a ZIP."
-
-    def is_subpath(name: str):
-        head, sep, _ = name.partition(posixpath.sep)
-        return head in base_dirs if sep else False
-
-    return is_subpath
-
-
 def is_file_uri(uri: str) -> bool:
     return uri.startswith('file://')
 

tests/test_archive.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from itertools import product
+
+import pytest
+
+from instawow.archive import find_addon_zip_tocs, make_zip_member_filter_fn
+
+
+def test_find_addon_zip_tocs_can_find_explicit_dirs():
+    assert {h for _, h in find_addon_zip_tocs(['b/', 'b/b.toc'])} == {'b'}
+
+
+def test_find_addon_zip_tocs_can_find_implicit_dirs():
+    assert {h for _, h in find_addon_zip_tocs(['b/b.toc'])} == {'b'}
+
+
+def test_find_addon_zip_tocs_discards_tocless_paths():
+    assert {h for _, h in find_addon_zip_tocs(['a', 'b/b.toc', 'c/'])} == {'b'}
+
+
+def test_find_addon_zip_tocs_discards_mismatched_tocs():
+    assert not {h for _, h in find_addon_zip_tocs(['a', 'a/b.toc'])}
+
+
+def test_find_addon_zip_tocs_accepts_multitoc():
+    assert {h for _, h in find_addon_zip_tocs(['a', 'a/a_mainline.toc'])} == {'a'}
+
+
+@pytest.mark.parametrize('ext', product('Tt', 'Oo', 'Cc'))
+def test_find_addon_zip_tocs_toc_is_case_insensitive(ext: tuple[str, ...]):
+    assert {h for _, h in find_addon_zip_tocs([f'a/a.{"".join(ext)}'])} == {'a'}
+
+
+def test_make_zip_member_filter_fn_discards_names_with_prefix_not_in_dirs():
+    is_member = make_zip_member_filter_fn({'b'})
+    assert list(filter(is_member, ['a/', 'b/', 'aa/', 'bb/', 'b/c', 'a/d'])) == ['b/', 'b/c']

tests/test_utils.py

@@ -3,12 +3,10 @@
 import asyncio
 import sys
 import time
-from itertools import product
 from pathlib import Path
 
 import pytest
 
-from instawow.pkg_management import find_addon_zip_tocs, make_zip_member_filter_fn
 from instawow.utils import (
     TocReader,
     bucketise,
@@ -24,36 +22,6 @@ def fake_addon_toc():
     return Path(__file__).parent / 'fixtures' / 'FakeAddon' / 'FakeAddon.toc'
 
 
-def test_find_addon_zip_tocs_can_find_explicit_dirs():
-    assert {h for _, h in find_addon_zip_tocs(['b/', 'b/b.toc'])} == {'b'}
-
-
-def test_find_addon_zip_tocs_can_find_implicit_dirs():
-    assert {h for _, h in find_addon_zip_tocs(['b/b.toc'])} == {'b'}
-
-
-def test_find_addon_zip_tocs_discards_tocless_paths():
-    assert {h for _, h in find_addon_zip_tocs(['a', 'b/b.toc', 'c/'])} == {'b'}
-
-
-def test_find_addon_zip_tocs_discards_mismatched_tocs():
-    assert not {h for _, h in find_addon_zip_tocs(['a', 'a/b.toc'])}
-
-
-def test_find_addon_zip_tocs_accepts_multitoc():
-    assert {h for _, h in find_addon_zip_tocs(['a', 'a/a_mainline.toc'])} == {'a'}
-
-
-@pytest.mark.parametrize('ext', product('Tt', 'Oo', 'Cc'))
-def test_find_addon_zip_tocs_toc_is_case_insensitive(ext: tuple[str, ...]):
-    assert {h for _, h in find_addon_zip_tocs([f'a/a.{"".join(ext)}'])} == {'a'}
-
-
-def test_make_zip_member_filter_fn_discards_names_with_prefix_not_in_dirs():
-    is_member = make_zip_member_filter_fn({'b'})
-    assert list(filter(is_member, ['a/', 'b/', 'aa/', 'bb/', 'b/c', 'a/d'])) == ['b/', 'b/c']
-
-
 def test_loading_toc_from_path(fake_addon_toc: Path):
     TocReader.from_path(fake_addon_toc)
     with pytest.raises(FileNotFoundError):