:recycle: [Refactor] Use tclogger, and configure envs better

apis/chat_api.py

@@ -16,7 +16,7 @@ from sse_starlette.sse import EventSourceResponse, ServerSentEvent
 from messagers.message_composer import MessageComposer
 from mocks.stream_chat_mocker import stream_chat_mock
 from networks.message_streamer import MessageStreamer
-from utils.logger import logger
+from tclogger import logger
 from constants.models import AVAILABLE_MODELS_DICTS
 
 

constants/env.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+from tclogger import logger, OSEnver
+
+secrets_path = Path(__file__).parents[1] / "secrets.json"
+ENVER = OSEnver(secrets_path)
+
+http_proxy = ENVER["http_proxy"]
+if http_proxy:
+    logger.note(f"> Using proxy: {http_proxy}")
+    PROXIES = {
+        "http": http_proxy,
+        "https": http_proxy,
+    }
+else:
+    PROXIES = None

messagers/message_composer.py

@@ -4,7 +4,7 @@ from pprint import pprint
 from transformers import AutoTokenizer
 
 from constants.models import AVAILABLE_MODELS, MODEL_MAP
-from utils.logger import logger
+from tclogger import logger
 
 
 class MessageComposer:

mocks/stream_chat_mocker.py

@@ -1,5 +1,5 @@
 import time
-from utils.logger import logger
+from tclogger import logger
 
 
 def stream_chat_mock(*args, **kwargs):

networks/message_streamer.py

@@ -2,6 +2,7 @@ import json
 import re
 import requests
 
+from tclogger import logger
 from tiktoken import get_encoding as tiktoken_get_encoding
 from transformers import AutoTokenizer
 
@@ -11,13 +12,11 @@ from constants.models import (
     TOKEN_LIMIT_MAP,
     TOKEN_RESERVED,
 )
+from constants.env import PROXIES
 from messagers.message_outputer import OpenaiStreamOutputer
-from utils.logger import logger
-from utils.enver import enver
 
 
 class MessageStreamer:
-
     def __init__(self, model: str):
         if model in MODEL_MAP.keys():
             self.model = model
@@ -120,12 +119,11 @@ class MessageStreamer:
         #     ]
 
         logger.back(self.request_url)
-        enver.set_envs(proxies=True)
         stream_response = requests.post(
             self.request_url,
             headers=self.request_headers,
             json=self.request_body,
-            proxies=enver.requests_proxies,
+            proxies=PROXIES,
             stream=True,
         )
         status_code = stream_response.status_code

utils/__init__.py

@@ -1,69 +0,0 @@
-import json
-import requests
-import os
-
-from pathlib import Path
-
-
-class OSEnver:
-    def __init__(self):
-        self.envs_stack = []
-        self.envs = os.environ.copy()
-
-    def store_envs(self):
-        self.envs_stack.append(self.envs)
-
-    def restore_envs(self):
-        self.envs = self.envs_stack.pop()
-        if self.global_scope:
-            os.environ = self.envs
-
-    def set_envs(self, secrets=True, proxies=None, store_envs=True):
-        # caller_info = inspect.stack()[1]
-        # logger.back(f"OS Envs is set by: {caller_info.filename}")
-
-        if store_envs:
-            self.store_envs()
-
-        if secrets:
-            secrets_path = Path(__file__).parents[1] / "secrets.json"
-            if secrets_path.exists():
-                with open(secrets_path, "r") as rf:
-                    secrets = json.load(rf)
-            else:
-                secrets = {}
-
-        if proxies:
-            for proxy_env in ["http_proxy", "https_proxy"]:
-                if isinstance(proxies, str):
-                    self.envs[proxy_env] = proxies
-                elif "http_proxy" in secrets.keys():
-                    self.envs[proxy_env] = secrets["http_proxy"]
-                elif os.getenv("http_proxy"):
-                    self.envs[proxy_env] = os.getenv("http_proxy")
-                else:
-                    continue
-
-        self.proxy = (
-            self.envs.get("all_proxy")
-            or self.envs.get("http_proxy")
-            or self.envs.get("https_proxy")
-            or None
-        )
-        self.requests_proxies = {
-            "http": self.proxy,
-            "https": self.proxy,
-        }
-
-        # https://www.proxynova.com/proxy-server-list/country-us/
-
-        print(f"Using proxy: [{self.proxy}]")
-        # r = requests.get(
-        #     "http://ifconfig.me/ip",
-        #     proxies=self.requests_proxies,
-        #     timeout=10,
-        # )
-        # print(f"[r.status_code] r.text")
-
-
-enver = OSEnver()

utils/enver.py

