-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapi_session.py
143 lines (108 loc) · 4.18 KB
/
api_session.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
import googleapi
import google.generativeai
import re
import time
import threading
from constants import * ## Global constants
class Session:
def __init__(self, client_name):
self.type = CLIENT_ID_TO_TYPE[client_name]
if self.type == TYPE_GEMINI:
self.client = GeminiClient(client_name)
elif self.type == TYPE_GOOGLE:
self.client = GoogleClient(client_name)
else:
self.client = None
def reset(self):
self.client.reset()
class GeminiClient:
def __init__(self, name):
self.name = name
self.model = google.generativeai.GenerativeModel(model_name=name)
self.chat = self.model.start_chat()
self.api_response = None
self.query = ''
self.response = ''
def reset(self):
self.api_response = None
self.query = ''
self.response = ''
def stop(self):
pass
def stopped(self):
pass
def send_request(self):
self.api_response = self.chat.send_message(self.query)
return True if self.api_response else False
def format_response(self):
response = self.api_response.text
response = self.f_code_blocks(response)
response = self.f_numbered_lists(response)
response = self.f_bold_text(response)
response = self.f_italicized_text(response)
response = self.f_general(response)
self.response = response
def f_code_blocks(self, response):
pattern = r'```(\w+)(.*?)```'
replacement = r'\t\1: - - - - -\2\t- - - - - - - - - -' # tab, language name, separator, code, tab, separator
return re.sub(pattern, replacement, response, flags=re.DOTALL)
def f_numbered_lists(self, response):
pattern = r'\*\*(\d)'
replacement = r'\t\1'
return re.sub(pattern, replacement, response)
def f_bold_text(self, response):
pattern = r'(?!\*\*\s)\*\*(.+?)(?!\s\*\*)\*\*'
replacement = r'\033[39;49;1m\1\033[0m' # ANSI: bold, \1, ANSI: reset
return re.sub(pattern, replacement, response)
def f_italicized_text(self, response):
pattern = r'(?!\*\s)\*(.+?)(?!\s\*)\*'
replacement = r'\033[39;49;3m\1\033[0m' # ANSI: italic, \1, ANSI: reset
return re.sub(pattern, replacement, response)
## REPLACE THIS FUNCTION WITH SMALLER FUNCTIONS
def f_general(self, response):
return response.replace('\t#', '\t\t').replace("* **", '- ').replace("**", '').replace("\n *", '\n -').replace("\n*", '\n\t-').replace("--", '')
class GoogleClient:
def __init__(self, name):
self.name = name
self._stop_event = threading.Event()
self.api_response = None
self.query = ''
self.response = ''
def reset(self):
self._stop_event.clear()
self.api_response = None
self.query = ''
self.response = ''
def stop(self):
self._stop_event.set()
def stopped(self):
self._stop_event.is_set()
def send_request(self):
wait = 0.5
while not self.api_response:
self.api_response = googleapi.google.search(self.query)
time.sleep(wait)
wait += 0.5
if wait > 3 or self.stopped():
break
return True if self.api_response else False
def format_response(self):
num_results = len(self.api_response)
response = ''
for i, result in enumerate(self.api_response):
## Remove link from title of search result
end_of_name = result.name.find("https://")
if not end_of_name:
end_of_name = result.name.find("http://")
name = result.name[:end_of_name]
## Populate response_string
response += f"{i+1} {name}\n"
response += f"\t{result.link}\n"
if result.description:
response += f"\n\t{result.description}\n"
else:
response += f"\n\tNo Description\n"
## Add newline to end of response_string
if i < (num_results - 1):
response += '\n\n'
self.response = response