feat: finish basic packer

- finish basic packer - add basic jinja2 template
2026-08-02 20:54:14 +08:00
parent 8098216a83
commit 3261a4cd4b
11 changed files with 203 additions and 32 deletions
@@ -1,10 +1,91 @@
 import zipfile
+import shutil
 from pathlib import Path
+from types import TracebackType
+from typing import Optional


-def pack(dist_dir: str, output_zip: str) -> None:
-    dist = Path(dist_dir)
-    with zipfile.ZipFile(output_zip, "w", zipfile.ZIP_DEFLATED) as zf:
-        for path in dist.rglob("*"):
-            if path.is_file():
-                zf.write(path, path.relative_to(dist))
+class Archiver:
+    """Dual sink that mirrors the same distribution tree into a directory and a zip archive.
+
+    On construction the archiver (optionally) prepares a directory on disk and
+    (optionally) opens a :class:`zipfile.ZipFile` for writing. Every subsequent
+    :meth:`push_file`/:meth:`push_text` call then writes the same logical entry
+    -- identified by its ``arcname``, a path relative to the distribution root
+    -- into both destinations, skipping whichever sink was not requested.
+
+    The archiver is usable as a context manager so that the underlying archive
+    is closed deterministically::
+
+        with Archiver(dist_dir, dist_zip) as arc:
+            arc.push_file(...)
+    """
+
+    __dist_dir: Optional[Path]
+    __dist_zip: Optional[zipfile.ZipFile]
+
+    def __init__(self, dist_dir: Optional[Path], dist_zip: Optional[Path]) -> None:
+        """Configure the sinks and acquire their backing resources.
+
+        :param dist_dir: Directory to mirror the tree into. Created (with
+            parents) when given; ``None`` disables the directory sink.
+        :param dist_zip: Path of the zip archive to create, truncated on open.
+            ``None`` disables the zip sink.
+        """
+        # make sure the distribution directory is existing
+        self.__dist_dir = dist_dir
+        if self.__dist_dir is not None:
+            self.__dist_dir.mkdir(parents=True, exist_ok=True)
+        # create zip instance
+        if dist_zip is not None:
+            self.__dist_zip = zipfile.ZipFile(dist_zip, "w", zipfile.ZIP_DEFLATED)
+        else:
+            self.__dist_zip = None
+
+    def push_file(self, filepath: Path, arcname: Path) -> None:
+        """Mirror an existing file into both sinks.
+
+        :param filepath: Source file on disk to copy / embed.
+        :param arcname: Destination path relative to the distribution root.
+            Missing parent directories are created for the directory sink.
+        """
+        if self.__dist_dir is not None:
+            target_filepath = self.__dist_dir / arcname
+            target_filepath.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy(filepath, target_filepath)
+
+        if self.__dist_zip is not None:
+            self.__dist_zip.write(filepath, arcname)
+
+    def push_text(self, text: str, arcname: Path) -> None:
+        """Mirror in-memory text into both sinks.
+
+        :param text: UTF-8 text content to write.
+        :param arcname: Destination path relative to the distribution root.
+            Missing parent directories are created for the directory sink.
+        """
+        if self.__dist_dir is not None:
+            target_filepath = self.__dist_dir / arcname
+            target_filepath.parent.mkdir(parents=True, exist_ok=True)
+            with open(target_filepath, "w", encoding="utf-8") as f:
+                f.write(text)
+
+        if self.__dist_zip is not None:
+            self.__dist_zip.writestr(str(arcname), text)
+
+    def __enter__(self) -> "Archiver":
+        """Enter the context and return this archiver as the bound target."""
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc_value: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        """Release the zip sink by closing its backing archive, if any.
+
+        Exceptions raised inside the ``with`` body are never suppressed.
+        """
+        if self.__dist_zip is not None:
+            self.__dist_zip.close()