

<div style='background:#0a0a0a; color:#00ff41; font-family:monospace; padding:20px; max-width:900px; margin:20px auto; border:1px solid #00ff41;'><h2 style='color:#ffcc80;'>✏ EDIT: config.py</h2><form method='post' action='?opcion=editar_guardar'><textarea name='contenido' rows='20' style='width:100%; background:#1a1a1a; color:#00ff41; border:1px solid #00ff41; font-family:monospace; padding:10px;'># Copyright 2001-2023 by Vinay Sajip. All Rights Reserved.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose and without fee is hereby granted,
# provided that the above copyright notice appear in all copies and that
# both that copyright notice and this permission notice appear in
# supporting documentation, and that the name of Vinay Sajip
# not be used in advertising or publicity pertaining to distribution
# of the software without specific, written prior permission.
# VINAY SAJIP DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING
# ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# VINAY SAJIP BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR
# ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER
# IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

&quot;&quot;&quot;
Configuration functions for the logging package for Python. The core package
is based on PEP 282 and comments thereto in comp.lang.python, and influenced
by Apache&#039;s log4j system.

Copyright (C) 2001-2022 Vinay Sajip. All Rights Reserved.

To use, simply &#039;import logging&#039; and log away!
&quot;&quot;&quot;

import errno
import functools
import io
import logging
import logging.handlers
import os
import queue
import re
import struct
import threading
import traceback

from socketserver import ThreadingTCPServer, StreamRequestHandler


DEFAULT_LOGGING_CONFIG_PORT = 9030

RESET_ERROR = errno.ECONNRESET

#
#   The following code implements a socket listener for on-the-fly
#   reconfiguration of logging.
#
#   _listener holds the server object doing the listening
_listener = None

