diff --git a/alwrity.py b/alwrity.py index f0081710..dbf3dc79 100644 --- a/alwrity.py +++ b/alwrity.py @@ -18,6 +18,7 @@ app = typer.Typer() from lib.ai_web_researcher.gpt_online_researcher import gpt_web_researcher from lib.ai_web_researcher.metaphor_basic_neural_web_search import metaphor_find_similar from lib.ai_writers.keywords_to_blog import write_blog_from_keywords +from lib.ai_writers.speech_to_blog.main_audio_to_blog import generate_audio_blog def prompt_for_time_range(): @@ -32,7 +33,7 @@ def prompt_for_time_range(): def write_blog_options(): choices = [ ("Keywords", "Keywords"), - ("Audio YouTube", "Audio YouTube"), + ("Audio To Blog", "Audio To Blog"), ("Programming", "Programming"), ("Scholar", "Scholar"), ("News/TBD", "News/TBD"), @@ -195,9 +196,10 @@ def write_blog(): if blog_type: if blog_type == 'Keywords': blog_from_keyword() - elif blog_type == 'Audio YouTube': - audio_youtube = prompt("Enter YouTube URL for audio blog generation:") - print(f"Write audio blog based on YouTube URL: {audio_youtube}") + + elif blog_type == 'Audio To Blog': + blog_from_audio() + elif blog_type == 'GitHub': github = prompt("Enter GitHub URL, CSV file, or topic:") print(f"Write blog based on GitHub: {github}") @@ -209,6 +211,30 @@ def write_blog(): raise typer.Exit() +def blog_from_audio(): + """ + Prompt the user to input either a YouTube URL, a file location, or keywords to search on YouTube. + Validate the input and take appropriate actions based on the input type. + """ + + while True: + print("https://github.com/AJaySi/AI-Blog-Writer/wiki/Audio-to-blog-AI-article-writer-%E2%80%90-Alwrity-Speech-To-Text-Feature") + audio_input = prompt("""Enter Youtube video URL OR provide Full-Path to audio file.\nšŸ‘‹ : """) + # If the user cancels, exit the loop and the application + if audio_input is None: + break + + # If the user presses OK without providing any input, prompt again + if not audio_input.strip(): + continue + + # Check if the input is a valid YouTube URL + if audio_input.startswith("https://www.youtube.com/") or audio_input.startswith("http://www.youtube.com/") or os.path.exists(audio_input): + # Validate YouTube URL, Process YouTube URL + generate_audio_blog(audio_input) + break + + def blog_from_keyword(): """ Input blog keywords, research and write a factual blog.""" while True: diff --git a/lib/ai_writers/speech_to_blog/main_audio_to_blog.py b/lib/ai_writers/speech_to_blog/main_audio_to_blog.py new file mode 100644 index 00000000..f3515220 --- /dev/null +++ b/lib/ai_writers/speech_to_blog/main_audio_to_blog.py @@ -0,0 +1,70 @@ +import os +import datetime #I wish +import sys +from textwrap import dedent +import openai +from tqdm import tqdm, trange +import time + +from loguru import logger +logger.remove() +logger.add(sys.stdout, + colorize=True, + format="{level}|{file}:{line}:{function}| {message}" + ) + +from .write_blogs_from_youtube_videos import youtube_to_blog +from ...ai_web_researcher.gpt_online_researcher import do_google_serp_search +from ...ai_writers.combine_research_and_blog import blog_with_research +from ...blog_metadata.get_blog_metadata import blog_metadata +from ...blog_postprocessing.save_blog_to_file import save_blog_to_file + + +def generate_audio_blog(audio_input): + """Takes a list of youtube videos and generates blog for each one of them. + """ + # Use to store the blog in a string, to save in a *.md file. + blog_markdown_str = "" + try: + logger.info(f"Starting to write blog on URL: {audio_input}") + yt_blog, yt_title = youtube_to_blog(audio_input) + except Exception as e: + logger.error(f"Error in youtube_to_blog: {e}") + sys.exit(1) + + try: + logger.info("Starting with online research for URL title.") + research_report = do_google_serp_search(yt_title) + print(research_report) + except Exception as e: + logger.error(f"Error in do_online_research: {e}") + sys.exit(1) + + try: + # Note: Check if the order of input matters for your function + logger.info("Preparing a blog content from audio script and online research content...") + blog_markdown_str = blog_with_research(research_report, yt_blog) + except Exception as e: + logger.error(f"Error in blog_with_research: {e}") + sys.exit(1) + + try: + blog_title, blog_meta_desc, blog_tags, blog_categories = blog_metadata(blog_markdown_str) + except Exception as err: + logger.error(f"Failed to generate blog metadata: {err}") + + try: + # TBD: Save the blog content as a .md file. Markdown or HTML ? + save_blog_to_file(blog_markdown_str, blog_title, blog_meta_desc, blog_tags, blog_categories, generated_image_filepath) + except Exception as err: + logger.error(f"Failed to save final blog in a file: {err}") + + blog_frontmatter = dedent(f"""\n\n\n\ + --- + title: {blog_title} + categories: [{blog_categories}] + tags: [{blog_tags}] + Meta description: {blog_meta_desc.replace(":", "-")} + ---\n\n""") + logger.info(f"{blog_frontmatter}{blog_markdown_str}") + logger.info(f"\n\n ################ Finished writing Blog for : {audio_input} #################### \n") diff --git a/lib/speech_to_blog/write_blogs_from_youtube_videos.py b/lib/ai_writers/speech_to_blog/write_blogs_from_youtube_videos.py similarity index 75% rename from lib/speech_to_blog/write_blogs_from_youtube_videos.py rename to lib/ai_writers/speech_to_blog/write_blogs_from_youtube_videos.py index b68d0216..d0847ab6 100644 --- a/lib/speech_to_blog/write_blogs_from_youtube_videos.py +++ b/lib/ai_writers/speech_to_blog/write_blogs_from_youtube_videos.py @@ -16,17 +16,15 @@ logger.add(sys.stdout, format="{level}|{file}:{line}:{function}| {message}" ) -from .gpt_providers.stt_audio_blog import speech_to_text -from .gpt_providers.openai_chat_completion import openai_chatgpt + +from ...gpt_providers.audio_to_text_generation.stt_audio_blog import speech_to_text +from ...gpt_providers.text_generation.main_text_generation import llm_text_gen def youtube_to_blog(video_url): """Function to transcribe a given youtube url """ # fixme: Doesnt work all types of yt urls. vid_id = video_url.split("=")[1] - #hti = Html2Image(output_path="../blog_images") - #hti.screenshot(url=video_url, save_as=f"yt-img-{vid_id}.png") - #yt_img_path = os.path.join("../blog_images", f"yt-img-{vid_id}.png") try: # Starting the speech-to-text process @@ -44,7 +42,6 @@ def youtube_to_blog(video_url): except Exception as e: logger.error(f"Error in summarize_youtube_video: {e}") sys.exit(1) # Exit the program due to error in summarize_youtube_video - return audio_blog_content def summarize_youtube_video(user_content, gpt_providers): @@ -77,21 +74,9 @@ def summarize_youtube_video(user_content, gpt_providers): that will rank well in search engine results and engage readers effectively. Follow above guidelines to craft a blog content from the following transcript:\n{user_content} """ - if 'gemini' in gpt_providers: - try: - genai.configure(api_key=os.getenv('GEMINI_API_KEY')) - except Exception as err: - logger.error("Failed in getting GEMINI_API_KEY") - # Use gemini-pro model for text and image. - model = genai.GenerativeModel('gemini-pro') - try: - response = model.generate_content(prompt) - return response.text - except Exception as err: - logger.error("Failed to get response from gemini.") - elif 'openai' in gpt_providers: - try: - response = openai_chatgpt(prompt) - return response - except Exception as err: - SystemError(f"Error in generating blog summary: {err}") + try: + response = llm_text_gen(prompt) + return response + except Exception as err: + logger.error(f"Failed to summarize_youtube_video: {err}") + exit(1) diff --git a/lib/blog_metadata/get_blog_metadata.py b/lib/blog_metadata/get_blog_metadata.py index 59e91dae..65ee63bf 100644 --- a/lib/blog_metadata/get_blog_metadata.py +++ b/lib/blog_metadata/get_blog_metadata.py @@ -13,7 +13,7 @@ logger.add(sys.stdout, ) -def blog_metadata(blog_content, search_keywords, blog_titles): +def blog_metadata(blog_content, search_keywords=None, blog_titles=None): """ Common function to get blog metadata """ blog_title = generate_blog_title(blog_content, search_keywords, blog_titles) blog_meta_desc = generate_blog_description(blog_content) diff --git a/lib/gpt_providers/audio_to_text_generation/stt_audio_blog.py b/lib/gpt_providers/audio_to_text_generation/stt_audio_blog.py index c3012ec7..3b403b27 100644 --- a/lib/gpt_providers/audio_to_text_generation/stt_audio_blog.py +++ b/lib/gpt_providers/audio_to_text_generation/stt_audio_blog.py @@ -34,22 +34,27 @@ def speech_to_text(video_url, output_path='.'): 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) + audio_file = None + if video_url.startswith("https://www.youtube.com/") or video_url.startswith("http://www.youtube.com/"): + 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() + 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 + 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}") + 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}") + # Audio filepath from local directory. + elif os.path.exists(audio_input): + audio_file = video_url # Checking file size max_file_size = 24 * 1024 * 1024 # 24MB @@ -59,6 +64,8 @@ def speech_to_text(video_url, output_path='.'): 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.") + # FIXME: We can chunk hour long videos, the code is not tested. + #long_video(audio_file) sys.exit("File size limit exceeded.") try: @@ -86,3 +93,43 @@ def speech_to_text(video_url, output_path='.'): if os.path.exists(audio_file): os.remove(audio_file) logger.info("Temporary audio file removed.") + + +def long_video(temp_file_name): + """ + Transcribes a YouTube video using OpenAI's Whisper API by processing the video in chunks. + + This function handles videos longer than the context limit of the Whisper API by dividing the video into + 10-minute segments, transcribing each segment individually, and then combining the results. + + Key Changes and Notes: + 1. Video Splitting: Splits the audio into 10-minute chunks using the moviepy library. + 2. Chunk Transcription: Each audio chunk is transcribed separately and the results are concatenated. + 3. Temporary Files for Chunks: Uses temporary files for each audio chunk for transcription. + 4. Error Handling: Exception handling is included to capture and return any errors during the process. + 5. Logging: Process steps are logged for debugging and monitoring. + 6. Cleaning Up: Removes temporary files for both the entire video and individual audio chunks after processing. + + Args: + video_url (str): URL of the YouTube video to be transcribed. + """ + # Extract audio and split into chunks + app.logger.info(f"Processing the YT video: {temp_file_name}") + full_audio = mp.AudioFileClip(temp_file_name) + duration = full_audio.duration + chunk_length = 600 # 10 minutes in seconds + chunks = [full_audio.subclip(start, min(start + chunk_length, duration)) for start in range(0, int(duration), chunk_length)] + + combined_transcript = "" + for i, chunk in enumerate(chunks): + with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as audio_chunk_file: + chunk.write_audiofile(audio_chunk_file.name, codec="mp3") + with open(audio_chunk_file.name, "rb") as audio_file: + # Transcribe each chunk using OpenAI's Whisper API + app.logger.info(f"Transcribing chunk {i+1}/{len(chunks)}") + transcript = openai.Audio.transcribe("whisper-1", audio_file) + combined_transcript += transcript['text'] + "\n\n" + + # Remove the chunk audio file + os.remove(audio_chunk_file.name) + diff --git a/lib/gpt_providers/text_generation/gemini_pro_text.py b/lib/gpt_providers/text_generation/gemini_pro_text.py index 974db369..a015dc3b 100644 --- a/lib/gpt_providers/text_generation/gemini_pro_text.py +++ b/lib/gpt_providers/text_generation/gemini_pro_text.py @@ -35,7 +35,7 @@ def gemini_text_response(prompt, temperature, top_p, n, max_tokens): "top_k": n, "max_output_tokens": max_tokens } - model = genai.GenerativeModel(model_name="gemini-pro", generation_config=generation_config) + model = genai.GenerativeModel(model_name="gemini-1.0-pro", generation_config=generation_config) try: response = model.generate_content(prompt, stream=True) for chunk in response: diff --git a/lib/speech_to_blog/main_audio_to_blog.py b/lib/speech_to_blog/main_audio_to_blog.py deleted file mode 100644 index 4d715985..00000000 --- a/lib/speech_to_blog/main_audio_to_blog.py +++ /dev/null @@ -1,114 +0,0 @@ -import json -import os -import datetime #I wish -import sys - -import openai -from tqdm import tqdm, trange -import time -import re -from textwrap import dedent -import nltk -nltk.download('punkt', quiet=True) -from nltk.corpus import stopwords -nltk.download('stopwords', quiet=True) - -from .write_blogs_from_youtube_videos import youtube_to_blog -from .wordpress_blog_uploader import compress_image, upload_blog_post, upload_media -from .gpt_online_researcher import do_online_research - -from loguru import logger -logger.remove() -logger.add(sys.stdout, - colorize=True, - format="{level}|{file}:{line}:{function}| {message}" - ) - - -def generate_youtube_blog(yt_url_list, output_format="markdown"): - """Takes a list of youtube videos and generates blog for each one of them. - """ - # Use to store the blog in a string, to save in a *.md file. - blog_markdown_str = "" - for a_yt_url in yt_url_list: - try: - logger.info(f"Starting to write blog on URL: {a_yt_url}") - yt_blog = youtube_to_blog(a_yt_url) - except Exception as e: - logger.error(f"Error in youtube_to_blog: {e}") - sys.exit(1) - - try: - logger.info("Starting with online research for URL title.") - research_report = do_online_research(yt_blog) - except Exception as e: - logger.error(f"Error in do_online_research: {e}") - sys.exit(1) - - try: - # Note: Check if the order of input matters for your function - logger.info("Preparing a blog content from audio script and online research content...") - blog_with_research(research_report, yt_blog) - except Exception as e: - logger.error(f"Error in blog_with_research: {e}") - sys.exit(1) - - try: - # Get the title and meta description of the blog. - blog_meta_desc = generate_blog_description(yt_blog) - title = generate_blog_title(blog_meta_desc) - logger.info(f"Title is {title} and description is {blog_meta_desc}") - blog_markdown_str = "# " + title.replace('"', '') + "\n\n" - # Get blog tags and categories. - blog_tags = get_blog_tags(blog_meta_desc) - logger.info(f"Blog tags are: {blog_tags}") - blog_categories = get_blog_categories(blog_meta_desc) - logger.info(f"Blog categories are: {blog_categories}") - - # Generate an introduction for the blog - blog_intro = get_blog_intro(title, yt_blog) - logger.info(f"The Blog intro is:\n {blog_intro}") - blog_markdown_str = blog_markdown_str + "\n\n" + f"{blog_intro}" + "\n\n" - - # Generate an image based on meta description - logger.info(f"Calling Image generation with prompt: {blog_meta_desc}") - main_img_path = generate_image(blog_meta_desc, image_dir, "dalle3") - - # Get a variation of the yt url screenshot to use in the blog. - #varied_img_path = gen_new_from_given_img(yt_img_path, image_dir) - #logger.info(f"Image path: {main_img_path} and varied path: {varied_img_path}") - #blog_markdown_str = blog_markdown_str + f'![img-description]({os.path.basename(varied_img_path)})' + '_Image Caption_' - - #stbdiff_img_path = generate_image(yt_img_path, image_dir, "stable_diffusion") - #logger.info(f"Image path: {main_img_path} from stable diffusion: {stbdiff_img_path}") - #blog_markdown_str = blog_markdown_str + f'![img-description]({os.path.basename(stbdiff_img_path)})' + f'_{title}_' - - # Add the body of the blog content. - blog_markdown_str = blog_markdown_str + "\n\n" + f'{yt_blog}' + "\n\n" - - # Get the Conclusion of the blog, by passing the generated blog. - blog_conclusion = get_blog_conclusion(blog_markdown_str) - # TBD: Add another image. - blog_markdown_str = blog_markdown_str + "### Conclusion" + "\n\n" + f"{blog_conclusion}" + "\n" - - # Proofread the blog, edit and remove dubplicates and refine it further. - # Presently, fixing the blog keywords to be tags and categories. - blog_keywords = f"{blog_tags} + {blog_categories}" - blog_markdown_str = blog_proof_editor(blog_markdown_str, blog_keywords) - - # Check the type of blog format needed by the user. - if 'html' in output_format: - blog_markdown_str = convert_tomarkdown_format(blog_markdown_str) - elif 'markdown' in output_path: - blog_markdown_str = convert_markdown_to_html(blog_markdown_str) - - # Try to save the blog content in a file, in whichever format. Just dump it. - try: - save_blog_to_file(blog_markdown_str, title, blog_meta_desc, blog_tags, blog_categories, main_img_path) - except Exception as err: - logger.error("Failed to Save blog content: {blog_markdown_str}") - - except Exception as e: - # raise assertionerror - logger.error(f"Error: Failed to generate_youtube_blog: {e}") - exit(1) diff --git a/lib/speech_to_blog/main_youtube_research_blog.py b/lib/speech_to_blog/main_youtube_research_blog.py deleted file mode 100644 index 43558fbc..00000000 --- a/lib/speech_to_blog/main_youtube_research_blog.py +++ /dev/null @@ -1,150 +0,0 @@ -import json -import os -import sys -from loguru import logger - -# Import from local packages -from .gpt_providers.openai_chat_completion import openai_chatgpt -from .gpt_providers.gpt_vision_img_details import analyze_and_extract_details_from_image -from .generate_image_from_prompt import generate_image -from .write_blogs_from_youtube_videos import youtube_to_blog -from .wordpress_blog_uploader import compress_image, upload_blog_post, upload_media -from .gpt_online_researcher import do_online_research -from .save_blog_to_file import save_blog_to_file -from .optimize_images_for_upload import optimize_image -from .combine_research_and_blog import blog_with_research -from .get_blog_meta_desc import generate_blog_description -from .get_blog_title import generate_blog_title -from .get_tags import get_blog_tags -from .get_blog_category import get_blog_categories -from .convert_content_to_markdown import convert_tomarkdown_format -from .convert_markdown_to_html import convert_markdown_to_html -from .utils.youtube_keyword_research import research_yt - -# Configuring the logger -logger.remove() -logger.add(sys.stdout, colorize=True, format="{level}|{file}:{line}:{function}| {message}") - -# Constants for directory paths -IMAGE_DIR = os.path.join(os.getcwd(), "blog_images") -OUTPUT_PATH = os.path.join(os.getcwd(), "blogs") - - -def generate_youtube_research_blog(yt_keywords): - """ - Research YouTube based on given keywords and get top video URLs. - """ - for ayt_keyword in yt_keywords: - yt_research_response = '' - data = {} - logger.info(f"Researching YouTube top videos for: {yt_keywords}") - try: - yt_research_response = research_yt(ayt_keyword) - if not yt_research_response: - yt_research_response = research_yt(ayt_keyword) - except Exception as err: - logger.error(f"Failed to do YouTube Research: {err}") - - if not yt_research_response.strip(): - logger.warning("Error: JSON data is empty.") - yt_research_response = research_yt(ayt_keyword) - else: - try: - aggregated_data = load_response_json(yt_research_response, ayt_keyword) - except Exception as err: - logger.error(f"Failed to load json response: {err}") - sys.exit(1) - - for title, a_yt_url, views, references, quickstart_code in zip( - aggregated_data["titles"], aggregated_data["urls"], aggregated_data["views"], - aggregated_data["references"], aggregated_data["quickstart_codes"]): - blog_markdown_str = "" - if a_yt_url != "No URL Provided": - # Transcribe the audio using whisper model. - try: - logger.info(f"Starting to write blog on URL: {a_yt_url}") - blog_markdown_str, yt_title = youtube_to_blog(a_yt_url) - logger.warning("\n\n--------------- First Draft of the Blog: --------\n\n") - logger.info(f"{blog_markdown_str}\n") - logger.warning("--------------------END of First draft----------\n\n") - if not yt_title or not blog_markdown_str: - logger.error("No content or title for audio to proceed.") - sys.exit(1) - except Exception as e: - logger.error(f"Error in youtube_to_blog: {e}") - sys.exit(1) - sys.exit(1) - - if title != "Unknown Title": - print(f"Title: {title}") - if url != "No URL Provided": - print(f"URL: {url}") - if views != "No View Count": - print(f"Views: {views}") - if references: # Checks if references list is not empty - print(f"References: {', '.join(references)}") - if quickstart_code != "Code coming soon": - print(f"Quickstart Code: {quickstart_code}") - print() # Adds a newline for separation between entries - - - -def load_response_json(yt_research_response, yt_keyword): - """ - Load and parse the YouTube research response JSON. - """ - try: - logger.info(f"Loading the JSON data for parsing: {yt_research_response}") - data = json.loads(yt_research_response.replace('`', '').strip()) - - if isinstance(data, dict): - results_key = next((key for key in data if key.lower().startswith("result")), None) - if results_key: - research_yt_dict = process_results(data[results_key]) - elif isinstance(data, list): - research_yt_dict = process_results(data) - - except json.JSONDecodeError as e: - logger.error(f"load_response_json: Failed to parse JSON data: {e}") - generate_youtube_research_blog([yt_keyword]) - - return research_yt_dict - - -def process_results(results): - """ - Process the results from the YouTube research JSON and return the aggregated data. - - Args: - results (list): List of dictionaries containing YouTube video details. - - Returns: - dict: A dictionary containing lists of titles, URLs, views, references, and quickstart codes. - - Raises: - Exception: If an error occurs during the processing of individual entries. - """ - titles = [] - urls = [] - views_list = [] - references_list = [] - quickstart_codes = [] - - for entry in results: - try: - titles.append(entry.get("Title", "Unknown Title")) - urls.append(entry.get("URL", "No URL Provided")) - views_list.append(entry.get("Views", "No View Count")) - references_list.append(entry.get("References", [])) - quickstart_codes.append(entry.get("Quickstart_Code", "Code coming soon")) - except Exception as e: - logger.error(f"Error processing yt resulr entry: {e}") - continue - - return { - "titles": titles, - "urls": urls, - "views": views_list, - "references": references_list, - "quickstart_codes": quickstart_codes - } diff --git a/lib/utils/youtube_keyword_research.py b/lib/utils/youtube_keyword_research.py deleted file mode 100644 index afe7880c..00000000 --- a/lib/utils/youtube_keyword_research.py +++ /dev/null @@ -1,57 +0,0 @@ -""" -At the command line, only need to run once to install the package via pip: - -$ pip install google-generativeai -""" -import os -import sys - -import google.generativeai as genai - -def research_yt(keywords): - """ Research top youtube videos for given keywords """ - try: - genai.configure(api_key=os.getenv('GEMINI_API_KEY')) - except Exception as err: - print("Google Gemini Error: {err}") - - # Set up the model - generation_config = { - "temperature": 0.9, - "top_p": 1, - "top_k": 1, - "max_output_tokens": 2048, - } - - safety_settings = [ - { - "category": "HARM_CATEGORY_HARASSMENT", - "threshold": "BLOCK_MEDIUM_AND_ABOVE" - }, - { - "category": "HARM_CATEGORY_HATE_SPEECH", - "threshold": "BLOCK_MEDIUM_AND_ABOVE" - }, - { - "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", - "threshold": "BLOCK_MEDIUM_AND_ABOVE" - }, - { - "category": "HARM_CATEGORY_DANGEROUS_CONTENT", - "threshold": "BLOCK_MEDIUM_AND_ABOVE" - }, - ] - - model = genai.GenerativeModel(model_name="gemini-pro", - generation_config=generation_config, - safety_settings=safety_settings) - - prompt_parts = [f"Research 5 latest youtube urls on {keywords}, released this week. Check the number of views and also get the references from youtube video description. REMEMBER to make sure, your response urls are available and valid. For each result, visit their webpages to write detailed quickstart code samples, preferably in python. Your response urls should consist of trending topics on latest {keywords}. Your response should be in json format, so that i can easily parse all the fields. For consistency, always use json key names as Title, URL, Views, References and Quickstart_Code."] - - try: - response = model.generate_content(prompt_parts) - except Exception as err: - print(f"Failed to get response from Gemini Pro.{response}") - sys.exit(1) - - return response.text