kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Blogen-V.000.0.1 Added features,Cleanup. WIP

Browse Source
This commit is contained in:
AjaySi
2023-12-09 18:07:18 +05:30
parent edc468f4aa
commit eaf13c2d16
164 changed files with 1859 additions and 71990 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
lib/gpt_providers/README.md
View File

@@ -1,19 +1,11 @@
gpt_providers are companies providing commercial/free GPT pre-trained models as saas.
These include openai, Azure, Goodle, FB, Anthrophic etc
# OpenAI ChatGPT Integration for Enhanced Blog Generation
- If you want to use chatgpt and its models, then use openai as gpt_provider
- We plan to integrate most the accurate, widely used models as gpt providers.
- These will also include text to image and video generations as blogging artifacts.
gpt_provider=openai
------------------------------------
Here are some tips for using LLMs to generate ideas:
- Be as specific as possible in your prompts. The more specific you are, the better the LLM will
be able to understand what you are asking for.
- Use keywords in your prompts. This will help the LLM to generate ideas that are relevant to your topic.
- Try different temperatures and top_p values. These parameters control the creativity and diversity of the generated ideas.
- Experiment with different prompts and settings to see what works best for you.
## Introduction
This toolkit, written in Python, integrates OpenAI's ChatGPT and other AI services for comprehensive blog generation. It allows for selecting and fine-tuning OpenAI models to suit various content creation needs, including text generation, image analysis, and speech-to-text conversion.
## Key Features
- **AI-Powered Text Generation**: Leverages OpenAI's ChatGPT for creating engaging and contextually relevant text based on user inputs.
- **Image Analysis and Detail Extraction**: Utilizes OpenAI's Vision API to analyze images and extract important details like Alt Text, Description, Title, and Caption.
- **Dynamic Image Generation**: Generates images from textual descriptions using DALL-E 2 and DALL-E 3 models, enhancing blog visual content.
- **Speech-to-Text Transcription**: Converts audio from YouTube videos to text, enabling easy content repurposing for blogs.
- **Image Variation Creation**: Produces variations of existing images, offering creative flexibility and maintaining topical relevance.

56
lib/gpt_providers/gen_dali2_images.py Normal file
View File

@@ -0,0 +1,56 @@
from openai import OpenAI
from loguru import logger
import sys
from .save_image import save_generated_image
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
) # for exponential backoff
@retry(wait=wait_random_exponential(min=1, max=120), stop=stop_after_attempt(6))
def generate_dalle3_images(img_prompt, image_dir, size="1024x1024", quality="hd", n=1):
"""
Generates images using the DALL-E 3 model based on a given text prompt.
Args:
img_prompt (str): Text prompt to generate the image.
image_dir (str): Directory where the generated image will be saved.
size (str, optional): Size of the generated images. Defaults to "1024x1024".
quality (str, optional): Quality of the generated images. Defaults to "hd".
n (int, optional): Number of images to generate. Defaults to 1.
Returns:
str: Path to the saved image.
Raises:
SystemExit: If an error occurs in image generation or saving.
"""
try:
logger.info("Generating Dall-e-3 image for the blog.")
client = OpenAI()
img_generation_response = client.images.generate(
model="dall-e-3",
prompt=img_prompt,
size=size,
quality=quality,
n=n
)
# Save the generated image locally.
try:
img_path = save_generated_image(img_generation_response, image_dir)
return img_path
except Exception as err:
logger.error(f"Failed to Save generated image: {err}")
except openai.OpenAIError as e:
logger.error(f"Dalle-3 image generation error: HTTP Status {e.http_status}, Error: {e.error}")
sys.exit("Exiting due to Dalle-3 image generation error.")
except Exception as e:
logger.error(f"Failed to generate images with Dalle3: {e}")
sys.exit("Exiting due to a general error in image generation.")

61
lib/gpt_providers/gen_dali3_images.py Normal file
View File

