mirror of
https://codeberg.org/vlw/3rd.git
synced 2026-01-11 22:36:01 +01:00
97 lines
2 KiB
Python
97 lines
2 KiB
Python
import os
|
|
import typing
|
|
import subprocess
|
|
|
|
from ..Cli import Cli
|
|
from ..Stdout import Stdout
|
|
from ..Config import Config
|
|
from .Filesystem import Filesystem
|
|
from ..Enums import Namespace, Format, StdoutLevel
|
|
|
|
class Archive():
|
|
preserve_archives = False
|
|
|
|
def __init__(self, item: Config):
|
|
"""
|
|
Create a new Archive instance for a target item
|
|
|
|
Args:
|
|
item (Config): Target item to archive
|
|
"""
|
|
|
|
self.item = item
|
|
self.fs = Filesystem(self.item.abspath_target)
|
|
|
|
self.__stdout = Stdout(Namespace.ARCHIVE)
|
|
|
|
@property
|
|
def output_path(self) -> str:
|
|
"""
|
|
Returns a pathname to the temporary zip file created by this class
|
|
|
|
Returns:
|
|
str: Absolute pathname to target zip file
|
|
"""
|
|
|
|
return f"{self.item.abspath_temp.rstrip('/')}/{self.fs.hash}.7z"
|
|
|
|
def cleanup(self) -> None:
|
|
"""
|
|
Remove archive file
|
|
"""
|
|
|
|
if Archive.preserve_archives:
|
|
return
|
|
|
|
os.remove(self.output_path)
|
|
self.__stdout.info(f"Archive removed: {self.output_path}")
|
|
|
|
def compress(self) -> None:
|
|
"""
|
|
Compress the target path
|
|
"""
|
|
|
|
self.__stdout.log(f"Starting compression for: {self.item.abspath_target}").sleep()
|
|
|
|
# Prepare command line arguments
|
|
args = [
|
|
"7z",
|
|
"a",
|
|
"-t7z",
|
|
f"-mx={self.item.compression}"
|
|
]
|
|
|
|
# Enable encryption if archive password is set
|
|
if self.item.password:
|
|
args.append("-mhe=on")
|
|
args.append(f"-p{self.item.password}")
|
|
|
|
# Append output path and file list manifest arguments for 7zip
|
|
args.append(self.output_path)
|
|
args.append(self.item.abspath_target)
|
|
|
|
# Exclude directories thats
|
|
for exclude in self.fs.common_relative_paths():
|
|
args.append(f"-xr!{exclude}")
|
|
|
|
cmd = Cli()
|
|
cmd.run(args)
|
|
|
|
if cmd.stderr:
|
|
cmd.cleanup()
|
|
return self.__die()
|
|
|
|
self.__stdout.info(f"Temporary archive placed at: {self.fs.path}").sleep()
|
|
self.__stdout.ok(f"Compression completed for: {self.item.abspath_target}")
|
|
|
|
cmd.cleanup()
|
|
self.cleanup()
|
|
|
|
def __die(self) -> None:
|
|
"""
|
|
Skip archiving of target item
|
|
"""
|
|
|
|
self.__stdout.warn(f"Archiving skipped for: {self.item.abspath_target}")
|
|
|
|
self.cleanup()
|