mirror of
https://github.com/IntelligenzaArtificiale/Free-Auto-GPT.git
synced 2026-07-03 13:15:41 +02:00
Delete hfAgent/FreeLLM directory
This commit is contained in:
IntelligenzaArtificiale
committed by
GitHub
GitHub
parent
12261a20e5
commit
e41dd7b7e3
@@ -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
|
||||
@@ -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<Response>\n" + str(response) + "\n</Response>\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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user