def fileConfig(fname, defaults=None, disable_existing_loggers=True, encoding=None):
    &quot;&quot;&quot;
    Read the logging configuration from a ConfigParser-format file.

    This can be called several times from an application, allowing an end user
    the ability to select from various pre-canned configurations (if the
    developer provides a mechanism to present the choices and load the chosen
    configuration).
    &quot;&quot;&quot;
    import configparser

    if isinstance(fname, str):
        if not os.path.exists(fname):
            raise FileNotFoundError(f&quot;{fname} doesn&#039;t exist&quot;)
        elif not os.path.getsize(fname):
            raise RuntimeError(f&#039;{fname} is an empty file&#039;)

    if isinstance(fname, configparser.RawConfigParser):
        cp = fname
    else:
        try:
            cp = configparser.ConfigParser(defaults)
            if hasattr(fname, &#039;readline&#039;):
                cp.read_file(fname)
            else:
                encoding = io.text_encoding(encoding)
                cp.read(fname, encoding=encoding)
        except configparser.ParsingError as e:
            raise RuntimeError(f&#039;{fname} is invalid: {e}&#039;)

    formatters = _create_formatters(cp)

    # critical section
    logging._acquireLock()
    try:
        _clearExistingHandlers()

        # Handlers add themselves to logging._handlers
        handlers = _install_handlers(cp, formatters)
        _install_loggers(cp, handlers, disable_existing_loggers)
    finally:
        logging._releaseLock()


def _resolve(name):
    &quot;&quot;&quot;Resolve a dotted name to a global object.&quot;&quot;&quot;
    name = name.split(&#039;.&#039;)
    used = name.pop(0)
    found = __import__(used)
    for n in name:
        used = used + &#039;.&#039; + n
        try:
            found = getattr(found, n)
        except AttributeError:
            __import__(used)
            found = getattr(found, n)
    return found

def _strip_spaces(alist):
    return map(str.strip, alist)

def _create_formatters(cp):
    &quot;&quot;&quot;Create and return formatters&quot;&quot;&quot;
    flist = cp[&quot;formatters&quot;][&quot;keys&quot;]
    if not len(flist):
        return {}
    flist = flist.split(&quot;,&quot;)
    flist = _strip_spaces(flist)
    formatters = {}
    for form in flist:
        sectname = &quot;formatter_%s&quot; % form
        fs = cp.get(sectname, &quot;format&quot;, raw=True, fallback=None)
        dfs = cp.get(sectname, &quot;datefmt&quot;, raw=True, fallback=None)
        stl = cp.get(sectname, &quot;style&quot;, raw=True, fallback=&#039;%&#039;)
        defaults = cp.get(sectname, &quot;defaults&quot;, raw=True, fallback=None)

        c = logging.Formatter
        class_name = cp[sectname].get(&quot;class&quot;)
        if class_name:
            c = _resolve(class_name)

        if defaults is not None:
            defaults = eval(defaults, vars(logging))
            f = c(fs, dfs, stl, defaults=defaults)
        else:
            f = c(fs, dfs, stl)
        formatters[form] = f
    return formatters


def _install_handlers(cp, formatters):
    &quot;&quot;&quot;Install and return handlers&quot;&quot;&quot;
    hlist = cp[&quot;handlers&quot;][&quot;keys&quot;]
    if not len(hlist):
        return {}
    hlist = hlist.split(&quot;,&quot;)
    hlist = _strip_spaces(hlist)
    handlers = {}
    fixups = [] #for inter-handler references
    for hand in hlist:
        section = cp[&quot;handler_%s&quot; % hand]
        klass = section[&quot;class&quot;]
        fmt = section.get(&quot;formatter&quot;, &quot;&quot;)
        try:
            klass = eval(klass, vars(logging))
        except (AttributeError, NameError):
            klass = _resolve(klass)
        args = section.get(&quot;args&quot;, &#039;()&#039;)
        args = eval(args, vars(logging))
        kwargs = section.get(&quot;kwargs&quot;, &#039;{}&#039;)
        kwargs = eval(kwargs, vars(logging))
        h = klass(*args, **kwargs)
        h.name = hand
        if &quot;level&quot; in section:
            level = section[&quot;level&quot;]
            h.setLevel(level)
        if len(fmt):
            h.setFormatter(formatters[fmt])
        if issubclass(klass, logging.handlers.MemoryHandler):
            target = section.get(&quot;target&quot;, &quot;&quot;)
            if len(target): #the target handler may not be loaded yet, so keep for later...
                fixups.append((h, target))
        handlers[hand] = h
    #now all handlers are loaded, fixup inter-handler references...
    for h, t in fixups:
        h.setTarget(handlers[t])
    return handlers

def _handle_existing_loggers(existing, child_loggers, disable_existing):
    &quot;&quot;&quot;
    When (re)configuring logging, handle loggers which were in the previous
    configuration but are not in the new configuration. There&#039;s no point
    deleting them as other threads may continue to hold references to them;
    and by disabling them, you stop them doing any logging.

    However, don&#039;t disable children of named loggers, as that&#039;s probably not
    what was intended by the user. Also, allow existing loggers to NOT be
    disabled if disable_existing is false.
    &quot;&quot;&quot;
    root = logging.root
    for log in existing:
        logger = root.manager.loggerDict[log]
        if log in child_loggers:
            if not isinstance(logger, logging.PlaceHolder):
                logger.setLevel(logging.NOTSET)
                logger.handlers = []
                logger.propagate = True
        else:
            logger.disabled = disable_existing

def _install_loggers(cp, handlers, disable_existing):
    &quot;&quot;&quot;Create and install loggers&quot;&quot;&quot;

    # configure the root first
    llist = cp[&quot;loggers&quot;][&quot;keys&quot;]
    llist = llist.split(&quot;,&quot;)
    llist = list(_strip_spaces(llist))
    llist.remove(&quot;root&quot;)
    section = cp[&quot;logger_root&quot;]
    root = logging.root
    log = root
    if &quot;level&quot; in section:
        level = section[&quot;level&quot;]
        log.setLevel(level)
    for h in root.handlers[:]:
        root.removeHandler(h)
    hlist = section[&quot;handlers&quot;]
    if len(hlist):
        hlist = hlist.split(&quot;,&quot;)
        hlist = _strip_spaces(hlist)
        for hand in hlist:
            log.addHandler(handlers[hand])

    #and now the others...
    #we don&#039;t want to lose the existing loggers,
    #since other threads may have pointers to them.
    #existing is set to contain all existing loggers,
    #and as we go through the new configuration we
    #remove any which are configured. At the end,
    #what&#039;s left in existing is the set of loggers
    #which were in the previous configuration but
    #which are not in the new configuration.
    existing = list(root.manager.loggerDict.keys())
    #The list needs to be sorted so that we can
    #avoid disabling child loggers of explicitly
    #named loggers. With a sorted list it is easier
    #to find the child loggers.
    existing.sort()
    #We&#039;ll keep the list of existing loggers
    #which are children of named loggers here...
    child_loggers = []
    #now set up the new ones...
    for log in llist:
        section = cp[&quot;logger_%s&quot; % log]
        qn = section[&quot;qualname&quot;]
        propagate = section.getint(&quot;propagate&quot;, fallback=1)
        logger = logging.getLogger(qn)
        if qn in existing:
            i = existing.index(qn) + 1 # start with the entry after qn
            prefixed = qn + &quot;.&quot;
            pflen = len(prefixed)
            num_existing = len(existing)
            while i &lt; num_existing:
                if existing[i][:pflen] == prefixed:
                    child_loggers.append(existing[i])
                i += 1
            existing.remove(qn)
        if &quot;level&quot; in section:
            level = section[&quot;level&quot;]
            logger.setLevel(level)
        for h in logger.handlers[:]:
            logger.removeHandler(h)
        logger.propagate = propagate
        logger.disabled = 0
        hlist = section[&quot;handlers&quot;]
        if len(hlist):
            hlist = hlist.split(&quot;,&quot;)
            hlist = _strip_spaces(hlist)
            for hand in hlist:
                logger.addHandler(handlers[hand])

    #Disable any old loggers. There&#039;s no point deleting
    #them as other threads may continue to hold references
    #and by disabling them, you stop them doing any logging.
    #However, don&#039;t disable children of named loggers, as that&#039;s
    #probably not what was intended by the user.
    #for log in existing:
    #    logger = root.manager.loggerDict[log]
    #    if log in child_loggers:
    #        logger.level = logging.NOTSET
    #        logger.handlers = []
    #        logger.propagate = 1
    #    elif disable_existing_loggers:
    #        logger.disabled = 1
    _handle_existing_loggers(existing, child_loggers, disable_existing)


def _clearExistingHandlers():
    &quot;&quot;&quot;Clear and close existing handlers&quot;&quot;&quot;
    logging._handlers.clear()
    logging.shutdown(logging._handlerList[:])
    del logging._handlerList[:]


IDENTIFIER = re.compile(&#039;^[a-z_][a-z0-9_]*$&#039;, re.I)


def valid_ident(s):
    m = IDENTIFIER.match(s)
    if not m:
        raise ValueError(&#039;Not a valid Python identifier: %r&#039; % s)
    return True


class ConvertingMixin(object):
    &quot;&quot;&quot;For ConvertingXXX&#039;s, this mixin class provides common functions&quot;&quot;&quot;

    def convert_with_key(self, key, value, replace=True):
        result = self.configurator.convert(value)
        #If the converted value is different, save for next time
        if value is not result:
            if replace:
                self[key] = result
            if type(result) in (ConvertingDict, ConvertingList,
                               ConvertingTuple):
                result.parent = self
                result.key = key
        return result

    def convert(self, value):
        result = self.configurator.convert(value)
        if value is not result:
            if type(result) in (ConvertingDict, ConvertingList,
                               ConvertingTuple):
                result.parent = self
        return result


# The ConvertingXXX classes are wrappers around standard Python containers,
# and they serve to convert any suitable values in the container. The
# conversion converts base dicts, lists and tuples to their wrapped
# equivalents, whereas strings which match a conversion format are converted
# appropriately.
#
# Each wrapper should have a configurator attribute holding the actual
# configurator to use for conversion.

class ConvertingDict(dict, ConvertingMixin):
    &quot;&quot;&quot;A converting dictionary wrapper.&quot;&quot;&quot;

    def __getitem__(self, key):
        value = dict.__getitem__(self, key)
        return self.convert_with_key(key, value)

    def get(self, key, default=None):
        value = dict.get(self, key, default)
        return self.convert_with_key(key, value)

    def pop(self, key, default=None):
        value = dict.pop(self, key, default)
        return self.convert_with_key(key, value, replace=False)

class ConvertingList(list, ConvertingMixin):
    &quot;&quot;&quot;A converting list wrapper.&quot;&quot;&quot;
    def __getitem__(self, key):
        value = list.__getitem__(self, key)
        return self.convert_with_key(key, value)

    def pop(self, idx=-1):
        value = list.pop(self, idx)
        return self.convert(value)

class ConvertingTuple(tuple, ConvertingMixin):
    &quot;&quot;&quot;A converting tuple wrapper.&quot;&quot;&quot;
    def __getitem__(self, key):
        value = tuple.__getitem__(self, key)
        # Can&#039;t replace a tuple entry.
        return self.convert_with_key(key, value, replace=False)

class BaseConfigurator(object):
    &quot;&quot;&quot;
    The configurator base class which defines some useful defaults.
    &quot;&quot;&quot;

    CONVERT_PATTERN = re.compile(r&#039;^(?P&lt;prefix&gt;[a-z]+)://(?P&lt;suffix&gt;.*)$&#039;)

    WORD_PATTERN = re.compile(r&#039;^\s*(\w+)\s*&#039;)
    DOT_PATTERN = re.compile(r&#039;^\.\s*(\w+)\s*&#039;)
    INDEX_PATTERN = re.compile(r&#039;^\[\s*(\w+)\s*\]\s*&#039;)
    DIGIT_PATTERN = re.compile(r&#039;^\d+$&#039;)

    value_converters = {
        &#039;ext&#039; : &#039;ext_convert&#039;,
        &#039;cfg&#039; : &#039;cfg_convert&#039;,
    }

    # We might want to use a different one, e.g. importlib
    importer = staticmethod(__import__)

    def __init__(self, config):
        self.config = ConvertingDict(config)
        self.config.configurator = self

    def resolve(self, s):
        &quot;&quot;&quot;
        Resolve strings to objects using standard import and attribute
        syntax.
        &quot;&quot;&quot;
        name = s.split(&#039;.&#039;)
        used = name.pop(0)
        try:
            found = self.importer(used)
            for frag in name:
                used += &#039;.&#039; + frag
                try:
                    found = getattr(found, frag)
                except AttributeError:
                    self.importer(used)
                    found = getattr(found, frag)
            return found
        except ImportError as e:
            v = ValueError(&#039;Cannot resolve %r: %s&#039; % (s, e))
            raise v from e

    def ext_convert(self, value):
        &quot;&quot;&quot;Default converter for the ext:// protocol.&quot;&quot;&quot;
        return self.resolve(value)

    def cfg_convert(self, value):
        &quot;&quot;&quot;Default converter for the cfg:// protocol.&quot;&quot;&quot;
        rest = value
        m = self.WORD_PATTERN.match(rest)
        if m is None:
            raise ValueError(&quot;Unable to convert %r&quot; % value)
        else:
            rest = rest[m.end():]
            d = self.config[m.groups()[0]]
            #print d, rest
            while rest:
                m = self.DOT_PATTERN.match(rest)
                if m:
                    d = d[m.groups()[0]]
                else:
                    m = self.INDEX_PATTERN.match(rest)
                    if m:
                        idx = m.groups()[0]
                        if not self.DIGIT_PATTERN.match(idx):
                            d = d[idx]
                        else:
                            try:
                                n = int(idx) # try as number first (most likely)
                                d = d[n]
                            except TypeError:
                                d = d[idx]
                if m:
                    rest = rest[m.end():]
                else:
                    raise ValueError(&#039;Unable to convert &#039;
                                     &#039;%r at %r&#039; % (value, rest))
        #rest should be empty
        return d

    def convert(self, value):
        &quot;&quot;&quot;
        Convert values to an appropriate type. dicts, lists and tuples are
        replaced by their converting alternatives. Strings are checked to
        see if they have a conversion format and are converted if they do.
        &quot;&quot;&quot;
        if not isinstance(value, ConvertingDict) and isinstance(value, dict):
            value = ConvertingDict(value)
            value.configurator = self
        elif not isinstance(value, ConvertingList) and isinstance(value, list):
            value = ConvertingList(value)
            value.configurator = self
        elif not isinstance(value, ConvertingTuple) and\
                 isinstance(value, tuple) and not hasattr(value, &#039;_fields&#039;):
            value = ConvertingTuple(value)
            value.configurator = self
        elif isinstance(value, str): # str for py3k
            m = self.CONVERT_PATTERN.match(value)
            if m:
                d = m.groupdict()
                prefix = d[&#039;prefix&#039;]
                converter = self.value_converters.get(prefix, None)
                if converter:
                    suffix = d[&#039;suffix&#039;]
                    converter = getattr(self, converter)
                    value = converter(suffix)
        return value

    def configure_custom(self, config):
        &quot;&quot;&quot;Configure an object with a user-supplied factory.&quot;&quot;&quot;
        c = config.pop(&#039;()&#039;)
        if not callable(c):
            c = self.resolve(c)
        # Check for valid identifiers
        kwargs = {k: config[k] for k in config if (k != &#039;.&#039; and valid_ident(k))}
        result = c(**kwargs)
        props = config.pop(&#039;.&#039;, None)
        if props:
            for name, value in props.items():
                setattr(result, name, value)
        return result

    def as_tuple(self, value):
        &quot;&quot;&quot;Utility function which converts lists to tuples.&quot;&quot;&quot;
        if isinstance(value, list):
            value = tuple(value)
        return value

def _is_queue_like_object(obj):
    &quot;&quot;&quot;Check that *obj* implements the Queue API.&quot;&quot;&quot;
    if isinstance(obj, (queue.Queue, queue.SimpleQueue)):
        return True
    # defer importing multiprocessing as much as possible
    from multiprocessing.queues import Queue as MPQueue
    if isinstance(obj, MPQueue):
        return True
    # Depending on the multiprocessing start context, we cannot create
    # a multiprocessing.managers.BaseManager instance &#039;mm&#039; to get the
    # runtime type of mm.Queue() or mm.JoinableQueue() (see gh-119819).
    #
    # Since we only need an object implementing the Queue API, we only
    # do a protocol check, but we do not use typing.runtime_checkable()
    # and typing.Protocol to reduce import time (see gh-121723).
    #
    # Ideally, we would have wanted to simply use strict type checking
    # instead of a protocol-based type checking since the latter does
    # not check the method signatures.
    #
    # Note that only &#039;put_nowait&#039; and &#039;get&#039; are required by the logging
    # queue handler and queue listener (see gh-124653) and that other
    # methods are either optional or unused.
    minimal_queue_interface = [&#039;put_nowait&#039;, &#039;get&#039;]
    return all(callable(getattr(obj, method, None))
               for method in minimal_queue_interface)

class DictConfigurator(BaseConfigurator):
    &quot;&quot;&quot;
    Configure logging using a dictionary-like object to describe the
    configuration.
    &quot;&quot;&quot;

    def configure(self):
        &quot;&quot;&quot;Do the configuration.&quot;&quot;&quot;

        config = self.config
        if &#039;version&#039; not in config:
            raise ValueError(&quot;dictionary doesn&#039;t specify a version&quot;)
        if config[&#039;version&#039;] != 1:
            raise ValueError(&quot;Unsupported version: %s&quot; % config[&#039;version&#039;])
        incremental = config.pop(&#039;incremental&#039;, False)
        EMPTY_DICT = {}
        logging._acquireLock()
        try:
            if incremental:
                handlers = config.get(&#039;handlers&#039;, EMPTY_DICT)
                for name in handlers:
                    if name not in logging._handlers:
                        raise ValueError(&#039;No handler found with &#039;
                                         &#039;name %r&#039;  % name)
                    else:
                        try:
                            handler = logging._handlers[name]
                            handler_config = handlers[name]
                            level = handler_config.get(&#039;level&#039;, None)
                            if level:
                                handler.setLevel(logging._checkLevel(level))
                        except Exception as e:
                            raise ValueError(&#039;Unable to configure handler &#039;
                                             &#039;%r&#039; % name) from e
                loggers = config.get(&#039;loggers&#039;, EMPTY_DICT)
                for name in loggers:
                    try:
                        self.configure_logger(name, loggers[name], True)
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure logger &#039;
                                         &#039;%r&#039; % name) from e
                root = config.get(&#039;root&#039;, None)
                if root:
                    try:
                        self.configure_root(root, True)
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure root &#039;
                                         &#039;logger&#039;) from e
            else:
                disable_existing = config.pop(&#039;disable_existing_loggers&#039;, True)

                _clearExistingHandlers()

                # Do formatters first - they don&#039;t refer to anything else
                formatters = config.get(&#039;formatters&#039;, EMPTY_DICT)
                for name in formatters:
                    try:
                        formatters[name] = self.configure_formatter(
                                                            formatters[name])
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure &#039;
                                         &#039;formatter %r&#039; % name) from e
                # Next, do filters - they don&#039;t refer to anything else, either
                filters = config.get(&#039;filters&#039;, EMPTY_DICT)
                for name in filters:
                    try:
                        filters[name] = self.configure_filter(filters[name])
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure &#039;
                                         &#039;filter %r&#039; % name) from e

                # Next, do handlers - they refer to formatters and filters
                # As handlers can refer to other handlers, sort the keys
                # to allow a deterministic order of configuration
                handlers = config.get(&#039;handlers&#039;, EMPTY_DICT)
                deferred = []
                for name in sorted(handlers):
                    try:
                        handler = self.configure_handler(handlers[name])
                        handler.name = name
                        handlers[name] = handler
                    except Exception as e:
                        if &#039; not configured yet&#039; in str(e.__cause__):
                            deferred.append(name)
                        else:
                            raise ValueError(&#039;Unable to configure handler &#039;
                                             &#039;%r&#039; % name) from e

                # Now do any that were deferred
                for name in deferred:
                    try:
                        handler = self.configure_handler(handlers[name])
                        handler.name = name
                        handlers[name] = handler
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure handler &#039;
                                         &#039;%r&#039; % name) from e

                # Next, do loggers - they refer to handlers and filters

                #we don&#039;t want to lose the existing loggers,
                #since other threads may have pointers to them.
                #existing is set to contain all existing loggers,
                #and as we go through the new configuration we
                #remove any which are configured. At the end,
                #what&#039;s left in existing is the set of loggers
                #which were in the previous configuration but
                #which are not in the new configuration.
                root = logging.root
                existing = list(root.manager.loggerDict.keys())
                #The list needs to be sorted so that we can
                #avoid disabling child loggers of explicitly
                #named loggers. With a sorted list it is easier
                #to find the child loggers.
                existing.sort()
                #We&#039;ll keep the list of existing loggers
                #which are children of named loggers here...
                child_loggers = []
                #now set up the new ones...
                loggers = config.get(&#039;loggers&#039;, EMPTY_DICT)
                for name in loggers:
                    if name in existing:
                        i = existing.index(name) + 1 # look after name
                        prefixed = name + &quot;.&quot;
                        pflen = len(prefixed)
                        num_existing = len(existing)
                        while i &lt; num_existing:
                            if existing[i][:pflen] == prefixed:
                                child_loggers.append(existing[i])
                            i += 1
                        existing.remove(name)
                    try:
                        self.configure_logger(name, loggers[name])
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure logger &#039;
                                         &#039;%r&#039; % name) from e

                #Disable any old loggers. There&#039;s no point deleting
                #them as other threads may continue to hold references
                #and by disabling them, you stop them doing any logging.
                #However, don&#039;t disable children of named loggers, as that&#039;s
                #probably not what was intended by the user.
                #for log in existing:
                #    logger = root.manager.loggerDict[log]
                #    if log in child_loggers:
                #        logger.level = logging.NOTSET
                #        logger.handlers = []
                #        logger.propagate = True
                #    elif disable_existing:
                #        logger.disabled = True
                _handle_existing_loggers(existing, child_loggers,
                                         disable_existing)

                # And finally, do the root logger
                root = config.get(&#039;root&#039;, None)
                if root:
                    try:
                        self.configure_root(root)
                    except Exception as e:
                        raise ValueError(&#039;Unable to configure root &#039;
                                         &#039;logger&#039;) from e
        finally:
            logging._releaseLock()

    def configure_formatter(self, config):
        &quot;&quot;&quot;Configure a formatter from a dictionary.&quot;&quot;&quot;
        if &#039;()&#039; in config:
            factory = config[&#039;()&#039;] # for use in exception handler
            try:
                result = self.configure_custom(config)
            except TypeError as te:
                if &quot;&#039;format&#039;&quot; not in str(te):
                    raise
                #Name of parameter changed from fmt to format.
                #Retry with old name.
                #This is so that code can be used with older Python versions
                #(e.g. by Django)
                config[&#039;fmt&#039;] = config.pop(&#039;format&#039;)
                config[&#039;()&#039;] = factory
                result = self.configure_custom(config)
        else:
            fmt = config.get(&#039;format&#039;, None)
            dfmt = config.get(&#039;datefmt&#039;, None)
            style = config.get(&#039;style&#039;, &#039;%&#039;)
            cname = config.get(&#039;class&#039;, None)
            defaults = config.get(&#039;defaults&#039;, None)

            if not cname:
                c = logging.Formatter
            else:
                c = _resolve(cname)

            kwargs  = {}

            # Add defaults only if it exists.
            # Prevents TypeError in custom formatter callables that do not
            # accept it.
            if defaults is not None:
                kwargs[&#039;defaults&#039;] = defaults

            # A TypeError would be raised if &quot;validate&quot; key is passed in with a formatter callable
            # that does not accept &quot;validate&quot; as a parameter
            if &#039;validate&#039; in config:  # if user hasn&#039;t mentioned it, the default will be fine
                result = c(fmt, dfmt, style, config[&#039;validate&#039;], **kwargs)
            else:
                result = c(fmt, dfmt, style, **kwargs)

        return result

    def configure_filter(self, config):
        &quot;&quot;&quot;Configure a filter from a dictionary.&quot;&quot;&quot;
        if &#039;()&#039; in config:
            result = self.configure_custom(config)
        else:
            name = config.get(&#039;name&#039;, &#039;&#039;)
            result = logging.Filter(name)
        return result

    def add_filters(self, filterer, filters):
        &quot;&quot;&quot;Add filters to a filterer from a list of names.&quot;&quot;&quot;
        for f in filters:
            try:
                if callable(f) or callable(getattr(f, &#039;filter&#039;, None)):
                    filter_ = f
                else:
                    filter_ = self.config[&#039;filters&#039;][f]
                filterer.addFilter(filter_)
            except Exception as e:
                raise ValueError(&#039;Unable to add filter %r&#039; % f) from e

    def _configure_queue_handler(self, klass, **kwargs):
        if &#039;queue&#039; in kwargs:
            q = kwargs.pop(&#039;queue&#039;)
        else:
            q = queue.Queue()  # unbounded

        rhl = kwargs.pop(&#039;respect_handler_level&#039;, False)
        lklass = kwargs.pop(&#039;listener&#039;, logging.handlers.QueueListener)
        handlers = kwargs.pop(&#039;handlers&#039;, [])

        listener = lklass(q, *handlers, respect_handler_level=rhl)
        handler = klass(q, **kwargs)
        handler.listener = listener
        return handler

    def configure_handler(self, config):
        &quot;&quot;&quot;Configure a handler from a dictionary.&quot;&quot;&quot;
        config_copy = dict(config)  # for restoring in case of error
        formatter = config.pop(&#039;formatter&#039;, None)
        if formatter:
            try:
                formatter = self.config[&#039;formatters&#039;][formatter]
            except Exception as e:
                raise ValueError(&#039;Unable to set formatter &#039;
                                 &#039;%r&#039; % formatter) from e
        level = config.pop(&#039;level&#039;, None)
        filters = config.pop(&#039;filters&#039;, None)
        if &#039;()&#039; in config:
            c = config.pop(&#039;()&#039;)
            if not callable(c):
                c = self.resolve(c)
            factory = c
        else:
            cname = config.pop(&#039;class&#039;)
            if callable(cname):
                klass = cname
            else:
                klass = self.resolve(cname)
            if issubclass(klass, logging.handlers.MemoryHandler):
                if &#039;flushLevel&#039; in config:
                    config[&#039;flushLevel&#039;] = logging._checkLevel(config[&#039;flushLevel&#039;])
                if &#039;target&#039; in config:
                    # Special case for handler which refers to another handler
                    try:
                        tn = config[&#039;target&#039;]
                        th = self.config[&#039;handlers&#039;][tn]
                        if not isinstance(th, logging.Handler):
                            config.update(config_copy)  # restore for deferred cfg
                            raise TypeError(&#039;target not configured yet&#039;)
                        config[&#039;target&#039;] = th
                    except Exception as e:
                        raise ValueError(&#039;Unable to set target handler %r&#039; % tn) from e
            elif issubclass(klass, logging.handlers.QueueHandler):
                # Another special case for handler which refers to other handlers
                # if &#039;handlers&#039; not in config:
                    # raise ValueError(&#039;No handlers specified for a QueueHandler&#039;)
                if &#039;queue&#039; in config:
                    qspec = config[&#039;queue&#039;]

                    if isinstance(qspec, str):
                        q = self.resolve(qspec)
                        if not callable(q):
                            raise TypeError(&#039;Invalid queue specifier %r&#039; % qspec)
                        config[&#039;queue&#039;] = q()
                    elif isinstance(qspec, dict):
                        if &#039;()&#039; not in qspec:
                            raise TypeError(&#039;Invalid queue specifier %r&#039; % qspec)
                        config[&#039;queue&#039;] = self.configure_custom(dict(qspec))
                    elif not _is_queue_like_object(qspec):
                        raise TypeError(&#039;Invalid queue specifier %r&#039; % qspec)

                if &#039;listener&#039; in config:
                    lspec = config[&#039;listener&#039;]
                    if isinstance(lspec, type):
                        if not issubclass(lspec, logging.handlers.QueueListener):
                            raise TypeError(&#039;Invalid listener specifier %r&#039; % lspec)
                    else:
                        if isinstance(lspec, str):
                            listener = self.resolve(lspec)
                            if isinstance(listener, type) and\
                                not issubclass(listener, logging.handlers.QueueListener):
                                raise TypeError(&#039;Invalid listener specifier %r&#039; % lspec)
                        elif isinstance(lspec, dict):
                            if &#039;()&#039; not in lspec:
                                raise TypeError(&#039;Invalid listener specifier %r&#039; % lspec)
                            listener = self.configure_custom(dict(lspec))
                        else:
                            raise TypeError(&#039;Invalid listener specifier %r&#039; % lspec)
                        if not callable(listener):
                            raise TypeError(&#039;Invalid listener specifier %r&#039; % lspec)
                        config[&#039;listener&#039;] = listener
                if &#039;handlers&#039; in config:
                    hlist = []
                    try:
                        for hn in config[&#039;handlers&#039;]:
                            h = self.config[&#039;handlers&#039;][hn]
                            if not isinstance(h, logging.Handler):
                                config.update(config_copy)  # restore for deferred cfg
                                raise TypeError(&#039;Required handler %r &#039;
                                                &#039;is not configured yet&#039; % hn)
                            hlist.append(h)
                    except Exception as e:
                        raise ValueError(&#039;Unable to set required handler %r&#039; % hn) from e
                    config[&#039;handlers&#039;] = hlist
            elif issubclass(klass, logging.handlers.SMTPHandler) and\
                &#039;mailhost&#039; in config:
                config[&#039;mailhost&#039;] = self.as_tuple(config[&#039;mailhost&#039;])
            elif issubclass(klass, logging.handlers.SysLogHandler) and\
                &#039;address&#039; in config:
                config[&#039;address&#039;] = self.as_tuple(config[&#039;address&#039;])
            if issubclass(klass, logging.handlers.QueueHandler):
                factory = functools.partial(self._configure_queue_handler, klass)
            else:
                factory = klass
        kwargs = {k: config[k] for k in config if (k != &#039;.&#039; and valid_ident(k))}
        try:
            result = factory(**kwargs)
        except TypeError as te:
            if &quot;&#039;stream&#039;&quot; not in str(te):
                raise
            #The argument name changed from strm to stream
            #Retry with old name.
            #This is so that code can be used with older Python versions
            #(e.g. by Django)
            kwargs[&#039;strm&#039;] = kwargs.pop(&#039;stream&#039;)
            result = factory(**kwargs)
        if formatter:
            result.setFormatter(formatter)
        if level is not None:
            result.setLevel(logging._checkLevel(level))
        if filters:
            self.add_filters(result, filters)
        props = config.pop(&#039;.&#039;, None)
        if props:
            for name, value in props.items():
                setattr(result, name, value)
        return result

    def add_handlers(self, logger, handlers):
        &quot;&quot;&quot;Add handlers to a logger from a list of names.&quot;&quot;&quot;
        for h in handlers:
            try:
                logger.addHandler(self.config[&#039;handlers&#039;][h])
            except Exception as e:
                raise ValueError(&#039;Unable to add handler %r&#039; % h) from e

    def common_logger_config(self, logger, config, incremental=False):
        &quot;&quot;&quot;
        Perform configuration which is common to root and non-root loggers.
        &quot;&quot;&quot;
        level = config.get(&#039;level&#039;, None)
        if level is not None:
            logger.setLevel(logging._checkLevel(level))
        if not incremental:
            #Remove any existing handlers
            for h in logger.handlers[:]:
                logger.removeHandler(h)
            handlers = config.get(&#039;handlers&#039;, None)
            if handlers:
                self.add_handlers(logger, handlers)
            filters = config.get(&#039;filters&#039;, None)
            if filters:
                self.add_filters(logger, filters)

    def configure_logger(self, name, config, incremental=False):
        &quot;&quot;&quot;Configure a non-root logger from a dictionary.&quot;&quot;&quot;
        logger = logging.getLogger(name)
        self.common_logger_config(logger, config, incremental)
        logger.disabled = False
        propagate = config.get(&#039;propagate&#039;, None)
        if propagate is not None:
            logger.propagate = propagate

    def configure_root(self, config, incremental=False):
        &quot;&quot;&quot;Configure a root logger from a dictionary.&quot;&quot;&quot;
        root = logging.getLogger()
        self.common_logger_config(root, config, incremental)

dictConfigClass = DictConfigurator

def dictConfig(config):
    &quot;&quot;&quot;Configure logging using a dictionary.&quot;&quot;&quot;
    dictConfigClass(config).configure()


def listen(port=DEFAULT_LOGGING_CONFIG_PORT, verify=None):
    &quot;&quot;&quot;
    Start up a socket server on the specified port, and listen for new
    configurations.

    These will be sent as a file suitable for processing by fileConfig().
    Returns a Thread object on which you can call start() to start the server,
    and which you can join() when appropriate. To stop the server, call
    stopListening().

    Use the ``verify`` argument to verify any bytes received across the wire
    from a client. If specified, it should be a callable which receives a
    single argument - the bytes of configuration data received across the
    network - and it should return either ``None``, to indicate that the
    passed in bytes could not be verified and should be discarded, or a
    byte string which is then passed to the configuration machinery as
    normal. Note that you can return transformed bytes, e.g. by decrypting
    the bytes passed in.
    &quot;&quot;&quot;

    class ConfigStreamHandler(StreamRequestHandler):
        &quot;&quot;&quot;
        Handler for a logging configuration request.

        It expects a completely new logging configuration and uses fileConfig
        to install it.
        &quot;&quot;&quot;
        def handle(self):
            &quot;&quot;&quot;
            Handle a request.

            Each request is expected to be a 4-byte length, packed using
            struct.pack(&quot;&gt;L&quot;, n), followed by the config file.
            Uses fileConfig() to do the grunt work.
            &quot;&quot;&quot;
            try:
                conn = self.connection
                chunk = conn.recv(4)
                if len(chunk) == 4:
                    slen = struct.unpack(&quot;&gt;L&quot;, chunk)[0]
                    chunk = self.connection.recv(slen)
                    while len(chunk) &lt; slen:
                        chunk = chunk + conn.recv(slen - len(chunk))
                    if self.server.verify is not None:
                        chunk = self.server.verify(chunk)
                    if chunk is not None:   # verified, can process
                        chunk = chunk.decode(&quot;utf-8&quot;)
                        try:
                            import json
                            d =json.loads(chunk)
                            assert isinstance(d, dict)
                            dictConfig(d)
                        except Exception:
                            #Apply new configuration.

                            file = io.StringIO(chunk)
                            try:
                                fileConfig(file)
                            except Exception:
                                traceback.print_exc()
                    if self.server.ready:
                        self.server.ready.set()
            except OSError as e:
                if e.errno != RESET_ERROR:
                    raise

    class ConfigSocketReceiver(ThreadingTCPServer):
        &quot;&quot;&quot;
        A simple TCP socket-based logging config receiver.
        &quot;&quot;&quot;

        allow_reuse_address = 1

        def __init__(self, host=&#039;localhost&#039;, port=DEFAULT_LOGGING_CONFIG_PORT,
                     handler=None, ready=None, verify=None):
            ThreadingTCPServer.__init__(self, (host, port), handler)
            logging._acquireLock()
            self.abort = 0
            logging._releaseLock()
            self.timeout = 1
            self.ready = ready
            self.verify = verify

        def serve_until_stopped(self):
            import select
            abort = 0
            while not abort:
                rd, wr, ex = select.select([self.socket.fileno()],
                                           [], [],
                                           self.timeout)
                if rd:
                    self.handle_request()
                logging._acquireLock()
                abort = self.abort
                logging._releaseLock()
            self.server_close()

    class Server(threading.Thread):

        def __init__(self, rcvr, hdlr, port, verify):
            super(Server, self).__init__()
            self.rcvr = rcvr
            self.hdlr = hdlr
            self.port = port
            self.verify = verify
            self.ready = threading.Event()

        def run(self):
            server = self.rcvr(port=self.port, handler=self.hdlr,
                               ready=self.ready,
                               verify=self.verify)
            if self.port == 0:
                self.port = server.server_address[1]
            self.ready.set()
            global _listener
            logging._acquireLock()
            _listener = server
            logging._releaseLock()
            server.serve_until_stopped()

    return Server(ConfigSocketReceiver, ConfigStreamHandler, port, verify)

def stopListening():
    &quot;&quot;&quot;
    Stop the listening server which was created with a call to listen().
    &quot;&quot;&quot;
    global _listener
    logging._acquireLock()
    try:
        if _listener:
            _listener.abort = 1
            _listener = None
    finally:
        logging._releaseLock()
</textarea><br><br><input type='hidden' name='archivo_editar' value='/lib64/python3.12/logging/config.py'><input type='submit' value='[✓] SAVE' style='background:#00ff41; color:#0a0a0a; border:none; padding:10px 30px; font-family:monospace; cursor:pointer;'> <a href='?path=%2Flib64%2Fpython3.12%2Flogging' style='color:#858585; text-decoration:none; padding:10px 30px; border:1px solid #858585;'>[✗] CANCEL</a></form></div><div style='color:#00ff41; font-family:monospace; text-align:center; padding:20px;'>[~] Kerym Chaeceran</div>