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

refactor: modularize code by creating separate utility modules for environment, configuration, UI setup, API key management, content generation, SEO tools, file processing, and voice processing

Browse Source
This commit is contained in:
ajaysi (aider)
2024-09-14 17:49:34 +05:30
parent 6fe6c52d99
commit 27072adbe5
10 changed files with 302 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
alwrity.py
View File

@@ -1,21 +1,12 @@
import os
import json
import base64
from datetime import datetime
import streamlit as st
# Load .env file
def load_environment():
from dotenv import load_dotenv
load_dotenv()
from lib.utils.alwrity_utils import (
blog_from_keyword, ai_agents_team, essay_writer, ai_news_writer, ai_seo_tools,
ai_finance_ta_writer, ai_social_writer, do_web_research, competitor_analysis
)
from lib.ai_writers.ai_story_writer.story_writer import story_input_section
from lib.ai_writers.ai_product_description_writer import write_ai_prod_desc
from lib.content_planning_calender.content_planning_agents_alwrity_crew import ai_agents_content_planner
from lib.utils.environment_utils import load_environment
from lib.utils.config_manager import save_config
from lib.utils.ui_setup import setup_ui, setup_tabs
from lib.utils.api_key_manager import check_api_keys, check_llm_environs
from lib.utils.content_generators import ai_writers, content_planning_tools
from lib.utils.seo_tools import ai_seo_tools
from lib.utils.file_processor import load_image, read_prompts, write_prompts
from lib.utils.voice_processing import record_voice
# Placeholder function definitions for missing functions
def blog_from_audio():

52
lib/utils/alwrity_utils.py
View File

@@ -1,27 +1,7 @@
import os
import re
import sys
import streamlit as st
from streamlit_mic_recorder import speech_to_text
import asyncio
import tempfile
from pathlib import Path
import configparser
from datetime import datetime
import uuid
from PIL import Image
import PyPDF2
import openai
import tiktoken
from docx import Document
from loguru import logger
logger.remove()
logger.add(sys.stdout,
colorize=True,
format="<level>{level}</level>|<green>{file}:{line}:{function}</green>| {message}"
)
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_streamlit import write_blog_from_keywords
@@ -38,43 +18,11 @@ from lib.ai_writers.youtube_ai_writer import write_yt_title, write_yt_descriptio
from lib.ai_writers.web_url_ai_writer import blog_from_url
from lib.ai_writers.image_ai_writer import blog_from_image
from lib.ai_writers.ai_essay_writer import ai_essay_generator
from lib.ai_seo_tools.seo_structured_data import ai_structured_data
from lib.ai_seo_tools.content_title_generator import ai_title_generator
from lib.ai_seo_tools.meta_desc_generator import metadesc_generator_main
from lib.ai_seo_tools.image_alt_text_generator import alt_text_gen
from lib.ai_seo_tools.opengraph_generator import og_tag_generator
from lib.ai_seo_tools.optimize_images_for_upload import main_img_optimizer
from lib.ai_seo_tools.google_pagespeed_insights import google_pagespeed_insights
from lib.ai_seo_tools.on_page_seo_analyzer import analyze_onpage_seo
from lib.ai_seo_tools.weburl_seo_checker import url_seo_checker
from lib.gpt_providers.text_to_image_generation.main_generate_image_from_prompt import generate_image
from lib.content_planning_calender.content_planning_agents_alwrity_crew import ai_agents_content_planner
from ..gpt_providers.text_generation.main_text_generation import llm_text_gen
@st.cache_data
def record_voice(language="en"):
# https://github.com/B4PT0R/streamlit-mic-recorder?tab=readme-ov-file#example
state = st.session_state
if "text_received" not in state:
state.text_received = []
text = speech_to_text(
start_prompt="🎙️Press & Speak🔊",
stop_prompt="🔇Stop Recording🚨",
language=language,
use_container_width=True,
just_once=False,
)
if text:
state.text_received.append(text)
result = ""
for text in state.text_received:
result += text
state.text_received = []
return result if result else None
def is_youtube_link(text):
if text is not None:
youtube_regex = re.compile(r'(https?://)?(www\.)?(youtube|youtu|youtube-nocookie)\.(com|be)/(watch\?v=|embed/|v/|.+\?v=)?([^&=%\?]{11})')

71
lib/utils/api_key_manager.py Normal file
View File

