Delete EdgeGPT

EdgeGPT/EdgeGPT.py

@@ -1,235 +0,0 @@
-"""
-Main.py
-"""
-from __future__ import annotations
-
-import json
-from pathlib import Path
-from typing import Generator
-
-from .chathub import *
-from .conversation import *
-from .conversation_style import *
-from .request import *
-from .utilities import *
-
-
-class Chatbot:
-    """
-    Combines everything to make it seamless
-    """
-
-    def __init__(
-        self,
-        proxy: str | None = None,
-        cookies: list[dict] | None = None,
-    ) -> None:
-        self.proxy: str | None = proxy
-        self.chat_hub: ChatHub = ChatHub(
-            Conversation(self.proxy, cookies=cookies),
-            proxy=self.proxy,
-            cookies=cookies,
-        )
-
-    @staticmethod
-    async def create(
-        proxy: str | None = None,
-        cookies: list[dict] | None = None,
-    ) -> Chatbot:
-        self = Chatbot.__new__(Chatbot)
-        self.proxy = proxy
-        self.chat_hub = ChatHub(
-            await Conversation.create(self.proxy, cookies=cookies),
-            proxy=self.proxy,
-            cookies=cookies,
-        )
-        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,
-        prompt: str,
-        wss_link: str = "wss://sydney.bing.com/sydney/ChatHub",
-        conversation_style: CONVERSATION_STYLE_TYPE = None,
-        webpage_context: str | None = None,
-        search_result: bool = False,
-        locale: str = guess_locale(),
-        simplify_response: bool = False,
-    ) -> dict:
-        """
-        Ask a question to the bot
-        Response:
-            {
-                item (dict):
-                    messages (list[dict]):
-                        adaptiveCards (list[dict]):
-                            body (list[dict]):
-                                text (str): Response
-            }
-        To get the response, you can do:
-            response["item"]["messages"][1]["adaptiveCards"][0]["body"][0]["text"]
-        """
-        async for final, response in self.chat_hub.ask_stream(
-            prompt=prompt,
-            conversation_style=conversation_style,
-            wss_link=wss_link,
-            webpage_context=webpage_context,
-            search_result=search_result,
-            locale=locale,
-        ):
-            if final:
-                if not simplify_response:
-                    return response
-                messages_left = response["item"]["throttling"][
-                    "maxNumUserMessagesInConversation"
-                ] - response["item"]["throttling"].get(
-                    "numUserMessagesInConversation", 0
-                )
-                if messages_left == 0:
-                    raise Exception("Max messages reached")
-                for msg in reversed(response["item"]["messages"]):
-                    if msg.get("adaptiveCards") and msg["adaptiveCards"][0]["body"][
-                        0
-                    ].get("text"):
-                        message = msg
-                        break
-                if not message:
-                    raise Exception("No message found")
-                suggestions = [
-                    suggestion["text"]
-                    for suggestion in message.get("suggestedResponses", [])
-                ]
-                adaptive_cards = message.get("adaptiveCards", [])
-                adaptive_text = (
-                    adaptive_cards[0]["body"][0].get("text") if adaptive_cards else None
-                )
-                sources = (
-                    adaptive_cards[0]["body"][0].get("text") if adaptive_cards else None
-                )
-                sources_text = (
-                    adaptive_cards[0]["body"][-1].get("text")
-                    if adaptive_cards
-                    else None
-                )
-                return {
-                    "text": message["text"],
-                    "author": message["author"],
-                    "sources": sources,
-                    "sources_text": sources_text,
-                    "suggestions": suggestions,
-                    "messages_left": messages_left,
-                    "max_messages": response["item"]["throttling"][
-                        "maxNumUserMessagesInConversation"
-                    ],
-                    "adaptive_text": adaptive_text,
-                }
-        return {}
-
-    async def ask_stream(
-        self,
-        prompt: str,
-        wss_link: str = "wss://sydney.bing.com/sydney/ChatHub",
-        conversation_style: CONVERSATION_STYLE_TYPE = None,
-        raw: bool = False,
-        webpage_context: str | None = None,
-        search_result: bool = False,
-        locale: str = guess_locale(),
-    ) -> Generator[bool, dict | str, None]:
-        """
-        Ask a question to the bot
-        """
-        async for response in self.chat_hub.ask_stream(
-            prompt=prompt,
-            conversation_style=conversation_style,
-            wss_link=wss_link,
-            raw=raw,
-            webpage_context=webpage_context,
-            search_result=search_result,
-            locale=locale,
-        ):
-            yield response
-
-    async def close(self) -> None:
-        """
-        Close the connection
-        """
-        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, delete=False) -> None:
-        """
-        Reset the conversation
-        """
-        if delete:
-            await self.remove_and_close()
-        else:
-            await self.close()
-        self.chat_hub = ChatHub(
-            await Conversation.create(self.proxy, cookies=self.chat_hub.cookies),
-            proxy=self.proxy,
-            cookies=self.chat_hub.cookies,
-        )
-
-
-if __name__ == "__main__":
-    from .main import main
-
-    main()

EdgeGPT/EdgeUtils.py