@@ -0,0 +1,61 @@
from openai import OpenAI
from loguru import logger
import sys
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
) # for exponential backoff
from .save_image import save_generated_image
@retry(wait=wait_random_exponential(min=1, max=120), stop=stop_after_attempt(6))
def generate_dalle3_images(img_prompt, image_dir, size="1024x1024", quality="hd", n=1):
"""
Generates images using the DALL-E 3 model based on a given text prompt.
Args:
img_prompt (str): Text prompt to generate the image.
image_dir (str): Directory where the generated image will be saved.
size (str, optional): Size of the generated images. Defaults to "1024x1024".
quality (str, optional): Quality of the generated images. Defaults to "hd".
n (int, optional): Number of images to generate. Defaults to 1.
Returns:
str: Path to the saved image.
Raises:
SystemExit: If an error occurs in image generation or saving.
"""
try:
logger.info("Generating Dall-e-3 image for the blog.")
client = OpenAI()
img_generation_response = client.images.generate(
model="dall-e-3",
prompt=img_prompt,
size=size,
quality=quality,
n=n
)
img_path = save_generated_image(img_generation_response, image_dir)
return img_path
except openai.OpenAIError as e:
logger.error(f"Dalle-3 image generation error: HTTP Status {e.http_status}, Error: {e.error}")
sys.exit("Exiting due to Dalle-3 image generation error.")
except Exception as e:
logger.error(f"Failed to generate images with Dalle3: {e}")
sys.exit("Exiting due to a general error in image generation.")
# Example usage
if __name__ == "__main__":
try:
image_path = generate_dalle3_images("A futuristic cityscape", "/path/to/image/dir")
print(f"Image generated and saved at: {image_path}")
except SystemExit as e:
print(f"Terminated: {e}")

51
lib/gpt_providers/gen_variation_img.py Normal file
View File

@@ -0,0 +1,51 @@
from loguru import logger
import sys
from PIL import Image
from openai import OpenAI
def gen_new_from_given_img(img_path, image_dir, num_img=1, img_size="1024x1024", response_format="url"):
"""
Generates variations of a given image using OpenAI's image variation API.
This function takes an existing image, processes it, and generates a specified number of new images based on it.
These generated images are variations of the original, providing creative flexibility.
Args:
img_path (str): Path to the original image file.
image_dir (str): Directory where the generated images will be saved.
num_img (int, optional): Number of image variations to generate. Defaults to 1.
img_size (str, optional): Size of the generated images. Defaults to "1024x1024".
response_format (str, optional): Format in which the generated images are returned. Defaults to "url".
Returns:
str: Path to the saved image variation.
Raises:
SystemExit: If a critical error occurs that prevents successful execution.
"""
try:
logger.info(f"Starting image variation generation for: {img_path}")
# Convert and prepare the image
png = Image.open(img_path).convert('RGBA')
background = Image.new('RGBA', png.size, (255, 255, 255))
alpha_composite = Image.alpha_composite(background, png)
alpha_composite.save(img_path, 'PNG', quality=80)
logger.info("Image prepared for variation generation.")
client = OpenAI()
variation_response = client.images.create_variation(
image=open(img_path, "rb"),
n=num_img,
size=img_size,
response_format=response_format
)
# Saving the generated image
generated_image_path = save_generated_image(variation_response, image_dir)
logger.info(f"Image variation generated and saved to: {generated_image_path}")
return generated_image_path
except Exception as e:
logger.error(f"Error occurred during image variation generation: {e}")
sys.exit(f"Exiting due to critical error: {e}")

106
lib/gpt_providers/gpt_vision_img_details.py Normal file
View File

