|
""" |
|
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) |