diff --git a/bardapi/common.py b/bardapi/common.py new file mode 100644 index 000000000..ba39443e2 --- /dev/null +++ b/bardapi/common.py @@ -0,0 +1,111 @@ +# Common function in async and sync core, core cookie modes +import requests +import browser_cookie3 + +IMG_UPLOAD_HEADERS = { + "authority": "content-push.googleapis.com", + "accept": "*/*", + "accept-language": "en-US,en;q=0.7", + "authorization": "Basic c2F2ZXM6cyNMdGhlNmxzd2F2b0RsN3J1d1U=", # constant authorization key + "content-type": "application/x-www-form-urlencoded;charset=UTF-8", + "origin": "https://bard.google.com", + "push-id": "feeds/mcudyrk2a4khkz", # constant + "referer": "https://bard.google.com/", + "x-goog-upload-command": "start", + "x-goog-upload-header-content-length": "", + "x-goog-upload-protocol": "resumable", + "x-tenant-id": "bard-storage", +} + + +def extract_links(data: list) -> list: + """ + Extract links from the given data. + + Args: + data: Data to extract links from. + + Returns: + list: Extracted links. + """ + links = [] + if isinstance(data, list): + for item in data: + if isinstance(item, list): + # recursive + links.extend(extract_links(item)) + elif ( + isinstance(item, str) + and item.startswith("http") + and "favicon" not in item + ): + links.append(item) + return links + + +def upload_image(image: bytes, filename="Photo.jpg"): + """ + Upload image into bard bucket on Google API, do not need session. + + Returns: + str: relative URL of image. + """ + resp = requests.options("https://content-push.googleapis.com/upload/") + resp.raise_for_status() + size = len(image) + + headers = IMG_UPLOAD_HEADERS + headers["size"] = str(size) + headers["x-goog-upload-command"] = "start" + + data = f"File name: {filename}" + resp = requests.post( + "https://content-push.googleapis.com/upload/", headers=headers, data=data + ) + resp.raise_for_status() + upload_url = resp.headers["X-Goog-Upload-Url"] + resp = requests.options(upload_url, headers=headers) + resp.raise_for_status() + headers["x-goog-upload-command"] = "upload, finalize" + + # It can be that we need to check returned offset + headers["X-Goog-Upload-Offset"] = "0" + resp = requests.post(upload_url, headers=headers, data=image) + resp.raise_for_status() + return resp.text + + +def extract_bard_cookie(): + """ + Extract token cookie from browser. + Supports all modern web browsers and OS + + + Returns: + str: __Secure-1PSID cookie value + """ + + # browser_cookie3.load is similar function but it's broken + # So here we manually search accross all browsers + browsers = [ + browser_cookie3.chrome, + browser_cookie3.chromium, + browser_cookie3.opera, + browser_cookie3.opera_gx, + browser_cookie3.brave, + browser_cookie3.edge, + browser_cookie3.vivaldi, + browser_cookie3.firefox, + browser_cookie3.librewolf, + browser_cookie3.safari, + ] + for browser_fn in browsers: + # if browser isn't installed browser_cookie3 raises exception + # hence we need to ignore it and try to find the right one + try: + cj = browser_fn(domain_name=".google.com") + for cookie in cj: + if cookie.name == "__Secure-1PSID" and cookie.value.endswith("."): + return cookie.value + except: + continue diff --git a/bardapi/constants.py b/bardapi/constants.py index 6f4669b36..2da6a6ce7 100644 --- a/bardapi/constants.py +++ b/bardapi/constants.py @@ -13,18 +13,3 @@ "Origin": "https://bard.google.com", "Referer": "https://bard.google.com/", } - -IMG_UPLOAD_HEADERS = { - "authority": "content-push.googleapis.com", - "accept": "*/*", - "accept-language": "en-US,en;q=0.7", - "authorization": "Basic c2F2ZXM6cyNMdGhlNmxzd2F2b0RsN3J1d1U=", # constant authorization key - "content-type": "application/x-www-form-urlencoded;charset=UTF-8", - "origin": "https://bard.google.com", - "push-id": "feeds/mcudyrk2a4khkz", # constant - "referer": "https://bard.google.com/", - "x-goog-upload-command": "start", - "x-goog-upload-header-content-length": "", - "x-goog-upload-protocol": "resumable", - "x-tenant-id": "bard-storage", -} diff --git a/bardapi/core.py b/bardapi/core.py index e168c672c..c1b6d2dec 100644 --- a/bardapi/core.py +++ b/bardapi/core.py @@ -6,10 +6,10 @@ import requests import base64 import uuid -import browser_cookie3 from deep_translator import GoogleTranslator from google.cloud import translate_v2 as translate -from bardapi.constants import ALLOWED_LANGUAGES, SESSION_HEADERS, IMG_UPLOAD_HEADERS +from bardapi.constants import ALLOWED_LANGUAGES, SESSION_HEADERS +from bardapi.common import extract_links, upload_image, extract_bard_cookie class Bard: @@ -44,7 +44,7 @@ def __init__( """ self.token = token or os.getenv("_BARD_API_KEY") if not self.token and token_from_browser: - self.token = self._extract_bard_cookie() + self.token = extract_bard_cookie() if not self.token: raise Exception( "\nCan't extract cookie from browsers.\nPlease sign in first at\nhttps://accounts.google.com/v3/signin/identifier?followup=https://bard.google.com/&flowName=GlifWebSignIn&flowEntry=ServiceLogin" @@ -210,7 +210,7 @@ def get_answer(self, input_text: str) -> dict: "factualityQueries": parsed_answer[3], "textQuery": parsed_answer[2][0] if parsed_answer[2] else "", "choices": [{"id": x[0], "content": x[1]} for x in parsed_answer[4]], - "links": self._extract_links(parsed_answer[4]), + "links": extract_links(parsed_answer[4]), "images": images, "langCode": langcode, "code": code, @@ -385,7 +385,7 @@ def ask_about_image(self, input_text: str, image: bytes, lang="en-GB") -> dict: """ # Supported format: jpeg, png, webp - image_url = self._upload_image(image) + image_url = upload_image(image) input_data_struct = [ None, @@ -437,7 +437,7 @@ def ask_about_image(self, input_text: str, image: bytes, lang="en-GB") -> dict: "factualityQueries": parsed_answer[3], "textQuery": parsed_answer[2][0] if parsed_answer[2] else "", "choices": [{"id": x[0], "content": x[1]} for x in parsed_answer[4]], - "links": self._extract_links(parsed_answer[4]), + "links": extract_links(parsed_answer[4]), "images": [""], "code": "", } @@ -449,37 +449,6 @@ def ask_about_image(self, input_text: str, image: bytes, lang="en-GB") -> dict: self._reqid += 100000 return bard_answer - def _upload_image(self, image: bytes, filename="Photo.jpg"): - """ - Upload image into bard bucket on Google API - - Returns: - str: relative URL of image. - """ - resp = requests.options("https://content-push.googleapis.com/upload/") - resp.raise_for_status() - size = len(image) - - headers = IMG_UPLOAD_HEADERS - headers["size"] = str(size) - headers["x-goog-upload-command"] = "start" - - data = "File name: Photo.jpg" - resp = requests.post( - "https://content-push.googleapis.com/upload/", headers=headers, data=data - ) - resp.raise_for_status() - upload_url = resp.headers["X-Goog-Upload-Url"] - resp = requests.options(upload_url, headers=headers) - resp.raise_for_status() - headers["x-goog-upload-command"] = "upload, finalize" - - # It can be that we need to check returned offset - headers["X-Goog-Upload-Offset"] = "0" - resp = requests.post(upload_url, headers=headers, data=image) - resp.raise_for_status() - return resp.text - def export_replit( self, code: str, langcode: str = None, filename: str = None, **kwargs ): @@ -596,61 +565,3 @@ def _get_snim0e(self) -> str: "SNlM0e value not found. Double-check __Secure-1PSID value or pass it as token='xxxxx'." ) return snim0e.group(1) - - def _extract_links(self, data: list) -> list: - """ - Extract links from the given data. - - Args: - data: Data to extract links from. - - Returns: - list: Extracted links. - """ - links = [] - if isinstance(data, list): - for item in data: - if isinstance(item, list): - links.extend(self._extract_links(item)) - elif ( - isinstance(item, str) - and item.startswith("http") - and "favicon" not in item - ): - links.append(item) - return links - - def _extract_bard_cookie(self): - """ - Extract token cookie from browser. - Supports all modern web browsers and OS - - - Returns: - str: __Secure-1PSID cookie value - """ - - # browser_cookie3.load is similar function but it's broken - # So here we manually search accross all browsers - browsers = [ - browser_cookie3.chrome, - browser_cookie3.chromium, - browser_cookie3.opera, - browser_cookie3.opera_gx, - browser_cookie3.brave, - browser_cookie3.edge, - browser_cookie3.vivaldi, - browser_cookie3.firefox, - browser_cookie3.librewolf, - browser_cookie3.safari, - ] - for browser_fn in browsers: - # if browser isn't installed browser_cookie3 raises exception - # hence we need to ignore it and try to find the right one - try: - cj = browser_fn(domain_name=".google.com") - for cookie in cj: - if cookie.name == "__Secure-1PSID" and cookie.value.endswith("."): - return cookie.value - except: - continue diff --git a/bardapi/core_async.py b/bardapi/core_async.py index cb6289631..163c9f2af 100644 --- a/bardapi/core_async.py +++ b/bardapi/core_async.py @@ -2,8 +2,10 @@ import string import random import json +import uuid from re import search from bardapi.constants import ALLOWED_LANGUAGES, SESSION_HEADERS +from bardapi.common import extract_links, upload_image, extract_bard_cookie from deep_translator import GoogleTranslator from google.cloud import translate_v2 as translate from httpx import AsyncClient @@ -22,6 +24,7 @@ def __init__( google_translator_api_key: str = None, language: str = None, run_code: bool = False, + token_from_browser=False, ): """ Initialize the Bard instance. @@ -33,6 +36,12 @@ def __init__( language (str): Language code for translation (e.g., "en", "ko", "ja"). """ self.token = token or os.getenv("_BARD_API_KEY") + if not self.token and token_from_browser: + self.token = extract_bard_cookie() + if not self.token: + raise Exception( + "\nCan't extract cookie from browsers.\nPlease sign in first at\nhttps://accounts.google.com/v3/signin/identifier?followup=https://bard.google.com/&flowName=GlifWebSignIn&flowEntry=ServiceLogin" + ) self.proxies = proxies self.timeout = timeout self._reqid = int("".join(random.choices(string.digits, k=4))) @@ -195,7 +204,7 @@ async def get_answer(self, input_text: str) -> dict: "factualityQueries": parsed_answer[3], "textQuery": parsed_answer[2][0] if parsed_answer[2] else "", "choices": [{"id": x[0], "content": x[1]} for x in parsed_answer[4]], - "links": self._extract_links(parsed_answer[4]), + "links": extract_links(parsed_answer[4]), "images": images, "code": code, "status_code": resp.status_code, @@ -253,25 +262,102 @@ async def _get_snim0e(self): ) return snim0e.group(1) - def _extract_links(self, data: list) -> list: + async def ask_about_image( + self, input_text: str, image: bytes, lang="en-GB" + ) -> dict: """ - Extract links from the given data. + Send Bard image along with question and get answer async mode + + Example: + >>> token = 'xxxxxxxxxx' + >>> bard = Bard(token=token) + >>> image = open('image.jpg', 'rb').read() + >>> bard_answer = bard.analyze_image("what is in the image?", image) Args: - data: Data to extract links from. + input_text (str): Input text for the query. + image (bytes): Input image bytes for the query, support image types: jpeg, png, webp + lang (str): Language to use. Returns: - list: Extracted links. + dict: Answer from the Bard API in the following format: + { + "content": str, + "conversation_id": str, + "response_id": str, + "factualityQueries": list, + "textQuery": str, + "choices": list, + "links": list, + "imgaes": set, + "code": str + } """ - links = [] - if isinstance(data, list): - for item in data: - if isinstance(item, list): - links.extend(self._extract_links(item)) - elif ( - isinstance(item, str) - and item.startswith("http") - and "favicon" not in item - ): - links.append(item) - return links + + if not isinstance(self.SNlM0e, str): + self.SNlM0e = await self.SNlM0e + + # Supported format: jpeg, png, webp + image_url = upload_image(image) + + input_data_struct = [ + None, + [ + [input_text, 0, None, [[[image_url, 1], "uploaded_photo.jpg"]]], + [lang], + ["", "", ""], + "", # Unknown random string value (1000 characters +) + uuid.uuid4().hex, # Should be random uuidv4 (32 characters) + None, + [1], + 0, + [], + [], + ], + ] + params = { + "bl": "boq_assistant-bard-web-server_20230716.16_p2", + "_reqid": str(self._reqid), + "rt": "c", + } + input_data_struct[1] = json.dumps(input_data_struct[1]) + data = { + "f.req": json.dumps(input_data_struct), + "at": self.SNlM0e, + } + + resp = await self.client.post( + "https://bard.google.com/_/BardChatUi/data/assistant.lamda.BardFrontendService/StreamGenerate", + params=params, + data=data, + ) + + # Post-processing of response + resp_dict = json.loads(resp.content.splitlines()[3])[0][2] + if not resp_dict: + return { + "content": f"Response Error: {resp.content}. " + f"\nUnable to get response." + f"\nPlease double-check the cookie values and verify your network environment or google account." + } + parsed_answer = json.loads(resp_dict) + + # Returnd dictionary object + bard_answer = { + "content": parsed_answer[4][0][1][0], + "conversation_id": parsed_answer[1][0], + "response_id": parsed_answer[1][1], + "factualityQueries": parsed_answer[3], + "textQuery": parsed_answer[2][0] if parsed_answer[2] else "", + "choices": [{"id": x[0], "content": x[1]} for x in parsed_answer[4]], + "links": extract_links(parsed_answer[4]), + "images": [""], + "code": "", + } + self.conversation_id, self.response_id, self.choice_id = ( + bard_answer["conversation_id"], + bard_answer["response_id"], + bard_answer["choices"][0]["id"], + ) + self._reqid += 100000 + return bard_answer diff --git a/bardapi/core_cookies.py b/bardapi/core_cookies.py index 2b9fc4df6..266bada7e 100644 --- a/bardapi/core_cookies.py +++ b/bardapi/core_cookies.py @@ -6,6 +6,7 @@ import requests from deep_translator import GoogleTranslator from bardapi.constants import ALLOWED_LANGUAGES, SESSION_HEADERS +from bardapi.common import extract_links class BardCookies: @@ -148,7 +149,7 @@ def get_answer(self, input_text: str) -> dict: "factualityQueries": parsed_answer[3], "textQuery": parsed_answer[2][0] if parsed_answer[2] else "", "choices": [{"id": x[0], "content": x[1]} for x in parsed_answer[4]], - "links": self._extract_links(parsed_answer[4]), + "links": extract_links(parsed_answer[4]), "images": images, "code": code, } @@ -192,29 +193,6 @@ def _get_snim0e(self) -> str: ) return snim0e.group(1) - def _extract_links(self, data: list) -> list: - """ - Extract links from the given data. - - Args: - data: Data to extract links from. - - Returns: - list: Extracted links. - """ - links = [] - if isinstance(data, list): - for item in data: - if isinstance(item, list): - links.extend(self._extract_links(item)) - elif ( - isinstance(item, str) - and item.startswith("http") - and "favicon" not in item - ): - links.append(item) - return links - # def auth(self): #Idea Contribution # url = 'https://bard.google.com' # driver_path = "/path/to/chromedriver"