Spaces:
Runtime error
Runtime error
File size: 4,362 Bytes
91525e6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
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
|