@@ -1,376 +0,0 @@
-import asyncio
-import json
-import re
-from pathlib import Path
-
-from .EdgeGPT import Chatbot, ConversationStyle
-from .ImageGen import ImageGen
-from log2d import Log
-
-Log("BingChat")
-log = Log.BingChat.debug  # shortcut to create a log entry
-
-class Cookie:
-    """
-    Convenience class for Bing Cookie files, data, and configuration. This Class
-    is updated dynamically by the Query class to allow cycling through >1
-    cookie/credentials file e.g. when daily request limits (current 200 per
-    account per day) are exceeded.
-    """
-    current_file_index = 0
-    dir_path = Path.home().resolve() / "bing_cookies"
-    current_file_path = dir_path  # Avoid Path errors when no cookie file used
-    search_pattern = 'bing_cookies_*.json'
-    ignore_files = set()
-    request_count = {}
-    supplied_files = set()
-    rotate_cookies = True
-
-    @classmethod
-    def files(cls):
-        """
-        Return a sorted list of all cookie files matching .search_pattern in
-        cls.dir_path, plus any supplied files, minus any ignored files.
-        """
-        all_files = set(Path(cls.dir_path).glob(cls.search_pattern))
-        if hasattr(cls, "supplied_files"):
-            supplied_files = {x for x in cls.supplied_files if x.is_file()}
-            all_files.update(supplied_files)
-        return sorted(list(all_files - cls.ignore_files))
-
-    @classmethod
-    def import_data(cls):
-        """
-        Read the active cookie file and populate the following attributes:
-
-          .current_file_path
-          .current_data
-          .image_token
-        """
-        if not cls.files():
-            log(f"No files in Cookie.dir_path")
-            return
-        try:
-            cls.current_file_path = cls.files()[cls.current_file_index]
-        except IndexError:
-            log(f"Invalid file index [{cls.current_file_index}]")
-            log("Files in Cookie.dir_path:")
-            for file in cls.files():
-                log(f"{file}")
-            return
-        log(f"Importing cookies from: {cls.current_file_path.name}")
-        with open(cls.current_file_path, encoding="utf-8") as file:
-            cls.current_data = json.load(file)
-        cls.image_token = [x for x in cls.current_data if x.get("name").startswith("_U")]
-        cls.image_token = cls.image_token[0].get("value")
-
-    @classmethod
-    def import_next(cls, discard=False):
-        """
-        Cycle through to the next cookies file then import it.
-
-        discard (bool): True -Mark the previous file to be ignored for the remainder of the current session.  Otherwise cycle through all available
-        cookie files (sharing the workload and 'resting' when not in use).
-        """
-        if not hasattr(cls, "current_file_path"):
-            cls.import_data()
-            return
-        try:
-            if discard:
-                cls.ignore_files.add(cls.current_file_path)
-            else:
-                Cookie.current_file_index += 1
-        except AttributeError:
-            # Will fail on first instantiation because no current_file_path
-            pass
-        if Cookie.current_file_index >= len(cls.files()):
-            Cookie.current_file_index = 0
-        Cookie.import_data()
-
-class Query:
-    """
-    A convenience class that wraps around EdgeGPT.Chatbot to encapsulate input,
-    config, and output all together.  Relies on Cookie class for authentication
-    unless ignore_cookies=True
-    """
-    index = []
-    image_dir_path = Path.cwd().resolve() / "bing_images"
-
-    def __init__(
-        self,
-        prompt,
-        style="precise",
-        content_type="text",
-        cookie_files=None,
-        ignore_cookies=False,
-        echo=True,
-        echo_prompt=False,
-        locale = "en-GB",
-        simplify_response = True,
-    ):
-        """
-        Arguments:
-
-        prompt: Text to enter into Bing Chat
-        style: creative, balanced, or precise
-        content_type: "text" for Bing Chat; "image" for Dall-e
-        ignore_cookies (bool): Ignore cookie data altogether
-        echo: Print something to confirm request made
-        echo_prompt: Print confirmation of the evaluated prompt
-        simplify_response: True -> single simplified prompt/response exchange
-        cookie_files: iterable of Paths or strings of cookie files (json)
-
-        Files in Cookie.dir_path will also be used if they exist.  This defaults
-        to the current working directory, so set Cookie.dir_path before
-        creating a Query if your cookie files are elsewhere.
-        """
-        self.__class__.index += [self]
-        self.prompt = prompt
-        self.locale = locale
-        self.simplify_response = simplify_response
-        self.ignore_cookies = ignore_cookies
-        if not ignore_cookies:
-            if cookie_files:
-                # Convert singular argument to an iterable:
-                if isinstance(cookie_files, (str, Path)):
-                    cookie_files = {cookie_files}
-                # Check all elements exist and are Paths:
-                cookie_files = {
-                    Path(x).resolve()
-                    for x in cookie_files
-                    if isinstance(x, (str, Path)) and x
-                }
-                Cookie.supplied_files = cookie_files
-            files = Cookie.files()  # includes .supplied_files
-            if Cookie.rotate_cookies:
-                Cookie.import_next()
-            else:
-                Cookie.import_data()
-        if content_type == "text":
-            self.style = style
-            self.log_and_send_query(echo, echo_prompt)
-        if content_type == "image":
-            self.create_image()
-
-    def log_and_send_query(self, echo, echo_prompt):
-        self.response = asyncio.run(self.send_to_bing(echo, echo_prompt))
-        if not hasattr(Cookie, "current_data"):
-            name = "<no_cookies>"
-        else:
-            name = Cookie.current_file_path.name
-        if not Cookie.request_count.get(name):
-            Cookie.request_count[name] = 1
-        else:
-            Cookie.request_count[name] += 1
-
-    def create_image(self):
-        image_generator = ImageGen(Cookie.image_token)
-        image_generator.save_images(
-            image_generator.get_images(self.prompt),
-            output_dir=self.__class__.image_dir_path,
-        )
-
-    async def send_to_bing(self, echo=True, echo_prompt=False):
-        """Creat, submit, then close a Chatbot instance.  Return the response"""
-        retries = len(Cookie.files()) or 1
-        while retries:
-            if not hasattr(Cookie, "current_data"):
-                bot = await Chatbot.create()
-            else:
-                bot = await Chatbot.create(cookies=Cookie.current_data)
-            if echo_prompt:
-                log(f"{self.prompt=}")
-            if echo:
-                log("Waiting for response...")
-            if self.style.lower() not in "creative balanced precise".split():
-                self.style = "precise"
-            try:
-                response = await bot.ask(
-                    prompt=self.prompt,
-                    conversation_style=getattr(ConversationStyle, self.style),simplify_response=self.simplify_response,
-                    locale=self.locale,
-                )
-                return response
-            except Exception as ex:
-                log(f"Exception: [{Cookie.current_file_path.name} may have exceeded the daily limit]\n{ex}")
-                Cookie.import_next(discard=True)
-                retries -= 1
-            finally:
-                await bot.close()
-
-    @property
-    def output(self):
-        """The response from a completed Chatbot request"""
-        if self.simplify_response:
-            try:
-                return self.response['text']
-            except TypeError as te:
-                raise TypeError(f"{te}\n(No response received - probably rate throttled...)")
-        else:
-            return [
-                x.get('text') or x.get('hiddenText')
-                for x in self.response['item']['messages']
-                if x['author']=='bot'
-            ]
-
-    @property
-    def sources(self):
-        """The source names and details parsed from a completed Chatbot request"""
-        if self.simplify_response:
-            return self.response['sources_text']
-        else:
-            return [
-                x.get('sourceAttributions') or []
-                for x in self.response['item']['messages']
-                if x['author']=='bot'
-            ]
-
-    @property
-    def sources_dict(self):
-        """The source names and details as a dictionary"""
-        if self.simplify_response:
-            text = self.response['sources_text']
-            sources = enumerate(re.findall(r'\((http.*?)\)', text))
-            return {index+1: value for index, value in sources}
-        else:
-            all_sources = []
-            name = 'providerDisplayName'
-            url = 'seeMoreUrl'
-            for sources in self.sources:
-                if not sources:
-                    continue
-                data = {}
-                for index, source in enumerate(sources):
-                    if name in source.keys() and url in source.keys():
-                        data[index+1] = source[url]
-                    else:
-                        continue
-                all_sources += [data]
-            return all_sources
-
-    @property
-    def code_block_formats(self):
-        """
-        Extract a list of programming languages/formats used in code blocks
-        """
-        regex = r"``` *(\b\w+\b\+*) *"
-        if self.simplify_response:
-            return re.findall(regex, self.output)
-        else:
-            return re.findall(regex, "\n".join(self.output))
-
-    @property
-    def code_blocks(self):
-        """
-        Return a list of code blocks (```) or snippets (`) as strings.
-
-        If the response contains a mix of snippets and code blocks, return the
-        code blocks only.
-
-        This method is not suitable if the main text response includes either of
-        the delimiters but not as part of an actual snippet or code block.
-
-        For example:
-        'In Markdown, the back-tick (`) is used to denote a code snippet'
-
-        """
-
-        final_blocks = []
-        if isinstance(self.output, str):  # I.e. simplify_response is True
-            separator = '```' if '```' in self.output else '`'
-            code_blocks = self.output.split(separator)[1:-1:2]
-            if separator == '`':
-                return code_blocks
-        else:
-            code_blocks = []
-            for response in self.output:
-                separator = '```' if '```' in response else '`'
-                code_blocks.extend(response.split(separator)[1:-1:2])
-            code_blocks = [x for x in code_blocks if x]
-        # Remove language name if present:
-        for block in code_blocks:
-            lines = block.splitlines()
-            code = lines[1:] if re.match(" *\w+ *", lines[0]) else lines
-            final_blocks += ["\n".join(code).removeprefix(separator)]
-        return [x for x in final_blocks if x]
-
-    @property
-    def code(self):
-        """
-        Extract and join any snippets of code or formatted data in the response
-        """
-        return "\n\n".join(self.code_blocks)
-
-
-    @property
-    def suggestions(self):
-        """Follow-on questions suggested by the Chatbot"""
-        if self.simplify_response:
-            return self.response['suggestions']
-        else:
-            try:
-                return [x['text'] for x in self.response['item']['messages'][1]['suggestedResponses']]
-            except KeyError:
-                return
-
-    def __repr__(self):
-        return f"<EdgeGPT.Query: {self.prompt}>"
-
-    def __str__(self):
-        if self.simplify_response:
-            return self.output
-        else:
-            return "\n\n".join(self.output)
-
-class ImageQuery(Query):
-    def __init__(self, prompt, **kwargs):
-        kwargs.update({"content_type": "image"})
-        super().__init__(prompt, **kwargs)
-
-    def __repr__(self):
-        return f"<EdgeGPT.ImageQuery: {self.prompt}>"
-
-def test_cookie_rotation():
-    for i in range(1, 50):
-        q = Query(f"What is {i} in Roman numerals?  Give the answer in JSON", style="precise")
-        log(f"{i}: {Cookie.current_file_path.name}")
-        log(q.code)
-        log(f"Cookie count: {Cookie.request_count.get(Cookie.current_file_path.name)}")
-
-def test_features():
-    try:
-        q = Query(f"What is {i} in Roman numerals?  Give the answer in JSON", style="precise")
-        log(f"{i}: {Cookie.current_file_path.name}")
-        print(f"{Cookie.current_file_index=}")
-        print(f"{Cookie.current_file_path=}")
-        print(f"{Cookie.current_data=}")
-        print(f"{Cookie.dir_path=}")
-        print(f"{Cookie.search_pattern=}")
-        print(f"{Cookie.files()=}")
-        print(f"{Cookie.image_token=}")
-        print(f"{Cookie.import_next(discard=True)=}")
-        print(f"{Cookie.rotate_cookies=}")
-        print(f"{Cookie.files()=}")
-        print(f"{Cookie.ignore_files=}")
-        print(f"{Cookie.supplied_files=}")
-        print(f"{Cookie.request_count=}")  # Keeps a tally of requests made in using each cookie file during this session
-        print(f"{q=}")
-        print(f"{q.prompt=}")
-        print(f"{q.ignore_cookies=}")
-        print(f"{q.style=}")
-        print(f"{q.simplify_response=}")
-        print(f"{q.locale=}")
-        print(f"{q.output=}")
-        print(q)
-        print(f"{q.sources=}")
-        print(f"{q.sources_dict=}")
-        print(f"{q.suggestions=}")
-        print(f"{q.code=}")  # All code as a single string
-        print(f"{q.code_blocks=}")  # Individual code blocks
-        print(f"{q.code_block_formats=}")  # The language/format of each code block (if given)
-        print(f"{Query.index=}")  # Keeps an index of Query objects created
-        print(f"{Query.image_dir_path=}")
-    except Exception as E:
-        raise Exception(E)    
-    finally:
-        return q