@@ -0,0 +1,71 @@
import os
import streamlit as st
@st.cache_data
def check_api_keys():
"""
Checks if the required API keys are present in the environment variables.
Prompts the user to enter missing keys and saves them in the .env file.
"""
api_keys = {
"METAPHOR_API_KEY": "https://dashboard.exa.ai/login",
"TAVILY_API_KEY": "https://tavily.com/#api",
"SERPER_API_KEY": "https://serper.dev/signup",
"STABILITY_API_KEY": "https://platform.stability.ai/",
"FIRECRAWL_API_KEY": "https://www.firecrawl.dev/account"
}
missing_keys = {
key: url for key, url in api_keys.items() if os.getenv(key) is None
}
if missing_keys:
st.error("🚨 Some API keys are missing! Please provide them below:")
for key, url in missing_keys.items():
api_key = st.text_input(f"Enter 🔏 {key}: 👉[Get it here]({url})👈")
if api_key:
os.environ[key] = api_key
try:
with open(".env", "a") as env_file:
env_file.write(f"{key}={api_key}\n")
except IOError as e:
st.error(f"Failed to write {key} to .env file: {e}")
st.success(f"✅ {key} added successfully!")
return False
return True
@st.cache_data
def check_llm_environs():
"""
Ensures that the LLM provider and corresponding API key are set.
Prompts the user to select a provider and enter the API key if missing.
"""
gpt_provider = os.getenv("GPT_PROVIDER")
supported_providers = {
'google': "GEMINI_API_KEY",
'openai': "OPENAI_API_KEY",
'mistralai': "MISTRAL_API_KEY"
}
if not gpt_provider or gpt_provider.lower() not in supported_providers:
gpt_provider = st.selectbox(
"Select your LLM Provider", options=list(supported_providers.keys())
)
os.environ["GPT_PROVIDER"] = gpt_provider
try:
with open(".env", "a") as env_file:
env_file.write(f"GPT_PROVIDER={gpt_provider}\n")
except IOError as e:
st.error(f"Failed to write GPT_PROVIDER to .env file: {e}")
st.success(f"GPT Provider set to {gpt_provider}")
api_key_var = supported_providers[gpt_provider.lower()]
if not os.getenv(api_key_var):
api_key = st.text_input(f"Enter {api_key_var}:")
if api_key:
os.environ[api_key_var] = api_key
with open(".env", "a") as env_file:
env_file.write(f"{api_key_var}={api_key}\n")
st.success(f"{api_key_var} added successfully!")
return False
return True

13
lib/utils/config_manager.py Normal file
View File

@@ -0,0 +1,13 @@
import json
import os
import streamlit as st
def save_config(config):
"""
Saves the provided configuration dictionary to a JSON file specified by the environment variable.
"""
try:
with open(os.getenv("ALWRITY_CONFIG"), "w") as config_file:
json.dump(config, config_file, indent=4)
except Exception as e:
st.error(f"An error occurred while saving the configuration: {e}")

67
lib/utils/content_generators.py Normal file
View File

@@ -0,0 +1,67 @@
import streamlit as st
from lib.utils.alwrity_utils import (
blog_from_keyword, ai_agents_team, essay_writer, ai_news_writer, ai_seo_tools,
ai_finance_ta_writer, ai_social_writer, do_web_research, competitor_analysis
)
from lib.ai_writers.ai_story_writer.story_writer import story_input_section
from lib.ai_writers.ai_product_description_writer import write_ai_prod_desc
from lib.content_planning_calender.content_planning_agents_alwrity_crew import ai_agents_content_planner
def ai_writers():
options = [
"AI Blog Writer",
"Story Writer",
"Essay writer",
"Write News reports",
"Write Financial TA report",
"AI Product Description Writer",
"AI Copywriter",
"Quit"
]
choice = st.selectbox("**👇Select a content creation type:**", options, index=0, format_func=lambda x: f"📝 {x}")
if choice == "AI Blog Writer":
blog_from_keyword()
elif choice == "Write from audio files":
blog_from_audio()
elif choice == "Story Writer":
story_input_section()
elif choice == "Essay writer":
essay_writer()
elif choice == "Write News reports":
ai_news_writer()
elif choice == "Write Financial TA report":
ai_finance_ta_writer()
elif choice == "AI Product Description Writer":
write_ai_prod_desc()
elif choice == "Quit":
st.subheader("Exiting, Getting Lost. But.... I have nowhere to go 🥹🥹")
def content_planning_tools():
st.markdown("""**Alwrity content Ideation & Planning** : Provide few keywords to do comprehensive web research.
Provide few keywords to get Google, Neural, pytrends analysis. Know keywords, blog titles to target.
Generate months long content calendar around given keywords.""")
options = [
"Keywords Researcher",
"Competitor Analysis",
"Content Calender Ideator"
]
choice = st.radio("Select a content planning tool:", options, index=0, format_func=lambda x: f"🔍 {x}")
if choice == "Keywords Researcher":
do_web_research()
elif choice == "Competitor Analysis":
competitor_analysis()
elif choice == "Content Calender Ideator":
plan_keywords = st.text_input(
"**Enter Your main Keywords to get 2 months content calendar:**",
placeholder="Enter 2-3 main keywords to generate AI content calendar with keyword researched blog titles",
help="The keywords are the ones where you would want to generate 50-60 blogs/articles on."
)
if st.button("**Ideate Content Calender**"):
if plan_keywords:
ai_agents_content_planner(plan_keywords)
else:
st.error("Come on, really, Enter some keywords to plan on..")

5
lib/utils/environment_utils.py Normal file
View File

@@ -0,0 +1,5 @@
import os
from dotenv import load_dotenv
def load_environment():
load_dotenv()

20
lib/utils/file_processor.py Normal file
View File

