:gem: [Feature] New logging system

conversation_creater.py

@@ -6,6 +6,7 @@ import pprint
 import urllib
 
 from chathub_request_constructor import ChathubRequestConstructor
+from logger.logger import logger
 
 
 http_proxy = "http://localhost:11111"  # Replace with yours
@@ -39,7 +40,7 @@ def serialize_websocket_message(msg: dict) -> str:
     return json.dumps(msg, ensure_ascii=False) + "\x1e"
 
 
-class ConversationChatter:
+class ConversationConnector:
     def __init__(
         self,
         sec_access_token=None,
@@ -129,23 +130,20 @@ class ConversationChatter:
                                 delta_content_pointer = len(message_html)
 
                                 if message.get("suggestedResponses"):
-                                    print("\nSuggested Questions: ", flush=True)
+                                    logger.note("\nSuggested Questions: ")
                                     for suggestion in message.get("suggestedResponses"):
                                         suggestion_text = suggestion.get("text")
-                                        print(f"- {suggestion_text}", flush=True)
+                                        logger.mesg(f"- {suggestion_text}")
 
                             elif message_type in ["InternalSearchQuery"]:
                                 message_hidden_text = message["hiddenText"]
-                                print(
-                                    f"\n[Searching: [{message_hidden_text}]]",
-                                    flush=True,
-                                )
+                                logger.note(f"\n[Searching: [{message_hidden_text}]]")
                             elif message_type in [
                                 "InternalSearchResult",
                             ]:
-                                print("[Analyzing search results ...]", flush=True)
+                                logger.note("[Analyzing search results ...]")
                             elif message_type in ["InternalLoaderMessage"]:
-                                print("[Generating answers ...]\n", flush=True)
+                                logger.note("[Generating answers ...]\n")
                             elif message_type in ["RenderCardRequest"]:
                                 continue
                             else:
@@ -156,12 +154,12 @@ class ConversationChatter:
                 elif data.get("type") == 2:
                     if data.get("item"):
                         item = data.get("item")
-                        print("\n[Saving chat messages ...]")
+                        logger.note("\n[Saving chat messages ...]")
                         # for message in item.get("messages"):
                         #     author = message["author"]
                         #     message_text = message["text"]
                 elif data.get("type") == 3:
-                    print("[Finished]")
+                    logger.success("[Finished]")
                     await wss.close()
                     break
                 elif data.get("type") == 6:
@@ -175,7 +173,7 @@ if __name__ == "__main__":
     creator = ConversationCreator()
     creator.create()
 
-    chatter = ConversationChatter(
+    conversation_connector = ConversationConnector(
         sec_access_token=creator.response_headers[
             "x-sydney-encryptedconversationsignature"
         ],
@@ -183,8 +181,8 @@ if __name__ == "__main__":
         conversation_id=creator.response_content["conversationId"],
     )
     prompt = "Today's weather of California"
-    print(f"\n[User]: {prompt}\n")
-    print(f"[Bing]:")
+    logger.info(f"\n[User]: {prompt}\n")
+    logger.info(f"[Bing]:")
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(chatter.stream_chat(prompt=prompt))
+    loop.run_until_complete(conversation_connector.stream_chat(prompt=prompt))
     loop.close()

logger/logger.py

@@ -0,0 +1,241 @@
+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 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__
+
+        self.logger = logging.getLogger(name)
+        self.logger.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")
+
+        self.handler = logging.StreamHandler()
+        self.handler.setLevel(logging.INFO)
+
+        self.handler.setFormatter(self.formatter)
+        self.logger.addHandler(self.handler)
+
+        self.log_indent = 0
+        self.log_indents = []
+
+        self.log_level = "info"
+        self.log_levels = []
+
+        self.bind_functions()
+
+    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.logger.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, method, msg, indent=0, fill=False, fill_side="both", *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]
+        msg_lines = msg_str.splitlines()
+
+        self.store_indent()
+        self.indent(indent)
+        indent_str = " " * self.log_indent
+        self.restore_indent()
+
+        level, color = self.LOG_METHODS[method]
+        indented_msg = "\n".join([f"{indent_str}{line}" for line in msg_lines])
+        if fill:
+            indented_msg = add_fillers(indented_msg, fill_side=fill_side)
+        getattr(self.logger, level)(colored(indented_msg, color), *args, **kwargs)
+
+    def bind_functions(self):
+        for method in self.LOG_METHODS:
+            setattr(self.logger, method, functools.partial(self.log, method))
+
+        for method in self.INDENT_METHODS:
+            setattr(self.logger, method, getattr(self, method))
+
+        for method in self.LEVEL_METHODS:
+            setattr(self.logger, method, getattr(self, method))
+
+
+logger = Logger().logger
+
+
+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)