EdgeGPT/ImageGen.py

@@ -1,11 +0,0 @@
-# Open pull requests and issues at https://github.com/acheong08/BingImageCreator
-import BingImageCreator
-
-ImageGen = BingImageCreator.ImageGen
-
-ImageGenAsync = BingImageCreator.ImageGenAsync
-
-main = BingImageCreator.main
-
-if __name__ == "__main__":
-    main()

EdgeGPT/chathub.py

@@ -1,275 +0,0 @@
-import asyncio
-import json
-import os
-import ssl
-import sys
-from time import time
-from typing import Generator
-from typing import List
-from typing import Union
-
-from websockets.client import connect, WebSocketClientProtocol
-import certifi
-import httpx
-from BingImageCreator import ImageGenAsync
-
-from .constants import DELIMITER
-from .constants import HEADERS
-from .constants import HEADERS_INIT_CONVER
-from .conversation import Conversation
-from .conversation_style import CONVERSATION_STYLE_TYPE
-from .request import ChatHubRequest
-from .utilities import append_identifier
-from .utilities import get_ran_hex
-from .utilities import guess_locale
-
-ssl_context = ssl.create_default_context()
-ssl_context.load_verify_locations(certifi.where())
-
-
-class ChatHub:
-    def __init__(
-        self,
-        conversation: Conversation,
-        proxy: str = None,
-        cookies: Union[List[dict], None] = None,
-    ) -> None:
-        self.request: ChatHubRequest
-        self.loop: bool
-        self.task: asyncio.Task
-        self.request = ChatHubRequest(
-            conversation_signature=conversation.struct["conversationSignature"],
-            client_id=conversation.struct["clientId"],
-            conversation_id=conversation.struct["conversationId"],
-        )
-        self.cookies = cookies
-        self.proxy: str = proxy
-        proxy = (
-            proxy
-            or os.environ.get("all_proxy")
-            or os.environ.get("ALL_PROXY")
-            or os.environ.get("https_proxy")
-            or os.environ.get("HTTPS_PROXY")
-            or None
-        )
-        if proxy is not None and proxy.startswith("socks5h://"):
-            proxy = "socks5://" + proxy[len("socks5h://") :]
-        self.session = httpx.AsyncClient(
-            proxies=proxy,
-            timeout=900,
-            headers=HEADERS_INIT_CONVER,
-        )
-
-    async def get_conversation(
-        self,
-        conversation_id: str = None,
-        conversation_signature: str = None,
-        client_id: str = None,
-    ) -> dict:
-        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?conversationId={conversation_id}&source=cib&participantId={client_id}&conversationSignature={conversation_signature}&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()
-
-    async def ask_stream(
-        self,
-        prompt: str,
-        wss_link: str = None,
-        conversation_style: CONVERSATION_STYLE_TYPE = None,
-        raw: bool = False,
-        webpage_context: Union[str, None] = None,
-        search_result: bool = False,
-        locale: str = guess_locale(),
-    ) -> Generator[bool, Union[dict, str], None]:
-        """ """
-        req_header = HEADERS
-        if self.cookies is not None:
-            ws_cookies = []
-            for cookie in self.cookies:
-                ws_cookies.append(f"{cookie['name']}={cookie['value']}")
-            req_header.update({
-                'Cookie': ';'.join(ws_cookies),
-            })
-
-        # Check if websocket is closed
-        async with connect(
-            wss_link or "wss://sydney.bing.com/sydney/ChatHub",
-            # extra_headers=HEADERS,
-            extra_headers=req_header,
-            max_size=None,
-            ssl=ssl_context,
-            ping_interval=None,
-        ) as wss:
-            await self._initial_handshake(wss)
-            # Construct a ChatHub request
-            self.request.update(
-                prompt=prompt,
-                conversation_style=conversation_style,
-                webpage_context=webpage_context,
-                search_result=search_result,
-                locale=locale,
-            )
-            # Send request
-            await wss.send(append_identifier(self.request.struct))
-            draw = False
-            resp_txt = ""
-            result_text = ""
-            resp_txt_no_link = ""
-            retry_count = 5
-            while True:
-                if wss.closed:
-                    break
-                msg = await wss.recv()
-                if not msg:
-                    retry_count -= 1
-                    if retry_count == 0:
-                        raise Exception("No response from server")
-                    continue
-                if isinstance(msg, str):
-                    objects = msg.split(DELIMITER)
-                else:
-                    continue
-                for obj in objects:
-                    if int(time()) % 6 == 0:
-                        await wss.send(append_identifier({"type": 6}))
-                    if obj is None or not obj:
-                        continue
-                    response = json.loads(obj)
-                    # print(response)
-                    if response.get("type") == 1 and response["arguments"][0].get(
-                        "messages",
-                    ):
-                        if not draw:
-                            if (
-                                response["arguments"][0]["messages"][0].get(
-                                    "messageType",
-                                )
-                                == "GenerateContentQuery"
-                            ):
-                                async with ImageGenAsync(
-                                    all_cookies=self.cookies
-                                ) as image_generator:
-                                    images = await image_generator.get_images(
-                                        response["arguments"][0]["messages"][0]["text"],
-                                    )
-                                for i, image in enumerate(images):
-                                    resp_txt = f"{resp_txt}\n![image{i}]({image})"
-                                draw = True
-                            if (
-                                (
-                                    response["arguments"][0]["messages"][0][
-                                        "contentOrigin"
-                                    ]
-                                    != "Apology"
-                                )
-                                and not draw
-                                and not raw
-                            ):
-                                resp_txt = result_text + response["arguments"][0][
-                                    "messages"
-                                ][0]["adaptiveCards"][0]["body"][0].get("text", "")
-                                resp_txt_no_link = result_text + response["arguments"][
-                                    0
-                                ]["messages"][0].get("text", "")
-                                if response["arguments"][0]["messages"][0].get(
-                                    "messageType",
-                                ):
-                                    resp_txt = (
-                                        resp_txt
-                                        + response["arguments"][0]["messages"][0][
-                                            "adaptiveCards"
-                                        ][0]["body"][0]["inlines"][0].get("text")
-                                        + "\n"
-                                    )
-                                    result_text = (
-                                        result_text
-                                        + response["arguments"][0]["messages"][0][
-                                            "adaptiveCards"
-                                        ][0]["body"][0]["inlines"][0].get("text")
-                                        + "\n"
-                                    )
-                            if not raw:
-                                yield False, resp_txt
-
-                    elif response.get("type") == 2:
-                        if response["item"]["result"].get("error"):
-                            await self.close()
-                            raise Exception(
-                                f"{response['item']['result']['value']}: {response['item']['result']['message']}",
-                            )
-                        if draw:
-                            cache = response["item"]["messages"][1]["adaptiveCards"][0][
-                                "body"
-                            ][0]["text"]
-                            response["item"]["messages"][1]["adaptiveCards"][0]["body"][
-                                0
-                            ]["text"] = (cache + resp_txt)
-                        if (
-                            response["item"]["messages"][-1]["contentOrigin"]
-                            == "Apology"
-                            and resp_txt
-                        ):
-                            response["item"]["messages"][-1]["text"] = resp_txt_no_link
-                            response["item"]["messages"][-1]["adaptiveCards"][0][
-                                "body"
-                            ][0]["text"] = resp_txt
-                            print(
-                                "Preserved the message from being deleted",
-                                file=sys.stderr,
-                            )
-                        await wss.close()
-                        yield True, response
-                        return
-                    if response.get("type") != 2:
-                        if response.get("type") == 6:
-                            await wss.send(append_identifier({"type": 6}))
-                        elif response.get("type") == 7:
-                            await wss.send(append_identifier({"type": 7}))
-                        elif raw:
-                            yield False, response
-
-    async def _initial_handshake(self, wss: WebSocketClientProtocol) -> None:
-        await wss.send(append_identifier({"protocol": "json", "version": 1}))
-        await wss.recv()
-        await wss.send(append_identifier({"type": 6}))
-
-    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()

