gpt4free/g4f/Provider/bing/upload_image.py

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