sarasacw-omrf/packer/src/sarasacw_omrf_packer/archiver.py

import zipfile
import shutil
from pathlib import Path
from types import TracebackType
from typing import Optional


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_dir(self, arcname: Path) -> None:
        if self.__dist_dir is not None:
            target_filepath = self.__dist_dir / arcname
            target_filepath.parent.mkdir(parents=True, exist_ok=True)

        if self.__dist_zip is not None:
            self.__dist_zip.mkdir(arcname.as_posix())

    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.as_posix())

    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(arcname.as_posix(), 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()