:recycle: [Refactor] Separate MessageParser for data type 1

conversation_connector.py

@@ -7,6 +7,7 @@ import urllib
 
 from conversation_creator import ConversationCreator
 from chathub_request_constructor import ChathubRequestConstructor
+from message_parser import MessageParser
 from logger.logger import logger
 
 
@@ -86,7 +87,7 @@ class ConversationConnector:
         await self.init_wss_connection()
         await self.send_chathub_request(prompt)
 
-        delta_content_pointer = 0
+        message_parser = MessageParser()
         while not self.wss.closed:
             response_lines_str = await self.wss.receive_str()
 
@@ -102,43 +103,7 @@ class ConversationConnector:
 
                 # Stream: Meaningful Messages
                 if data.get("type") == 1:
-                    arguments = data["arguments"][0]
-                    if arguments.get("throttling"):
-                        throttling = arguments.get("throttling")
-                        # pprint.pprint(throttling)
-                    if arguments.get("messages"):
-                        for message in arguments.get("messages"):
-                            message_type = message.get("messageType")
-                            # Message: Displayed answer
-                            if message_type is None:
-                                content = message["adaptiveCards"][0]["body"][0]["text"]
-                                delta_content = content[delta_content_pointer:]
-                                logger.line(delta_content, end="")
-                                delta_content_pointer = len(content)
-                                # Message: Suggested Questions
-                                if message.get("suggestedResponses"):
-                                    logger.note("\n\nSuggested Questions: ")
-                                    for suggestion in message.get("suggestedResponses"):
-                                        suggestion_text = suggestion.get("text")
-                                        logger.file(f"- {suggestion_text}")
-                            # Message: Search Query
-                            elif message_type in ["InternalSearchQuery"]:
-                                message_hidden_text = message["hiddenText"]
-                                logger.note(f"\n[Searching: [{message_hidden_text}]]")
-                            # Message: Internal Search Results
-                            elif message_type in ["InternalSearchResult"]:
-                                logger.note("[Analyzing search results ...]")
-                            # Message: Loader status, such as "Generating Answers"
-                            elif message_type in ["InternalLoaderMessage"]:
-                                logger.note("[Generating answers ...]\n")
-                            # Message: Render Cards for Webpages
-                            elif message_type in ["RenderCardRequest"]:
-                                continue
-                            # Message: Not Implemented
-                            else:
-                                raise NotImplementedError(
-                                    f"Not Supported Message Type: {message_type}"
-                                )
+                    message_parser.parse(data)
                 # Stream: List of whole conversation messages
                 elif data.get("type") == 2:
                     if data.get("item"):

message_parser.py

@@ -0,0 +1,47 @@
+from logger.logger import logger
+
+
+class MessageParser:
+    def __init__(self) -> None:
+        self.delta_content_pointer = 0
+
+    def parse(self, data):
+        arguments = data["arguments"][0]
+
+        if arguments.get("throttling"):
+            throttling = arguments.get("throttling")
+            # pprint.pprint(throttling)
+
+        if arguments.get("messages"):
+            for message in arguments.get("messages"):
+                message_type = message.get("messageType")
+                # Message: Displayed answer
+                if message_type is None:
+                    content = message["adaptiveCards"][0]["body"][0]["text"]
+                    delta_content = content[self.delta_content_pointer :]
+                    logger.line(delta_content, end="")
+                    self.delta_content_pointer = len(content)
+                    # Message: Suggested Questions
+                    if message.get("suggestedResponses"):
+                        logger.note("\n\nSuggested Questions: ")
+                        for suggestion in message.get("suggestedResponses"):
+                            suggestion_text = suggestion.get("text")
+                            logger.file(f"- {suggestion_text}")
+                # Message: Search Query
+                elif message_type in ["InternalSearchQuery"]:
+                    message_hidden_text = message["hiddenText"]
+                    logger.note(f"\n[Searching: [{message_hidden_text}]]")
+                # Message: Internal Search Results
+                elif message_type in ["InternalSearchResult"]:
+                    logger.note("[Analyzing search results ...]")
+                # Message: Loader status, such as "Generating Answers"
+                elif message_type in ["InternalLoaderMessage"]:
+                    logger.note("[Generating answers ...]\n")
+                # Message: Render Cards for Webpages
+                elif message_type in ["RenderCardRequest"]:
+                    continue
+                # Message: Not Implemented
+                else:
+                    raise NotImplementedError(
+                        f"Not Supported Message Type: {message_type}"
+                    )