diff --git a/bardapi/core.py b/bardapi/core.py index 264e78652..dd31a76c4 100644 --- a/bardapi/core.py +++ b/bardapi/core.py @@ -40,7 +40,7 @@ def __init__( session (requests.Session): Requests session object. conversation_id: ID to fetch conversational context google_translator_api_key (str): Google cloud translation API key. - language (str): Language code for translation (e.g., "en", "ko", "ja"). + language (str): Natural language code for translation (e.g., "en", "ko", "ja"). run_code (bool): Whether to directly execute the code included in the answer (Python only) token_from_browser (bool): Gets a token from the browser """ @@ -91,12 +91,12 @@ def get_answer(self, input_text: str) -> dict: "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list, "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -111,7 +111,7 @@ def get_answer(self, input_text: str) -> dict: api_key=self.google_translator_api_key ) - # Set language (optional) + # [Optional] Set language if ( self.language is not None and self.language not in ALLOWED_LANGUAGES @@ -158,7 +158,7 @@ def get_answer(self, input_text: str) -> dict: } resp_json = json.loads(resp_dict) - # Gather image links (optional) + # [Optional] gather image links images = set() try: if len(resp_json) >= 3: @@ -171,8 +171,8 @@ def get_answer(self, input_text: str) -> dict: # Parsed Answer Object parsed_answer = json.loads(resp_dict) - # Translated by Google Translator (optional) - # Unofficial for testing + # [Optional] translated by google translator + # Unofficial if ( self.language is not None and self.language not in ALLOWED_LANGUAGES @@ -183,7 +183,8 @@ def get_answer(self, input_text: str) -> dict: [x[0], [translator_to_lang.translate(x[1][0])] + x[1][1:], x[2]] for x in parsed_answer[4] ] - # Official Google Cloud Translation API + + # Official google cloud translation API elif ( self.language is not None and self.language not in ALLOWED_LANGUAGES @@ -199,24 +200,24 @@ def get_answer(self, input_text: str) -> dict: for x in parsed_answer[4] ] - # Get langcode & code (optional) + # [Optional] get program_lang & code try: - langcode = parsed_answer[4][0][1][0].split("```")[1].split("\n")[0].strip() - code = parsed_answer[4][0][1][0].split("```")[1][len(langcode) :] + program_lang = parsed_answer[4][0][1][0].split("```")[1].split("\n")[0].strip() + code = parsed_answer[4][0][1][0].split("```")[1][len(program_lang) :] except Exception: - langcode, code = None, None + program_lang, code = None, None # 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 "", + "factuality_queries": parsed_answer[3], + "text_query": 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": images, - "langCode": langcode, + "program_lang": program_lang, "code": code, "status_code": resp.status_code, } @@ -227,7 +228,7 @@ def get_answer(self, input_text: str) -> dict: ) self._reqid += 100000 - # Execute Code + # Execute code if self.run_code and bard_answer["code"] is not None: try: print(bard_answer["code"]) @@ -391,12 +392,12 @@ def ask_about_image(self, input_text: str, image: bytes, lang: str = None) -> di "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list, "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -408,7 +409,7 @@ def ask_about_image(self, input_text: str, image: bytes, lang: str = None) -> di else: translator_to_eng = GoogleTranslator(source="auto", target="en") - # Set language (optional) + # [Optional] Set language if ( (self.language is not None or lang is not None) and self.language not in ALLOWED_LANGUAGES @@ -520,12 +521,12 @@ def ask_about_image(self, input_text: str, image: bytes, lang: str = None) -> di "content": translated_content, "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 "", + "factuality_queries": parsed_answer[3], + "text_query": 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": [""], - "langCode": "", + "program_lang": "", "code": "", "status_code": resp.status_code, } @@ -538,22 +539,22 @@ def ask_about_image(self, input_text: str, image: bytes, lang: str = None) -> di return bard_answer def export_replit( - self, code: str, langcode: str = None, filename: str = None, **kwargs + self, code: str, program_lang: str = None, filename: str = None, **kwargs ): """ - Get Export URL to repl.it from code + Get export URL to repl.it from code Example: >>> token = 'xxxxxx' >>> bard = Bard(token=token) - >>> bard_answer = bard.get_answer("code python to print hello world") - >>> url = bard.export_replit(bard_answer['code'], bard_answer['langCode']) + >>> bard_answer = bard.get_answer("Give me python code to print hello world") + >>> url = bard.export_replit(bard_answer['code'], bard_answer['program_lang']) >>> print(url['url']) Args: code (str): source code - langcode (str): code language - filename (str): filename for code language + program_lang (str): programming language + filename (str): filename **kwargs: instructions, source_path Returns: dict: Answer from the Bard API in the following format: @@ -592,14 +593,15 @@ def export_replit( "rust": "main.rs", "perl": "main.pl", } + # Reference: https://github.com/jincheng9/markdown_supported_languages - if langcode not in support_langs and filename is None: + if program_lang not in support_langs and filename is None: raise Exception( - f"Language {langcode} not supported, please set filename manually." + f"Language {program_lang} not supported, please set filename manually." ) filename = ( - support_langs.get(langcode, filename) if filename is None else filename + support_langs.get(program_lang, filename) if filename is None else filename ) input_data_struct = [ [ @@ -625,11 +627,10 @@ def export_replit( timeout=self.timeout, proxies=self.proxies, ) - resp_dict = json.loads(resp.content.splitlines()[3]) print(resp_dict) url = json.loads(resp_dict[0][2])[0] - # increment request ID + # Increment request ID self._reqid += 100000 return {"url": url, "status_code": resp.status_code} @@ -652,7 +653,7 @@ def _get_snim0e(self) -> str: ) if resp.status_code != 200: raise Exception( - f"Response code not 200. Response Status is {resp.status_code}" + f"Response status code is not 200. Response Status is {resp.status_code}" ) snim0e = re.search(r"SNlM0e\":\"(.*?)\"", resp.text) if not snim0e: diff --git a/bardapi/core_async.py b/bardapi/core_async.py index cf85e1d36..1aabd4949 100644 --- a/bardapi/core_async.py +++ b/bardapi/core_async.py @@ -1,11 +1,11 @@ import os +import json +import uuid import string import random -import json import base64 -import uuid -from langdetect import detect from re import search +from langdetect import detect from httpx import AsyncClient from deep_translator import GoogleTranslator from google.cloud import translate_v2 as translate @@ -36,7 +36,7 @@ def __init__( timeout (int): Request timeout in seconds. proxies (dict): Proxy configuration for requests. google_translator_api_key (str): Google cloud translation API key. - language (str): Language code for translation (e.g., "en", "ko", "ja"). + language (str): Natural language code for translation (e.g., "en", "ko", "ja"). run_code (bool): Whether to directly execute the code included in the answer (Python only) token_from_browser (bool): Gets a token from the browser """ @@ -53,7 +53,7 @@ def __init__( self.conversation_id = "" self.response_id = "" self.choice_id = "" - # Making Httpx Async Client that will be used for all API calls + # Making httpx async client that will be used for all API calls self.client = AsyncClient( http2=True, headers=SESSION_HEADERS, @@ -90,12 +90,12 @@ async def get_answer(self, input_text: str) -> dict: "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -113,7 +113,7 @@ async def get_answer(self, input_text: str) -> dict: api_key=self.google_translator_api_key ) - # Set language (optional) + # [Optional] Set language if ( self.language is not None and self.language not in ALLOWED_LANGUAGES @@ -158,7 +158,7 @@ async def get_answer(self, input_text: str) -> dict: return {"content": f"Response Error: {resp.content}."} resp_json = json.loads(resp_dict) - # Gather image links (optional) + # [Optional] Gather image links try: images = set() if len(resp_json) >= 3: @@ -174,7 +174,7 @@ async def get_answer(self, input_text: str) -> dict: pass parsed_answer = json.loads(resp_dict) - # Translated by Google Translator (optional) + # [Optional] Translated by google translator ## Unofficial for testing if ( self.language is not None @@ -186,7 +186,7 @@ async def get_answer(self, input_text: str) -> dict: [x[0], [translator_to_lang.translate(x[1][0])] + x[1][1:], x[2]] for x in parsed_answer[4] ] - ## Official Google Cloud Translation API + ## Official google cloud translation API elif ( self.language is not None and self.language not in ALLOWED_LANGUAGES @@ -204,22 +204,22 @@ async def get_answer(self, input_text: str) -> dict: # Get code try: - langcode = parsed_answer[4][0][1][0].split("```")[1].split("\n")[0].strip() - code = parsed_answer[4][0][1][0].split("```")[1][len(langcode) :] + program_lang = parsed_answer[4][0][1][0].split("```")[1].split("\n")[0].strip() + code = parsed_answer[4][0][1][0].split("```")[1][len(program_lang) :] except Exception: - langcode, code = None, None + program_lang, code = None, None # Returned 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 "", + "factuality_queries": parsed_answer[3], + "text_query": 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": images, - "langCode": langcode, + "program_lang": program_lang, "code": code, "status_code": resp.status_code, } @@ -231,13 +231,13 @@ async def get_answer(self, input_text: str) -> dict: ) self._reqid += 100000 - # Execute Code + # Execute code if self.run_code and bard_answer["code"] is not None: try: print(bard_answer["code"]) # TODO: # find a way to handle this following warning - # EX100: Use of builtin exec function for dynamic input is insecure and can leave your application + # EX100: use of builtin exec function for dynamic input is insecure and can leave your application # open to arbitrary code execution. Found in 'exec(bard_answer['code'])'. exec(bard_answer["code"]) except Exception as e: @@ -330,7 +330,7 @@ async def _get_snim0e(self): ) if resp.status_code != 200: raise Exception( - f"Response code not 200. Response Status is {resp.status_code}" + f"Response status code is not 200. Response Status is {resp.status_code}" ) snim0e = search(r"SNlM0e\":\"(.*?)\"", resp.text) if not snim0e: @@ -422,10 +422,10 @@ async def export_conversation(self, bard_answer, title: str = "") -> str: } async def export_replit( - self, code: str, langcode: str = None, filename: str = None, **kwargs + self, code: str, program_lang: str = None, filename: str = None, **kwargs ) -> str: """ - Get Export URL to repl.it from code + Get export URL to repl.it from code Example: >>> import asyncio @@ -433,16 +433,16 @@ async def export_replit( >>> async def main(): >>> token = 'xxxxxx' >>> bard = BardAsync(token=token) - >>> bard_answer = await bard.get_answer("code python to print hello world") - >>> url = await bard.export_replit(bard_answer['code'], bard_answer['langCode']) + >>> bard_answer = await bard.get_answer("Give me python code to print hello world") + >>> url = await bard.export_replit(bard_answer['code'], bard_answer['program_lang']) >>> print(url['url']) >>> >>> asyncio.run(main()) Args: code (str): source code - langcode (str): code language - filename (str): filename for code language + program_lang (str): programming language + filename (str): filename **kwargs: instructions, source_path Returns: dict: Answer from the Bard API in the following format: @@ -483,13 +483,13 @@ async def export_replit( "perl": "main.pl", } # Reference: https://github.com/jincheng9/markdown_supported_languages - if langcode not in support_langs and filename is None: + if program_lang not in support_langs and filename is None: raise Exception( - f"Language {langcode} not supported, please set filename manually." + f"Language {program_lang} not supported, please set filename manually." ) filename = ( - support_langs.get(langcode, filename) if filename is None else filename + support_langs.get(program_lang, filename) if filename is None else filename ) input_data_struct = [ [ @@ -517,7 +517,7 @@ async def export_replit( resp_dict = json.loads(resp.content.splitlines()[3]) url = json.loads(resp_dict[0][2])[0] - # increment request ID + # Increment request ID self._reqid += 100000 return {"url": url, "status_code": resp.status_code} @@ -550,12 +550,12 @@ async def ask_about_image( "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list, "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -571,7 +571,7 @@ async def ask_about_image( else: translator_to_eng = GoogleTranslator(source="auto", target="en") - # Set language (optional) + # [Optional] Set language if ( (self.language is not None or lang is not None) and self.language not in ALLOWED_LANGUAGES @@ -682,12 +682,12 @@ async def ask_about_image( "content": translated_content, "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 "", + "factuality_queries": parsed_answer[3], + "text_query": 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": [""], - "langCode": "", + "program_lang": "", "code": "", "status_code": resp.status_code, } diff --git a/bardapi/core_cookies.py b/bardapi/core_cookies.py index 9842eeb2f..c9787375a 100644 --- a/bardapi/core_cookies.py +++ b/bardapi/core_cookies.py @@ -1,9 +1,8 @@ import os +import re import string import random -import re import requests -from bardapi.constants import SESSION_HEADERS from httpx import AsyncClient from bardapi.core import Bard from bardapi.core_async import BardAsync @@ -34,7 +33,7 @@ def __init__( proxies (dict): Proxy configuration for requests. session (requests.Session): Requests session object. google_translator_api_key (str): Google cloud translation API key. - language (str): Language code for translation (e.g., "en", "ko", "ja"). + language (str): Natural language code for translation (e.g., "en", "ko", "ja"). run_code (bool): Whether to directly execute the code included in the answer (Python only) """ self.cookie_dict = cookie_dict @@ -80,12 +79,12 @@ def get_answer(self, input_text: str) -> dict: "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list, "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -142,12 +141,12 @@ def ask_about_image(self, input_text: str, image: bytes, lang: str = None) -> di "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list, "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -181,10 +180,10 @@ def export_conversation(self, bard_answer, title: str = ""): return super().export_conversation(bard_answer, title) def export_replit( - self, code: str, langcode: str = None, filename: str = None, **kwargs + self, code: str, program_lang: str = None, filename: str = None, **kwargs ): """ - Get Export URL to repl.it from code + Get export URL to repl.it from code Example: >>> cookies = { @@ -193,13 +192,13 @@ def export_replit( >>> } >>> bard = BardCookies(cookie_dict=cookies) >>> bard_answer = bard.get_answer("code python to print hello world") - >>> url = bard.export_replit(bard_answer['code'], bard_answer['langCode']) + >>> url = bard.export_replit(bard_answer['code'], bard_answer['program_lang']) >>> print(url['url']) Args: code (str): source code - langcode (str): code language - filename (str): filename for code language + program_lang (str): programming language + filename (str): filename for code **kwargs: instructions, source_path Returns: dict: Answer from the Bard API in the following format: @@ -208,7 +207,7 @@ def export_replit( "status_code": int } """ - return super().export_replit(code, langcode, filename, **kwargs) + return super().export_replit(code, program_lang, filename, **kwargs) def _get_snim0e(self) -> str: """ @@ -224,7 +223,7 @@ def _get_snim0e(self) -> str: ) if resp.status_code != 200: raise Exception( - f"Response code not 200. Response Status is {resp.status_code}" + f"Response status code is not 200. Response Status is {resp.status_code}" ) snim0e = re.search(r"SNlM0e\":\"(.*?)\"", resp.text) if not snim0e: @@ -256,7 +255,7 @@ def __init__( timeout (int): Request timeout in seconds. proxies (dict): Proxy configuration for requests. google_translator_api_key (str): Google cloud translation API key. - language (str): Language code for translation (e.g., "en", "ko", "ja"). + language (str): Natural language code for translation (e.g., "en", "ko", "ja"). run_code (bool): Whether to directly execute the code included in the answer (Python only) """ self.cookie_dict = cookie_dict @@ -266,7 +265,7 @@ def __init__( self.conversation_id = "" self.response_id = "" self.choice_id = "" - # Making Httpx Async Client that will be used for all API calls + # Making httpx async client that will be used for all API calls self.client = AsyncClient( http2=True, cookies=self.cookie_dict, @@ -306,12 +305,12 @@ async def get_answer(self, input_text: str) -> dict: "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -383,12 +382,12 @@ async def ask_about_image( "content": str, "conversation_id": str, "response_id": str, - "factualityQueries": list, - "textQuery": str, + "factuality_queries": list, + "text_query": str, "choices": list, "links": list, "images": set, - "langCode": str, + "program_lang": str, "code": str, "status_code": int } @@ -427,10 +426,10 @@ async def export_conversation(self, bard_answer, title: str = "") -> str: return await super().export_conversation(bard_answer, title) async def export_replit( - self, code: str, langcode: str = None, filename: str = None, **kwargs + self, code: str, program_lang: str = None, filename: str = None, **kwargs ) -> str: """ - Get Export URL to repl.it from code + Get export URL to repl.it from code Example: >>> import asyncio @@ -442,15 +441,15 @@ async def export_replit( >>> } >>> bard = BardAsyncCookies(cookie_dict=cookies) >>> bard_answer = await bard.get_answer("code python to print hello world") - >>> url = await bard.export_replit(bard_answer['code'], bard_answer['langCode']) + >>> url = await bard.export_replit(bard_answer['code'], bard_answer['program_lang']) >>> print(url['url']) >>> >>> asyncio.run(main()) Args: code (str): source code - langcode (str): code language - filename (str): filename for code language + program_lang (str): programming language + filename (str): filename **kwargs: instructions, source_path Returns: dict: Answer from the Bard API in the following format: @@ -460,7 +459,7 @@ async def export_replit( } """ - return await super().export_replit(code, langcode, filename, **kwargs) + return await super().export_replit(code, program_lang, filename, **kwargs) async def _get_snim0e(self): """ @@ -478,7 +477,7 @@ async def _get_snim0e(self): ) if resp.status_code != 200: raise Exception( - f"Response code not 200. Response Status is {resp.status_code}" + f"Response status code is not 200. Response Status is {resp.status_code}" ) snim0e = re.search(r"SNlM0e\":\"(.*?)\"", resp.text) if not snim0e: diff --git a/translate_to/core_async.py b/translate_to/core_async.py deleted file mode 100644 index 27275feda..000000000 --- a/translate_to/core_async.py +++ /dev/null @@ -1,238 +0,0 @@ -import json -import os -import random -import string -from re import search - -from bardapi.constants import ALLOWED_LANGUAGES, SESSION_HEADERS -from deep_translator import GoogleTranslator -from httpx import AsyncClient - - -class BardAsync: - """ - Bard class for interacting with the Bard API using httpx[http2] - Tested and working (could break in the future, if not kept up to date) - """ - - def __init__( - self, - token: str = None, - timeout: int = 20, - proxies: dict = None, - language: str = None, - run_code: bool = False, - ): - """ - Initialize the Bard instance. - - Args: - token (str): Bard API token. - timeout (int): Request timeout in seconds. - proxies (dict): Proxy configuration for requests. - language (str): Language code for translation (e.g., "en", "ko", "ja"). - """ - self.token = token or os.getenv("_BARD_API_KEY") - self.proxies = proxies - self.timeout = timeout - self._reqid = int("".join(random.choices(string.digits, k=4))) - self.conversation_id = "" - self.response_id = "" - self.choice_id = "" - # Making Httpx Async Client that will be used for all API calls - self.client = AsyncClient( - http2=True, - headers=SESSION_HEADERS, - cookies={"__Secure-1PSID": self.token}, - timeout=self.timeout, - proxies=self.proxies, - ) - self.SNlM0e = self._get_snim0e() - self.language = language or os.getenv("_BARD_API_LANG", "en") - self.run_code = run_code or False - - async def get_answer(self, input_text: str) -> dict: - """ - Get an answer from the Bard API for the given input text. - - Example: - >>> token = 'xxxxxxxxxx' - >>> bard = BardAsync(token=token) - >>> response = await bard.get_answer("나와 내 동년배들이 좋아하는 뉴진스에 대해서 알려줘") - >>> print(response['content']) - - Args: - input_text (str): Input text for the query. - - Returns: - 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 - } - """ - if not isinstance(self.SNlM0e, str): - self.SNlM0e = await self.SNlM0e - params = { - "bl": "boq_assistant-bard-web-server_20230419.00_p1", - "_reqid": str(self._reqid), - "rt": "c", - } - - # Set language (optional) - if self.language is not None and self.language not in ALLOWED_LANGUAGES: - translator_to_eng = GoogleTranslator(source="auto", target="en") - input_text = translator_to_eng.translate(input_text) - - # Make post data structure and insert prompt - input_text_struct = [ - [input_text], - None, - [self.conversation_id, self.response_id, self.choice_id], - ] - data = { - "f.req": json.dumps([None, json.dumps(input_text_struct)]), - "at": self.SNlM0e, - } - - resp = await self.client.post( - "https://bard.google.com/_/BardChatUi/data/assistant.lamda.BardFrontendService/StreamGenerate", - params=params, - data=data, - timeout=self.timeout, - follow_redirects=True, - headers=SESSION_HEADERS, - cookies={"__Secure-1PSID": self.token}, - ) - - # 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}."} - resp_json = json.loads(resp_dict) - - # Gather image links - images = set() - if len(resp_json) >= 3: - if len(resp_json[4][0]) >= 4 and resp_json[4][0][4] is not None: - for img in resp_json[4][0][4]: - try: - images.add(img[0][0][0]) - except Exception as e: - # TODO: - # handle exception using logging instead - print(f"Unable to parse image from the response: {e}") - parsed_answer = json.loads(resp_dict) - - # Translated by Google Translator (optional) - if self.language is not None and self.language not in ALLOWED_LANGUAGES: - translator_to_lang = GoogleTranslator(source="auto", target=self.language) - parsed_answer[0][0] = translator_to_lang.translate(parsed_answer[0][0]) - parsed_answer[4] = [ - (x[0], translator_to_lang.translate(x[1][0])) for x in parsed_answer[4] - ] - - # Get code - try: - code = parsed_answer[0][0].split("```")[1][6:] - except Exception as e: - # TODO: - # handle exception using logging instead - code = None - print(f"Unable to parse answer from the response: {e}") - - # Returned dictionary object - bard_answer = { - "content": parsed_answer[0][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": self._extract_links(parsed_answer[4]), - "images": images, - "code": code, - "status_code": resp.status_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 - - # Execute Code - if self.run_code and bard_answer["code"] is not None: - try: - print(bard_answer["code"]) - # TODO: - # find a way to handle this following warning - # EX100: Use of builtin exec function for dynamic input is insecure and can leave your application - # open to arbitrary code execution. Found in 'exec(bard_answer['code'])'. - exec(bard_answer["code"]) - except Exception as e: - # TODO: - # handle exception using logging instead - print(f"Unable to execute the code: {e}") - - return bard_answer - - async def _get_snim0e(self): - """ - The _get_snim0e function is used to get the SNlM0e value from the Bard website. - - The function uses a regular expression to search for the SNlM0e value in the response text. - If it finds it, then it returns that value. - - :param self: Represent the instance of the class - :return: (`str`) The snlm0e value - """ - if not self.token or self.token[-1] != ".": - raise Exception( - "__Secure-1PSID value must end with a single dot. Enter correct __Secure-1PSID value." - ) - - resp = await self.client.get( - "https://bard.google.com/", timeout=self.timeout, follow_redirects=True - ) - if resp.status_code != 200: - raise Exception( - f"Response code not 200. Response Status is {resp.status_code}" - ) - snim0e = search(r"SNlM0e\":\"(.*?)\"", resp.text) - if not snim0e: - raise Exception( - "SNlM0e value not found in response. Check __Secure-1PSID value." - ) - 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