123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
# -*- 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() |