feat: add backupper

Author: egvimo

package.py

@@ -22,9 +22,16 @@ def archiver() -> None:
     _create_archive("archiver", ["archiver.py", "common.py"])
 
 
+def backupper() -> None:
+    _create_archive(
+        "backupper", ["backupper.py", "common.py", "archiver.py", "checksum.py"]
+    )
+
+
 if __name__ == "__main__":
     if len(sys.argv) > 1:
         globals()[sys.argv[1]]()
     else:
         # TODO Make generic  # pylint: disable=fixme
         archiver()
+        backupper()

scripts/backupper.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+
+from argparse import ArgumentParser, ArgumentTypeError, Namespace
+import os
+from typing import Iterator
+
+from common import Config, Logger
+from archiver import Archiver
+import checksum
+
+
+logger = Logger()
+
+
+class Backupper:  # pylint: disable=too-few-public-methods
+    def __init__(self, verbose: bool = False) -> None:
+        self._config = Config("backupper.json")
+        self._archiver = Archiver()
+        self._verbose = verbose
+        if verbose:
+            logger.verbose()
+
+    def run(
+        self, destination: str | None = None, password: str | None = None
+    ) -> Iterator[tuple[str, bool]]:
+        config = self._config.get_config()
+        sources = config["sources"]
+
+        if not sources:
+            logger.warning("No sources defined")
+            return iter(())
+
+        paths = [s["path"] for s in sources]
+        checksums = checksum.generate(paths)
+
+        archives: list[str] = []
+        for path, digest in checksums:
+            for source in sources:
+                if not os.path.samefile(path, source["path"]):
+                    continue
+                if digest != source.get("checksum", None):
+                    source["checksum"] = digest
+                    archives.append(path)
+
+        self._config.set_config(config)
+
+        return self._archiver.create_archives(
+            archives, destination=destination, password=password, verbose=self._verbose
+        )
+
+
+def _check_directory(directory_string):
+    if not os.path.isdir(directory_string):
+        raise ArgumentTypeError("invalid destination directory")
+    return directory_string
+
+
+def _create_argument_parser() -> ArgumentParser:
+    parser: ArgumentParser = ArgumentParser(
+        description="Backups data by creating 7zip archives"
+    )
+    parser.add_argument("-p", "--password", type=str, help="password")
+    parser.add_argument("-v", "--verbose", action="store_true", help="verbose")
+    parser.add_argument(
+        "-d", "--destination", type=_check_directory, help="destination directory"
+    )
+    return parser
+
+
+def main() -> None:
+    parser: ArgumentParser = _create_argument_parser()
+    args: Namespace = parser.parse_args()
+
+    backupper = Backupper(args.verbose)
+    backupper.run(args.destination, args.password)
+
+
+if __name__ == "__main__":
+    main()

tests/test_backupper.py

@@ -0,0 +1,45 @@
+import json
+import os
+from pathlib import Path
+import shutil
+import uuid
+import pytest
+
+from scripts.backupper import Backupper
+from scripts.common import Config
+
+
+TEST_PATH = "./out"
+TARGET_DIR = f"{TEST_PATH}/test"
+TEST_ARCHIVE = f"{TEST_PATH}/test.7z"
+PASSWORD = "test"
+
+
+@pytest.fixture(autouse=True)
+def prepare():
+    # pylint: disable=duplicate-code
+    if os.path.exists(TEST_PATH):
+        shutil.rmtree(TEST_PATH)
+    Path(TARGET_DIR).mkdir(parents=True, exist_ok=True)
+    for i in range(1, 4):
+        with open(f"{TARGET_DIR}/file{i}.txt", "w", encoding="utf-8") as file:
+            file.write(f"Random string in file {i} {uuid.uuid4()}")
+
+
+def test_backup():
+    with open(f"{TEST_PATH}/backupper.json", "w", encoding="utf-8") as file:
+        json.dump({"sources": [{"path": TARGET_DIR}]}, file)
+
+    config = Config(f"{TEST_PATH}/backupper.json")
+
+    backupper = Backupper()
+    backupper._config = config  # pylint: disable=protected-access
+
+    result = dict(backupper.run(destination=TEST_PATH, password=PASSWORD))
+
+    assert os.path.isfile(TEST_ARCHIVE)
+    assert result[os.path.normpath(TEST_ARCHIVE)]
+
+    with open(f"{TEST_PATH}/backupper.json", "r", encoding="utf-8") as file:
+        source = json.load(file)["sources"][0]
+        assert len(source["checksum"]) == 40