EdgeGPT/constants.py

@@ -1,51 +0,0 @@
-import random
-import uuid
-
-DELIMITER = "\x1e"
-
-
-# Generate random IP between range 13.104.0.0/14
-FORWARDED_IP = f"1.0.0.{random.randint(0, 255)}"
-
-HEADERS = {
-    "accept": "application/json",
-    "accept-language": "en-US,en;q=0.9",
-    "content-type": "application/json",
-    "sec-ch-ua": '"Not_A Brand";v="99", Microsoft Edge";v="110", "Chromium";v="110"',
-    "sec-ch-ua-arch": '"x86"',
-    "sec-ch-ua-bitness": '"64"',
-    "sec-ch-ua-full-version": '"109.0.1518.78"',
-    "sec-ch-ua-full-version-list": '"Chromium";v="110.0.5481.192", "Not A(Brand";v="24.0.0.0", "Microsoft Edge";v="110.0.1587.69"',
-    "sec-ch-ua-mobile": "?0",
-    "sec-ch-ua-model": "",
-    "sec-ch-ua-platform": '"Windows"',
-    "sec-ch-ua-platform-version": '"15.0.0"',
-    "sec-fetch-dest": "empty",
-    "sec-fetch-mode": "cors",
-    "sec-fetch-site": "same-origin",
-    "x-ms-client-request-id": str(uuid.uuid4()),
-    "x-ms-useragent": "azsdk-js-api-client-factory/1.0.0-beta.1 core-rest-pipeline/1.10.0 OS/Win32",
-    "Referer": "https://www.bing.com/search?q=Bing+AI&showconv=1&FORM=hpcodx",
-    "Referrer-Policy": "origin-when-cross-origin",
-    "x-forwarded-for": FORWARDED_IP,
-}
-
-HEADERS_INIT_CONVER = {
-    "authority": "www.bing.com",
-    "accept": "application/json",
-    "accept-language": "en-US,en;q=0.9",
-    "cache-control": "max-age=0",
-    "sec-ch-ua": '"Chromium";v="110", "Not A(Brand";v="24", "Microsoft Edge";v="110"',
-    "sec-ch-ua-arch": '"x86"',
-    "sec-ch-ua-bitness": '"64"',
-    "sec-ch-ua-full-version": '"110.0.1587.69"',
-    "sec-ch-ua-full-version-list": '"Chromium";v="110.0.5481.192", "Not A(Brand";v="24.0.0.0", "Microsoft Edge";v="110.0.1587.69"',
-    "sec-ch-ua-mobile": "?0",
-    "sec-ch-ua-model": '""',
-    "sec-ch-ua-platform": '"Windows"',
-    "sec-ch-ua-platform-version": '"15.0.0"',
-    "upgrade-insecure-requests": "1",
-    "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36 Edg/112.0.1722.46",
-    "x-edge-shopping-flag": "1",
-    "x-forwarded-for": FORWARDED_IP,
-}

