LogManager.py

import os
import os.path as path
import gzip
import tarfile
import time
from functools import partial
from typing import Generator, List, Callable, Tuple
import logging


logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s',
    filename="daemon.log",
    level=logging.INFO
)


def func_call_handler(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.error(e)
            return None
        finally:
            logging.info(f"{func.__name__} {kwargs}")
            return

    return wrapper


def get_file_paths(dir_path) -> Generator[None, str, None]:
    abs_dir = path.abspath(dir_path)
    entrys = map(lambda entry: path.join(abs_dir, entry), os.listdir(abs_dir))
    return (e for e in entrys if path.isfile(e))


def auto_remove(file_path: str) -> bool:
    is_expire = False # TODO
    if is_expire:
        os.remove(file_path)
        return True
    return False


def split_path(_path) -> Tuple[str]:
    root, ext = path.splitext(_path)
    *paths, name = path.split(root)
    _dir = "/".join(paths)
    return _dir, name, ext


@func_call_handler
def gz_to_log(gz_path, save_path="", original_remove=False) -> str:
    _dir, name, _ = split_path(gz_path)
    save_path = save_path if save_path else path.join(_dir, f"{name}.log")
    with open(save_path, mode="w") as temp_file:
        with gzip.open(gz_path, mode="rt") as gz_file:
            temp_file.write(gz_file.read())
        if original_remove:
            os.remove(gz_path)
    return save_path


@func_call_handler
def log_to_targz(log_path, save_path="", original_remove=False) -> str:
    _dir, name, ext = split_path(log_path)
    save_path = save_path if save_path else path.join(_dir, f"{name}.tar.gz")
    with tarfile.open(save_path, mode="w:gz") as targz_file:
        targz_file.add(log_path, arcname=name+ext)
        if original_remove:
            os.remove(log_path)
    return save_path


@func_call_handler
def gz_to_targz(gz_path: str, save_path="", original_remove=False) -> str:
    _dir, name, _ = split_path(gz_path)

    temp_path = path.join(_dir, name)
    save_path = save_path if save_path else path.join(_dir, f"{name}.tar.gz")

    gz_to_log(gz_path=gz_path, save_path=temp_path, original_remove=original_remove)
    log_to_targz(log_path=temp_path, save_path=save_path, original_remove=original_remove)

    return save_path


def watch(source_dir_path: str, collect_exts: List[str]) -> None:
    file_paths = get_file_paths(source_dir_path)
    for file_path in file_paths:
        if auto_remove(file_path):
            continue
        if any(map(lambda ext: file_path.endswith(ext), collect_exts)):
            continue
        if file_path.endswith('.gz'):
            _dir, name, _ = split_path(file_path)
            save_path = path.join(_dir, f"{name}.tar.gz")
            gz_to_targz(gz_path=file_path, save_path=save_path, original_remove=True)


def strategy(source_dir_paths: List[str], collect_exts: List[str]) -> None:
    for source_dir_path in source_dir_paths:
        watch(source_dir_path, collect_exts)


def batch(batch_sec: int, func: Callable) -> None:
    while True:
        start_time = time.time()
        func()
        end_time = time.time()
        sleep_time = max(0, batch_sec - abs(end_time - start_time))
        time.sleep(sleep_time)


def main():

    BATCH_SEC = 2
    COLLECT_EXTS = [".tar.gz"]
    SOURCE_DIR_PATHS = tuple(map(path.abspath, [
        "/home/dns_sftp_it/logs",
        "/home/dns_sftp_svc/logs",
        "/home/dns_sftp_auth/logs",
    ]))
    BATCH_FUNC = partial(strategy, **{
        "collect_exts": COLLECT_EXTS,
        "source_dir_paths": SOURCE_DIR_PATHS,
    })

    batch(BATCH_SEC, BATCH_FUNC)


if __name__ == "__main__":
    main()

LogManager.service

[Unit]
Description=Log File Manager
 
[Service]
ExecStart=/usr/bin/python3 <FILE PATH>/LogManager.py
 
[Install]
WantedBy=multi-user.target