@@ -0,0 +1,106 @@
import requests
import re
import base64
import os
import sys
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
) # for exponential backoff
@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
def analyze_and_extract_details_from_image(image_path):
"""
Analyzes an image using OpenAI's Vision API to extract Alt Text, Description, Title, and Caption.
This function encodes an image to a base64 string and sends a request to the OpenAI API.
It interprets the contents of the image, returning a textual description.
Args:
image_path (str): Path to the image file.
Returns:
dict: A dictionary with extracted details including Alt Text, Description, Title, and Caption.
None: If an error occurs during processing.
Raises:
SystemExit: If a critical error occurs that prevents the function from executing successfully.
"""
try:
logger.info("Starting image analysis using OpenAI's Vision API.")
def encode_image(path):
""" Encodes an image to a base64 string. """
with open(path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
base64_image = encode_image(image_path)
logger.info("Image encoded to base64 successfully.")
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY')}"
}
payload = {
"model": "gpt-4-vision-preview",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Analyze the given image and suggest the following: Alternative text(Alt Text), description, title, caption."
},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
}
]
}
],
"max_tokens": 300
}
response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
response.raise_for_status()
assistant_message = response.json()['choices'][0]['message']['content']
logger.info("Received response from OpenAI API.")
# Extracting details using regular expressions
alt_text_match = re.search(r'Alt Text: "(.*?)"', assistant_message)
description_match = re.search(r'Description: (.*?)\n\n', assistant_message)
title_match = re.search(r'Title: "(.*?)"', assistant_message)
caption_match = re.search(r'Caption: "(.*?)"', assistant_message)
image_details = {
'alt_text': alt_text_match.group(1) if alt_text_match else "N/A",
'description': description_match.group(1) if description_match else "N/A",
'title': title_match.group(1) if title_match else "N/A",
'caption': caption_match.group(1) if caption_match else "N/A"
}
logger.info("Image analysis completed successfully.")
return image_details
except requests.RequestException as e:
logger.error(f"GPT-Vision API communication failure. Error: {e}")
sys.exit(f"Exiting due to GPT-Vision API communication failure: {e}")
except Exception as e:
logger.error(f"Unexpected error occurred during image analysis: {e}")
sys.exit(f"Exiting due to an unexpected error: {e}")
# Example usage
if __name__ == "__main__":
image_path = "path/to/your/image.jpg"
try:
details = analyze_and_extract_details_from_image(image_path)
if details:
print(f"Extracted image details: {details}")
else:
print("No details extracted from the image.")
except SystemExit as e:
print(f"Terminated: {e}")

63
lib/gpt_providers/openai_chat_completion.py Normal file
View File

@@ -0,0 +1,63 @@
import time
import logging
import openai
import os
# Configure standard logging
logging.basicConfig(level=logging.INFO, format='[%(asctime)s-%(levelname)s-%(module)s-%(lineno)d]- %(message)s')
logger = logging.getLogger(__name__)
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
) # for exponential backoff
@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
def openai_chatgpt(prompt, model="gpt-4-1106-preview", temperature=0.2, max_tokens=4096, top_p=0.9, n=1):
"""
Wrapper function for OpenAI's ChatGPT completion.
Args:
prompt (str): The input text to generate completion for.
model (str, optional): Model to be used for the completion. Defaults to "gpt-4-1106-preview".
temperature (float, optional): Controls randomness. Lower values make responses more deterministic. Defaults to 0.2.
max_tokens (int, optional): Maximum number of tokens to generate. Defaults to 8192.
top_p (float, optional): Controls diversity. Defaults to 0.9.
n (int, optional): Number of completions to generate. Defaults to 1.
Returns:
str: The generated text completion.
Raises:
SystemExit: If an API error, connection error, or rate limit error occurs.
"""
# Wait for 10 seconds to comply with rate limits
for _ in range(10):
time.sleep(1)
try:
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
n=n,
top_p=top_p
# Additional parameters can be included here
)
return response.choices[0].message.content
except openai.APIError as e:
logger.error(f"OpenAI API Error: {e}")
raise SystemExit from e
except openai.APIConnectionError as e:
logger.error(f"Failed to connect to OpenAI API: {e}")
raise SystemExit from e
except openai.RateLimitError as e:
logger.error(f"Rate limit exceeded on OpenAI API request: {e}")
raise SystemExit from e
except Exception as err:
logger.error(f"OpenAI error: {err}")
raise SystemExit from e

53
lib/gpt_providers/openai_chat_completion_streaming.py Normal file
View File