@@ -1,60 +0,0 @@
-import json
-import os
-
-from pathlib import Path
-from utils.logger import logger
-
-
-class OSEnver:
-    def __init__(self):
-        self.envs_stack = []
-        self.envs = os.environ.copy()
-
-    def store_envs(self):
-        self.envs_stack.append(self.envs)
-
-    def restore_envs(self):
-        self.envs = self.envs_stack.pop()
-
-    def set_envs(self, secrets=True, proxies=None, store_envs=True):
-        # caller_info = inspect.stack()[1]
-        # logger.back(f"OS Envs is set by: {caller_info.filename}")
-
-        if store_envs:
-            self.store_envs()
-
-        if secrets:
-            secrets_path = Path(__file__).parents[1] / "secrets.json"
-            if secrets_path.exists():
-                with open(secrets_path, "r") as rf:
-                    secrets = json.load(rf)
-            else:
-                secrets = {}
-
-        if proxies:
-            for proxy_env in ["http_proxy", "https_proxy"]:
-                if isinstance(proxies, str):
-                    self.envs[proxy_env] = proxies
-                elif "http_proxy" in secrets.keys():
-                    self.envs[proxy_env] = secrets["http_proxy"]
-                elif os.getenv("http_proxy"):
-                    self.envs[proxy_env] = os.getenv("http_proxy")
-                else:
-                    continue
-
-        self.proxy = (
-            self.envs.get("all_proxy")
-            or self.envs.get("http_proxy")
-            or self.envs.get("https_proxy")
-            or None
-        )
-        self.requests_proxies = {
-            "http": self.proxy,
-            "https": self.proxy,
-        }
-
-        if self.proxy:
-            logger.note(f"Using proxy: [{self.proxy}]")
-
-
-enver = OSEnver()

utils/logger.py