EdgeGPT/conversation.py

@@ -1,122 +0,0 @@
-import json
-import os
-from typing import List
-from typing import Union
-
-import httpx
-
-from .constants import HEADERS_INIT_CONVER
-from .exceptions import NotAllowedToAccess
-
-
-class Conversation:
-    def __init__(
-        self,
-        proxy: Union[str, None] = None,
-        async_mode: bool = False,
-        cookies: Union[List[dict], None] = None,
-    ) -> None:
-        if async_mode:
-            return
-        self.struct: dict = {
-            "conversationId": None,
-            "clientId": None,
-            "conversationSignature": None,
-            "result": {"value": "Success", "message": None},
-        }
-        self.proxy = proxy
-        proxy = (
-            proxy
-            or os.environ.get("all_proxy")
-            or os.environ.get("ALL_PROXY")
-            or os.environ.get("https_proxy")
-            or os.environ.get("HTTPS_PROXY")
-            or None
-        )
-        if proxy is not None and proxy.startswith("socks5h://"):
-            proxy = "socks5://" + proxy[len("socks5h://") :]
-        self.session = httpx.Client(
-            proxies=proxy,
-            timeout=900,
-            headers=HEADERS_INIT_CONVER,
-        )
-        if cookies:
-            for cookie in cookies:
-                self.session.cookies.set(cookie["name"], cookie["value"])
-        # Send GET request
-        response = self.session.get(
-            url=os.environ.get("BING_PROXY_URL")
-            or "https://edgeservices.bing.com/edgesvc/turing/conversation/create",
-        )
-        if response.status_code != 200:
-            print(f"Status code: {response.status_code}")
-            print(response.text)
-            print(response.url)
-            raise Exception("Authentication failed")
-        try:
-            self.struct = response.json()
-        except (json.decoder.JSONDecodeError, NotAllowedToAccess) as exc:
-            raise Exception(
-                "Authentication failed. You have not been accepted into the beta.",
-            ) from exc
-        if self.struct["result"]["value"] == "UnauthorizedRequest":
-            raise NotAllowedToAccess(self.struct["result"]["message"])
-
-    @staticmethod
-    async def create(
-        proxy: Union[str, None] = None,
-        cookies: Union[List[dict], None] = None,
-    ) -> "Conversation":
-        self = Conversation(async_mode=True)
-        self.struct = {
-            "conversationId": None,
-            "clientId": None,
-            "conversationSignature": None,
-            "result": {"value": "Success", "message": None},
-        }
-        self.proxy = proxy
-        proxy = (
-            proxy
-            or os.environ.get("all_proxy")
-            or os.environ.get("ALL_PROXY")
-            or os.environ.get("https_proxy")
-            or os.environ.get("HTTPS_PROXY")
-            or None
-        )
-        if proxy is not None and proxy.startswith("socks5h://"):
-            proxy = "socks5://" + proxy[len("socks5h://") :]
-        transport = httpx.AsyncHTTPTransport(retries=900)
-        # Convert cookie format to httpx format
-        formatted_cookies = None
-        if cookies:
-            formatted_cookies = httpx.Cookies()
-            for cookie in cookies:
-                formatted_cookies.set(cookie["name"], cookie["value"])
-        async with httpx.AsyncClient(
-            proxies=proxy,
-            timeout=30,
-            headers=HEADERS_INIT_CONVER,
-            transport=transport,
-            cookies=formatted_cookies,
-        ) as client:
-            # Send GET request
-            response = await client.get(
-                url=os.environ.get("BING_PROXY_URL")
-                or "https://www.bing.com/turing/conversation/create",
-                follow_redirects=True,
-            )
-        if response.status_code != 200:
-            print(f"Status code: {response.status_code}")
-            print(response.text)
-            print(response.url)
-            raise Exception("Authentication failed")
-        try:
-            self.struct = response.json()
-        except (json.decoder.JSONDecodeError, NotAllowedToAccess) as exc:
-            print(response.text)
-            raise Exception(
-                "Authentication failed. You have not been accepted into the beta.",
-            ) from exc
-        if self.struct["result"]["value"] == "UnauthorizedRequest":
-            raise NotAllowedToAccess(self.struct["result"]["message"])
-        return self

