Somewhat messy, but ready for production.

This commit is contained in:
ForeverPyrite
2024-10-15 20:45:01 -04:00
parent 05b7c247a7
commit d66fe2155c
5 changed files with 238 additions and 37 deletions

29
main.py
View File

@@ -29,7 +29,6 @@ class StreamOutput:
self.response = ""
self.done = False
self.buffer: list = []
print("Reset stream output obj")
async def send_delta(self, delta):
self.delta = delta
@@ -44,9 +43,8 @@ class StreamOutput:
if self.delta != self.buffer[get_index(self.buffer)]:
self.buffer.append(delta)
except IndexError as index_error:
print(index_error)
log(f"\nCaught IndexError: {str(index_error)}")
self.buffer.append(delta)
else: self.buffer.append(delta)
# To get the env var
@@ -55,6 +53,13 @@ import os
load_dotenv()
# For logging
import pytz
from datetime import datetime
def log(str):
with open("log.md", "at") as file:
file.write(str)
### OpenAI Config
@@ -73,7 +78,7 @@ asst_screw_bardo_id = "asst_JGFaX6uOIotqy5mIJnu3Yyp7"
class EventHandler(AssistantEventHandler):
@override
def on_text_created(self, text) -> None:
awaiter(output_stream.send_delta("Response Recieved:\nScrew-Bardo: "))
awaiter(output_stream.send_delta("Response Recieved:\n\nScrew-Bardo:\n\n"))
@override
def on_text_delta(self, delta, snapshot):
@@ -93,17 +98,6 @@ def create_and_stream(transcript):
stream.until_done()
output_stream.done = True
def fake_stream():
i = 0
STREAM = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 18]
print("Starting fake stream.")
while i <= len(STREAM)-1:
awaiter(asyncio.sleep(0.05))
awaiter(output_stream.send_delta(str(STREAM[i])))
i += 1
output_stream.done = True
return
def get_video_id(url):
youtu_be = r'(?<=youtu.be/)([A-Za-z0-9_-]{11})'
youtube_com = r'(?<=youtube\.com\/watch\?v=)([A-Za-z0-9_-]{11})'
@@ -113,7 +107,7 @@ def get_video_id(url):
id = re.search(youtube_com, url)
if not id:
print("Couldn't parse video ID from URL")
# Couldn't parse video ID from URL
return None
return id.group(1)
@@ -123,7 +117,8 @@ def get_auto_transcript(video_id):
trans_api_errors = youtube_transcript_api._errors
try:
transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['en'], proxies=None, cookies=None, preserve_formatting=False)
except trans_api_errors.TranscriptsDisabled:
except trans_api_errors.TranscriptsDisabled as e:
log(f'\n\n# Exception while fetching transcript:\n \n{e}\n')
return None
formatter = TextFormatter() # Ensure that you create an instance of TextFormatter