@@ -0,0 +1,20 @@
import os
import base64
import streamlit as st
def load_image(image_path):
with open(image_path, "rb") as img_file:
b64_string = base64.b64encode(img_file.read()).decode()
return b64_string
def read_prompts(file_path="prompt_llm.txt"):
if os.path.exists(file_path):
with open(file_path, "r") as file:
prompts = file.readlines()
return [prompt.strip() for prompt in prompts]
return []
def write_prompts(prompts, file_path="prompt_llm.txt"):
with open(file_path, "w") as file:
for prompt in prompts:
file.write(f"{prompt}\n")

47
lib/utils/seo_tools.py Normal file
View File

@@ -0,0 +1,47 @@
import streamlit as st
from lib.ai_seo_tools.seo_structured_data import ai_structured_data
from lib.ai_seo_tools.content_title_generator import ai_title_generator
from lib.ai_seo_tools.meta_desc_generator import metadesc_generator_main
from lib.ai_seo_tools.image_alt_text_generator import alt_text_gen
from lib.ai_seo_tools.opengraph_generator import og_tag_generator
from lib.ai_seo_tools.optimize_images_for_upload import main_img_optimizer
from lib.ai_seo_tools.google_pagespeed_insights import google_pagespeed_insights
from lib.ai_seo_tools.on_page_seo_analyzer import analyze_onpage_seo
from lib.ai_seo_tools.weburl_seo_checker import url_seo_checker
def ai_seo_tools():
""" Collection SEO tools for content creators. """
options = [
"Generate Structured Data - Rich Snippet",
"Generate SEO optimized Blog Titles",
"Generate Meta Description for SEO",
"Generate Image Alt Text",
"Generate OpenGraph Tags",
"Optimize/Resize Image",
"Run Google PageSpeed Insights",
"Analyze On Page SEO",
"URL SEO Checker"
]
# Using st.radio instead of st.selectbox
choice = st.radio("**👇 Select AI SEO Tool:**", options, index=0, format_func=lambda x: f"📝 {x}")
# Handle choices based on the selected option
if choice == "Generate Structured Data - Rich Snippet":
ai_structured_data()
elif choice == "Generate Meta Description for SEO":
metadesc_generator_main()
elif choice == "Generate SEO optimized Blog Titles":
ai_title_generator()
elif choice == "Generate Image Alt Text":
alt_text_gen()
elif choice == "Generate OpenGraph Tags":
og_tag_generator()
elif choice == "Optimize/Resize Image":
main_img_optimizer()
elif choice == "Run Google PageSpeed Insights":
google_pagespeed_insights()
elif choice == "Analyze On Page SEO":
analyze_onpage_seo()
elif choice == "URL SEO Checker":
url_seo_checker()

47
lib/utils/ui_setup.py Normal file
View File

@@ -0,0 +1,47 @@
import os
import streamlit as st
from .file_processor import load_image
def setup_ui():
"""Sets up the Streamlit UI with custom CSS and logo."""
try:
css_file_path = os.path.join('lib', 'workspace', 'alwrity_ui_styling.css')
with open(css_file_path) as f:
custom_css = f.read()
st.set_page_config(page_title="Alwrity", layout="wide")
st.markdown(f'<style>{custom_css}</style>', unsafe_allow_html=True)
except Exception as err:
st.error(f"Failed in setting up Alwrity Streamlit UI: {err}")
image_base64 = load_image("lib/workspace/alwrity_logo.png")
st.markdown(f"""
<div class='main-header'>
<img src='data:image/png;base64,{image_base64}' alt='Alwrity Logo' style='height: 50px; margin-right: 10px; vertical-align: middle;'>
Welcome to Alwrity!
</div>
""", unsafe_allow_html=True)
def setup_tabs():
"""Sets up the main tabs in the Streamlit app."""
tab1, tab2, tab3, tab4, tab5, tab6 = st.tabs(
["📅Content Planning", " 📝🤖AI Writers", "🤝🤖Agents Teams", "🛠️🔍AI SEO tools", "📱AI Social Tools", " 💬Ask Alwrity"])
with tab1:
content_planning_tools()
with tab2:
ai_writers()
with tab3:
ai_agents_team()
with tab4:
ai_seo_tools()
with tab5:
ai_social_writer()
with tab6:
st.subheader("Chat with your Data, Chat with any Data.. COMING SOON !")
st.markdown("Create a collection by uploading files (PDF, MD, CSV, etc), or crawl a data source (Websites, more sources coming soon.")
st.markdown("One can ask/chat, summarize and do semantic search over the uploaded data")
# alwrity_chat_docqa()

24
lib/utils/voice_processing.py Normal file
View File

@@ -0,0 +1,24 @@
import streamlit as st
from streamlit_mic_recorder import speech_to_text
@st.cache_data
def record_voice(language="en"):
# https://github.com/B4PT0R/streamlit-mic-recorder?tab=readme-ov-file#example
state = st.session_state
if "text_received" not in state:
state.text_received = []
text = speech_to_text(
start_prompt="🎙️Press & Speak🔊",
stop_prompt="🔇Stop Recording🚨",
language=language,
use_container_width=True,
just_once=False,
)
if text:
state.text_received.append(text)
result = ""
for text in state.text_received:
result += text
state.text_received = []
return result if result else None