AutoZoomDiscord/modules/functions.py

# -*- coding: utf-8 -*-
"""Some set of functions needed for discord/telegram bots and other types of apps"""

from os import sep, stat, makedirs, kill
from os import name as osname
from datetime import datetime
from typing import Union
from ujson import loads, dumps
from shutil import copyfileobj
from gzip import open as gzipopen
from psutil import Process


def nowtimeGet(format="%H:%M:%S | %d.%m.%Y") -> str:
    """Return current local time formatted as arg.

    ### Args:
        * format (str, optional): Format that should be returned. Defaults to "%H:%M:%S | %d.%m.%Y".

    ### Returns:
        * str: Local time formatted as arg.
    """    
    return datetime.now().strftime(format)


def checkSize(logs_folder=f"logs{sep}", log_size=1024) -> None:
    """Checks latest log file size and rotates it if needed.

    ### Args:
        * logs_folder (str, optional): Folder where logs stored. Defaults to "logs/".
        * log_size (int, optional): How many bytes should file containt to be rotated. Defaults to 1024.
    """    
    i = 0
    while i < 2:
        try:
            log = stat(logs_folder + 'latest.log')
            if (log.st_size / 1024) > log_size:
                with open(logs_folder + 'latest.log', 'rb') as f_in:
                    with gzipopen(f'{logs_folder}{datetime.now().strftime("%d.%m.%Y_%H:%M:%S")}.zip', 'wb') as f_out:
                        copyfileobj(f_in, f_out)  
                open(logs_folder + 'latest.log', 'w').close()
            i = 2
        except FileNotFoundError:
            try:
                log = open(logs_folder + 'latest.log', 'a')
                open(logs_folder + 'latest.log', 'a').close()
            except:
                try:
                    makedirs(logs_folder, exist_ok=True)
                    log = open(logs_folder + 'latest.log', 'a')
                    open(logs_folder + 'latest.log', 'a').close()
                except:
                    pass
            i += 1


def logWrite(message: str, logs_folder=f"logs{sep}", level="INFO") -> None:
    """Append some message to latest log file.

    ### Args:
        * message (str): Something you want to add to log
        * logs_folder (str, optional): Folder where logs stored. Defaults to "logs/".
        * level (str, optional): Log level (INFO, WARN, ERRO, CRIT, DEBG). Defaults to "INFO".
    """    

    checkSize(logs_folder=logs_folder)

    try:
        log = open(logs_folder + 'latest.log', 'a')
        open(logs_folder + 'latest.log', 'a').close()
    except:
        try:
            makedirs(logs_folder, exist_ok=True)
            log = open(logs_folder + 'latest.log', 'a')
            open(logs_folder + 'latest.log', 'a').close()
        except:
            print(f'[{nowtimeGet()}] [ERRO] Log file could not be written.', flush=True)
            return

    log.write(f'[{nowtimeGet()}] [{level}] {message}\n')
    print(f'[{nowtimeGet()}] [{level}] {message}', flush=True)
    log.close()


def jsonSave(filename: str, value: Union[list, dict]) -> None:
    """Save some list or dict as json file.

    ### Args:
        * filename (str): File to which value will be written.
        * value (Union[list, dict]): Some object that will be written to filename.
    """    
    with open(filename, 'w', encoding="utf-8") as f:
        f.write(dumps(value, indent=4, ensure_ascii=False))
        f.close()


def jsonLoad(filename: str) -> any:
    """Load json file and return python dict or list.

    ### Args:
        * filename (str): File which should be loaded.

    ### Returns:
        * any: Content of json file provided.
    """    
    with open(filename, 'r', encoding="utf-8") as f:
        value = loads(f.read())
        f.close()
    return value


def killProc(pid: int) -> None:
    """Kill the process by its PID

    ### Args:
        * pid (int): Process ID to be killed
    """    
    if osname == "posix":
        from signal import SIGKILL
        kill(pid, SIGKILL)
    else:
        p = Process(pid)
        p.kill()