archive_to_archive.py

Go to the documentation of this file.

r"""
# +==== BEGIN CatFeeder =================+
# LOGO:
# ..............(..../\
# ...............)..(.')
# ..............(../..)
# ...............\‍(__)|
# Inspired by Joan Stark
# source https://www.asciiart.eu/
# animals/cats
# /STOP
# PROJECT: CatFeeder
# FILE: archive_to_archive.py
# CREATION DATE: 15-01-2026
# LAST Modified: 1:33:19 17-01-2026
# DESCRIPTION:
# This is the backend server in charge of making the actual website work.
# /STOP
# COPYRIGHT: (c) Cat Feeder
# PURPOSE: The file containing the code for converting archives from one format to another.
# // AR
# +==== END CatFeeder =================+
"""
 
from typing import Optional
from pathlib import Path
import tempfile
 
import patoolib
from display_tty import Disp, initialise_logger
 
from . import converters_constants as CONV_CONST
from .aliases import ARCHIVE_FORMAT_ALIASES
 
from ..http_constants import DataTypes, MEDIA_TYPES
 
 
from ...core import FinalClass
from ...utils import CONST
 
 
class ArchiveToArchive(metaclass=FinalClass):
    """Class used to convert archives from one format to another using patoolib."""
 
    disp: Disp = initialise_logger(__qualname__, CONST.DEBUG)
 
    _instance: Optional["ArchiveToArchive"] = None
 
    def __new__(cls) -> "ArchiveToArchive":
        if cls._instance is None:
            cls._instance = super(ArchiveToArchive, cls).__new__(cls)
        return cls._instance
 
    def __init__(self) -> None:
        self.disp.log_debug("Initialising...")
        self.disp.log_debug("Initialised.")
 
    def __call__(self, data: bytes, source_format: DataTypes) -> CONV_CONST.ConversionResult:
        return self.archive_to_archive(data, source_format)
 
    @staticmethod
    def is_archive(path: Path) -> bool:
        """Check if a file is a valid archive."""
        try:
            return patoolib.is_archive(str(path))
        except Exception:
            return False
 
    @staticmethod
    def split_name_and_ext(path: Path) -> tuple[str, str]:
        """
        Handles multi-part extensions like .tar.gz
 
        Args:
            path: Path object to split
 
        Returns:
            Tuple of (base_name, extension)
        """
        name = path.name
        parts = name.split(".")
 
        if len(parts) <= 2:
            return path.stem, path.suffix.lstrip(".")
 
        # heuristic: treat last 2 parts as extension for compound extensions
        return parts[0], ".".join(parts[1:])
 
    def get_archive_extension(self, data_type: DataTypes) -> Optional[str]:
        """
        Get the file extension for a given archive DataType.
 
        Args:
            data_type: The DataType to get extension for
 
        Returns:
            Extension string or None
        """
        # Map DataTypes to archive extensions
        return ARCHIVE_FORMAT_ALIASES.get(data_type)
 
    def _validate_conversion_params(
        self,
        data: bytes,
        source_format: DataTypes
    ) -> tuple[Optional[DataTypes], Optional[str], Optional[str]]:
        """
        Validate conversion parameters and get destination format and extensions.
 
        Args:
            data: The archive data to validate
            source_format: The source archive format
 
        Returns:
            Tuple of (destination_format, source_ext, dest_ext)
            Returns (None, None, None) if validation fails
        """
        try:
            destination_format = MEDIA_TYPES.get_conversion_target(
                source_format)
        except (AttributeError, NameError):
            destination_format = None
 
        if destination_format is None:
            self.disp.log_debug(f"No conversion target for {source_format}")
            return None, None, None
 
        if source_format == destination_format:
            self.disp.log_debug(
                f"Source and destination formats are the same: {source_format}"
            )
            return destination_format, None, None
 
        source_ext = self.get_archive_extension(source_format)
        dest_ext = self.get_archive_extension(destination_format)
 
        if not source_ext or not dest_ext:
            self.disp.log_warning(
                f"Unknown archive extension for {source_format} -> {destination_format}"
            )
            return destination_format, None, None
 
        return destination_format, source_ext, dest_ext
 
    def _create_temp_files(
        self,
        data: bytes,
        source_ext: str,
        dest_ext: str
    ) -> tuple[Path, Path]:
        """
        Create temporary files for archive conversion.
 
        Args:
            data: The source archive data
            source_ext: The source file extension
            dest_ext: The destination file extension
 
        Returns:
            Tuple of (source_path, destination_path)
        """
        with tempfile.NamedTemporaryFile(
            suffix=f".{source_ext}",
            delete=False
        ) as src_file:
            src_file.write(data)
            src_path = Path(src_file.name)
 
        with tempfile.NamedTemporaryFile(
            suffix=f".{dest_ext}",
            delete=False
        ) as dst_file:
            dst_path = Path(dst_file.name)
 
        return src_path, dst_path
 
    def _perform_conversion(
        self,
        src_path: Path,
        dst_path: Path,
        source_format: DataTypes,
        destination_format: DataTypes
    ) -> Optional[bytes]:
        """
        Perform the actual archive conversion using patoolib.
 
        Args:
            src_path: Path to source archive file
            dst_path: Path to destination archive file
            source_format: Source archive format
            destination_format: Destination archive format
 
        Returns:
            Converted archive data as bytes, or None if conversion failed
        """
        if not self.is_archive(src_path):
            self.disp.log_warning("Input data is not a valid archive")
            return None
 
        self.disp.log_debug(
            f"Converting archive: {source_format} -> {destination_format}"
        )
 
        patoolib.repack_archive(
            str(src_path),
            str(dst_path),
            verbosity=-1  # Suppress output
        )
 
        with open(dst_path, 'rb') as f:
            converted_data = f.read()
 
        self.disp.log_debug("Archive conversion successful")
        return converted_data
 
    @staticmethod
    def _cleanup_temp_files(src_path: Path, dst_path: Path) -> None:
        """
        Clean up temporary files.
 
        Args:
            src_path: Source file path to delete
            dst_path: Destination file path to delete
        """
        if src_path.exists():
            src_path.unlink()
        if dst_path.exists():
            dst_path.unlink()
 
    def _create_failed_result(
        self,
        data: bytes,
        source_format: DataTypes,
        destination_format: DataTypes,
        result: Optional[bytes] = None
    ) -> CONV_CONST.ConversionResult:
        """
        Create a failed conversion result.
 
        Args:
            data: Original archive data
            source_format: Source archive format
            destination_format: Destination archive format
            result: Optional result data
 
        Returns:
            ConversionResult indicating failure
        """
        return CONV_CONST.ConversionResult(
            data=data,
            converted=False,
            from_type=source_format,
            to_type=destination_format,
            result=result
        )
 
    def _create_success_result(
        self,
        data: bytes,
        source_format: DataTypes,
        destination_format: DataTypes,
        converted_data: bytes
    ) -> CONV_CONST.ConversionResult:
        """
        Create a successful conversion result.
 
        Args:
            data: Original archive data
            source_format: Source archive format
            destination_format: Destination archive format
            converted_data: Converted archive data
 
        Returns:
            ConversionResult indicating success
        """
        return CONV_CONST.ConversionResult(
            data=data,
            converted=True,
            from_type=source_format,
            to_type=destination_format,
            result=converted_data
        )
 
    def _handle_validation_failure(
        self,
        data: bytes,
        source_format: DataTypes,
        destination_format: Optional[DataTypes],
        source_ext: Optional[str],
        dest_ext: Optional[str]
    ) -> Optional[CONV_CONST.ConversionResult]:
        """
        Handle validation failures and return appropriate result.
 
        Args:
            data: Original archive data
            source_format: Source archive format
            destination_format: Destination format (may be None)
            source_ext: Source extension (may be None)
            dest_ext: Destination extension (may be None)
 
        Returns:
            ConversionResult if validation failed, None if validation passed
        """
        if destination_format is None:
            return self._create_failed_result(
                data, source_format, source_format, None
            )
 
        if source_ext is None or dest_ext is None:
            result_data = data if source_format == destination_format else data
            return self._create_failed_result(
                data, source_format, destination_format, result_data
            )
 
        return None
 
    def _convert_with_temp_files(
        self,
        data: bytes,
        source_format: DataTypes,
        destination_format: DataTypes,
        source_ext: str,
        dest_ext: str
    ) -> CONV_CONST.ConversionResult:
        """
        Perform conversion using temporary files.
 
        Args:
            data: Original archive data
            source_format: Source archive format
            destination_format: Destination archive format
            source_ext: Source file extension
            dest_ext: Destination file extension
 
        Returns:
            ConversionResult with conversion outcome
        """
        src_path: Optional[Path] = None
        dst_path: Optional[Path] = None
 
        try:
            src_path, dst_path = self._create_temp_files(
                data, source_ext, dest_ext
            )
 
            converted_data = self._perform_conversion(
                src_path, dst_path, source_format, destination_format
            )
 
            if converted_data is None:
                return self._create_failed_result(
                    data, source_format, destination_format, None
                )
 
            return self._create_success_result(
                data, source_format, destination_format, converted_data
            )
 
        except Exception as e:
            self.disp.log_error(f"Archive conversion error: {e}")
            return self._create_failed_result(
                data, source_format, destination_format, None
            )
 
        finally:
            if src_path is not None and dst_path is not None:
                self._cleanup_temp_files(src_path, dst_path)
 
    def archive_to_archive(self, data: bytes, source_format: DataTypes) -> CONV_CONST.ConversionResult:
        """
        Convert archive data from one format to another using patoolib.
 
        Args:
            data (bytes): The archive data to convert.
            source_format (DataTypes): The original archive format.
 
        Returns:
            ConversionResult: The converted archive data in a contained dataclass.
        """
        destination_format, source_ext, dest_ext = self._validate_conversion_params(
            data, source_format
        )
 
        validation_result = self._handle_validation_failure(
            data, source_format, destination_format, source_ext, dest_ext
        )
        if validation_result is not None:
            return validation_result
 
        # Type guard: at this point we know these are not None due to validation
        if destination_format is None or source_ext is None or dest_ext is None:
            # This should never happen after validation, but satisfies type checker
            return self._create_failed_result(
                data, source_format, source_format, None
            )
 
        return self._convert_with_temp_files(
            data, source_format, destination_format, source_ext, dest_ext
        )