WIP - Streamlit UI, firecrawl - V0.5

lib/ai_writers/image_ai_writer.py

@@ -0,0 +1,112 @@
+import sys
+import os
+
+from textwrap import dedent
+from PIL import Image
+import json
+from pathlib import Path
+from datetime import datetime
+import streamlit as st
+
+from dotenv import load_dotenv
+load_dotenv(Path('../../.env'))
+from loguru import logger
+logger.remove()
+logger.add(sys.stdout,
+        colorize=True,
+        format="<level>{level}</level>|<green>{file}:{line}:{function}</green>| {message}"
+    )
+
+from ..ai_web_researcher.firecrawl_web_crawler import scrape_url
+from ..blog_metadata.get_blog_metadata import blog_metadata
+from ..blog_postprocessing.save_blog_to_file import save_blog_to_file
+from ..gpt_providers.text_to_image_generation.main_generate_image_from_prompt import generate_image
+from ..gpt_providers.text_generation.main_text_generation import llm_text_gen
+
+import google.generativeai as genai
+
+
+def blog_from_image(prompt, uploaded_img):
+    """
+    This function will take a blog Topic to first generate sections for it
+    and then generate content for each section.
+    """
+    # Use to store the blog in a string, to save in a *.md file.
+    blog_markdown_str = None
+    logger.info(f"Researching and Writing Blog on {uploaded_img} and {prompt}")
+    # FIXME: Implement support for Openai.
+    if not os.getenv("GEMINI_API_KEY"):
+        st.error("Only Gemini supported, Open Issue ticket on github for Openai, others.")
+        st.stop()
+
+    with st.status("Started Writing from Image..", expanded=True) as status:
+        st.empty()
+        status.update(label=f"Researching and Writing Blog on given Image")
+        try:
+            blog_markdown_str = write_blog_from_image(prompt, uploaded_img)
+        except Exception as err:
+            st.error(f"Failed to write blog from Image - Error: {err}")
+            logger.error(f"Failed to write blog from image: {err}")
+            st.stop()
+        status.update(label="Successfully wrote blog from image.", expanded=False, state="complete")
+
+        try:
+            status.update(label="🙎 Generating - Title, Meta Description, Tags, Categories for the content.")
+            blog_title, blog_meta_desc, blog_tags, blog_categories = blog_metadata(blog_markdown_str)
+        except Exception as err:
+            st.error(f"Failed to get blog metadata: {err}")
+
+        try:
+            status.update(label="🙎 Generating Image for the new blog.")
+            generated_image_filepath = generate_image(f"{blog_title} + ' ' + {blog_meta_desc}")
+        except Exception as err:
+            st.warning(f"Failed in Image generation: {err}")
+
+        saved_blog_to_file = save_blog_to_file(blog_markdown_str, blog_title, blog_meta_desc, 
+                            blog_tags, blog_categories, generated_image_filepath)
+        status.update(label=f"Saved the content in this file: {saved_blog_to_file}")
+        logger.info(f"\n\n --------- Finished writing Blog -------------- \n")
+        st.image(generated_image_filepath, caption=blog_title)
+        st.markdown(f"{blog_markdown_str}")
+        status.update(label=f"Finished, Review & Use your Original Content Below: {saved_blog_to_file}", state="complete")
+
+        # Clean up the temporary file after processing (optional)
+        os.remove(uploaded_img)
+
+
+def write_blog_from_image(prompt, uploaded_img):
+    """Combine the given online research and GPT blog content"""
+    try:
+        config_path = Path(os.environ["ALWRITY_CONFIG"])
+        with open(config_path, 'r', encoding='utf-8') as file:
+            config = json.load(file)
+    except Exception as err:
+        logger.error(f"Error: Failed to read values from config: {err}")
+        exit(1)
+
+    blog_characteristics = config['Blog Content Characteristics']
+    
+    if not prompt:
+        prompt = f"""
+            As expert Creative Content writer, analyse the given image carefully.
+            I want you to write a detailed {blog_characteristics['Blog Type']} blog post including 5 FAQs.
+        
+            Below are the guidelines to follow:
+            1). You must respond in {blog_characteristics['Blog Language']} language.
+            2). Tone and Brand Alignment: Adjust your tone, voice, personality for {blog_characteristics['Blog Tone']} audience.
+            3). Make sure your response content length is of {blog_characteristics['Blog Length']} words.
+        """
+    logger.info("Generating blog and FAQs from Google web search results.")
+    
+    try:
+        #response = llm_text_gen(prompt)
+        genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
+        version = 'models/gemini-1.5-flash'
+        model = genai.GenerativeModel(version)
+        model_info = genai.get_model(version)
+        print(f'{version} - input limit: {model_info.input_token_limit}, output limit: {model_info.output_token_limit}')
+        response = model.generate_content([prompt, Image.open(uploaded_img)])
+        return response.text
+    except Exception as err:
+        logger.error(f"Exit: Failed to get response from LLM: {err}")
+        exit(1)

lib/ai_writers/web_url_ai_writer.py

@@ -71,19 +71,10 @@ def blog_from_url(weburl):
         saved_blog_to_file = save_blog_to_file(blog_markdown_str, blog_title, blog_meta_desc, 
                             blog_tags, blog_categories, generated_image_filepath)
         status.update(label=f"Saved the content in this file: {saved_blog_to_file}")
-        blog_frontmatter = dedent(f"""
-        \n---------------------------------------------------------------------
-        title: {blog_title}\n
-        categories: [{blog_categories}]\n
-        tags: [{blog_tags}]\n
-        Meta description: {blog_meta_desc.replace(":", "-")}\n
-        ---------------------------------------------------------------------\n
-        """)
         logger.info(f"\n\n --------- Finished writing Blog for : {weburl} -------------- \n")
-        st.markdown(f"{blog_frontmatter}")
         st.image(generated_image_filepath)
         st.markdown(f"{blog_markdown_str}")
-        status.update(label=f"Finished, Review & Use your Original Content Below: {saved_blog_to_file}")
+        status.update(label=f"Finished, Review & Use your Original Content Below: {saved_blog_to_file}", state="complete")
         
 
 def write_blog_from_weburl(scraped_website):