EdgeGPT/conversation_style.py

@@ -1,56 +0,0 @@
-from enum import Enum
-
-try:
-    from typing import Literal, Union
-except ImportError:
-    from typing_extensions import Literal
-from typing import Optional
-
-
-class ConversationStyle(Enum):
-    creative = [
-        "nlu_direct_response_filter",
-        "deepleo",
-        "disable_emoji_spoken_text",
-        "responsible_ai_policy_235",
-        "enablemm",
-        "h3imaginative",
-        "objopinion",
-        "dsblhlthcrd",
-        "dv3sugg",
-        "autosave",
-        "clgalileo",
-        "gencontentv3",
-    ]
-    balanced = [
-        "nlu_direct_response_filter",
-        "deepleo",
-        "disable_emoji_spoken_text",
-        "responsible_ai_policy_235",
-        "enablemm",
-        "galileo",
-        "saharagenconv5",
-        "objopinion",
-        "dsblhlthcrd",
-        "dv3sugg",
-        "autosave",
-    ]
-    precise = [
-        "nlu_direct_response_filter",
-        "deepleo",
-        "disable_emoji_spoken_text",
-        "responsible_ai_policy_235",
-        "enablemm",
-        "h3precise",
-        "objopinion",
-        "dsblhlthcrd",
-        "dv3sugg",
-        "autosave",
-        "clgalileo",
-        "gencontentv3",
-    ]
-
-
-CONVERSATION_STYLE_TYPE = Optional[
-    Union[ConversationStyle, Literal["creative", "balanced", "precise"]]
-]

EdgeGPT/exceptions.py

@@ -1,2 +0,0 @@
-class NotAllowedToAccess(Exception):
-    pass

EdgeGPT/locale.py

@@ -1,85 +0,0 @@
-from enum import Enum
-
-try:
-    from typing import Literal, Union
-except ImportError:
-    from typing_extensions import Literal
-from typing import Optional
-
-
-class LocationHint(Enum):
-    USA = {
-        "locale": "en-US",
-        "LocationHint": [
-            {
-                "country": "United States",
-                "state": "California",
-                "city": "Los Angeles",
-                "timezoneoffset": 8,
-                "countryConfidence": 8,
-                "Center": {
-                    "Latitude": 34.0536909,
-                    "Longitude": -118.242766,
-                },
-                "RegionType": 2,
-                "SourceType": 1,
-            },
-        ],
-    }
-    CHINA = {
-        "locale": "zh-CN",
-        "LocationHint": [
-            {
-                "country": "China",
-                "state": "",
-                "city": "Beijing",
-                "timezoneoffset": 8,
-                "countryConfidence": 8,
-                "Center": {
-                    "Latitude": 39.9042,
-                    "Longitude": 116.4074,
-                },
-                "RegionType": 2,
-                "SourceType": 1,
-            },
-        ],
-    }
-    EU = {
-        "locale": "en-IE",
-        "LocationHint": [
-            {
-                "country": "Norway",
-                "state": "",
-                "city": "Oslo",
-                "timezoneoffset": 1,
-                "countryConfidence": 8,
-                "Center": {
-                    "Latitude": 59.9139,
-                    "Longitude": 10.7522,
-                },
-                "RegionType": 2,
-                "SourceType": 1,
-            },
-        ],
-    }
-    UK = {
-        "locale": "en-GB",
-        "LocationHint": [
-            {
-                "country": "United Kingdom",
-                "state": "",
-                "city": "London",
-                "timezoneoffset": 0,
-                "countryConfidence": 8,
-                "Center": {
-                    "Latitude": 51.5074,
-                    "Longitude": -0.1278,
-                },
-                "RegionType": 2,
-                "SourceType": 1,
-            },
-        ],
-    }
-
-
-LOCATION_HINT_TYPES = Optional[Union[LocationHint, Literal["USA", "CHINA", "EU", "UK"]]]

EdgeGPT/main.py

