better results with "https://cn.bing.com"

[manatlan]

In some cases, BIC doesn't work (http connexion errors). (in threads of gunicorn/h11 workers)
I dive into the code, and reach to fix it ... just by changing the global var "BING_URL" from "https://www.bing.com" to "https://cn.bing.com" ... (in that cases, when using the original one, the first POST couldn't return the id, and breaks the following .... with the 'cn' version : it worked ootb) ... perhaps It could help someone.

BTW, you should concentrate your effort on the async one .... and make the sync one based on using the async one. (currently you have the 2 methods to maintain, when api/http changes)

BTW2: relations between edgegpt & bingimagecreator are confusing. For my needs, I've made a class, which use both, and expose functions from both in one instance. It's the way to go. And since, "creative mode" can return images too, it has a lot of sense ! Because it could replace the BIC apis. Because in chat mode, it's a lot easier to make multiple requests to make images, with context keeped. (the are no context in BIC)

[manatlan]

in fact, sometimes .... the redirect_url, returned by the first post, can be an absolute or relative one.
so it breaks all the rest, making the redirection rt4/rt3 fails

[manatlan]

@acheong08 ...

using just httpx, it could be written as :

import httpx,asyncio,re,json

HEADERS= {} # doesn't seem to be needed (?!)

COOKIES={}
for cookie in json.loads(open('bing.json', encoding="utf-8").read()):
    COOKIES[ cookie.get("name") ] = cookie.get("value")

async def get_images(prompt:str) -> list:
    async with httpx.AsyncClient(base_url="https://cn.bing.com", headers=HEADERS, cookies=COOKIES) as bing:
        r = await bing.post("/images/create",data={"q":prompt,"qs":"ds"},params={"q":prompt,"rt":4,"FORM":"GENCRE"})
        assert r.status_code==302, f"not a redirect: http/{r.status_code} ?!"
        request_id = r.headers.get("location","").split("id=")[-1]
    
        assert request_id.isalnum(),f"the redirect location doesn't contain 'id' {r.headers.get('location')}?!"
        
        await asyncio.sleep(5)   # seems it takes always more than 5sec

        for nb_retry in range(30):
            await asyncio.sleep(1)
            
            r=await bing.get(f"/images/create/async/results/{request_id}",params={"q":prompt})
            assert r.status_code==200,f"the polling url returns http/{r.status_code} ?!"
            
            if image_links := re.findall(r'src="([^"]+)"', r.text):
                return [ x.split("?")[0] for x in image_links ]
            
        return [] # 30 poll calls, and nothing ;-(

ll=asyncio.run(  get_images("cat with a ball")  )
print( ll )

I'm from France ... but I need to post to "cn.bing.com" to have a good redirection (which contains an id).
If I post to "www.bing.com" : it redirects me to cn.bing.com, without an id ?!?
(tested with differents headers, but no luck)

It's same behaviour with your code ... perhaps you know why ?!

[acheong08]

I'm from France ... but I need to post to "cn.bing.com" to have a good redirection (which contains an id).

This is extremely weird. I've never had this issue despite being in China (using a VPN).

The fact that it tries to redirect you to China suggests that the server sees a Chinese IP rather than French

[spike014]

@acheong08 ...

using just httpx, it could be written as :

import httpx,asyncio,re,json

HEADERS= {} # doesn't seem to be needed (?!)

COOKIES={}
for cookie in json.loads(open('bing.json', encoding="utf-8").read()):
    COOKIES[ cookie.get("name") ] = cookie.get("value")

async def get_images(prompt:str) -> list:
    async with httpx.AsyncClient(base_url="https://cn.bing.com", headers=HEADERS, cookies=COOKIES) as bing:
        r = await bing.post("/images/create",data={"q":prompt,"qs":"ds"},params={"q":prompt,"rt":4,"FORM":"GENCRE"})
        assert r.status_code==302, f"not a redirect: http/{r.status_code} ?!"
        request_id = r.headers.get("location","").split("id=")[-1]
    
        assert request_id.isalnum(),f"the redirect location doesn't contain 'id' {r.headers.get('location')}?!"
        
        await asyncio.sleep(5)   # seems it takes always more than 5sec

        for nb_retry in range(30):
            await asyncio.sleep(1)
            
            r=await bing.get(f"/images/create/async/results/{request_id}",params={"q":prompt})
            assert r.status_code==200,f"the polling url returns http/{r.status_code} ?!"
            
            if image_links := re.findall(r'src="([^"]+)"', r.text):
                return [ x.split("?")[0] for x in image_links ]
            
        return [] # 30 poll calls, and nothing ;-(

ll=asyncio.run(  get_images("cat with a ball")  )
print( ll )

I'm from France ... but I need to post to "cn.bing.com" to have a good redirection (which contains an id). If I post to "www.bing.com" : it redirects me to cn.bing.com, without an id ?!? (tested with differents headers, but no luck)

It's same behaviour with your code ... perhaps you know why ?!

You ran this project on a Chinese machine, or your bing account Settings > Country/region is China?

[manatlan]

You ran this project on a Chinese machine ?

no ... it's a hosted linux/vm on a french hosting service ( https://www.alwaysdata.com/ )

or your bing account Settings > Country/region is China?

no, it's clearly france

[spike014]

You ran this project on a Chinese machine ?

no ... it's a hosted linux/vm on a french hosting service ( https://www.alwaysdata.com/ )

or your bing account Settings > Country/region is China?

no, it's clearly france

Sorry. I can't figure it out yet. 🤔

[vansh77swami]

@manatlan this is the code of my bing image generator ai-

to run this i ahve to tun this command - python -m BingImageCreator -U <YOUR_TOKEN> --prompt "YOUR_IMAGE_PROMPT" --output-dir "OUTPUT_PATH"

now, i want to run it just by giving prompt like a chatbot to generate images

code -

import argparse
import asyncio
import contextlib
import json
import os
import random
import sys
import time
from functools import partial
from typing import Dict
from typing import List
from typing import Union

import httpx
import pkg_resources
import regex
import requests

if os.environ.get("BING_URL") == None:
BING_URL = "https://www.bing.com"
else:
BING_URL = os.environ.get("BING_URL")

Generate random IP between range 13.104.0.0/14

FORWARDED_IP = (
f"13.{random.randint(104, 107)}.{random.randint(0, 255)}.{random.randint(0, 255)}"
)
HEADERS = {
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,/;q=0.8,application/signed-exchange;v=b3;q=0.7",
"accept-language": "en-US,en;q=0.9",
"cache-control": "max-age=0",
"content-type": "application/x-www-form-urlencoded",
"referrer": "https://www.bing.com/images/create/",
"origin": "https://www.bing.com",
"user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.63",
"x-forwarded-for": FORWARDED_IP,
}

Error messages

error_timeout = "Your request has timed out."
error_redirect = "Redirect failed"
error_blocked_prompt = (
"Your prompt has been blocked by Bing. Try to change any bad words and try again."
)
error_being_reviewed_prompt = "Your prompt is being reviewed by Bing. Try to change any sensitive words and try again."
error_noresults = "Could not get results"
error_unsupported_lang = "\nthis language is currently not supported by bing"
error_bad_images = "Bad images"
error_no_images = "No images"

sending_message = "Sending request..."
wait_message = "Waiting for results..."
download_message = "\nDownloading images..."

def debug(debug_file, text_var):
"""helper function for debug"""
with open(f"{debug_file}", "a", encoding="utf-8") as f:
f.write(str(text_var))

class ImageGen:
"""
Image generation by Microsoft Bing
Parameters:
auth_cookie: str
Optional Parameters:
debug_file: str
quiet: bool
all_cookies: List[Dict]
"""

def __init__(
    self,
    auth_cookie: str,
    debug_file: Union[str, None] = None,
    quiet: bool = False,
    all_cookies: List[Dict] = None,
) -> None:
    self.session: requests.Session = requests.Session()
    self.session.headers = HEADERS
    self.session.cookies.set("_U", auth_cookie)
    if all_cookies:
        for cookie in all_cookies:
            self.session.cookies.set(cookie["name"], cookie["value"])
    self.quiet = quiet
    self.debug_file = debug_file
    if self.debug_file:
        self.debug = partial(debug, self.debug_file)

def get_images(self, prompt: str) -> list:
    """
    Fetches image links from Bing
    Parameters:
        prompt: str
    """
    if not self.quiet:
        print(sending_message)
    if self.debug_file:
        self.debug(sending_message)
    url_encoded_prompt = requests.utils.quote(prompt)
    payload = f"q={url_encoded_prompt}&qs=ds"
    # https://www.bing.com/images/create?q=<PROMPT>&rt=3&FORM=GENCRE
    url = f"{BING_URL}/images/create?q={url_encoded_prompt}&rt=4&FORM=GENCRE"
    response = self.session.post(
        url,
        allow_redirects=False,
        data=payload,
        timeout=200,
    )
    # check for content waring message
    if "this prompt is being reviewed" in response.text.lower():
        if self.debug_file:
            self.debug(f"ERROR: {error_being_reviewed_prompt}")
        raise Exception(
            error_being_reviewed_prompt,
        )
    if "this prompt has been blocked" in response.text.lower():
        if self.debug_file:
            self.debug(f"ERROR: {error_blocked_prompt}")
        raise Exception(
            error_blocked_prompt,
        )
    if (
        "we're working hard to offer image creator in more languages"
        in response.text.lower()
    ):
        if self.debug_file:
            self.debug(f"ERROR: {error_unsupported_lang}")
        raise Exception(error_unsupported_lang)
    if response.status_code != 302:
        # if rt4 fails, try rt3
        url = f"{BING_URL}/images/create?q={url_encoded_prompt}&rt=3&FORM=GENCRE"
        response = self.session.post(url, allow_redirects=False, timeout=200)
        if response.status_code != 302:
            if self.debug_file:
                self.debug(f"ERROR: {error_redirect}")
            print(f"ERROR: {response.text}")
            raise Exception(error_redirect)
    # Get redirect URL
    redirect_url = response.headers["Location"].replace("&nfy=1", "")
    request_id = redirect_url.split("id=")[-1]
    self.session.get(f"{BING_URL}{redirect_url}")
    # https://www.bing.com/images/create/async/results/{ID}?q={PROMPT}
    polling_url = f"{BING_URL}/images/create/async/results/{request_id}?q={url_encoded_prompt}"
    # Poll for results
    if self.debug_file:
        self.debug("Polling and waiting for result")
    if not self.quiet:
        print("Waiting for results...")
    start_wait = time.time()
    while True:
        if int(time.time() - start_wait) > 200:
            if self.debug_file:
                self.debug(f"ERROR: {error_timeout}")
            raise Exception(error_timeout)
        if not self.quiet:
            print(".", end="", flush=True)
        response = self.session.get(polling_url)
        if response.status_code != 200:
            if self.debug_file:
                self.debug(f"ERROR: {error_noresults}")
            raise Exception(error_noresults)
        if not response.text or response.text.find("errorMessage") != -1:
            time.sleep(1)
            continue
        else:
            break
    # Use regex to search for src=""
    image_links = regex.findall(r'src="([^"]+)"', response.text)
    # Remove size limit
    normal_image_links = [link.split("?w=")[0] for link in image_links]
    # Remove duplicates
    normal_image_links = list(set(normal_image_links))

    # Bad images
    bad_images = [
        "https://r.bing.com/rp/in-2zU3AJUdkgFe7ZKv19yPBHVs.png",
        "https://r.bing.com/rp/TX9QuO3WzcCJz1uaaSwQAz39Kb0.jpg",
    ]
    for img in normal_image_links:
        if img in bad_images:
            raise Exception("Bad images")
    # No images
    if not normal_image_links:
        raise Exception(error_no_images)
    return normal_image_links

def save_images(self, links: list, output_dir: str, file_name: str = None) -> None:
    """
    Saves images to output directory
    Parameters:
        links: list[str]
        output_dir: str
        file_name: str
    """
    if self.debug_file:
        self.debug(download_message)
    if not self.quiet:
        print(download_message)
    with contextlib.suppress(FileExistsError):
        os.mkdir(output_dir)
    try:
        fn = f"{file_name}_" if file_name else ""
        jpeg_index = 0
        for link in links:
            while os.path.exists(
                os.path.join(output_dir, f"{fn}{jpeg_index}.jpeg"),
            ):
                jpeg_index += 1
            with self.session.get(link, stream=True) as response:
                # save response to file
                response.raise_for_status()
                with open(
                    os.path.join(output_dir, f"{fn}{jpeg_index}.jpeg"),
                    "wb",
                ) as output_file:
                    for chunk in response.iter_content(chunk_size=8192):
                        output_file.write(chunk)
    except requests.exceptions.MissingSchema as url_exception:
        raise Exception(
            "Inappropriate contents found in the generated images. Please try again or try another prompt.",
        ) from url_exception

class ImageGenAsync:
"""
Image generation by Microsoft Bing
Parameters:
auth_cookie: str
Optional Parameters:
debug_file: str
quiet: bool
all_cookies: list[dict]
"""

def __init__(
    self,
    auth_cookie: str = None,
    debug_file: Union[str, None] = None,
    quiet: bool = False,
    all_cookies: List[Dict] = None,
) -> None:
    if auth_cookie is None and not all_cookies:
        raise Exception("No auth cookie provided")
    self.session = httpx.AsyncClient(
        headers=HEADERS,
        trust_env=True,
    )
    if auth_cookie:
        self.session.cookies.update({"_U": auth_cookie})
    if all_cookies:
        for cookie in all_cookies:
            self.session.cookies.update(
                {cookie["name"]: cookie["value"]},
            )
    self.quiet = quiet
    self.debug_file = debug_file
    if self.debug_file:
        self.debug = partial(debug, self.debug_file)

async def __aenter__(self):
    return self

async def __aexit__(self, *excinfo) -> None:
    await self.session.aclose()

async def get_images(self, prompt: str) -> list:
    """
    Fetches image links from Bing
    Parameters:
        prompt: str
    """
    if not self.quiet:
        print("Sending request...")
    url_encoded_prompt = requests.utils.quote(prompt)
    # https://www.bing.com/images/create?q=<PROMPT>&rt=3&FORM=GENCRE
    url = f"{BING_URL}/images/create?q={url_encoded_prompt}&rt=3&FORM=GENCRE"
    payload = f"q={url_encoded_prompt}&qs=ds"
    response = await self.session.post(
        url,
        follow_redirects=False,
        data=payload,
    )
    content = response.text
    if "this prompt has been blocked" in content.lower():
        raise Exception(
            "Your prompt has been blocked by Bing. Try to change any bad words and try again.",
        )
    if response.status_code != 302:
        # if rt4 fails, try rt3
        url = f"{BING_URL}/images/create?q={url_encoded_prompt}&rt=4&FORM=GENCRE"
        response = await self.session.post(
            url,
            follow_redirects=False,
            timeout=200,
        )
        if response.status_code != 302:
            print(f"ERROR: {response.text}")
            raise Exception("Redirect failed")
    # Get redirect URL
    redirect_url = response.headers["Location"].replace("&nfy=1", "")
    request_id = redirect_url.split("id=")[-1]
    await self.session.get(f"{BING_URL}{redirect_url}")
    # https://www.bing.com/images/create/async/results/{ID}?q={PROMPT}
    polling_url = f"{BING_URL}/images/create/async/results/{request_id}?q={url_encoded_prompt}"
    # Poll for results
    if not self.quiet:
        print("Waiting for results...")
    while True:
        if not self.quiet:
            print(".", end="", flush=True)
        # By default, timeout is 300s, change as needed
        response = await self.session.get(polling_url)
        if response.status_code != 200:
            raise Exception("Could not get results")
        content = response.text
        if content and content.find("errorMessage") == -1:
            break

        await asyncio.sleep(1)
        continue
    # Use regex to search for src=""
    image_links = regex.findall(r'src="([^"]+)"', content)
    # Remove size limit
    normal_image_links = [link.split("?w=")[0] for link in image_links]
    # Remove duplicates
    normal_image_links = list(set(normal_image_links))

    # Bad images
    bad_images = [
        "https://r.bing.com/rp/in-2zU3AJUdkgFe7ZKv19yPBHVs.png",
        "https://r.bing.com/rp/TX9QuO3WzcCJz1uaaSwQAz39Kb0.jpg",
    ]
    for im in normal_image_links:
        if im in bad_images:
            raise Exception("Bad images")
    # No images
    if not normal_image_links:
        raise Exception("No images")
    return normal_image_links

async def save_images(
    self,
    links: list,
    output_dir: str,
    file_name: str = None,
) -> None:
    """
    Saves images to output directory
    """
    if self.debug_file:
        self.debug(download_message)
    if not self.quiet:
        print(download_message)
    with contextlib.suppress(FileExistsError):
        os.mkdir(output_dir)
    try:
        fn = f"{file_name}_" if file_name else ""
        jpeg_index = 0
        for link in links:
            while os.path.exists(
                os.path.join(output_dir, f"{fn}{jpeg_index}.jpeg"),
            ):
                jpeg_index += 1
            response = await self.session.get(link)
            if response.status_code != 200:
                raise Exception("Could not download image")
            # save response to file
            with open(
                os.path.join(output_dir, f"{fn}{jpeg_index}.jpeg"),
                "wb",
            ) as output_file:
                output_file.write(response.content)
    except httpx.InvalidURL as url_exception:
        raise Exception(
            "Inappropriate contents found in the generated images. Please try again or try another prompt.",
        ) from url_exception

async def async_image_gen(
prompt: str,
output_dir: str,
u_cookie=None,
debug_file=None,
quiet=False,
all_cookies=None,
):
async with ImageGenAsync(
u_cookie,
debug_file=debug_file,
quiet=quiet,
all_cookies=all_cookies,
) as image_generator:
images = await image_generator.get_images(prompt)
await image_generator.save_images(images, output_dir=output_dir)

def main():
parser = argparse.ArgumentParser()
parser.add_argument("-U", help="Auth cookie from browser", type=str)
parser.add_argument("--cookie-file", help="File containing auth cookie", type=str)
parser.add_argument(
"--prompt",
help="Prompt to generate images for",
type=str,
required=True,
)

parser.add_argument(
    "--output-dir",
    help="Output directory",
    type=str,
    default="./output",
)

parser.add_argument(
    "--debug-file",
    help="Path to the file where debug information will be written.",
    type=str,
)

parser.add_argument(
    "--quiet",
    help="Disable pipeline messages",
    action="store_true",
)
parser.add_argument(
    "--asyncio",
    help="Run ImageGen using asyncio",
    action="store_true",
)
parser.add_argument(
    "--version",
    action="store_true",
    help="Print the version number",
)

args = parser.parse_args()

if args.version:
    print(pkg_resources.get_distribution("BingImageCreator").version)
    sys.exit()

# Load auth cookie
cookie_json = None
if args.cookie_file is not None:
    with contextlib.suppress(Exception):
        with open(args.cookie_file, encoding="utf-8") as file:
            cookie_json = json.load(file)

if args.U is None and args.cookie_file is None:
    raise Exception("Could not find auth cookie")

if not args.asyncio:
    # Create image generator
    image_generator = ImageGen(
        args.U,
        args.debug_file,
        args.quiet,
        all_cookies=cookie_json,
    )
    image_generator.save_images(
        image_generator.get_images(args.prompt),
        output_dir=args.output_dir,
    )
else:
    asyncio.run(
        async_image_gen(
            args.prompt,
            args.output_dir,
            args.U,
            args.debug_file,
            args.quiet,
            all_cookies=cookie_json,
        ),
    )

if name == "main":
main()

Sir, @manatlan , could you please tell me how I can simplify it? For example, by inputting a prompt in the chatbot along with predefined token values and specifying the output path

[manatlan]

@UseLEss213 ...
I don't understand what you want ... Sorry
If you want a minimal version, just use
#31 (comment)
If I can help, I'll do ...
But I don't understand your needs

[vansh77swami]

"Actually, sir @manatlan , I am running the BingImageGenerator in Python using Visual Studio Code, and it is working without any issues. However, the problem is that it requires running the module command like this: 'python -m BingImageCreator -U <YOUR_TOKEN> --prompt "YOUR_IMAGE_PROMPT" --output-dir "OUTPUT_IMAGE_PATH"'.

Now, I want to define the output path and token directly in the code, so I don't have to write them in the terminal. Essentially, I would like to create a chatbot in the terminal where I can provide just the prompt, and it will generate and save the images accordingly."

[manatlan]

@UseLEss213 ... ok, you are not developper !

The best way, for you ... without python modifications.
is to use "alias" under a bash/console (if you are on linux)
(on windows, it should be possible too)

alias mycommand=python -m BingImageCreator -U U8HDHDS88782838972398 --prompt "$1" --output-dir "YOUR_FOLDER"

(replace "U8HDHDS88782838972398" by your token .... and "YOUR_FOLDER" by your destination folder)

thus, in console, you'll be be able to call it like that:

$ mycommand a cat with sunglasses

[vansh77swami]

Sir @manatlan , I just want to retrieve the final image URL. I don't want to save the image itself, but instead, I need a simple code where I can input my token. Based on my prompt, it should generate images and provide me with the corresponding image URL.

[acheong08]

@UseLEss213

image_generator = ImageGen(...)
print(image_generator.get_images(prompt))

[vansh77swami]

Actually, @acheong08 , sir, can you please provide me with the entire short code that retrieves the final image URL from the generated images on the Bing site? In the code, the "u" token is already included; we just need to provide the prompt in the terminal.

sorry to bother, but i am unable to understand the code by myself.

[acheong08]

Just don't call save_images. get_images returns the URLs