Kate-03/GenAI-Arena

1"""2Common utilities.3"""4from asyncio import AbstractEventLoop5import json6import logging7import logging.handlers8import os9import platform10import sys11from typing import AsyncGenerator, Generator12import warnings13from pathlib import Path14 15import requests16 17from .constants import LOGDIR, LOG_SERVER_ADDR, SAVE_LOG18from .utils import save_log_str_on_log_server19 20 21handler = None22visited_loggers = set()23 24 25# Assuming LOGDIR and other necessary imports and global variables are defined26 27class APIHandler(logging.Handler):28    """Custom logging handler that sends logs to an API."""29 30    def __init__(self, apiUrl, log_path, *args, **kwargs):31        super(APIHandler, self).__init__(*args, **kwargs)32        self.apiUrl = apiUrl33        self.log_path = log_path34 35    def emit(self, record):36        log_entry = self.format(record)37        try:38            save_log_str_on_log_server(log_entry, self.log_path)39        except requests.RequestException as e:40            print(f"Error sending log to API: {e}", file=sys.stderr)41 42def build_logger(logger_name, logger_filename, add_remote_handler=False):43    global handler44 45    formatter = logging.Formatter(46        fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s",47        datefmt="%Y-%m-%d %H:%M:%S",48    )49 50    # Set the format of root handlers51    if not logging.getLogger().handlers:52        if sys.version_info[1] >= 9:53            # This is for windows54            logging.basicConfig(level=logging.INFO, encoding="utf-8")55        else:56            if platform.system() == "Windows":57                warnings.warn(58                    "If you are running on Windows, "59                    "we recommend you use Python >= 3.9 for UTF-8 encoding."60                )61            logging.basicConfig(level=logging.INFO)62    logging.getLogger().handlers[0].setFormatter(formatter)63 64    # Redirect stdout and stderr to loggers65    stdout_logger = logging.getLogger("stdout")66    stdout_logger.setLevel(logging.INFO)67    sl = StreamToLogger(stdout_logger, logging.INFO)68    sys.stdout = sl69 70    stderr_logger = logging.getLogger("stderr")71    stderr_logger.setLevel(logging.ERROR)72    sl = StreamToLogger(stderr_logger, logging.ERROR)73    sys.stderr = sl74 75    # Get logger76    logger = logging.getLogger(logger_name)77    logger.setLevel(logging.INFO)78    79    if add_remote_handler:80        # Add APIHandler to send logs to your API81        api_url = f"{LOG_SERVER_ADDR}/{SAVE_LOG}"82        83        remote_logger_filename = str(Path(logger_filename).stem + "_remote.log")84        api_handler = APIHandler(apiUrl=api_url, log_path=f"{LOGDIR}/{remote_logger_filename}")85        api_handler.setFormatter(formatter)86        logger.addHandler(api_handler)87        88        stdout_logger.addHandler(api_handler)89        stderr_logger.addHandler(api_handler)90 91    # if LOGDIR is empty, then don't try output log to local file92    if LOGDIR != "":93        os.makedirs(LOGDIR, exist_ok=True)94        filename = os.path.join(LOGDIR, logger_filename)95        handler = logging.handlers.TimedRotatingFileHandler(96            filename, when="D", utc=True, encoding="utf-8"97        )98        handler.setFormatter(formatter)99 100        for l in [stdout_logger, stderr_logger, logger]:101            if l in visited_loggers:102                continue103            visited_loggers.add(l)104            l.addHandler(handler)105 106    return logger107 108 109class StreamToLogger(object):110    """111    Fake file-like stream object that redirects writes to a logger instance.112    """113 114    def __init__(self, logger, log_level=logging.INFO):115        self.terminal = sys.stdout116        self.logger = logger117        self.log_level = log_level118        self.linebuf = ""119 120    def __getattr__(self, attr):121        return getattr(self.terminal, attr)122 123    def write(self, buf):124        temp_linebuf = self.linebuf + buf125        self.linebuf = ""126        for line in temp_linebuf.splitlines(True):127            # From the io.TextIOWrapper docs:128            #   On output, if newline is None, any '\n' characters written129            #   are translated to the system default line separator.130            # By default sys.stdout.write() expects '\n' newlines and then131            # translates them so this is still cross platform.132            if line[-1] == "\n":133                encoded_message = line.encode("utf-8", "ignore").decode("utf-8")134                self.logger.log(self.log_level, encoded_message.rstrip())135            else:136                self.linebuf += line137 138    def flush(self):139        if self.linebuf != "":140            encoded_message = self.linebuf.encode("utf-8", "ignore").decode("utf-8")141            self.logger.log(self.log_level, encoded_message.rstrip())142        self.linebuf = ""