Open manatlan opened 1 year ago
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
@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 ?!
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
@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
?
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
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. 🤔
@manatlan this is the code of my bing image generator ai-
to run this i ahve to tun this command - python -m BingImageCreator -U
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")
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_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
@UseLEss213 ... I don't understand what you want ... Sorry If you want a minimal version, just use https://github.com/acheong08/BingImageCreator/issues/31#issuecomment-1561716705 If I can help, I'll do ... But I don't understand your needs
"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
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."
@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
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.
@UseLEss213
image_generator = ImageGen(...)
print(image_generator.get_images(prompt))
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.
Just don't call save_images
. get_images
returns the URLs
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)