Serialize ChatData to XML instead JSON

chatbot/networking/dto/chatdata.py

@@ -1,8 +1,10 @@
 from typing import List, Optional
 
 from jsonpickle import json
+from lxml import etree
 
 from chatbot.core import SerializableInterface
+from chatbot.utils.fn import create_xml_element
 
 
 class ChatData(SerializableInterface):
@@ -35,3 +37,18 @@ def deserialize(self, data: str):
             self.answer = json_data["answer"]
         if "reporter" in json_data:
             self.reporter = json_data["reporter"]
+
+    def to_xml(self) -> str:
+        """Return XML representation of self."""
+        root = etree.Element('data')
+        for key, val in vars(self).items():
+            if isinstance(val, list):
+                sub_root = etree.Element(key, {'type': 'array'})
+                for txt in val:
+                    elem = create_xml_element('value', txt)
+                    sub_root.append(elem)
+                root.append(sub_root)
+            else:
+                elem = create_xml_element(key, val)
+                root.append(elem)
+        return etree.tostring(root, encoding='utf-8', pretty_print=True).decode()

chatbot/services/v1/rag_chatbot/service.py

@@ -1,4 +1,3 @@
-import json
 import threading
 from pathlib import Path
 
@@ -168,16 +167,7 @@ def _api_feedback(self, request_data) -> Response:
             },
         )
 
-        json_content = {
-            "chat_id": feedback.chat_id,
-            "history": feedback.chat_history,
-            "question": feedback.question,
-            "answer": feedback.answer,
-            "user": feedback.reporter,
-            "feedback": feedback.feedback,
-        }
-        with safe_chat_file_path.open("w") as feedback_file:
-            json.dump(json_content, feedback_file, indent=2, ensure_ascii=False)
+        safe_chat_file_path.write_text(data=feedback.to_xml(), encoding='utf-8')
 
         return Response(jsonpickle.encode({"message": "success"}), mimetype="application/json")
 

chatbot/utils/fn.py

@@ -1,7 +1,8 @@
 import re
 from os.path import normpath
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Any
+from lxml import etree
 
 import jsonpickle
 from langchain_core.load import dumps, loads
@@ -14,7 +15,7 @@ def _get_safe_file_path(chats_path, chat_id: str) -> Optional[Path]:
 
 def _get_safe_feedback_path(chat_id: str) -> Optional[Path]:
     if not re.search(r"[^A-Za-z0-9_\-\\]", chat_id):
-        return Path(normpath(f"feedback/{chat_id}.json"))
+        return Path(normpath(f"feedback/{chat_id}.xml"))
 
 
 def load_json(path):
@@ -40,3 +41,9 @@ def reciprocal_rank_fusion(results: list[list], k=60, top_n=5):
         reranked_results.append((loads(doc), score))
 
     return reranked_results[:top_n]
+
+
+def create_xml_element(tag:str, text:str) -> etree.Element:
+    elem = etree.Element(tag)
+    elem.text = text
+    return elem