@@ -0,0 +1,53 @@
import sys
import logging
import openai
# Configure standard logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def openai_chatgpt_streaming_text(user_prompt):
"""
Uses streaming functionality to get real-time output from OpenAI's GPT model.
Args:
user_prompt (str): The prompt to send to the model.
Returns:
str: The complete text generated by the model in response to the prompt.
Raises:
SystemExit: If an error occurs in connecting to the OpenAI API or during streaming.
"""
try:
client = openai.OpenAI()
response = client.chat.completions.create(
model="gpt-3.5-turbo-16k",
messages=[{"role": "user", "content": user_prompt}],
max_tokens=8192,
temperature=0.9,
n=1,
stream=True
)
collected_events = []
completion_text = ''
logger.info("Starting to receive streaming responses...")
for chunk in response:
collected_events.append(chunk) # Save the event response
event_text = chunk.choices[0].delta.content # Extract the text
completion_text += event_text # Append the text
sys.stdout.write(event_text)
sys.stdout.flush()
logger.info("Completed receiving streaming responses.")
return completion_text
except openai.OpenAIError as e:
logger.error(f"OpenAI API Error: {e}")
sys.exit("Exiting due to OpenAI API error.")
except Exception as e:
logger.error(f"Unexpected error during streaming: {e}")
sys.exit("Exiting due to an unexpected error.")

89
lib/gpt_providers/openai_gpt_provider.py
View File

