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 ..archives import find_archive_addon_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,
 )
 
 
@@ -204,7 +204,7 @@ async def _find_matching_asset_from_zip_contents(self, assets: list[_GithubRelea
 
             toc_filenames = {
                 n
-                for n, h in find_addon_zip_tocs(f.filename for f in dynamic_addon_zip.filelist)
+                for n, h in find_archive_addon_tocs(f.filename for f in dynamic_addon_zip.filelist)
                 if h in candidate['name']  # Folder name is a substring of zip name.
             }
             if not toc_filenames:

src/instawow/archives.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import posixpath
+import zipfile
+from collections.abc import Callable, Iterable, Set
+from contextlib import AbstractContextManager, contextmanager
+from pathlib import Path
+from typing import NamedTuple, Protocol
+
+
+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_archive_addon_tocs(names: Iterable[str]):
+    "Find top-level folders in a list of archive 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_archive_member_filter_fn(base_dirs: Set[str]):
+    "Filter out items which are not sub-paths of top-level folders in an archive."
+
+    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):
+    with zipfile.ZipFile(path) as archive:
+        names = archive.namelist()
+        top_level_folders = {h for _, h in find_archive_addon_tocs(names)}
+
+        def extract(parent: Path) -> None:
+            should_extract = make_archive_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)

src/instawow/manager_ctx.py

@@ -24,6 +24,7 @@
 from ._sources.tukui import TukuiResolver
 from ._sources.wago import WagoResolver
 from ._sources.wowi import WowiResolver
+from .archives import ArchiveOpener, open_zip_archive
 from .catalogue.cataloguer import (
     CATALOGUE_VERSION,
     ComputedCatalogue,
@@ -49,6 +50,10 @@ async def __aexit__(self, *args: object):
 
 
 class _Resolvers(dict[str, Resolver]):
+    @cached_property
+    def archive_opener_dict(self) -> _ResolverArchiveOpenerDict:
+        return _ResolverArchiveOpenerDict(self)
+
     @cached_property
     def priority_dict(self) -> _ResolverPriorityDict:
         return _ResolverPriorityDict(self)
@@ -70,6 +75,16 @@ def __missing__(self, key: str) -> float:
         return float('inf')
 
 
+class _ResolverArchiveOpenerDict(dict[str, ArchiveOpener]):
+    def __init__(self, resolvers: _Resolvers) -> None:
+        super().__init__(
+            (r.metadata.id, r.archive_opener) for r in resolvers.values() if r.archive_opener
+        )
+
+    def __missing__(self, key: str) -> ArchiveOpener:
+        return open_zip_archive
+
+
 _web_client = cv.ContextVar[http.ClientSessionType]('_web_client')
 
 LocksType: TypeAlias = Mapping[object, AbstractAsyncContextManager[None]]
@@ -139,7 +154,7 @@ 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)
 
     @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,7 @@ 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):
+    with ctx.resolvers.archive_opener_dict[pkg.source](archive) as (top_level_folders, extract):
         with ctx.database.connect() as connection:
             installed_conflicts = connection.execute(
                 sa.select(pkg_db.pkg)
@@ -162,7 +145,10 @@ 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):
+    with ctx.resolvers.archive_opener_dict[new_pkg.source](archive) as (
+        top_level_folders,
+        extract,
+    ):
         with ctx.database.connect() as connection:
             installed_conflicts = connection.execute(
                 sa.select(pkg_db.pkg)

src/instawow/resolvers.py

@@ -10,7 +10,7 @@
 from typing_extensions import Self
 from yarl import URL
 
-from . import http, manager_ctx, pkg_models
+from . import archives, http, manager_ctx, pkg_models
 from . import results as R
 from .catalogue.cataloguer import CatalogueEntry
 from .common import AddonHashMethod, Defn, SourceMetadata
@@ -37,6 +37,7 @@ class HeadersIntent(enum.IntEnum):
 class Resolver(Protocol):  # pragma: no cover
     metadata: ClassVar[SourceMetadata]
     requires_access_token: ClassVar[str | None]
+    archive_opener: ClassVar[archives.ArchiveOpener | None]
 
     def __init__(self, manager_ctx: manager_ctx.ManagerCtx) -> None:
         ...
@@ -81,6 +82,8 @@ def catalogue(cls, web_client: http.ClientSessionType) -> AsyncIterator[Catalogu
 class BaseResolver(Resolver, Protocol):
     _manager_ctx: manager_ctx.ManagerCtx
 
+    archive_opener = None
+
     def __init__(self, manager_ctx: manager_ctx.ManagerCtx) -> None:
         self._manager_ctx = manager_ctx
 

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_archives.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from itertools import product
+
+import pytest
+
+from instawow.archives import find_archive_addon_tocs, make_archive_member_filter_fn
+
+
+def test_find_archive_addon_tocs_can_find_explicit_dirs():
+    assert {h for _, h in find_archive_addon_tocs(['b/', 'b/b.toc'])} == {'b'}
+
+
+def test_find_archive_addon_tocs_can_find_implicit_dirs():
+    assert {h for _, h in find_archive_addon_tocs(['b/b.toc'])} == {'b'}
+
+
+def test_find_archive_addon_tocs_discards_tocless_paths():
+    assert {h for _, h in find_archive_addon_tocs(['a', 'b/b.toc', 'c/'])} == {'b'}
+
+
+def test_find_archive_addon_tocs_discards_mismatched_tocs():
+    assert not {h for _, h in find_archive_addon_tocs(['a', 'a/b.toc'])}
+
+
+def test_find_archive_addon_tocs_accepts_multitoc():
+    assert {h for _, h in find_archive_addon_tocs(['a', 'a/a_mainline.toc'])} == {'a'}
+
+
+@pytest.mark.parametrize('ext', product('Tt', 'Oo', 'Cc'))
+def test_find_archive_addon_tocs_toc_is_case_insensitive(ext: tuple[str, ...]):
+    assert {h for _, h in find_archive_addon_tocs([f'a/a.{"".join(ext)}'])} == {'a'}
+
+
+def test_make_archive_member_filter_fn_discards_names_with_prefix_not_in_dirs():
+    is_member = make_archive_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):