@@ -1,244 +0,0 @@
-import argparse
-import asyncio
-import json
-import re
-import sys
-from pathlib import Path
-
-from EdgeGPT.EdgeGPT import Chatbot
-from prompt_toolkit import PromptSession
-from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
-from prompt_toolkit.completion import WordCompleter
-from prompt_toolkit.history import InMemoryHistory
-from prompt_toolkit.key_binding import KeyBindings
-from rich.live import Live
-from rich.markdown import Markdown
-
-
-def create_session() -> PromptSession:
-    kb = KeyBindings()
-
-    @kb.add("enter")
-    def _(event) -> None:
-        buffer_text = event.current_buffer.text
-        if buffer_text.startswith("!"):
-            event.current_buffer.validate_and_handle()
-        else:
-            event.current_buffer.insert_text("\n")
-
-    @kb.add("escape")
-    def _(event) -> None:
-        if event.current_buffer.complete_state:
-            # event.current_buffer.cancel_completion()
-            event.current_buffer.text = ""
-
-    return PromptSession(key_bindings=kb, history=InMemoryHistory())
-
-
-def create_completer(commands: list, pattern_str: str = "$") -> WordCompleter:
-    return WordCompleter(words=commands, pattern=re.compile(pattern_str))
-
-
-def _create_history_logger(f):
-    def logger(*args, **kwargs) -> None:
-        tmp = sys.stdout
-        sys.stdout = f
-        print(*args, **kwargs, flush=True)
-        sys.stdout = tmp
-
-    return logger
-
-
-async def get_input_async(
-    session: PromptSession = None,
-    completer: WordCompleter = None,
-) -> str:
-    """
-    Multiline input function.
-    """
-    return await session.prompt_async(
-        completer=completer,
-        multiline=True,
-        auto_suggest=AutoSuggestFromHistory(),
-    )
-
-
-async def async_main(args: argparse.Namespace) -> None:
-    """
-    Main function
-    """
-    print("Initializing...")
-    print("Enter `alt+enter` or `escape+enter` to send a message")
-    # Read and parse cookies
-    cookies = None
-    if args.cookie_file:
-        file_path = Path(args.cookie_file)
-        if file_path.exists():
-            with file_path.open("r", encoding="utf-8") as f:
-                cookies = json.load(f)
-    bot = await Chatbot.create(proxy=args.proxy, cookies=cookies)
-    session = create_session()
-    completer = create_completer(["!help", "!exit", "!reset"])
-    initial_prompt = args.prompt
-
-    # Log chat history
-    def p_hist(*args, **kwargs) -> None:
-        pass
-
-    if args.history_file:
-        history_file_path = Path(args.history_file)
-        f = history_file_path.open("a+", encoding="utf-8")
-        p_hist = _create_history_logger(f)
-
-    while True:
-        print("\nYou:")
-        p_hist("\nYou:")
-        if initial_prompt:
-            question = initial_prompt
-            print(question)
-            initial_prompt = None
-        else:
-            question = (
-                input()
-                if args.enter_once
-                else await get_input_async(session=session, completer=completer)
-            )
-        print()
-        p_hist(question + "\n")
-        if question == "!exit":
-            break
-        if question == "!help":
-            print(
-                """
-            !help - Show this help message
-            !exit - Exit the program
-            !reset - Reset the conversation
-            """,
-            )
-            continue
-        if question == "!reset":
-            await bot.reset()
-            continue
-        print("Bot:")
-        p_hist("Bot:")
-        if args.no_stream:
-            response = (
-                await bot.ask(
-                    prompt=question,
-                    conversation_style=args.style,
-                    wss_link=args.wss_link,
-                    search_result=args.search_result,
-                    locale=args.locale,
-                )
-            )["item"]["messages"][-1]["adaptiveCards"][0]["body"][0]["text"]
-            print(response)
-            p_hist(response)
-        else:
-            wrote = 0
-            if args.rich:
-                md = Markdown("")
-                with Live(md, auto_refresh=False) as live:
-                    async for final, response in bot.ask_stream(
-                        prompt=question,
-                        conversation_style=args.style,
-                        wss_link=args.wss_link,
-                        search_result=args.search_result,
-                        locale=args.locale,
-                    ):
-                        if not final:
-                            if not wrote:
-                                p_hist(response, end="")
-                            else:
-                                p_hist(response[wrote:], end="")
-                            if wrote > len(response):
-                                print(md)
-                                print(Markdown("***Bing revoked the response.***"))
-                            wrote = len(response)
-                            md = Markdown(response)
-                            live.update(md, refresh=True)
-            else:
-                async for final, response in bot.ask_stream(
-                    prompt=question,
-                    conversation_style=args.style,
-                    wss_link=args.wss_link,
-                    search_result=args.search_result,
-                    locale=args.locale,
-                ):
-                    if not final:
-                        if not wrote:
-                            print(response, end="", flush=True)
-                            p_hist(response, end="")
-                        else:
-                            print(response[wrote:], end="", flush=True)
-                            p_hist(response[wrote:], end="")
-                        wrote = len(response)
-                print()
-                p_hist()
-    if args.history_file:
-        f.close()
-    await bot.close()
-
-
-def main() -> None:
-    print(
-        """
-        EdgeGPT - A demo of reverse engineering the Bing GPT chatbot
-        Repo: github.com/acheong08/EdgeGPT
-        By: Antonio Cheong
-
-        !help for help
-
-        Type !exit to exit
-    """,
-    )
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--enter-once", action="store_true")
-    parser.add_argument("--search-result", action="store_true")
-    parser.add_argument("--no-stream", action="store_true")
-    parser.add_argument("--rich", action="store_true")
-    parser.add_argument(
-        "--proxy",
-        help="Proxy URL (e.g. socks5://127.0.0.1:1080)",
-        type=str,
-    )
-    parser.add_argument(
-        "--wss-link",
-        help="WSS URL(e.g. wss://sydney.bing.com/sydney/ChatHub)",
-        type=str,
-        default="wss://sydney.bing.com/sydney/ChatHub",
-    )
-    parser.add_argument(
-        "--style",
-        choices=["creative", "balanced", "precise"],
-        default="balanced",
-    )
-    parser.add_argument(
-        "--prompt",
-        type=str,
-        default="",
-        required=False,
-        help="prompt to start with",
-    )
-    parser.add_argument(
-        "--cookie-file",
-        type=str,
-        default="",
-        required=False,
-        help="path to cookie file",
-    )
-    parser.add_argument(
-        "--history-file",
-        type=str,
-        default="",
-        required=False,
-        help="path to history file",
-    )
-    parser.add_argument(
-        "--locale",
-        type=str,
-        default="en-US",
-        required=False,
-        help="your locale",
-    )
-    args = parser.parse_args()
-    asyncio.run(async_main(args))