@@ -20,7 +20,12 @@ import tempfile
from html2image import Html2Image
import datetime
from PIL import Image
import moviepy.editor as mp
import requests
from moviepy.editor import AudioFileClip
from concurrent.futures import ThreadPoolExecutor
from ..gpt_online_researcher import do_online_research
from loguru import logger
logger.remove()
@@ -29,8 +34,6 @@ logger.add(sys.stdout,
format="<level>{level}</level>|<green>{file}:{line}:{function}</green>| {message}"
)
def analyze_and_extract_details_from_image(image_path):
"""
Analyzes an image using OpenAI's Vision API and extracts Alt Text, Description, Title, and Caption.
@@ -103,12 +106,14 @@ def analyze_and_extract_details_from_image(image_path):
return image_details
except requests.RequestException as e:
sys.exit(f"Error: Failed to communicate with OpenAI API. Error: {e}")
#sys.exit(f"Error: GPT-Vision: Failed to communicate with OpenAI API. Error: {e}")
logger.error(f"Error: GPT-Vision: Failed to communicate with OpenAI API. Error: {e}")
except Exception as e:
sys.exit(f"Error occurred: {e}")
#sys.exit(f"Error occurred- GPT-Vision: {e}")
logger.error(f"Error occurred- GPT-Vision: {e}")
def openai_chatgpt(prompt, model="gpt-3.5-turbo-16k", temperature=0.2, max_tokens=8192, top_p=0.9, n=1):
def openai_chatgpt(prompt, model="gpt-4-1106-preview", temperature=0.2, max_tokens=4096, top_p=0.9, n=1):
"""
Wrapper function for openai chat Completion
"""
@@ -119,6 +124,10 @@ def openai_chatgpt(prompt, model="gpt-3.5-turbo-16k", temperature=0.2, max_token
try:
client = OpenAI()
except Exception as err:
print("Error: OpenAI Client.")
exit(1)
try:
# using OpenAI's Completion module that helps execute any tasks involving text
response = client.chat.completions.create(
# model name used, there are many other models available under the umbrella of GPT-3
@@ -142,6 +151,8 @@ def openai_chatgpt(prompt, model="gpt-3.5-turbo-16k", temperature=0.2, max_token
except openai.RateLimitError as e:
#Handle rate limit error (we recommend using exponential backoff)
SystemError(f"OpenAI API request exceeded rate limit: {e}")
except Exception as err:
SystemError(f"OpenAI client Error: {err}")
return response.choices[0].message.content
@@ -231,39 +242,57 @@ def generate_dalle3_images(img_prompt, image_dir, size="1024x1024", quality="hd"
return img_path
def speech_to_text(video_url):
""" Common openai function for speech to text. """
client = OpenAI()
def speech_to_text(video_url, output_path='.'):
""" Transcribes speech to text from a YouTube video URL. """
try:
# Download YouTube video
logger.info(f"Download YouTube video: {video_url}")
# Create a YouTube object
print(f"Accessing YouTube URL: {video_url}")
yt = YouTube(video_url)
stream = yt.streams.filter(only_audio=True).first()
# Save the video in a temporary file
logger.info(f"Finished Downloading, Saving video for transcription.")
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_file:
temp_file_name = temp_file.name
# Select the highest quality audio stream
print("Fetching audio stream. Select the highest quality audio stream")
audio_stream = yt.streams.filter(only_audio=True).first()
stream.download(output_path=os.path.dirname(temp_file_name), filename=os.path.basename(temp_file_name))
try:
# Transcribe the video using OpenAI's Whisper API
logger.info(f"Transcribe the video using OpenAI's Whisper API")
with open(temp_file_name, "rb") as audio_file:
if audio_stream is None:
print("No audio stream found for this video.")
return
else:
# Download the audio stream
print(f"Downloading audio for: {yt.title}")
audio_file = audio_stream.download(output_path)
print(f"Downloaded: {yt.title} to {output_path}")
try:
# Check if the audio file size is less than 24MB
max_file_size = 24 * 1024 * 1024 # 24MB in bytes
file_size = os.path.getsize(audio_file)
if file_size > max_file_size:
print("Error: File size exceeds 24MB limit.")
exit(1)
# File uploads are currently limited to 25 MB and the following input
# file types are supported: mp3, mp4, mpeg, mpga, m4a, wav, and webm.
try:
client = OpenAI()
except Exception as err:
SystemExit("Unable to get openai client object: {err}")
print("Transcribing using Openai whisper.")
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
model="whisper-1",
file=open(audio_file, "rb"),
response_format="text"
)
except Exception as err:
logger.error(f"Failed to transcribe using whisper model: {err}")
logger.info("Finished Transcribing. Creating a blog from the transcript.")
# Remove the temporary file after transcription
os.remove(temp_file_name)
return(transcript)
return transcript
except Exception as err:
print(f"Failed in whisper transcription: {err}")
exit(1)
except Exception as e:
logger.error(f"Error: speech-to-text, Failed to transcribe url: {video_url} with error: {e}")
print(f"YT video download, An error occurred: {e}")
exit(1)
os.remove(audio_file)
# The idea is to download images from other blogs and recreate from it.

35
lib/gpt_providers/save_image.py Normal file
View File

@@ -0,0 +1,35 @@
import datetime
import os
import requests
from PIL import Image
import logging
def save_generated_image(img_generation_response, image_dir):
"""
Save generated images for blog, ensuring unique names for SEO.
"""
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
generated_image_name = f"generated_image_{datetime.datetime.now():%Y-%m-%d-%H-%M-%S}.png"
generated_image_filepath = os.path.join(image_dir, generated_image_name)
generated_image_url = img_generation_response.data[0].url
logger.info(f"Fetch the image from url: {generated_image_url}")
try:
response = requests.get(generated_image_url, stream=True)
response.raise_for_status()
with open(generated_image_filepath, "wb") as image_file:
image_file.write(response.content)
except requests.exceptions.RequestException as e:
logger.error(f"Failed to get generated image content: {e}")
return None
logger.info(f"Saved image at path: {generated_image_filepath}")
if os.environ.get('DISPLAY', ''): # Check if display is supported
img = Image.open(generated_image_filepath)
img.show()
return generated_image_filepath

88
lib/gpt_providers/stt_audio_blog.py Normal file
View File

@@ -0,0 +1,88 @@
from pytube import YouTube
import os
import sys
from loguru import logger
from openai import OpenAI
from tqdm import tqdm
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
) # for exponential backoff
def progress_function(stream, chunk, bytes_remaining):
# Calculate the percentage completion
current = ((stream.filesize - bytes_remaining) / stream.filesize)
progress_bar.update(current - progress_bar.n) # Update the progress bar
@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
def speech_to_text(video_url, output_path='.'):
"""
Transcribes speech to text from a YouTube video URL using OpenAI's Whisper model.
Args:
video_url (str): URL of the YouTube video to transcribe.
output_path (str, optional): Directory where the audio file will be saved. Defaults to '.'.
Returns:
str: The transcribed text from the video.
Raises:
SystemExit: If a critical error occurs that prevents successful execution.
"""
try:
logger.info(f"Accessing YouTube URL: {video_url}")
yt = YouTube(video_url, on_progress_callback=progress_function)
logger.info("Fetching the highest quality audio stream")
audio_stream = yt.streams.filter(only_audio=True).first()
if audio_stream is None:
logger.warning("No audio stream found for this video.")
return None
#logger.info(f"Downloading audio for: {yt.title}")
global progress_bar
progress_bar = tqdm(total=1.0, unit='iB', unit_scale=True, desc=yt.title)
audio_file = audio_stream.download(output_path)
progress_bar.close()
logger.info(f"Audio downloaded: {yt.title} to {output_path}")
# Checking file size
max_file_size = 24 * 1024 * 1024 # 24MB
file_size = os.path.getsize(audio_file)
# Convert file size to MB for logging
file_size_MB = file_size / (1024 * 1024) # Convert bytes to MB
logger.info(f"Downloaded Audio Size is: {file_size_MB:.2f} MB")
if file_size > max_file_size:
logger.error("File size exceeds 24MB limit.")
sys.exit("File size limit exceeded.")
try:
logger.info("Initializing OpenAI client for transcription.")
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
logger.info("Transcribing using OpenAI's Whisper model.")
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=open(audio_file, "rb"),
response_format="text"
)
logger.info("\nYouTube video transcription:\n\n{transcript}\n")
return transcript, yt.title
except Exception as e:
logger.error(f"Failed in Whisper transcription: {e}")
sys.exit("Transcription failure.")
except Exception as e:
logger.error(f"An error occurred during YouTube video processing: {e}")
sys.exit("Video processing failure.")
finally:
if os.path.exists(audio_file):
os.remove(audio_file)
logger.info("Temporary audio file removed.")

74
lib/gpt_providers/stt_audio_blog.py.bk Normal file
View File

@@ -0,0 +1,74 @@
from pytube import YouTube
import os
import sys
from loguru import logger
from openai import OpenAI
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
) # for exponential backoff
@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
def speech_to_text(video_url, output_path='.'):
"""
Transcribes speech to text from a YouTube video URL using OpenAI's Whisper model.
Args:
video_url (str): URL of the YouTube video to transcribe.
output_path (str, optional): Directory where the audio file will be saved. Defaults to '.'.
Returns:
str: The transcribed text from the video.
Raises:
SystemExit: If a critical error occurs that prevents successful execution.
"""
try:
logger.info(f"Accessing YouTube URL: {video_url}")
yt = YouTube(video_url)
logger.info("Fetching the highest quality audio stream")
audio_stream = yt.streams.filter(only_audio=True).first()
if audio_stream is None:
logger.warning("No audio stream found for this video.")
return None
logger.info(f"Downloading audio for: {yt.title}")
audio_file = audio_stream.download(output_path)
logger.info(f"Audio downloaded: {yt.title} to {output_path}")
# Checking file size
max_file_size = 24 * 1024 * 1024 # 24MB
logger.info(f"Downloaded Audio Size is: {max_file_size}")
file_size = os.path.getsize(audio_file)
if file_size > max_file_size:
logger.error("File size exceeds 24MB limit.")
sys.exit("File size limit exceeded.")
try:
logger.info("Initializing OpenAI client for transcription.")
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
logger.info("Transcribing using OpenAI's Whisper model.")
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=open(audio_file, "rb"),
response_format="text"
)
return transcript, yt.title
except Exception as e:
logger.error(f"Failed in Whisper transcription: {e}")
sys.exit("Transcription failure.")
except Exception as e:
logger.error(f"An error occurred during YouTube video processing: {e}")
sys.exit("Video processing failure.")
finally:
if os.path.exists(audio_file):
os.remove(audio_file)
logger.info("Temporary audio file removed.")