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

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 create_default_logger(level=logging.DEBUG,
                          fname='/tmp/ethereumd-proxy.log'):
    handler = logging.FileHandler(fname)
    formatter = colorlog.ColoredFormatter(
        "%(log_color)s[%(asctime)s %(levelname)s "
        "%(name)s - %(module)s:%(funcName)s:%(lineno)d]"
        "%(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='%'
    )
    handler.setFormatter(formatter)
    handler.setLevel(level)
    logger = colorlog.getLogger()
    logger.addHandler(handler)
    logger.setLevel(level)
    return logger

def cb_crawler_after_finish(self, queue):
        """Crawler callback (called after the crawler finished).

        Args:
            queue (obj): The current crawling queue.

        """

        if queue.get_all(QueueItem.STATUS_CANCELLED):
            colorlog.getLogger().warning("Detective scanner finished (but some requests were cancelled).")
        else:
            colorlog.getLogger().info("Detective scanner finished.")

        if self.__vulnerable_items:
            colorlog.getLogger().success("Found " + str(len(self.__vulnerable_items)) + " endpoint(s) with interesting information.")
            colorlog.getLogger().success("Listing endpoint(s) with interesting information.")

            for vulnerable_item in self.__vulnerable_items:
                colorlog.getLogger().success(vulnerable_item.request.url)
        else:
            colorlog.getLogger().warning("Couldn't find any endpoints with interesting information.")

def cb_request_before_start(self, queue, queue_item):
        """Crawler callback (called before a request starts).

        Args:
            queue (:class:`nyawc.Queue`): The current crawling queue.
            queue_item (:class:`nyawc.QueueItem`): The queue item that's about to start.

        Returns:
            str: A crawler action (either DO_SKIP_TO_NEXT, DO_STOP_CRAWLING or DO_CONTINUE_CRAWLING).

        """

        colorlog.getLogger().info("Investigating " + queue_item.request.url)

        if self.__vulnerable_items and self.__args.stop_if_vulnerable:
            self.stopping = True
            return CrawlerActions.DO_STOP_CRAWLING

        if self.stopping:
            return CrawlerActions.DO_STOP_CRAWLING

        return CrawlerActions.DO_CONTINUE_CRAWLING

def set_level(self):
        # Set the level of the logging system
        pref = self.preferences_dialog.ui
        level = pref.preferences_options_general_logging_value.currentText()
        level =  getattr(colorlog.logging.logging, level)
        colorlog.getLogger().setLevel(level)

    # ---------------------------------------------------------------
    # More GUI / reading the snakefile (sequana or generic)
    # ---------------------------------------------------------------

def sequana_debug_level(level="WARNING"):
    """A deubg level setter at top level of the library"""
    assert level in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
    logging_level = getattr(logger.logging.logging, level)
    logger.getLogger().setLevel(logging_level)

def create_module_dbg_logger(module_name: str):
        module_name_simple = module_name.split(".")[-1]

        def module_dbg_logger(msg: str):
            if ({module_name_simple, "*"} & set(config.CFG.glob["LOG_DBG_MODULES"])) and (config.CFG.glob["LOG_LEVEL"] == "debug"):
                logger = getLogger()
                logger.setLevel(logging.DEBUG)
                debug(module_name_simple + ": " + msg)
                logger.setLevel(logging.INFO)

        return module_dbg_logger

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 test_colorlog_basicConfig(test_logger):
    colorlog.basicConfig()
    test_logger(colorlog.getLogger())

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 __signal_handler(self, signum, frame):
        """On sigint (e.g. CTRL+C) stop the crawler.

        Args:
            signum (int): The signal number.
            frame (obj): The current stack frame.

        """

        if self.stopping:
            return

        self.stopping = True

        colorlog.getLogger().warning("Received SIGINT, stopping the crawling threads safely. This could take up to 15 seconds (the thread timeout).")

def cb_crawler_before_start(self):
        """Called before the crawler starts crawling."""

        colorlog.getLogger().info("Detective scanner started.")

def cb_request_on_error(self, queue_item, message):
        """Crawler callback (called when a request error occurs).

        Args:
            queue_item (:class:`nyawc.QueueItem`): The queue item that failed.
            message (str): The error message.

        """

        colorlog.getLogger().error(message)

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 setup_logging(self, fd, level):  # pylint: disable=C0103
        logger = logging.getLogger()
        logger.handlers.clear()
        logger.addHandler(logging.StreamHandler(fd))
        logger.setLevel(max(logging.ERROR - (level * 10), 1))
        try:
            import colorlog

            handler = logging.getLogger().handlers[0]
            handler.setFormatter(colorlog.ColoredFormatter('%(log_color)s' + self.LOGGING_FORMAT))
            colorlog.getLogger().addHandler(handler)
        except ImportError:
            # color log is just optional feature
            logging.getLogger().handlers[0].setFormatter(logging.Formatter(self.LOGGING_FORMAT))

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 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