Add get and set conversation

Add get and set conversation

Author: JE-Chen

pyproject.toml

@@ -5,8 +5,8 @@ requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "re_edge_gpt"
-version = "0.0.18"
+name = "re_edge_gpt_dev"
+version = "0.0.19"
 authors = [
     { name = "JE-Chen", email = "jechenmailman@gmail.com" },
 ]

re_edge_gpt/chathub.py

@@ -21,7 +21,6 @@
 from .request import ChatHubRequest
 from .upload_image import upload_image, upload_image_url
 from .utilities import append_identifier
-from .utilities import get_ran_hex
 from .utilities import guess_locale
 
 ssl_context = ssl.create_default_context()
@@ -62,36 +61,7 @@ def __init__(
             self.encrypted_conversation_signature = conversation.struct["encryptedConversationSignature"]
         else:
             self.encrypted_conversation_signature = None
-
-    async def get_conversation(
-            self,
-            conversation_id: str = None,
-            conversation_signature: str = None,
-            client_id: str = None,
-    ) -> dict:
-        self.conversation_id = conversation_id or self.request.conversation_id
-        conversation_signature = (
-                conversation_signature or self.request.conversation_signature
-        )
-        client_id = client_id or self.request.client_id
-        url = f"https://sydney.bing.com/sydney/GetConversation" \
-              f"?conversationId={conversation_id}" \
-              f"&source=cib&participantId={client_id}" \
-              f"&conversationSignature={conversation_signature}" \
-              f"&traceId={get_ran_hex()}"
-        response = await self.session.get(url)
-        return response.json()
-
-    async def get_activity(self) -> dict:
-        url = "https://www.bing.com/turing/conversation/chats"
-        headers = HEADERS_INIT_CONVER.copy()
-        if self.cookies is not None:
-            for cookie in self.cookies:
-                if cookie["name"] == "_U":
-                    headers["Cookie"] = f"SUID=A; _U={cookie['value']};"
-                    break
-        response = await self.session.get(url, headers=headers)
-        return response.json()
+        self.conversation = conversation
 
     async def ask_stream(
             self,
@@ -221,28 +191,20 @@ async def ask_stream(
                     elif raw:
                         yield False, response
 
-    async def delete_conversation(
-            self,
-            conversation_id: str = None,
-            conversation_signature: str = None,
-            client_id: str = None,
-    ) -> None:
-        conversation_id = conversation_id or self.request.conversation_id
-        conversation_signature = (
-                conversation_signature or self.request.conversation_signature
-        )
-        client_id = client_id or self.request.client_id
-        url = "https://sydney.bing.com/sydney/DeleteSingleConversation"
-        await self.session.post(
-            url,
-            json={
-                "conversationId": conversation_id,
-                "conversationSignature": conversation_signature,
-                "participant": {"id": client_id},
-                "source": "cib",
-                "optionsSets": ["autosave"],
-            },
-        )
-
     async def close(self) -> None:
         await self.session.aclose()
+
+    async def get_conversation(self):
+        return {
+            "conversation_id": self.conversation_id,
+            "client_id": self.request.client_id,
+            "encrypted_conversation_signature": self.encrypted_conversation_signature,
+            "conversation_signature": self.request.conversation_signature,
+        }
+
+    async def set_conversation(self, conversation_dict: dict):
+        self.conversation.struct["conversationId"] = conversation_dict.get("conversation_id")
+        self.conversation.struct["client_id"] = conversation_dict.get("client_id")
+        self.conversation.struct[
+            "encrypted_conversation_signature"] = conversation_dict.get("encrypted_conversation_signature")
+        self.conversation.struct["conversation_signature"] = conversation_dict.get("conversation_signature")

re_edge_gpt/re_edge_gpt.py

@@ -40,50 +40,6 @@ async def create(
         )
         return self
 
-    async def save_conversation(self, filename: str) -> None:
-        """
-        Save the conversation to a file
-        """
-        with open(filename, "w") as f:
-            conversation_id = self.chat_hub.request.conversation_id
-            conversation_signature = self.chat_hub.request.conversation_signature
-            client_id = self.chat_hub.request.client_id
-            invocation_id = self.chat_hub.request.invocation_id
-            f.write(
-                json.dumps(
-                    {
-                        "conversation_id": conversation_id,
-                        "conversation_signature": conversation_signature,
-                        "client_id": client_id,
-                        "invocation_id": invocation_id,
-                    },
-                ),
-            )
-
-    async def load_conversation(self, filename: str) -> None:
-        """
-        Load the conversation from a file
-        """
-        with open(filename) as f:
-            conversation = json.load(f)
-            self.chat_hub.request = ChatHubRequest(
-                conversation_signature=conversation["conversation_signature"],
-                client_id=conversation["client_id"],
-                conversation_id=conversation["conversation_id"],
-                invocation_id=conversation["invocation_id"],
-            )
-
-    async def get_conversation(self) -> dict:
-        """
-        Gets the conversation history from conversation_id (requires load_conversation)
-        """
-        return await self.chat_hub.get_conversation()
-
-    async def get_activity(self) -> dict:
-        """
-        Gets the recent activity (requires cookies)
-        """
-        return await self.chat_hub.get_activity()
 
     async def ask(
             self,
@@ -198,21 +154,6 @@ async def close(self) -> None:
         """
         await self.chat_hub.close()
 
-    async def delete_conversation(
-            self,
-            conversation_id: str = None,
-            conversation_signature: str = None,
-            client_id: str = None,
-    ) -> None:
-        """
-        Delete the chat in the server
-        """
-        await self.chat_hub.delete_conversation(
-            conversation_id=conversation_id,
-            conversation_signature=conversation_signature,
-            client_id=client_id,
-        )
-
     async def reset(self) -> None:
         """
         Reset the conversation

re_edge_gpt/utilities.py

@@ -19,11 +19,11 @@ def get_ran_hex(length: int = 32) -> str:
 
 def get_location_hint_from_locale(locale_str: str) -> Union[dict, None]:
     locale_str = locale_str.lower()
-    if locale == "en-gb":
+    if locale_str == "en-gb":
         hint = LocationHint.UK.value
-    elif locale == "en-ie":
+    elif locale_str == "en-ie":
         hint = LocationHint.EU.value
-    elif locale == "zh-cn":
+    elif locale_str == "zh-cn":
         hint = LocationHint.CHINA.value
     else:
         hint = LocationHint.USA.value

dev.toml renamed to stable.toml

@@ -5,8 +5,8 @@ requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "re_edge_gpt_dev"
-version = "0.0.18"
+name = "re_edge_gpt"
+version = "0.0.19"
 authors = [
     { name = "JE-Chen", email = "jechenmailman@gmail.com" },
 ]

test/unit_test/manual_test/test_bot_manual.py

@@ -17,7 +17,7 @@ async def test_ask() -> None:
             str(Path(str(Path.cwd()) + "/bing_cookies.json")), encoding="utf-8").read())
         bot = await Chatbot.create(cookies=cookies)
         response = await bot.ask(
-            prompt="How to get extract iron on mine",
+            prompt="How to extract gold",
             conversation_style=ConversationStyle.balanced,
             simplify_response=True
         )

test/unit_test/manual_test/test_conversation.py

@@ -0,0 +1,65 @@
+import asyncio
+import json
+from pathlib import Path
+
+from re_edge_gpt import Chatbot
+from re_edge_gpt import ConversationStyle
+
+# If you are using jupyter pls install this package
+# from nest_asyncio import apply
+
+conversation_dict = {}
+
+
+async def test_ask() -> None:
+    bot = None
+    try:
+        cookies: list[dict] = json.loads(open(
+            str(Path(str(Path.cwd()) + "/bing_cookies.json")), encoding="utf-8").read())
+        bot = await Chatbot.create(cookies=cookies)
+        response = await bot.ask(
+            prompt="Translate next word what I say to english",
+            conversation_style=ConversationStyle.balanced,
+            simplify_response=True
+        )
+        # If you are using non ascii char you need set ensure_ascii=False
+        print(json.dumps(response, indent=2, ensure_ascii=False))
+        print(await bot.chat_hub.get_conversation())
+        conversation_dict.update(await bot.chat_hub.get_conversation())
+    except Exception as error:
+        raise error
+    finally:
+        if bot is not None:
+            await bot.close()
+
+
+async def test_ask_conversation() -> None:
+    bot = None
+    try:
+        cookies: list[dict] = json.loads(open(
+            str(Path(str(Path.cwd()) + "/bing_cookies.json")), encoding="utf-8").read())
+        bot = await Chatbot.create(cookies=cookies)
+        await bot.chat_hub.set_conversation(conversation_dict=conversation_dict)
+        response = await bot.ask(
+            prompt="піца",
+            conversation_style=ConversationStyle.balanced,
+            simplify_response=True
+        )
+        # If you are using non ascii char you need set ensure_ascii=False
+        print(json.dumps(response, indent=2, ensure_ascii=False))
+    except Exception as error:
+        raise error
+    finally:
+        if bot is not None:
+            await bot.close()
+
+
+if __name__ == "__main__":
+    # If you are using jupyter pls use nest_asyncio apply()
+    # apply()
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = asyncio.get_event_loop()
+    loop.run_until_complete(test_ask())
+    loop.run_until_complete(test_ask_conversation())