diff --git a/hfAgent/FreeLLM/BardChatAPI.py b/hfAgent/FreeLLM/BardChatAPI.py deleted file mode 100644 index d44303c..0000000 --- a/hfAgent/FreeLLM/BardChatAPI.py +++ /dev/null @@ -1,59 +0,0 @@ -from Bard import Chatbot -import asyncio - -import requests -from langchain.llms.base import LLM -from typing import Optional, List, Mapping, Any -import pydantic -import os -from langchain import PromptTemplate, LLMChain -from time import sleep - - - -class BardChat(LLM): - - history_data: Optional[List] = [] - cookie : Optional[str] - chatbot : Optional[Chatbot] = None - - - @property - def _llm_type(self) -> str: - return "custom" - - async def call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - if stop is not None: - pass - #raise ValueError("stop kwargs are not permitted.") - #cookie is a must check - if self.chatbot is None: - if self.cookie is None: - raise ValueError("Need a COOKIE , check https://github.com/acheong08/EdgeGPT/tree/master#getting-authentication-required for get your COOKIE AND SAVE") - else: - #if self.chatbot == None: - self.chatbot = Chatbot(self.cookie) - - response = self.chatbot.ask(prompt) - #print(response) - response_text = response['content'] - #add to history - self.history_data.append({"prompt":prompt,"response":response_text}) - - return response_text - - def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - return asyncio.run(self.call(prompt=prompt)) - - @property - def _identifying_params(self) -> Mapping[str, Any]: - """Get the identifying parameters.""" - return {"model": "BardCHAT", "cookie": self.cookie} - - - -#llm = BardChat(cookie = "YOURCOOKIE") #for start new chat - -#print(llm("Hello, how are you?")) -#print(llm("what is AI?")) -#print(llm("Can you resume your previus answer?")) #now memory work well diff --git a/hfAgent/FreeLLM/BingChatAPI.py b/hfAgent/FreeLLM/BingChatAPI.py deleted file mode 100644 index b35f37e..0000000 --- a/hfAgent/FreeLLM/BingChatAPI.py +++ /dev/null @@ -1,104 +0,0 @@ -from EdgeGPT import Chatbot, ConversationStyle -import asyncio - -import requests -from langchain.llms.base import LLM -from typing import Optional, List, Mapping, Any -import pydantic -import os -from langchain import PromptTemplate, LLMChain -from time import sleep - - - -class BingChat(LLM): - - history_data: Optional[List] = [] - cookiepath : Optional[str] - chatbot : Optional[Chatbot] = None - conversation_style : Optional[str] - conversation_style_on : Optional[ConversationStyle] = ConversationStyle.precise - search_result : Optional[bool] = False - - @property - def _llm_type(self) -> str: - return "custom" - - def select_conversation(self, conversation_style: str): - if conversation_style == "precise": - self.conversation_style_on = ConversationStyle.precise - elif conversation_style == "creative": - self.conversation_style_on = ConversationStyle.creative - elif conversation_style == "balanced": - self.conversation_style_on = ConversationStyle.balanced - else: - raise ValueError("conversation_style must be precise, creative or balaced") - self.conversation_style = conversation_style - - async def call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - if stop is not None: - raise ValueError("stop kwargs are not permitted.") - #cookiepath is a must check - if self.chatbot is None: - if self.cookiepath is None: - raise ValueError("Need a COOKIE , check https://github.com/acheong08/EdgeGPT/tree/master#getting-authentication-required for get your COOKIE AND SAVE") - else: - #if self.chatbot == None: - self.chatbot = await Chatbot.create(cookie_path=self.cookiepath) - - if self.conversation_style is not None: - self.conversation_style_on = self.select_conversation(self.conversation_style) - - response = await self.chatbot.ask(prompt=prompt, conversation_style=self.conversation_style, search_result=self.search_result) - """ - this is a sample response. - {'type': 2, 'invocationId': '0', - 'item': {'messages': [{'text': 'Hello, how are you?', 'author': 'user', 'from': {'id': '985157152860707', 'name': None}, 'createdAt': '2023-05-03T19:51:39.5491558+00:00', 'timestamp': '2023-05-03T19:51:39.5455787+00:00', 'locale': 'en-us', 'market': 'en-us', 'region': 'us', 'messageId': '87f90c57-b2ad-4b3a-b24f-99f633f5332f', 'requestId': '87f90c57-b2ad-4b3a-b24f-99f633f5332f', 'nlu': {'scoredClassification': {'classification': 'CHAT_GPT', 'score': None}, 'classificationRanking': [{'classification': 'CHAT_GPT', 'score': None}], 'qualifyingClassifications': None, 'ood': None, 'metaData': None, 'entities': None}, 'offense': 'None', 'feedback': {'tag': None, 'updatedOn': None, 'type': 'None'}, 'contentOrigin': 'cib', 'privacy': None, 'inputMethod': 'Keyboard'}, {'text': "Hello! I'm doing well, thank you. How can I assist you today?", 'author': 'bot', 'createdAt': '2023-05-03T19:51:41.5176164+00:00', 'timestamp': '2023-05-03T19:51:41.5176164+00:00', 'messageId': '1d013e71-408b-4031-a131-2f5c009fe938', 'requestId': '87f90c57-b2ad-4b3a-b24f-99f633f5332f', 'offense': 'None', 'adaptiveCards': [{'type': 'AdaptiveCard', 'version': '1.0', 'body': [{'type': 'TextBlock', 'text': "Hello! I'm doing well, thank you. How can I assist you today?\n", 'wrap': True}]}], - 'sourceAttributions': [], - 'feedback': {'tag': None, 'updatedOn': None, 'type': 'None'}, - 'contentOrigin': 'DeepLeo', - 'privacy': None, - 'suggestedResponses': [{'text': 'What is the weather like today?', 'author': 'user', 'createdAt': '2023-05-03T19:51:42.7502696+00:00', 'timestamp': '2023-05-03T19:51:42.7502696+00:00', 'messageId': 'cd7a84d3-f9bc-47ff-9897-077b2de12e21', 'messageType': 'Suggestion', 'offense': 'Unknown', 'feedback': {'tag': None, 'updatedOn': None, 'type': 'None'}, 'contentOrigin': 'DeepLeo', 'privacy': None}, {'text': 'What is the latest news?', 'author': 'user', 'createdAt': '2023-05-03T19:51:42.7502739+00:00', 'timestamp': '2023-05-03T19:51:42.7502739+00:00', 'messageId': 'b611632a-9a8e-42de-86eb-8eb3b7b8ddbb', 'messageType': 'Suggestion', 'offense': 'Unknown', 'feedback': {'tag': None, 'updatedOn': None, 'type': 'None'}, 'contentOrigin': 'DeepLeo', 'privacy': None}, {'text': 'Tell me a joke.', 'author': 'user', 'createdAt': '2023-05-03T19:51:42.7502743+00:00', 'timestamp': '2023-05-03T19:51:42.7502743+00:00', 'messageId': '70232e45-d7e8-4d77-83fc-752b3cd3355c', 'messageType': 'Suggestion', 'offense': 'Unknown', 'feedback': {'tag': None, 'updatedOn': None, 'type': 'None'}, 'contentOrigin': 'DeepLeo', 'privacy': None}], 'spokenText': 'How can I assist you today?'}], 'firstNewMessageIndex': 1, 'defaultChatName': None, 'conversationId': '51D|BingProd|3E1274E188350D7BE273FFE95E02DD2984DAB52F95260300D0A2937162F98FDA', 'requestId': '87f90c57-b2ad-4b3a-b24f-99f633f5332f', 'conversationExpiryTime': '2023-05-04T01:51:42.8260286Z', 'shouldInitiateConversation': True, 'telemetry': {'metrics': None, 'startTime': '2023-05-03T19:51:39.5456555Z'}, 'throttling': {'maxNumUserMessagesInConversation': 20, 'numUserMessagesInConversation': 1}, 'result': {'value': 'Success', 'serviceVersion': '20230501.30'}}} - """ - response_messages = response.get("item", {}).get("messages", []) - response_text = response_messages[1].get("text", "") - - if response_text == "": - hidden_text = response_messages[1].get("hiddenText", "") - print(">>>> [DEBBUGGER] hidden_text = " + str(hidden_text) + " [DEBBUGGER] <<<<") - print(">>>> [DEBBUGGER] BING CHAT dont is open Like CHATGPT , BingCHAT have refused to respond. [DEBBUGGER] <<<<") - response_text = hidden_text - """ - # reset the chatbot and remake the call - print("[DEBUGGER] Chatbot failed to respond. Resetting and trying again. [DEBUGGER]") - print("[ INFO DEBUGGER ] \n\n" + str(response) + "\n\n\n") - sleep(10) - self.chatbot = await Chatbot.create(cookie_path=self.cookiepath) - sleep(2) - response = await self.chatbot.ask(prompt=prompt) - response_messages = response.get("item", {}).get("messages", []) - response_text = response_messages[1].get("text", "") - """ - - #add to history - self.history_data.append({"prompt":prompt,"response":response_text}) - - return response_text - - def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - return asyncio.run(self.call(prompt=prompt)) - - @property - def _identifying_params(self) -> Mapping[str, Any]: - """Get the identifying parameters.""" - return {"model": "BingCHAT", "cookiepath": self.cookiepath} - - - -#llm = BingChat(cookiepath = "YOUR-COOKIE") #for start new chat -#llm = BingChat(cookiepath = "YOUR-COOKIE", conversation_style = "precise") #precise, creative or balaced -#llm = BingChat(cookiepath = "YOUR-COOKIE" , conversation_style = "precise" , search_result=True) #with web access - -#print(llm("Hello, how are you?")) -#print(llm("what is AI?")) -#print(llm("Can you resume your previus answer?")) #now memory work well diff --git a/hfAgent/FreeLLM/ChatGPTAPI-unsafe.py b/hfAgent/FreeLLM/ChatGPTAPI-unsafe.py deleted file mode 100644 index ccc8538..0000000 --- a/hfAgent/FreeLLM/ChatGPTAPI-unsafe.py +++ /dev/null @@ -1,75 +0,0 @@ - -from revChatGPT.V1 import Chatbot -import requests -from langchain.llms.base import LLM -from typing import Optional, List, Mapping, Any -import pydantic -import os -from langchain import PromptTemplate, LLMChain -from time import sleep - - - -class ChatGPT(LLM): - - history_data: Optional[List] = [] - token : Optional[str] - chatbot : Optional[Chatbot] = None - call : int = 0 - conversation : Optional[str] = "" - prev_message : str = "" - - #### WARNING : for each api call this library will create a new chat on chat.openai.com - - - @property - def _llm_type(self) -> str: - return "custom" - - def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - if stop is not None: - raise ValueError("stop kwargs are not permitted.") - #token is a must check - if self.token is None: - raise ValueError("Need a token , check https://chat.openai.com/api/auth/session for get your token") - else: - if self.conversation == "": - self.chatbot = Chatbot(config={"access_token": self.token}) - elif self.conversation != "" and self.prev_message == "": - self.chatbot = Chatbot(config={"access_token": self.token}, conversation_id=self.conversation) - elif self.conversation != "" and self.prev_message != "": - self.chatbot = Chatbot(config={"access_token": self.token}, conversation_id=self.conversation, parent_id=self.prev_message) - else: - raise ValueError("Something went wrong") - - response = "" - # OpenAI: 50 requests / hour for each account - if self.call >= 45: - raise ValueError("You have reached the maximum number of requests per hour ! Help me to Improve. Abusing this tool is at your own risk") - else: - sleep(2) - for data in self.chatbot.ask( prompt , conversation_id=self.conversation, parent_id=self.prev_message): - response = data["message"] - FullResponse = data - - self.conversation = FullResponse["conversation_id"] - self.prev_message = FullResponse["parent_id"] - self.call += 1 - - #add to history - self.history_data.append({"prompt":prompt,"response":response}) - - return response - - @property - def _identifying_params(self) -> Mapping[str, Any]: - """Get the identifying parameters.""" - return {"model": "ChatGPT", "token": self.token} - - -#llm = ChatGPT(token = "YOUR_TOKEN") #for start new chat -#llm = ChatGPT(token = "YOUR-TOKEN", conversation = "Add-XXXX-XXXX-Convesation-ID") #for use a chat already started - -#print(llm("Hello, how are you?")) -#print(llm("what is AI?")) -#print(llm("Can you resume your previus answer?")) #now memory work well diff --git a/hfAgent/FreeLLM/ChatGPTAPI.py b/hfAgent/FreeLLM/ChatGPTAPI.py deleted file mode 100644 index 6191080..0000000 --- a/hfAgent/FreeLLM/ChatGPTAPI.py +++ /dev/null @@ -1,83 +0,0 @@ -from .pyChatGPT import ChatGPT as Chatbot -#from pyChatGPT import ChatGPT as Chatbot - -import requests -from langchain.llms.base import LLM -from typing import Optional, List, Mapping, Any -import pydantic -import os -from langchain import PromptTemplate, LLMChain -from time import sleep - - - -class ChatGPT(LLM): - - history_data: Optional[List] = [] - token : Optional[str] - chatbot : Optional[Chatbot] = None - call : int = 0 - conversation : Optional[str] = "" - model : str = "default" # or gpt4 - - #### WARNING : for each api call this library will create a new chat on chat.openai.com - - - @property - def _llm_type(self) -> str: - return "custom" - - def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - if stop is not None: - pass - #raise ValueError("stop kwargs are not permitted.") - #token is a must check - if self.chatbot is None: - if self.token is None: - raise ValueError("Need a token , check https://chat.openai.com/api/auth/session for get your token") - else: - if self.conversation == "": - self.chatbot = Chatbot(self.token) - elif self.conversation != "" : - self.chatbot = Chatbot(self.token, conversation_id=self.conversation) - else: - raise ValueError("Something went wrong") - - response = "" - # OpenAI: 50 requests / hour for each account - if (self.call >= 45 and self.model == "default") or (self.call >= 23 and self.model == "gpt4"): - raise ValueError("You have reached the maximum number of requests per hour ! Help me to Improve. Abusing this tool is at your own risk") - else: - sleep(2) - if self.model == "default": - data = self.chatbot.send_message(prompt) - elif self.model == "gpt4": - data = self.chatbot.send_message(prompt, model="gpt4") - #print(data) - response = data["message"] - self.conversation = data["conversation_id"] - FullResponse = data - - self.call += 1 - - #add to history - self.history_data.append({"prompt":prompt,"response":response}) - - return response - - @property - def _identifying_params(self) -> Mapping[str, Any]: - """Get the identifying parameters.""" - return {"model": "ChatGPT", "token": self.token, "model": self.model} - - - -#llm = ChatGPT(token = "YOUR-COOKIE") #for start new chat - -#llm = ChatGPT(token = "YOUR-COOKIE" , model="gpt4") # REQUIRED CHATGPT PLUS subscription - -#llm = ChatGPT(token = "YOUR-COOKIE", conversation = "Add-XXXX-XXXX-Convesation-ID") #for use a chat already started - -#print(llm("Hello, how are you?")) -#print(llm("what is AI?")) -#print(llm("Can you resume your previus answer?")) #now memory work well diff --git a/hfAgent/FreeLLM/HuggingChatAPI.py b/hfAgent/FreeLLM/HuggingChatAPI.py deleted file mode 100644 index 95c97ff..0000000 --- a/hfAgent/FreeLLM/HuggingChatAPI.py +++ /dev/null @@ -1,64 +0,0 @@ - -from hugchat import hugchat -import requests -from langchain.llms.base import LLM -from typing import Optional, List, Mapping, Any -import pydantic -import os -from langchain import PromptTemplate, LLMChain -from time import sleep - - - -class HuggingChat(LLM): - - history_data: Optional[List] = [] - chatbot : Optional[hugchat.ChatBot] = None - conversation : Optional[str] = "" - cookiepath : Optional[str] - #### WARNING : for each api call this library will create a new chat on chat.openai.com - - - @property - def _llm_type(self) -> str: - return "custom" - - def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: - if stop is not None: - pass - #raise ValueError("stop kwargs are not permitted.") - #token is a must check - if self.chatbot is None: - if self.cookiepath is None: - ValueError("Cookie path is required, pls check the documentation on github") - else: - if self.conversation == "": - self.chatbot = hugchat.ChatBot(cookie_path=self.cookiepath) - else: - raise ValueError("Something went wrong") - - - sleep(2) - data = self.chatbot.chat(prompt, temperature=0.5, stream=False) - #conversation_list = self.chatbot.get_conversation_list() - #print(conversation_list) - - #add to history - self.history_data.append({"prompt":prompt,"response":data}) - - return data - - @property - def _identifying_params(self) -> Mapping[str, Any]: - """Get the identifying parameters.""" - return {"model": "HuggingCHAT"} - - - -#llm = HuggingChat(cookiepath = "YOUR-COOKIES-PATH") #for start new chat - - -#print(llm("Hello, how are you?")) -#print(llm("what is AI?")) -#print(llm("Can you resume your previus answer?")) #now memory work well - diff --git a/hfAgent/FreeLLM/pyChatGPT.py b/hfAgent/FreeLLM/pyChatGPT.py deleted file mode 100644 index 64d8f31..0000000 --- a/hfAgent/FreeLLM/pyChatGPT.py +++ /dev/null @@ -1,591 +0,0 @@ -from selenium.webdriver.support import expected_conditions as EC -from selenium.common import exceptions as SeleniumExceptions -from selenium.webdriver.support.ui import WebDriverWait -from selenium.webdriver.common.keys import Keys -from selenium.webdriver.common.by import By - -import undetected_chromedriver as uc -from markdownify import markdownify -from threading import Thread -import platform -import logging -import weakref -import json -import time -import re -import os -import random - -cf_challenge_form = (By.ID, 'challenge-form') - -chatgpt_textbox = (By.TAG_NAME, 'textarea') -chatgpt_streaming = (By.CLASS_NAME, 'result-streaming') -chatgpt_big_response = (By.XPATH, '//div[@class="flex-1 overflow-hidden"]//div[p]') -chatgpt_small_response = ( - By.XPATH, - '//div[starts-with(@class, "markdown prose w-full break-words")]', -) -chatgpt_alert = (By.XPATH, '//div[@role="alert"]') -chatgpt_dialog = (By.XPATH, '//div[@role="dialog"]') -chatgpt_intro = (By.ID, 'headlessui-portal-root') -chatgpt_login_btn = (By.XPATH, '//button[text()="Log in"]') -chatgpt_login_h1 = (By.XPATH, '//h1[text()="Welcome back"]') -chatgpt_logged_h1 = (By.XPATH, '//h1[text()="ChatGPT"]') - -chatgpt_new_chat = (By.LINK_TEXT, 'New chat') -chatgpt_clear_convo = (By.LINK_TEXT, 'Clear conversations') -chatgpt_confirm_clear_convo = (By.LINK_TEXT, 'Confirm clear conversations') -chatgpt_chats_list_first_node = ( - By.XPATH, - "//li[@class='relative z-[15]']//a", -) - -model_selector = (By.XPATH, "//button[starts-with(@id, 'headlessui-listbox-button')]") -gpt4_selector = (By.XPATH, "//ul[@role='listbox']/li[contains(.//text(), 'GPT-4')][1]") - -chatgpt_chat_url = 'https://chat.openai.com' - - -class ChatGPT: - ''' - An unofficial Python wrapper for OpenAI's ChatGPT API - ''' - - def __init__( - self, - session_token: str = None, - conversation_id: str = '', - auth_type: str = None, - email: str = None, - password: str = None, - login_cookies_path: str = '', - captcha_solver: str = 'pypasser', - solver_apikey: str = '', - proxy: str = None, - chrome_args: list = [], - moderation: bool = True, - verbose: bool = False, - ): - ''' - Initialize the ChatGPT object\n - :param session_token: The session token to use for authentication - :param conversation_id: The conversation ID to use for the chat session - :param auth_type: The authentication type to use (`google`, `microsoft`, `openai`) - :param email: The email to use for authentication - :param password: The password to use for authentication - :param login_cookies_path: The path to the cookies file to use for authentication - :param captcha_solver: The captcha solver to use (`pypasser`, `2captcha`) - :param solver_apikey: The apikey of the captcha solver to use (if any) - :param proxy: The proxy to use for the browser (`https://ip:port`) - :param chrome_args: The arguments to pass to the browser - :param moderation: Whether to enable message moderation - :param verbose: Whether to enable verbose logging - ''' - self.__init_logger(verbose) - - self.__session_token = session_token - self.__conversation_id = conversation_id - self.__auth_type = auth_type - self.__email = email - self.__password = password - self.__login_cookies_path = login_cookies_path - self.__captcha_solver = captcha_solver - self.__solver_apikey = solver_apikey - self.__proxy = proxy - self.__chrome_args = chrome_args - self.__moderation = moderation - - if not self.__session_token and ( - not self.__email or not self.__password or not self.__auth_type - ): - raise ValueError( - 'Please provide either a session token or login credentials' - ) - if self.__auth_type not in [None, 'google', 'microsoft', 'openai']: - raise ValueError('Invalid authentication type') - if self.__captcha_solver not in [None, 'pypasser', '2captcha']: - raise ValueError('Invalid captcha solver') - if self.__captcha_solver == '2captcha' and not self.__solver_apikey: - raise ValueError('Please provide a 2captcha apikey') - if self.__proxy and not re.findall( - r'(https?|socks(4|5)?):\/\/.+:\d{1,5}', self.__proxy - ): - raise ValueError('Invalid proxy format') - if self.__auth_type == 'openai' and self.__captcha_solver == 'pypasser': - try: - import ffmpeg_downloader as ffdl - except ModuleNotFoundError: - raise ValueError( - 'Please install ffmpeg_downloader, PyPasser, and pocketsphinx by running `pip install ffmpeg_downloader PyPasser pocketsphinx`' - ) - - ffmpeg_installed = bool(ffdl.ffmpeg_version) - self.logger.debug(f'ffmpeg installed: {ffmpeg_installed}') - if not ffmpeg_installed: - import subprocess - - subprocess.run(['ffdl', 'install']) - os.environ['PATH'] += os.pathsep + ffdl.ffmpeg_dir - - self.__init_browser() - weakref.finalize(self, self.__del__) - - def __del__(self): - ''' - Close the browser and display - ''' - self.__is_active = False - if hasattr(self, 'driver'): - self.logger.debug('Closing browser...') - self.driver.quit() - if hasattr(self, 'display'): - self.logger.debug('Closing display...') - self.display.stop() - - def __init_logger(self, verbose: bool) -> None: - ''' - Initialize the logger\n - :param verbose: Whether to enable verbose logging - ''' - self.logger = logging.getLogger('pyChatGPT') - self.logger.setLevel(logging.DEBUG) - if verbose: - formatter = logging.Formatter('[%(funcName)s] %(message)s') - stream_handler = logging.StreamHandler() - stream_handler.setFormatter(formatter) - self.logger.addHandler(stream_handler) - - def __init_browser(self) -> None: - ''' - Initialize the browser - ''' - if platform.system() == 'Linux' and 'DISPLAY' not in os.environ: - self.logger.debug('Starting virtual display...') - try: - from pyvirtualdisplay import Display - - self.display = Display() - except ModuleNotFoundError: - raise ValueError( - 'Please install PyVirtualDisplay to start a virtual display by running `pip install PyVirtualDisplay`' - ) - except FileNotFoundError as e: - if 'No such file or directory: \'Xvfb\'' in str(e): - raise ValueError( - 'Please install Xvfb to start a virtual display by running `sudo apt install xvfb`' - ) - raise e - self.display.start() - - self.logger.debug('Initializing browser...') - options = uc.ChromeOptions() - options.add_argument('--window-size=1024,768') - options.add_argument("--verbose") - options.add_argument('--no-sandbox') - options.add_argument('--headless') - options.add_argument('--disable-gpu') - options.add_argument('--disable-dev-shm-usage') - if self.__proxy: - options.add_argument(f'--proxy-server={self.__proxy}') - for arg in self.__chrome_args: - options.add_argument(arg) - try: - self.driver = uc.Chrome(options=options, version_main=112) - except TypeError as e: - if str(e) == 'expected str, bytes or os.PathLike object, not NoneType': - raise ValueError('Chrome installation not found') - raise e - - if self.__login_cookies_path and os.path.exists(self.__login_cookies_path): - self.logger.debug('Restoring cookies...') - try: - with open(self.__login_cookies_path, 'r', encoding='utf-8') as f: - cookies = json.load(f) - for cookie in cookies: - if cookie['name'] == '__Secure-next-auth.session-token': - self.__session_token = cookie['value'] - except json.decoder.JSONDecodeError: - self.logger.debug(f'Invalid cookies file: {self.__login_cookies_path}') - - if self.__session_token: - self.logger.debug('Restoring session_token...') - self.driver.execute_cdp_cmd( - 'Network.setCookie', - { - 'domain': 'chat.openai.com', - 'path': '/', - 'name': '__Secure-next-auth.session-token', - 'value': self.__session_token, - 'httpOnly': True, - 'secure': True, - }, - ) - - if not self.__moderation: - self.logger.debug('Blocking moderation...') - self.driver.execute_cdp_cmd( - 'Network.setBlockedURLs', - {'urls': ['https://chat.openai.com/backend-api/moderations']}, - ) - - self.logger.debug('Ensuring Cloudflare cookies...') - self.__ensure_cf() - - self.logger.debug('Opening chat page...') - self.driver.get(f'{chatgpt_chat_url}/{self.__conversation_id}') - self.__check_blocking_elements() - - self.__is_active = True - Thread(target=self.__keep_alive, daemon=True).start() - - def __ensure_cf(self, retry: int = 3) -> None: - ''' - Ensure Cloudflare cookies are set\n - :param retry: Number of retries - ''' - self.logger.debug('Opening new tab...') - original_window = self.driver.current_window_handle - self.driver.switch_to.new_window('tab') - - self.logger.debug('Getting Cloudflare challenge...') - self.driver.get('https://chat.openai.com/api/auth/session') - try: - WebDriverWait(self.driver, 10).until_not( - EC.presence_of_element_located(cf_challenge_form) - ) - except SeleniumExceptions.TimeoutException: - self.logger.debug(f'Cloudflare challenge failed, retrying {retry}...') - self.driver.save_screenshot(f'cf_failed_{retry}.png') - if retry > 0: - self.logger.debug('Closing tab...') - self.driver.close() - self.driver.switch_to.window(original_window) - return self.__ensure_cf(retry - 1) - raise ValueError('Cloudflare challenge failed') - self.logger.debug('Cloudflare challenge passed') - - self.logger.debug('Validating authorization...') - response = self.driver.page_source - if response[0] != '{': - response = self.driver.find_element(By.TAG_NAME, 'pre').text - response = json.loads(response) - if (not response) or ( - 'error' in response and response['error'] == 'RefreshAccessTokenError' - ): - self.logger.debug('Authorization is invalid') - if not self.__auth_type: - raise ValueError('Invalid session token') - self.__login() - self.logger.debug('Authorization is valid') - - self.logger.debug('Closing tab...') - self.driver.close() - self.driver.switch_to.window(original_window) - - def __check_capacity(self, target_url: str): - ''' - Check if ChatGPT is at capacity\n - :param target_url: URL to retry if ChatGPT is at capacity - ''' - while True: - try: - self.logger.debug('Checking if ChatGPT is at capacity...') - WebDriverWait(self.driver, 3).until( - EC.presence_of_element_located( - (By.XPATH, '//div[text()="ChatGPT is at capacity right now"]') - ) - ) - self.logger.debug('ChatGPT is at capacity, retrying...') - self.driver.get(target_url) - except SeleniumExceptions.TimeoutException: - self.logger.debug('ChatGPT is not at capacity') - break - - def __login(self) -> None: - ''' - Login to ChatGPT - ''' - self.logger.debug('Opening new tab...') - original_window = self.driver.current_window_handle - self.driver.switch_to.new_window('tab') - - self.logger.debug('Opening login page...') - self.driver.get('https://chat.openai.com/auth/login') - self.__check_capacity('https://chat.openai.com/auth/login') - - self.logger.debug('Clicking login button...') - WebDriverWait(self.driver, 5).until( - EC.element_to_be_clickable(chatgpt_login_btn) - ).click() - - WebDriverWait(self.driver, 5).until( - EC.presence_of_element_located(chatgpt_login_h1) - ) - - from . import Auth0 - - Auth0.login(self) - - self.logger.debug('Checking if login was successful') - try: - WebDriverWait(self.driver, 5).until( - EC.presence_of_element_located(chatgpt_logged_h1) - ) - if self.__login_cookies_path: - self.logger.debug('Saving cookies...') - with open(self.__login_cookies_path, 'w', encoding='utf-8') as f: - json.dump( - [ - i - for i in self.driver.get_cookies() - if i['name'] == '__Secure-next-auth.session-token' - ], - f, - ) - except SeleniumExceptions.TimeoutException as e: - self.driver.save_screenshot('login_failed.png') - raise e - - self.logger.debug('Closing tab...') - self.driver.close() - self.driver.switch_to.window(original_window) - - def __keep_alive(self) -> None: - ''' - Keep the session alive by updating the local storage\n - Credit to Rawa#8132 in the ChatGPT Hacking Discord server - ''' - while self.__is_active: - self.logger.debug('Updating session...') - payload = ( - '{"event":"session","data":{"trigger":"getSession"},"timestamp":%d}' - % int(time.time()) - ) - try: - self.driver.execute_script( - 'window.localStorage.setItem("nextauth.message", arguments[0])', - payload, - ) - except Exception as e: - self.logger.debug(f'Failed to update session: {str(e)}') - self.__sleep(60) - - def __check_blocking_elements(self) -> None: - ''' - Check for blocking elements and dismiss them - ''' - self.logger.debug('Looking for blocking elements...') - - try: - # FInd a button to dismiss the dialog with class="btn relative btn-primary" inside the div[@role="dialog"] - btn_to_dismiss = WebDriverWait(self.driver, 5).until( - EC.presence_of_element_located((By.XPATH, '//div[@role="dialog"]//button[@class="btn relative btn-primary"]')) - ) - if btn_to_dismiss: - self.logger.debug('Dismissing dialog...') - self.driver.execute_script('arguments[0].click()', btn_to_dismiss) - except: - pass - - try: - # for 3 times - i = 0 - while i<=2: - self.__sleep(0.4) - if i !=2: - #get the button with class="btn relative btn-neutral ml-auto" - btn = WebDriverWait(self.driver, 5).until( - EC.presence_of_element_located((By.XPATH, '//button[@class="btn relative btn-neutral ml-auto"]')) - ) - else: - #get the button with class="btn relative btn-primary ml-auto" - btn = WebDriverWait(self.driver, 5).until( - EC.presence_of_element_located((By.XPATH, '//button[@class="btn relative btn-primary ml-auto"]')) - ) - if btn: - self.logger.debug('Dismissing dialog...') - self.driver.execute_script('arguments[0].click()', btn) - i+=1 - except: - pass - - def __stream_message(self): - prev_content = '' - while True: - result_streaming = self.driver.find_elements(*chatgpt_streaming) - responses = self.driver.find_elements(*chatgpt_big_response) - if responses: - response = responses[-1] - if 'text-red' in response.get_attribute('class'): - self.logger.debug('Response is an error') - raise ValueError(response.text) - response = self.driver.find_elements(*chatgpt_small_response)[-1] - content = response.text - if content != prev_content: - yield content[len(prev_content) :] - prev_content = content - if not result_streaming: - break - - def send_message(self, message: str, stream: bool = False, model: str = "default") -> dict: - ''' - Send a message to ChatGPT\n - :param message: Message to send - :return: Dictionary with keys `message` and `conversation_id` - ''' - self.logger.debug('Ensuring Cloudflare cookies...') - self.__ensure_cf() - #self.__check_blocking_elements() - - # Wait for page to load - try: - textbox = WebDriverWait(self.driver, 10).until( - EC.element_to_be_clickable(chatgpt_textbox) - ) - except SeleniumExceptions.ElementClickInterceptedException(): - pass - - # If we have paid access, we should select GPT4 model - try: - if model == "gpt4": - self.logger.debug('Trying to select model...') - WebDriverWait(self.driver, 3).until( - EC.presence_of_element_located(model_selector) - ).click() - self.logger.debug('Paid access detected, selecting GPT4 model...') - self.__sleep(1.0) - WebDriverWait(self.driver, 3).until( - EC.presence_of_element_located(gpt4_selector) - ).click() - self.logger.debug('GPT4 model selected') - except SeleniumExceptions.TimeoutException: - print(">>> WARNING <<<\n>> You don't have paid access to GPT4, using default model...") - self.logger.debug('Paid access not detected, using default model...') - - self.logger.debug('Sending message...') - try: - textbox = WebDriverWait(self.driver, 3).until( - EC.element_to_be_clickable(chatgpt_textbox) - ) - textbox.click() - except SeleniumExceptions.ElementClickInterceptedException(): - self.__check_blocking_elements() - textbox = WebDriverWait(self.driver, 10).until( - EC.element_to_be_clickable(chatgpt_textbox) - ) - textbox.click() - - self.driver.execute_script( - ''' - var element = arguments[0], txt = arguments[1]; - element.value += txt; - element.dispatchEvent(new Event("change")); - ''', - textbox, - message, - ) - textbox.send_keys(Keys.ENTER) - - if stream: - for i in self.__stream_message(): - print(i, end='') - self.__sleep(0.1) - return print() - - self.logger.debug('Waiting for completion...') - WebDriverWait(self.driver, 120).until_not( - EC.presence_of_element_located(chatgpt_streaming) - ) - - self.logger.debug('Getting response...') - responses = self.driver.find_elements(*chatgpt_big_response) - if responses: - response = responses[-1] - if 'text-red' in response.get_attribute('class'): - self.logger.debug('Response is an error') - raise ValueError(response.text) - response = self.driver.find_elements(*chatgpt_small_response)[-1] - - content = markdownify(response.get_attribute('innerHTML')).replace( - 'Copy code`', '`' - ) - - pattern = re.compile( - r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}' - ) - matches = pattern.search(self.driver.current_url) - if not matches: - self.driver.refresh() - self.__sleep(2) - #self.__check_blocking_elements() - self.driver.execute_script("arguments[0].click();", WebDriverWait(self.driver, 10).until( - EC.element_to_be_clickable(chatgpt_chats_list_first_node) - )) - self.__sleep(2) - #print(self.driver.current_url) - matches = pattern.search(self.driver.current_url) - - - #get current url and print it - #print(self.driver.current_url) - #print(matches) - - conversation_id = matches.group() - return {'message': content, 'conversation_id': conversation_id} - #return {'message': content} - - def __sleep(self, sec=1.0, multiplier=2) -> None: - """ - Random sleep to avoid detection - """ - time.sleep(random.uniform(sec, sec * multiplier)) - - def reset_conversation(self) -> None: - ''' - Reset the conversation - ''' - if not self.driver.current_url.startswith(chatgpt_chat_url): - return self.logger.debug('Current URL is not chat page, skipping reset') - - self.logger.debug('Resetting conversation...') - try: - self.driver.find_element(*chatgpt_new_chat).click() - except SeleniumExceptions.NoSuchElementException: - self.logger.debug('New chat button not found') - self.driver.save_screenshot('reset_conversation_failed.png') - - def clear_conversations(self) -> None: - ''' - Clear all conversations - ''' - if not self.driver.current_url.startswith(chatgpt_chat_url): - return self.logger.debug('Current URL is not chat page, skipping clear') - - self.logger.debug('Clearing conversations...') - try: - self.driver.find_element(*chatgpt_clear_convo).click() - except SeleniumExceptions.NoSuchElementException: - self.logger.debug('Clear conversations button not found') - - try: - self.driver.find_element(*chatgpt_confirm_clear_convo).click() - except SeleniumExceptions.NoSuchElementException: - return self.logger.debug('Confirm clear conversations button not found') - try: - WebDriverWait(self.driver, 20).until_not( - EC.presence_of_element_located(chatgpt_chats_list_first_node) - ) - self.logger.debug('Cleared conversations') - except SeleniumExceptions.TimeoutException: - self.logger.debug('Clear conversations failed') - - def refresh_chat_page(self) -> None: - ''' - Refresh the chat page - ''' - if not self.driver.current_url.startswith(chatgpt_chat_url): - return self.logger.debug('Current URL is not chat page, skipping refresh') - - self.driver.get(chatgpt_chat_url) - self.__check_capacity(chatgpt_chat_url) - self.__check_blocking_elements() \ No newline at end of file