@@ -1,269 +0,0 @@
-import datetime
-import functools
-import inspect
-import logging
-import os
-import shutil
-import subprocess
-from termcolor import colored
-
-
-def add_fillers(text, filler="=", fill_side="both"):
-    terminal_width = shutil.get_terminal_size().columns
-    text = text.strip()
-    text_width = len(text)
-    if text_width >= terminal_width:
-        return text
-
-    if fill_side[0].lower() == "b":
-        leading_fill_str = filler * ((terminal_width - text_width) // 2 - 1) + " "
-        trailing_fill_str = " " + filler * (
-            terminal_width - text_width - len(leading_fill_str) - 1
-        )
-    elif fill_side[0].lower() == "l":
-        leading_fill_str = filler * (terminal_width - text_width - 1) + " "
-        trailing_fill_str = ""
-    elif fill_side[0].lower() == "r":
-        leading_fill_str = ""
-        trailing_fill_str = " " + filler * (terminal_width - text_width - 1)
-    else:
-        raise ValueError("Invalid fill_side")
-
-    filled_str = f"{leading_fill_str}{text}{trailing_fill_str}"
-    return filled_str
-
-
-class OSLogger(logging.Logger):
-    LOG_METHODS = {
-        "err": ("error", "red"),
-        "warn": ("warning", "light_red"),
-        "note": ("info", "light_magenta"),
-        "mesg": ("info", "light_cyan"),
-        "file": ("info", "light_blue"),
-        "line": ("info", "white"),
-        "success": ("info", "light_green"),
-        "fail": ("info", "light_red"),
-        "back": ("debug", "light_cyan"),
-    }
-    INDENT_METHODS = [
-        "indent",
-        "set_indent",
-        "reset_indent",
-        "store_indent",
-        "restore_indent",
-        "log_indent",
-    ]
-    LEVEL_METHODS = [
-        "set_level",
-        "store_level",
-        "restore_level",
-        "quiet",
-        "enter_quiet",
-        "exit_quiet",
-    ]
-    LEVEL_NAMES = {
-        "critical": logging.CRITICAL,
-        "error": logging.ERROR,
-        "warning": logging.WARNING,
-        "info": logging.INFO,
-        "debug": logging.DEBUG,
-    }
-
-    def __init__(self, name=None, prefix=False):
-        if not name:
-            frame = inspect.stack()[1]
-            module = inspect.getmodule(frame[0])
-            name = module.__name__
-
-        super().__init__(name)
-        self.setLevel(logging.INFO)
-
-        if prefix:
-            formatter_prefix = "[%(asctime)s] - [%(name)s] - [%(levelname)s]\n"
-        else:
-            formatter_prefix = ""
-
-        self.formatter = logging.Formatter(formatter_prefix + "%(message)s")
-
-        stream_handler = logging.StreamHandler()
-        stream_handler.setLevel(logging.INFO)
-        stream_handler.setFormatter(self.formatter)
-        self.addHandler(stream_handler)
-
-        self.log_indent = 0
-        self.log_indents = []
-
-        self.log_level = "info"
-        self.log_levels = []
-
-    def indent(self, indent=2):
-        self.log_indent += indent
-
-    def set_indent(self, indent=2):
-        self.log_indent = indent
-
-    def reset_indent(self):
-        self.log_indent = 0
-
-    def store_indent(self):
-        self.log_indents.append(self.log_indent)
-
-    def restore_indent(self):
-        self.log_indent = self.log_indents.pop(-1)
-
-    def set_level(self, level):
-        self.log_level = level
-        self.setLevel(self.LEVEL_NAMES[level])
-
-    def store_level(self):
-        self.log_levels.append(self.log_level)
-
-    def restore_level(self):
-        self.log_level = self.log_levels.pop(-1)
-        self.set_level(self.log_level)
-
-    def quiet(self):
-        self.set_level("critical")
-
-    def enter_quiet(self, quiet=False):
-        if quiet:
-            self.store_level()
-            self.quiet()
-
-    def exit_quiet(self, quiet=False):
-        if quiet:
-            self.restore_level()
-
-    def log(
-        self,
-        level,
-        color,
-        msg,
-        indent=0,
-        fill=False,
-        fill_side="both",
-        end="\n",
-        *args,
-        **kwargs,
-    ):
-        if type(msg) == str:
-            msg_str = msg
-        else:
-            msg_str = repr(msg)
-            quotes = ["'", '"']
-            if msg_str[0] in quotes and msg_str[-1] in quotes:
-                msg_str = msg_str[1:-1]
-
-        indent_str = " " * (self.log_indent + indent)
-        indented_msg = "\n".join([indent_str + line for line in msg_str.split("\n")])
-
-        if fill:
-            indented_msg = add_fillers(indented_msg, fill_side=fill_side)
-
-        handler = self.handlers[0]
-        handler.terminator = end
-
-        getattr(self, level)(colored(indented_msg, color), *args, **kwargs)
-
-    def route_log(self, method, msg, *args, **kwargs):
-        level, method = method
-        functools.partial(self.log, level, method, msg)(*args, **kwargs)
-
-    def err(self, msg: str = "", *args, **kwargs):
-        self.route_log(("error", "red"), msg, *args, **kwargs)
-
-    def warn(self, msg: str = "", *args, **kwargs):
-        self.route_log(("warning", "light_red"), msg, *args, **kwargs)
-
-    def note(self, msg: str = "", *args, **kwargs):
-        self.route_log(("info", "light_magenta"), msg, *args, **kwargs)
-
-    def mesg(self, msg: str = "", *args, **kwargs):
-        self.route_log(("info", "light_cyan"), msg, *args, **kwargs)
-
-    def file(self, msg: str = "", *args, **kwargs):
-        self.route_log(("info", "light_blue"), msg, *args, **kwargs)
-
-    def line(self, msg: str = "", *args, **kwargs):
-        self.route_log(("info", "white"), msg, *args, **kwargs)
-
-    def success(self, msg: str = "", *args, **kwargs):
-        self.route_log(("info", "light_green"), msg, *args, **kwargs)
-
-    def fail(self, msg: str = "", *args, **kwargs):
-        self.route_log(("info", "light_red"), msg, *args, **kwargs)
-
-    def back(self, msg: str = "", *args, **kwargs):
-        self.route_log(("debug", "light_cyan"), msg, *args, **kwargs)
-
-
-logger = OSLogger()
-
-
-def shell_cmd(cmd, getoutput=False, showcmd=True, env=None):
-    if showcmd:
-        logger.info(colored(f"\n$ [{os.getcwd()}]", "light_blue"))
-        logger.info(colored(f"  $ {cmd}\n", "light_cyan"))
-    if getoutput:
-        output = subprocess.getoutput(cmd, env=env)
-        return output
-    else:
-        subprocess.run(cmd, shell=True, env=env)
-
-
-class Runtimer:
-    def __enter__(self):
-        self.t1, _ = self.start_time()
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.t2, _ = self.end_time()
-        self.elapsed_time(self.t2 - self.t1)
-
-    def start_time(self):
-        t1 = datetime.datetime.now()
-        self.logger_time("start", t1)
-        return t1, self.time2str(t1)
-
-    def end_time(self):
-        t2 = datetime.datetime.now()
-        self.logger_time("end", t2)
-        return t2, self.time2str(t2)
-
-    def elapsed_time(self, dt=None):
-        if dt is None:
-            dt = self.t2 - self.t1
-        self.logger_time("elapsed", dt)
-        return dt, self.time2str(dt)
-
-    def logger_time(self, time_type, t):
-        time_types = {
-            "start": "Start",
-            "end": "End",
-            "elapsed": "Elapsed",
-        }
-        time_str = add_fillers(
-            colored(
-                f"{time_types[time_type]} time: [ {self.time2str(t)} ]",
-                "light_magenta",
-            ),
-            fill_side="both",
-        )
-        logger.line(time_str)
-
-    # Convert time to string
-    def time2str(self, t):
-        datetime_str_format = "%Y-%m-%d %H:%M:%S"
-        if isinstance(t, datetime.datetime):
-            return t.strftime(datetime_str_format)
-        elif isinstance(t, datetime.timedelta):
-            hours = t.seconds // 3600
-            hour_str = f"{hours} hr" if hours > 0 else ""
-            minutes = (t.seconds // 60) % 60
-            minute_str = f"{minutes:>2} min" if minutes > 0 else ""
-            seconds = t.seconds % 60
-            second_str = f"{seconds:>2} s"
-            time_str = " ".join([hour_str, minute_str, second_str]).strip()
-            return time_str
-        else:
-            return str(t)