Skip to content

Instantly share code, notes, and snippets.

@arduinka55055
Created October 2, 2021 20:53
Show Gist options
  • Save arduinka55055/62b69c238e88c56d2d613b1cd34915d6 to your computer and use it in GitHub Desktop.
Save arduinka55055/62b69c238e88c56d2d613b1cd34915d6 to your computer and use it in GitHub Desktop.
Python lightweight logger
from __future__ import annotations
from abc import ABC, abstractmethod
import sys
from threading import Lock
from typing import Dict, List, Set, Tuple, Type, Union
from datetime import datetime, time
from traceback import format_exception
import asyncio
import time as ttt
class LoggerHandler(ABC):
@abstractmethod
def perform(self, time: str, lvl: int, message: str, args: str):
raise NotImplementedError("Perform not implemented!")
class FileHandler(LoggerHandler):
__file:str
def __init__(self,filename) -> None:
self.__file=filename
def write(self,data:str)->int:
with open(self.__file,'a') as file:
ret=file.write(data)
file.close()
return ret
def perform(self, time: str, lvl: int, message: str, args=None):
out=time+" "+Logger.formatlevel(lvl)[0]+" "+message
if args:
out+=" "+args
out+='\r\n'
self.write(out)
defaultlevel=3
loggers:Dict[str,Logger]=dict()
handlers:Set[LoggerHandler]=set()
def addHandler(handler:LoggerHandler)->None:
"""Example usage::
logger.addHandler(logger.FileHandler("log.log"))
"""
handlers.add(handler)
def removeHandler(handler:LoggerHandler)->None:
handlers.remove(handler)
def getLogger(name:str,level:int=defaultlevel)->Logger:
"""Example Usage::
from logger import getLogger
getLogger(name="main", level=0).debug("hello",data)
"""
if loggers.get(name) != None: return loggers[name]
ret=Logger(name,level=level)
loggers[name]=ret
return ret
class Logger():
DEBUG=1
INFO=2
WARNING=3
ERROR=4
CRITICAL=5
__TIME = '\033[94m'
__MODULE = '\033[92m'
__ARGS="\033[90m"
__RST="\033[0m"
__level=3
def __init__(self,name="root",level=defaultlevel):
self.__level=level
self.name=name
def getLevel(self)->int:
return self.__level
def setLevel(self,setter):
self.__level=setter
def write(self,time,lvl,message,args=None):
for x in handlers:
x.perform(time,lvl,message,args)
def print(self,message):
print(message)
@staticmethod
def formatlevel(lvl)->Tuple[str,str]:
if lvl==1:
return "DEBUG:",''
if lvl==2:
return "INFO:",'\033[96m'
if lvl==3:
return "WARNING:",'\033[93m'
if lvl==4:
return "ERROR:",'\033[91m'
if lvl>=5:
return "CRITICAL:",'\033[1;91m'
return "None",''
def __check_level(self,lvl)->bool:
return True if lvl>=self.__level else False
def __colorize(self,time:str,message:str,lvl:int,args:Union[str,None])->str:
ret=self.__TIME+time+self.__RST+" "
ret+=self.__MODULE+self.name+self.__RST+" "
ret+=self.formatlevel(lvl)[1]+self.formatlevel(lvl)[0]+" "
ret+=message+self.__RST+" "
if args:
ret+=self.__ARGS+args+self.__RST
return ret
def log(self,level,message,*args,Logger_doPrint=True,**kwargs):
if self.__check_level(level):
time='['+str(datetime.now().replace(microsecond=0))+']:'
outargs=None
if args or kwargs:
outargs="With Args:"
for arg in args:
outargs += str(arg)+" "
for kwarg in kwargs:
outargs += "%s=%s, " % (kwarg,kwargs[kwarg])
if Logger_doPrint:
self.print(self.__colorize(time,message,level,outargs))
self.write(time,level,message,args=outargs)
def debug(self,message,*args,**kwargs): self.log(self.DEBUG,message,*args,**kwargs);
def info(self,message,*args,**kwargs): self.log(self.INFO,message,*args,**kwargs);
def warning(self,message,*args,**kwargs): self.log(self.WARNING,message,*args,**kwargs);
def error(self,message,*args,**kwargs): self.log(self.ERROR,message,*args,**kwargs);
def critical(self,message,*args,**kwargs): self.log(self.CRITICAL,message,*args,**kwargs);
def trace(self,sys_exc_info:sys._OptExcInfo,level=3):
#sys.exc_info()
self.log(2,"Here is caught traceback, don't worry. it is additional info!",Logger_doPrint=False)
self.log(2,''.join(format_exception(*sys_exc_info)),Logger_doPrint=False)
class BufferedData:
limit:int
timeout:int#TODO: timeout. send and clear buffer every 10 secs
buf:List[str]=[]
__timer:asyncio.Task=None
def __init__(self,limit:int,timeout:int) -> None:
self.limit=limit
self.timeout=timeout
def __getbuflen(self):
return sum([len(x) for x in self.buf])
async def __mytask(self):
await asyncio.sleep(self.timeout)
self.flush()
print("COMPLETED TIMER TASK!")
def triggertimer(self):
if self.__timer != None:
self.__timer.cancel(self.__timer)
self.__timer=asyncio.ensure_future(self.__mytask())
#asyncio.gather(self.__timer)
def write(self,data:str):
self.triggertimer()
if(self.__getbuflen()+len(data)>=self.limit):
self.flush()
self.buf.append(data)
def flush(self):
tmp=""
for x in self.buf:
tmp+=x+"\r\n"
self.buf=[]
return tmp
class BufferedLoggerHandler(BufferedData,LoggerHandler):
def __init__(self, limit: int,timeout:int) -> None:
LoggerHandler.__init__(self)
BufferedData.__init__(self,limit,timeout)
def perform(self, time: str, lvl: int, message: str, args: str):
self.write(time+str(lvl)+message+args)
class DiscordHandler(BufferedLoggerHandler):
def __init__(self, limit: int,channel:int,timeout:int) -> None:
super().__init__(limit,timeout)
self.__channel=channel
def flush(self)->None:
try:
asyncio.ensure_future(self.__channel.send("```m\r\n" + super().flush() + "```"))
except:
print("ERROR WITH DISCORD LOGGER!")
def perform(self, time: str, lvl: int, message: str, args=None):
out=time+" "+Logger.formatlevel(lvl)[0]+" "+message
if args:
out+=" "+args
self.write(out)
@Maksikos-ctrl
Copy link

Wow....

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment