File size: 4,109 Bytes
db4810d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
import logging
import os
import sys
from dotenv import load_dotenv
load_dotenv()
def addLoggingLevel(levelName, levelNum, methodName=None):
"""
Comprehensively adds a new logging level to the `logging` module and the
currently configured logging class.
`levelName` becomes an attribute of the `logging` module with the value
`levelNum`. `methodName` becomes a convenience method for both `logging`
itself and the class returned by `logging.getLoggerClass()` (usually just
`logging.Logger`). If `methodName` is not specified, `levelName.lower()` is
used.
To avoid accidental clobberings of existing attributes, this method will
raise an `AttributeError` if the level name is already an attribute of the
`logging` module or if the method name is already present
Example
-------
>>> addLoggingLevel('TRACE', logging.DEBUG - 5)
>>> logging.getLogger(__name__).setLevel('TRACE')
>>> logging.getLogger(__name__).trace('that worked')
>>> logging.trace('so did this')
>>> logging.TRACE
5
"""
if not methodName:
methodName = levelName.lower()
if hasattr(logging, levelName):
raise AttributeError('{} already defined in logging module'.format(levelName))
if hasattr(logging, methodName):
raise AttributeError('{} already defined in logging module'.format(methodName))
if hasattr(logging.getLoggerClass(), methodName):
raise AttributeError('{} already defined in logger class'.format(methodName))
# This method was inspired by the answers to Stack Overflow post
# http://stackoverflow.com/q/2183233/2988730, especially
# http://stackoverflow.com/a/13638084/2988730
def logForLevel(self, message, *args, **kwargs):
if self.isEnabledFor(levelNum):
self._log(levelNum, message, args, **kwargs)
def logToRoot(message, *args, **kwargs):
logging.log(levelNum, message, *args, **kwargs)
logging.addLevelName(levelNum, levelName)
setattr(logging, levelName, levelNum)
setattr(logging.getLoggerClass(), methodName, logForLevel)
setattr(logging, methodName, logToRoot)
def setup_logging():
# Try to add RESULT level, but ignore if it already exists
try:
addLoggingLevel('RESULT', 35) # This allows ERROR, FATAL and CRITICAL
except AttributeError:
pass # Level already exists, which is fine
log_type = os.getenv('BROWSER_USE_LOGGING_LEVEL', 'info').lower()
# Check if handlers are already set up
if logging.getLogger().hasHandlers():
return
# Clear existing handlers
root = logging.getLogger()
root.handlers = []
class BrowserUseFormatter(logging.Formatter):
def format(self, record):
if type(record.name) == str and record.name.startswith('browser_use.'):
record.name = record.name.split('.')[-2]
return super().format(record)
# Setup single handler for all loggers
console = logging.StreamHandler(sys.stdout)
# adittional setLevel here to filter logs
if log_type == 'result':
console.setLevel('RESULT')
console.setFormatter(BrowserUseFormatter('%(message)s'))
else:
console.setFormatter(BrowserUseFormatter('%(levelname)-8s [%(name)s] %(message)s'))
# Configure root logger only
root.addHandler(console)
# switch cases for log_type
if log_type == 'result':
root.setLevel('RESULT') # string usage to avoid syntax error
elif log_type == 'debug':
root.setLevel(logging.DEBUG)
else:
root.setLevel(logging.INFO)
# Configure browser_use logger
browser_use_logger = logging.getLogger('browser_use')
browser_use_logger.propagate = False # Don't propagate to root logger
browser_use_logger.addHandler(console)
browser_use_logger.setLevel(root.level) # Set same level as root logger
logger = logging.getLogger('browser_use')
logger.info('BrowserUse logging setup complete with level %s', log_type)
# Silence third-party loggers
for logger in [
'WDM',
'httpx',
'selenium',
'playwright',
'urllib3',
'asyncio',
'langchain',
'openai',
'httpcore',
'charset_normalizer',
'anthropic._base_client',
'PIL.PngImagePlugin',
'trafilatura.htmlprocessing',
'trafilatura',
]:
third_party = logging.getLogger(logger)
third_party.setLevel(logging.ERROR)
third_party.propagate = False
|