-
Notifications
You must be signed in to change notification settings - Fork 0
/
webproxy.py
149 lines (120 loc) · 6.13 KB
/
webproxy.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import os
import socket
import threading
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
class ChatAutomaton:
def __init__(self, filepath, chrome_path, chrome_driver_path):
"""
This constructor automates the following steps:
1. Open a Chrome browser with remote debugging enabled at a specified URL.
2. Prompt the user to complete the log-in/registration/human verification, if required.
3. Connect a Selenium WebDriver to the browser instance after human verification is completed.
:param chrome_path: file path to chrome.exe (ex. C:\\Users\\User\\...\\chromedriver.exe)
:param chrome_driver_path: file path to chromedriver.exe (ex. C:\\Users\\User\\...\\chromedriver.exe)
"""
self.chrome_path = chrome_path
self.chrome_driver_path = chrome_driver_path
url = r"https://chat.openai.com"
free_port = self.find_available_port()
self.launch_chrome_with_remote_debugging(free_port, url)
self.wait_for_human_verification()
self.driver = self.setup_webdriver(free_port)
self.cookie = self.get_cookie()
@staticmethod
def find_available_port():
""" This function finds and returns an available port number on the local machine by creating a temporary
socket, binding it to an ephemeral port, and then closing the socket. """
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('', 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
return s.getsockname()[1]
def launch_chrome_with_remote_debugging(self, port, url):
""" Launches a new Chrome instance with remote debugging enabled on the specified port and navigates to the
provided url """
def open_chrome():
chrome_cmd = f"{self.chrome_path} --remote-debugging-port={port} --user-data-dir=remote-profile {url}"
os.system(chrome_cmd)
chrome_thread = threading.Thread(target=open_chrome)
chrome_thread.start()
def setup_webdriver(self, port):
""" Initializes a Selenium WebDriver instance, connected to an existing Chrome browser
with remote debugging enabled on the specified port"""
chrome_options = webdriver.ChromeOptions()
chrome_options.binary_location = self.chrome_driver_path
chrome_options.add_experimental_option("debuggerAddress", f"127.0.0.1:{port}")
driver = webdriver.Chrome(options=chrome_options)
return driver
def get_cookie(self):
"""
Get chat.openai.com cookie from the running chrome instance.
"""
cookies = self.driver.get_cookies()
cookie = [elem for elem in cookies if elem["name"] == '__Secure-next-auth.session-token'][0]['value']
return cookie
def send_prompt_to_chatgpt(self, prompt):
""" Sends a message to ChatGPT and waits for 20 seconds for the response """
input_box = self.driver.find_element(by=By.XPATH, value='//textarea[contains(@id, "prompt-textarea")]')
self.driver.execute_script(f"arguments[0].value = '{prompt}';", input_box)
input_box.send_keys(Keys.RETURN)
input_box.submit()
self.check_response_ended()
def check_response_ended(self):
""" Checks if ChatGPT response ended """
start_time = time.time()
while len(self.driver.find_elements(by=By.CSS_SELECTOR, value='div.text-base')[-1].find_elements(
by=By.CSS_SELECTOR, value='button.text-token-text-tertiary')) < 1:
time.sleep(0.5)
# Exit the while loop after 60 seconds anyway
if time.time() - start_time > 60:
break
time.sleep(1) # the length should be =4, so it's better to wait a moment to be sure it's really finished
def return_chatgpt_conversation(self):
"""
:return: returns a list of items, even items are the submitted questions (prompts) and odd items are chatgpt response
"""
return self.driver.find_elements(by=By.CSS_SELECTOR, value='div.text-base')
def save_conversation(self, file_name):
"""
It saves the full chatgpt conversation of the tab open in chrome into a text file, with the following format:
prompt: ...
response: ...
delimiter
prompt: ...
response: ...
:param file_name: name of the file where you want to save
"""
directory_name = "conversations"
if not os.path.exists(directory_name):
os.makedirs(directory_name)
delimiter = "|^_^|"
chatgpt_conversation = self.return_chatgpt_conversation()
with open(os.path.join(directory_name, file_name), "a") as file:
for i in range(0, len(chatgpt_conversation), 2):
file.write(
f"prompt: {chatgpt_conversation[i].text}\nresponse: {chatgpt_conversation[i + 1].text}\n\n{delimiter}\n\n")
def return_last_response(self):
""" :return: the text of the last chatgpt response """
response_elements = self.driver.find_elements(by=By.CSS_SELECTOR, value='div.text-base')
return response_elements[-1].text
@staticmethod
def wait_for_human_verification():
print("You need to manually complete the log-in or the human verification if required.")
while True:
user_input = input(
"Enter 'y' if you have completed the log-in or the human verification, or 'n' to check again: ").lower().strip()
if user_input == 'y':
print("Continuing with the automation process...")
break
elif user_input == 'n':
print("Waiting for you to complete the human verification...")
time.sleep(5) # You can adjust the waiting time as needed
else:
print("Invalid input. Please enter 'y' or 'n'.")
def quit(self):
""" Closes the browser and terminates the WebDriver session."""
print("Closing the browser...")
self.driver.close()
self.driver.quit()