Spaces:
Running
Running
""" | |
Module to handle image uploading and processing for Bing AI integrations. | |
""" | |
from __future__ import annotations | |
import string | |
import random | |
import json | |
import math | |
from aiohttp import ClientSession | |
from PIL import Image | |
from ...typing import ImageType, Tuple | |
from ...image import to_image, process_image, to_base64, ImageResponse | |
IMAGE_CONFIG = { | |
"maxImagePixels": 360000, | |
"imageCompressionRate": 0.7, | |
"enableFaceBlurDebug": False, | |
} | |
async def upload_image( | |
session: ClientSession, | |
image_data: ImageType, | |
tone: str, | |
proxy: str = None | |
) -> ImageResponse: | |
""" | |
Uploads an image to Bing's AI service and returns the image response. | |
Args: | |
session (ClientSession): The active session. | |
image_data (bytes): The image data to be uploaded. | |
tone (str): The tone of the conversation. | |
proxy (str, optional): Proxy if any. Defaults to None. | |
Raises: | |
RuntimeError: If the image upload fails. | |
Returns: | |
ImageResponse: The response from the image upload. | |
""" | |
image = to_image(image_data) | |
new_width, new_height = calculate_new_dimensions(image) | |
processed_img = process_image(image, new_width, new_height) | |
img_binary_data = to_base64(processed_img, IMAGE_CONFIG['imageCompressionRate']) | |
data, boundary = build_image_upload_payload(img_binary_data, tone) | |
headers = prepare_headers(session, boundary) | |
async with session.post("https://www.bing.com/images/kblob", data=data, headers=headers, proxy=proxy) as response: | |
if response.status != 200: | |
raise RuntimeError("Failed to upload image.") | |
return parse_image_response(await response.json()) | |
def calculate_new_dimensions(image: Image.Image) -> Tuple[int, int]: | |
""" | |
Calculates the new dimensions for the image based on the maximum allowed pixels. | |
Args: | |
image (Image): The PIL Image object. | |
Returns: | |
Tuple[int, int]: The new width and height for the image. | |
""" | |
width, height = image.size | |
max_image_pixels = IMAGE_CONFIG['maxImagePixels'] | |
if max_image_pixels / (width * height) < 1: | |
scale_factor = math.sqrt(max_image_pixels / (width * height)) | |
return int(width * scale_factor), int(height * scale_factor) | |
return width, height | |
def build_image_upload_payload(image_bin: str, tone: str) -> Tuple[str, str]: | |
""" | |
Builds the payload for image uploading. | |
Args: | |
image_bin (str): Base64 encoded image binary data. | |
tone (str): The tone of the conversation. | |
Returns: | |
Tuple[str, str]: The data and boundary for the payload. | |
""" | |
boundary = "----WebKitFormBoundary" + ''.join(random.choices(string.ascii_letters + string.digits, k=16)) | |
data = f"--{boundary}\r\n" \ | |
f"Content-Disposition: form-data; name=\"knowledgeRequest\"\r\n\r\n" \ | |
f"{json.dumps(build_knowledge_request(tone), ensure_ascii=False)}\r\n" \ | |
f"--{boundary}\r\n" \ | |
f"Content-Disposition: form-data; name=\"imageBase64\"\r\n\r\n" \ | |
f"{image_bin}\r\n" \ | |
f"--{boundary}--\r\n" | |
return data, boundary | |
def build_knowledge_request(tone: str) -> dict: | |
""" | |
Builds the knowledge request payload. | |
Args: | |
tone (str): The tone of the conversation. | |
Returns: | |
dict: The knowledge request payload. | |
""" | |
return { | |
'invokedSkills': ["ImageById"], | |
'subscriptionId': "Bing.Chat.Multimodal", | |
'invokedSkillsRequestData': { | |
'enableFaceBlur': True | |
}, | |
'convoData': { | |
'convoid': "", | |
'convotone': tone | |
} | |
} | |
def prepare_headers(session: ClientSession, boundary: str) -> dict: | |
""" | |
Prepares the headers for the image upload request. | |
Args: | |
session (ClientSession): The active session. | |
boundary (str): The boundary string for the multipart/form-data. | |
Returns: | |
dict: The headers for the request. | |
""" | |
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' | |
return headers | |
def parse_image_response(response: dict) -> ImageResponse: | |
""" | |
Parses the response from the image upload. | |
Args: | |
response (dict): The response dictionary. | |
Raises: | |
RuntimeError: If parsing the image info fails. | |
Returns: | |
ImageResponse: The parsed image response. | |
""" | |
if not response.get('blobId'): | |
raise RuntimeError("Failed to parse image info.") | |
result = {'bcid': response.get('blobId', ""), 'blurredBcid': response.get('processedBlobId', "")} | |
result["imageUrl"] = f"https://www.bing.com/images/blob?bcid={result['blurredBcid'] or result['bcid']}" | |
result['originalImageUrl'] = ( | |
f"https://www.bing.com/images/blob?bcid={result['blurredBcid']}" | |
if IMAGE_CONFIG["enableFaceBlurDebug"] else | |
f"https://www.bing.com/images/blob?bcid={result['bcid']}" | |
) | |
return ImageResponse(result["imageUrl"], "", result) |