from __future__ import annotations
import string
import random
import json
import os
import re
import io
import base64
import numpy as np
import uuid
import urllib.parse
from PIL import Image
from aiohttp import ClientSession, ClientTimeout
from ..typing import AsyncResult, Messages
from .base_provider import AsyncGeneratorProvider
class Tones():
creative = "Creative"
balanced = "Balanced"
precise = "Precise"
default_cookies = {
'SRCHD' : 'AF=NOFORM',
'PPLState' : '1',
'KievRPSSecAuth': '',
'SUID' : '',
'SRCHUSR' : '',
'SRCHHPGUSR' : '',
}
class Bing(AsyncGeneratorProvider):
url = "https://bing.com/chat"
working = True
supports_message_history = True
supports_gpt_4 = True
@staticmethod
def create_async_generator(
model: str,
messages: Messages,
proxy: str = None,
cookies: dict = None,
tone: str = Tones.creative,
image: str = None,
**kwargs
) -> AsyncResult:
if len(messages) < 2:
prompt = messages[0]["content"]
context = None
else:
prompt = messages[-1]["content"]
context = create_context(messages[:-1])
if not cookies or "SRCHD" not in cookies:
cookies = default_cookies
return stream_generate(prompt, tone, image, context, proxy, cookies)
def create_context(messages: Messages):
return "".join(
f"[{message['role']}]" + ("(#message)" if message['role']!="system" else "(#additional_instructions)") + f"\n{message['content']}\n\n"
for message in messages
)
class Conversation():
def __init__(self, conversationId: str, clientId: str, conversationSignature: str, imageInfo: dict=None) -> None:
self.conversationId = conversationId
self.clientId = clientId
self.conversationSignature = conversationSignature
self.imageInfo = imageInfo
async def create_conversation(session: ClientSession, tone: str, image: str = None, proxy: str = None) -> Conversation:
url = 'https://www.bing.com/turing/conversation/create?bundleVersion=1.1199.4'
async with session.get(url, proxy=proxy) as response:
data = await response.json()
conversationId = data.get('conversationId')
clientId = data.get('clientId')
conversationSignature = response.headers.get('X-Sydney-Encryptedconversationsignature')
if not conversationId or not clientId or not conversationSignature:
raise Exception('Failed to create conversation.')
conversation = Conversation(conversationId, clientId, conversationSignature, None)
if isinstance(image,str):
try:
config = {
"visualSearch": {
"maxImagePixels": 360000,
"imageCompressionRate": 0.7,
"enableFaceBlurDebug": 0,
}
}
is_data_uri_an_image(image)
img_binary_data = extract_data_uri(image)
is_accepted_format(img_binary_data)
img = Image.open(io.BytesIO(img_binary_data))
width, height = img.size
max_image_pixels = config['visualSearch']['maxImagePixels']
compression_rate = config['visualSearch']['imageCompressionRate']
if max_image_pixels / (width * height) < 1:
new_width = int(width * np.sqrt(max_image_pixels / (width * height)))
new_height = int(height * np.sqrt(max_image_pixels / (width * height)))
else:
new_width = width
new_height = height
try:
orientation = get_orientation(img)
except Exception:
orientation = None
new_img = process_image(orientation, img, new_width, new_height)
new_img_binary_data = compress_image_to_base64(new_img, compression_rate)
data, boundary = build_image_upload_api_payload(new_img_binary_data, conversation, tone)
headers = session.headers.copy()
headers["content-type"] = f'multipart/form-data; boundary={boundary}'
headers["referer"] = 'https://www.bing.com/search?q=Bing+AI&showconv=1&FORM=hpcodx'
headers["origin"] = 'https://www.bing.com'
async with session.post("https://www.bing.com/images/kblob", data=data, headers=headers, proxy=proxy) as image_upload_response:
if image_upload_response.status != 200:
raise Exception("Failed to upload image.")
image_info = await image_upload_response.json()
if not image_info.get('blobId'):
raise Exception("Failed to parse image info.")
result = {'bcid': image_info.get('blobId', "")}
result['blurredBcid'] = image_info.get('processedBlobId', "")
if result['blurredBcid'] != "":
result["imageUrl"] = "https://www.bing.com/images/blob?bcid=" + result['blurredBcid']
elif result['bcid'] != "":
result["imageUrl"] = "https://www.bing.com/images/blob?bcid=" + result['bcid']
result['originalImageUrl'] = (
"https://www.bing.com/images/blob?bcid="
+ result['blurredBcid']
if config['visualSearch']["enableFaceBlurDebug"]
else "https://www.bing.com/images/blob?bcid="
+ result['bcid']
)
conversation.imageInfo = result
except Exception as e:
print(f"An error happened while trying to send image: {str(e)}")
return conversation
async def list_conversations(session: ClientSession) -> list:
url = "https://www.bing.com/turing/conversation/chats"
async with session.get(url) as response:
response = await response.json()
return response["chats"]
async def delete_conversation(session: ClientSession, conversation: Conversation, proxy: str = None) -> list:
url = "https://sydney.bing.com/sydney/DeleteSingleConversation"
json = {
"conversationId": conversation.conversationId,
"conversationSignature": conversation.conversationSignature,
"participant": {"id": conversation.clientId},
"source": "cib",
"optionsSets": ["autosave"]
}
async with session.post(url, json=json, proxy=proxy) as response:
try:
response = await response.json()
return response["result"]["value"] == "Success"
except:
return False
class Defaults:
delimiter = "\x1e"
ip_address = f"13.{random.randint(104, 107)}.{random.randint(0, 255)}.{random.randint(0, 255)}"
allowedMessageTypes = [
"ActionRequest",
"Chat",
"Context",
# "Disengaged", unwanted
"Progress",
# "AdsQuery", unwanted
"SemanticSerp",
"GenerateContentQuery",
"SearchQuery",
# The following message types should not be added so that it does not flood with
# useless messages (such as "Analyzing images" or "Searching the web") while it's retrieving the AI response
# "InternalSearchQuery",
# "InternalSearchResult",
"RenderCardRequest",
# "RenderContentRequest"
]
sliceIds = [
'abv2',
'srdicton',
'convcssclick',
'stylewv2',
'contctxp2tf',
'802fluxv1pc_a',
'806log2sphs0',
'727savemem',
'277teditgnds0',
'207hlthgrds0',
]
location = {
"locale": "en-US",
"market": "en-US",
"region": "US",
"locationHints": [
{
"country": "United States",
"state": "California",
"city": "Los Angeles",
"timezoneoffset": 8,
"countryConfidence": 8,
"Center": {"Latitude": 34.0536909, "Longitude": -118.242766},
"RegionType": 2,
"SourceType": 1,
}
],
}
headers = {
'accept': '*/*',
'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"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.69',
'x-edge-shopping-flag': '1',
'x-forwarded-for': ip_address,
}
optionsSets = [
'nlu_direct_response_filter',
'deepleo',
'disable_emoji_spoken_text',
'responsible_ai_policy_235',
'enablemm',
'iyxapbing',
'iycapbing',
'gencontentv3',
'fluxsrtrunc',
'fluxtrunc',
'fluxv1',
'rai278',
'replaceurl',
'eredirecturl',
'nojbfedge'
]
def format_message(msg: dict) -> str:
return json.dumps(msg, ensure_ascii=False) + Defaults.delimiter
def build_image_upload_api_payload(image_bin: str, conversation: Conversation, tone: str):
payload = {
'invokedSkills': ["ImageById"],
'subscriptionId': "Bing.Chat.Multimodal",
'invokedSkillsRequestData': {
'enableFaceBlur': True
},
'convoData': {
'convoid': "",
'convotone': tone
}
}
knowledge_request = {
'imageInfo': {},
'knowledgeRequest': payload
}
boundary="----WebKitFormBoundary" + ''.join(random.choices(string.ascii_letters + string.digits, k=16))
data = (
f'--{boundary}'
+ '\r\nContent-Disposition: form-data; name="knowledgeRequest"\r\n\r\n'
+ json.dumps(knowledge_request, ensure_ascii=False)
+ "\r\n--"
+ boundary
+ '\r\nContent-Disposition: form-data; name="imageBase64"\r\n\r\n'
+ image_bin
+ "\r\n--"
+ boundary
+ "--\r\n"
)
return data, boundary
def is_data_uri_an_image(data_uri: str):
try:
# Check if the data URI starts with 'data:image' and contains an image format (e.g., jpeg, png, gif)
if not re.match(r'data:image/(\w+);base64,', data_uri):
raise ValueError("Invalid data URI image.")
# Extract the image format from the data URI
image_format = re.match(r'data:image/(\w+);base64,', data_uri).group(1)
# Check if the image format is one of the allowed formats (jpg, jpeg, png, gif)
if image_format.lower() not in ['jpeg', 'jpg', 'png', 'gif']:
raise ValueError("Invalid image format (from mime file type).")
except Exception as e:
raise e
def is_accepted_format(binary_data: bytes) -> bool:
try:
check = False
if binary_data.startswith(b'\xFF\xD8\xFF'):
check = True # It's a JPEG image
elif binary_data.startswith(b'\x89PNG\r\n\x1a\n'):
check = True # It's a PNG image
elif binary_data.startswith(b'GIF87a') or binary_data.startswith(b'GIF89a'):
check = True # It's a GIF image
elif binary_data.startswith(b'\x89JFIF') or binary_data.startswith(b'JFIF\x00'):
check = True # It's a JPEG image
elif binary_data.startswith(b'\xFF\xD8'):
check = True # It's a JPEG image
elif binary_data.startswith(b'RIFF') and binary_data[8:12] == b'WEBP':
check = True # It's a WebP image
# else we raise ValueError
if not check:
raise ValueError("Invalid image format (from magic code).")
except Exception as e:
raise e
def extract_data_uri(data_uri: str) -> bytes:
try:
data = data_uri.split(",")[1]
data = base64.b64decode(data)
return data
except Exception as e:
raise e
def get_orientation(data: bytes) -> int:
try:
if data[:2] != b'\xFF\xD8':
raise Exception('NotJpeg')
with Image.open(data) as img:
exif_data = img._getexif()
if exif_data is not None:
orientation = exif_data.get(274) # 274 corresponds to the orientation tag in EXIF
if orientation is not None:
return orientation
except Exception:
pass
def process_image(orientation: int, img: Image.Image, new_width: int, new_height: int) -> Image.Image:
try:
# Initialize the canvas
new_img = Image.new("RGB", (new_width, new_height), color="#FFFFFF")
if orientation:
if orientation > 4:
img = img.transpose(Image.FLIP_LEFT_RIGHT)
if orientation in [3, 4]:
img = img.transpose(Image.ROTATE_180)
if orientation in [5, 6]:
img = img.transpose(Image.ROTATE_270)
if orientation in [7, 8]:
img = img.transpose(Image.ROTATE_90)
new_img.paste(img, (0, 0))
return new_img
except Exception as e:
raise e
def compress_image_to_base64(img, compression_rate) -> str:
try:
output_buffer = io.BytesIO()
img.save(output_buffer, format="JPEG", quality=int(compression_rate * 100))
return base64.b64encode(output_buffer.getvalue()).decode('utf-8')
except Exception as e:
raise e
def create_message(conversation: Conversation, prompt: str, tone: str, context: str=None) -> str:
options_sets = Defaults.optionsSets
if tone == Tones.creative:
options_sets.append("h3imaginative")
elif tone == Tones.precise:
options_sets.append("h3precise")
elif tone == Tones.balanced:
options_sets.append("galileo")
else:
options_sets.append("harmonyv3")
request_id = str(uuid.uuid4())
struct = {
'arguments': [
{
'source': 'cib',
'optionsSets': options_sets,
'allowedMessageTypes': Defaults.allowedMessageTypes,
'sliceIds': Defaults.sliceIds,
'traceId': os.urandom(16).hex(),
'isStartOfSession': True,
'requestId': request_id,
'message': Defaults.location | {
'author': 'user',
'inputMethod': 'Keyboard',
'text': prompt,
'messageType': 'Chat',
'requestId': request_id,
'messageId': request_id,
},
"scenario": "SERP",
'tone': tone,
'spokenTextMode': 'None',
'conversationId': conversation.conversationId,
'participant': {
'id': conversation.clientId
},
}
],
'invocationId': '1',
'target': 'chat',
'type': 4
}
if conversation.imageInfo != None and "imageUrl" in conversation.imageInfo and "originalImageUrl" in conversation.imageInfo:
struct['arguments'][0]['message']['originalImageUrl'] = conversation.imageInfo['originalImageUrl']
struct['arguments'][0]['message']['imageUrl'] = conversation.imageInfo['imageUrl']
struct['arguments'][0]['experienceType'] = None
struct['arguments'][0]['attachedFileInfo'] = {"fileName": None, "fileType": None}
if context:
struct['arguments'][0]['previousMessages'] = [{
"author": "user",
"description": context,
"contextType": "WebPage",
"messageType": "Context",
"messageId": "discover-web--page-ping-mriduna-----"
}]
return format_message(struct)
async def stream_generate(
prompt: str,
tone: str,
image: str = None,
context: str = None,
proxy: str = None,
cookies: dict = None
):
async with ClientSession(
timeout=ClientTimeout(total=900),
cookies=cookies,
headers=Defaults.headers,
) as session:
conversation = await create_conversation(session, tone, image, proxy)
try:
async with session.ws_connect('wss://sydney.bing.com/sydney/ChatHub', autoping=False, params={'sec_access_token': conversation.conversationSignature}, proxy=proxy) as wss:
await wss.send_str(format_message({'protocol': 'json', 'version': 1}))
await wss.receive(timeout=900)
await wss.send_str(create_message(conversation, prompt, tone, context))
response_txt = ''
returned_text = ''
final = False
while not final:
msg = await wss.receive(timeout=900)
objects = msg.data.split(Defaults.delimiter)
for obj in objects:
if obj is None or not obj:
continue
response = json.loads(obj)
if response.get('type') == 1 and response['arguments'][0].get('messages'):
message = response['arguments'][0]['messages'][0]
if (message['contentOrigin'] != 'Apology'):
if 'adaptiveCards' in message:
card = message['adaptiveCards'][0]['body'][0]
if "text" in card:
response_txt = card.get('text')
if message.get('messageType'):
inline_txt = card['inlines'][0].get('text')
response_txt += inline_txt + '\n'
elif message.get('contentType') == "IMAGE":
query = urllib.parse.quote(message.get('text'))
url = f"\nhttps://www.bing.com/images/create?q={query}"
response_txt += url
final = True
if response_txt.startswith(returned_text):
new = response_txt[len(returned_text):]
if new != "\n":
yield new
returned_text = response_txt
elif response.get('type') == 2:
result = response['item']['result']
if result.get('error'):
raise Exception(f"{result['value']}: {result['message']}")
return
finally:
await delete_conversation(session, conversation, proxy)
