def get(cls, tag):
"""
Get the global logger instance for the given tag
:param tag: the log tag
:return: the logger instance
"""
if tag not in cls._LOGGERS:
if cls._use_color:
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter( \
' %(log_color)s%(name)s/%(levelname)-8s%(reset)s |'
' %(log_color)s%(message)s%(reset)s'))
else:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter( \
' %(name)s/%(levelname)-8s | %(message)s'))
logger = logging.getLogger(tag)
logger.addHandler(handler)
cls._LOGGERS[tag] = logger
return cls._LOGGERS[tag]
python类StreamHandler()的实例源码
def init_logger(self):
self._mylogger = colorlog.getLogger("sequanix")
"""self._fh = open(self._logger_output, "w")
self._handler = colorlog.StreamHandler(self._fh)
formatter = colorlog.ColoredFormatter(
"%(log_color)s%(levelname)-8s%(reset)s %(blue)s%(message)s",
datefmt=None,
reset=True,
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red',
}
)
self._handler.setFormatter(formatter)
self._mylogger.addHandler(self._handler)
"""
def __init__(self):
if _colorlog:
self.handler = colorlog.StreamHandler()
self.handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(message)s',
datefmt=None,
reset=True,
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red,bg_white',
},
style='%'
))
else:
# Default to basic logging
self.handler = logging.StreamHandler()
self.logger = logging.getLogger('cli')
self.logger.addHandler(self.handler)
self.logger.setLevel(logging.DEBUG)
self.logger.propagate = False
def __init__(self, name='', fname='', level='INFO'):
if fname:
try:
self.fout = open(fname, 'w')
except IOError:
print('[ERROR] Failed to open %s for writing'%fname)
self.fout = None
else:
self.fout = None
self.logger = logging.getLogger(name)
self.handler = colorlog.StreamHandler()
self.handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(asctime)s %(levelname)s:%(name)s:%(message)s%(reset)s',
log_colors={
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red'
}))
self.logger.addHandler(self.handler)
self.setLevel(level)
def init_logger(logger):
"""Initialize a logger object."""
colored_handler = colorlog.StreamHandler()
colored_handler.setFormatter(colorlog.ColoredFormatter( \
"%(green)s%(asctime)s%(reset)s " + \
"%(blue)s[%(cyan)s%(name)s%(blue)s]%(reset)s " + \
"%(bold)s%(levelname)s%(reset)s " + \
"%(message)s"))
logger.setLevel(logging.INFO)
logger.addHandler(colored_handler)
def set_logger(cls, level=logging.DEBUG):
"""
Configure logging
:param level: expected level of verbosity
This utility function should probably be put elsewhere
"""
handler = colorlog.StreamHandler()
formatter = colorlog.ColoredFormatter(
"%(asctime)-2s %(log_color)s%(message)s",
datefmt='%H:%M:%S',
reset=True,
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red,bg_white',
},
secondary_log_colors={},
style='%'
)
handler.setFormatter(formatter)
logging.getLogger('').handlers = []
logging.getLogger('').addHandler(handler)
logging.getLogger('').setLevel(level=level)
def test_example():
"""Tests the usage example from the README"""
import colorlog
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(levelname)s:%(name)s:%(message)s'))
logger = colorlog.getLogger('example')
logger.addHandler(handler)
def setup_logger():
"""Setup ColorLog to enable colored logging output."""
# Colored logging
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
"%(log_color)s[%(levelname)s] %(message)s",
log_colors={
"DEBUG": "cyan",
"INFO": "white",
"SUCCESS": "green",
"WARNING": "yellow",
"ERROR": "red",
"CRITICAL": "red,bg_white"
}
))
logger = colorlog.getLogger()
logger.addHandler(handler)
# Also show INFO logs
logger.setLevel(logging.INFO)
# Add SUCCESS logging
logging.SUCCESS = 25
logging.addLevelName(
logging.SUCCESS,
"SUCCESS"
)
setattr(
logger,
"success",
lambda message, *args: logger._log(logging.SUCCESS, message, args)
)
def init(level=logging.INFO):
logger = colorlog.getLogger('elliptic.Mesh.Mesh')
logger.setLevel(level)
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s:%(name)s:%(message)s'))
logger.addHandler(handler)
def __init__(self, scope=None, proxies=None, browser=None, log_level='DEBUG'):
self._scope = (scope or Scope())
self.proxies = (proxies or {})
self.script = None
self.registered_names = set()
self.browser = browser
self._is_ready = False
self._exec_options = {}
self.register_defaults()
self.sync_scopes()
self.ready()
self._line = 0
self._col = 0
self._log_handler = colorlog.StreamHandler(stream=sys.stderr)
self._log_handler.setFormatter(colorlog.ColoredFormatter('%(log_color)s%(message)s'))
self.log = colorlog.getLogger('friendscript')
self.log.propagate = False
self.log.setLevel(logging.getLevelName(log_level.upper()))
self.log.addHandler(self._log_handler)
def __init__(self, path: str, recursive=False, worker_count=cpu_count() * 2, verbose=0) -> None:
"""
Create an OCyara object that can scan the specified directory or file and store the results.
Arguments:
path -- File or directory to be processed
Keyword Arguments:
recursive -- Whether the specified path should be recursivly searched for images (default False)
worker_count -- The number of worker processes that should be spawned when
run() is executed (default available CPU cores * 2)
verbose -- An int() from 0-2 that sets the verbosity level.
0 is default, 1 is information and 2 is debug
"""
self.path = path
self.recursive = recursive
self.q = JoinableQueue()
self.results = {}
self.threads = worker_count
self.lock = Lock()
self.workers = []
if os.path.isdir(self.path):
if self.path[-1] is '/':
self.path += '*'
else:
self.path += '/*'
self.manager = Manager()
self.matchedfiles = None
self.total_items_to_queue = None
self.total_added_to_queue = None
self.queue_items_completed = None
self.tempdir = tempfile.TemporaryDirectory()
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(levelname)s:%(name)s:%(message)s'))
self.logger = colorlog.getLogger('OCyara')
self.logger.addHandler(handler)
if verbose == 1:
self.logger.setLevel('INFO')
elif verbose > 1:
self.logger.setLevel('DEBUG')
else:
self.logger.setLevel('WARN')
def _process_image(self, yara_rule: str, save_context: bool) -> None:
"""
Perform OCR and yara rule matching as a worker.
process_image() is used by the run() method to create multiple worker processes for
parallel execution. process_image normally will not be called directly.
Arguments:
yara_rule -- File path pointing to a Yara rule file
"""
context = None
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(levelname)s:%(name)s:%(message)s'))
# Creates a logger object for the individual workers that contains the PID as part of the message header
worker_logger = colorlog.getLogger('worker_'+str(os.getpid()))
worker_logger.addHandler(handler)
worker_logger.setLevel(self.logger.level)
worker_logger.info('PID {0} created to process queue'.format(str(os.getpid())))
while True:
try:
image, filepath = self.q.get(timeout=.25)
except Empty:
if self.total_added_to_queue[0] == self.total_items_to_queue[0]:
worker_logger.debug('Queue Empty PID %d exiting' % os.getpid())
return
else:
worker_logger.debug('Queue still loading')
continue
ocrtext = tesserocr.image_to_text(image)
try:
rules = yara.compile(yara_rule)
except yara.Error:
worker_logger.error('Invalid rule file/rule file not found: {0}'.format(yara_rule))
matches = rules.match(data=ocrtext)
# If there is a match, store the filename and the rule in a dictionary local to this process.
# Create an editable copy of the master manager dict.
with self.lock:
local_results_dict = self.matchedfiles[0]
if matches:
if save_context:
context = ocrtext
for x in matches:
try:
local_results_dict[filepath].append((x.rule, context))
except KeyError:
local_results_dict[filepath] = [(x.rule, context)]
self.matchedfiles[0] = local_results_dict
self.queue_items_completed[0] += 1
self.q.task_done()
def pytest_configure(config):
if config.option.logging_level is None:
logging_level = logging.INFO
else:
logging_level = config.option.logging_level
if logging_level not in logging_level_map:
raise PytestConfigurationError(
'{} not a recognised logging level'.format(logging_level)
)
logging_level = logging_level_map[logging_level]
sh = colorlog.StreamHandler()
sh.setLevel(logging_level)
formatter = colorlog.ColoredFormatter(
"%(white)s%(name)s %(reset)s %(log_color)s%"
"(levelname)-8s%(reset)s %(blue)s%(message)s",
datefmt=None,
reset=True,
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red,bg_white',
},
secondary_log_colors={},
style='%'
)
sh.setFormatter(formatter)
root = logging.getLogger()
# remove the default streamhandler
handler = next(
(
handler for handler in root.handlers if
isinstance(handler, logging.StreamHandler)
), None
)
if handler:
index = root.handlers.index(handler)
root.handlers.pop(index)
# and add our fancy coloured one
root.addHandler(sh)
if config.option.file_logging is True:
add_file_logging()
def setup_logger(name="Logger"):
""" Sets up a logging.Logger with the name $name. If the colorlog module
is available, the logger will use colors, otherwise it will be in b/w.
The colorlog module is available at
https://github.com/borntyping/python-colorlog
but can also easily be installed with e.g. 'sudo pip3 colorlog' or similar
commands.
Arguments:
name: name of the logger
Returns:
Logger
"""
if colorlog:
_logger = colorlog.getLogger(name)
else:
_logger = logging.getLogger(name)
if _logger.handlers:
# the logger already has handlers attached to it, even though
# we didn't add it ==> logging.get_logger got us an existing
# logger ==> we don't need to do anything
return _logger
_logger.setLevel(logging.DEBUG)
if colorlog is not None:
sh = colorlog.StreamHandler()
log_colors = {'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'red'}
formatter = colorlog.ColoredFormatter(
'%(log_color)s%(name)s:%(levelname)s:%(message)s',
log_colors=log_colors)
else:
# no colorlog available:
sh = logging.StreamHandler()
formatter = logging.Formatter('%(name)s:%(levelname)s:%(message)s')
sh.setFormatter(formatter)
sh.setLevel(logging.DEBUG)
_logger.addHandler(sh)
if colorlog is None:
_logger.debug("Module colorlog not available. Log will be b/w.")
return _logger