EdgeGPT/request.py

@@ -1,156 +0,0 @@
-import uuid
-from datetime import datetime
-from typing import Union
-
-from .conversation_style import CONVERSATION_STYLE_TYPE
-from .conversation_style import ConversationStyle
-from .utilities import get_location_hint_from_locale
-from .utilities import get_ran_hex
-from .utilities import guess_locale
-
-
-class ChatHubRequest:
-    def __init__(
-        self,
-        conversation_signature: str,
-        client_id: str,
-        conversation_id: str,
-        invocation_id: int = 3,
-    ) -> None:
-        self.struct: dict = {}
-
-        self.client_id: str = client_id
-        self.conversation_id: str = conversation_id
-        self.conversation_signature: str = conversation_signature
-        self.invocation_id: int = invocation_id
-
-    def update(
-        self,
-        prompt: str,
-        conversation_style: CONVERSATION_STYLE_TYPE,
-        webpage_context: Union[str, None] = None,
-        search_result: bool = False,
-        locale: str = guess_locale(),
-    ) -> None:
-        options = [
-            "deepleo",
-            "enable_debug_commands",
-            "disable_emoji_spoken_text",
-            "enablemm",
-        ]
-        if conversation_style:
-            if not isinstance(conversation_style, ConversationStyle):
-                conversation_style = getattr(ConversationStyle, conversation_style)
-            options = conversation_style.value
-        message_id = str(uuid.uuid4())
-        # Get the current local time
-        now_local = datetime.now()
-
-        # Get the current UTC time
-        now_utc = datetime.utcnow()
-
-        # Calculate the time difference between local and UTC time
-        timezone_offset = now_local - now_utc
-
-        # Get the offset in hours and minutes
-        offset_hours = int(timezone_offset.total_seconds() // 3600)
-        offset_minutes = int((timezone_offset.total_seconds() % 3600) // 60)
-
-        # Format the offset as a string
-        offset_string = f"{offset_hours:+03d}:{offset_minutes:02d}"
-
-        # Get current time
-        timestamp = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") + offset_string
-        self.struct = {
-            "arguments": [
-                {
-                    "source": "cib",
-                    "optionsSets": options,
-                    "allowedMessageTypes": [
-                        "ActionRequest",
-                        "Chat",
-                        "Context",
-                        "InternalSearchQuery",
-                        "InternalSearchResult",
-                        "Disengaged",
-                        "InternalLoaderMessage",
-                        "Progress",
-                        "RenderCardRequest",
-                        "AdsQuery",
-                        "SemanticSerp",
-                        "GenerateContentQuery",
-                        "SearchQuery",
-                    ],
-                    "sliceIds": [
-                        "winmuid1tf",
-                        "styleoff",
-                        "ccadesk",
-                        "smsrpsuppv4cf",
-                        "ssrrcache",
-                        "contansperf",
-                        "crchatrev",
-                        "winstmsg2tf",
-                        "creatgoglt",
-                        "creatorv2t",
-                        "sydconfigoptt",
-                        "adssqovroff",
-                        "530pstho",
-                        "517opinion",
-                        "418dhlth",
-                        "512sprtic1s0",
-                        "emsgpr",
-                        "525ptrcps0",
-                        "529rweas0",
-                        "515oscfing2s0",
-                        "524vidansgs0",
-                    ],
-                    "verbosity": "verbose",
-                    "traceId": get_ran_hex(32),
-                    "isStartOfSession": self.invocation_id == 3,
-                    "message": {
-                        "locale": locale,
-                        "market": locale,
-                        "region": locale[-2:],  # en-US -> US
-                        "locationHints": get_location_hint_from_locale(locale),
-                        "timestamp": timestamp,
-                        "author": "user",
-                        "inputMethod": "Keyboard",
-                        "text": prompt,
-                        "messageType": "Chat",
-                        "messageId": message_id,
-                        "requestId": message_id,
-                    },
-                    "tone": conversation_style.name.capitalize(),  # Make first letter uppercase
-                    "requestId": message_id,
-                    "conversationSignature": self.conversation_signature,
-                    "participant": {
-                        "id": self.client_id,
-                    },
-                    "conversationId": self.conversation_id,
-                },
-            ],
-            "invocationId": str(self.invocation_id),
-            "target": "chat",
-            "type": 4,
-        }
-        if search_result:
-            have_search_result = [
-                "InternalSearchQuery",
-                "InternalSearchResult",
-                "InternalLoaderMessage",
-                "RenderCardRequest",
-            ]
-            self.struct["arguments"][0]["allowedMessageTypes"] += have_search_result
-        if webpage_context:
-            self.struct["arguments"][0]["previousMessages"] = [
-                {
-                    "author": "user",
-                    "description": webpage_context,
-                    "contextType": "WebPage",
-                    "messageType": "Context",
-                    "messageId": "discover-web--page-ping-mriduna-----",
-                },
-            ]
-        self.invocation_id += 1
-
-        # print(timestamp)

EdgeGPT/utilities.py

@@ -1,39 +0,0 @@
-import json
-import locale
-import random
-import sys
-from typing import Union
-
-from .constants import DELIMITER
-from .locale import LocationHint
-
-
-def append_identifier(msg: dict) -> str:
-    # Convert dict to json string
-    return json.dumps(msg, ensure_ascii=False) + DELIMITER
-
-
-def get_ran_hex(length: int = 32) -> str:
-    return "".join(random.choice("0123456789abcdef") for _ in range(length))
-
-
-def get_location_hint_from_locale(locale: str) -> Union[dict, None]:
-    locale = locale.lower()
-    if locale == "en-gb":
-        hint = LocationHint.UK.value
-    elif locale == "en-ie":
-        hint = LocationHint.EU.value
-    elif locale == "zh-cn":
-        hint = LocationHint.CHINA.value
-    else:
-        hint = LocationHint.USA.value
-    return hint.get("LocationHint")
-
-
-def guess_locale() -> str:
-    if sys.platform.startswith("win"):
-        return "en-us"
-    loc, _ = locale.getlocale()
-    if not loc:
-        return "en-us"
-    return loc.replace("_", "-")