from __future__ import annotations import string import random import json import io import base64 import math from PIL import Image from ...typing import ImageType from aiohttp import ClientSession from ...image import to_image, process_image, to_base64 image_config = { "maxImagePixels": 360000, "imageComp.ssionRate": 0.7, "enableFaceBlurDebug": 0, } async def upload_image( session: ClientSession, image: ImageType, tone: str, proxy: str = None ) -> dict: image = to_image(image) width, height = image.size max_image_pixels = image_config['maxImagePixels'] if max_image_pixels / (width * height) < 1: new_width = int(width * math.sqrt(max_image_pixels / (width * height))) new_height = int(height * math.sqrt(max_image_pixels / (width * height))) else: new_width = width new_height = height new_img = process_image(image, new_width, new_height) new_img_binary_data = to_base64(new_img, image_config['imageCompressionRate']) data, boundary = build_image_upload_api_payload(new_img_binary_data, 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 response: if response.status != 200: raise RuntimeError("Failed to upload image.") image_info = await response.json() if not image_info.get('blobId'): raise RuntimeError("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 image_config["enableFaceBlurDebug"] else "https://www.bing.com/images/blob?bcid=" + result['bcid'] ) return result def build_image_upload_api_payload(image_bin: str, 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