Python colorlog 模块，StreamHandler() 实例源码

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]

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