prashanth-criodo/SPARC
0
1# Standard library imports2import os3from typing import Dict, Optional4from datetime import datetime5 6# Third-party imports7import streamlit as st8from dotenv import load_dotenv9 10# Local imports11from components.persona_manager import PersonaManager12from components.campaign_manager import CampaignManager13from components.social_media_manager import SocialMediaManager14from components.email_service import EmailService15from components.analytics_manager import AnalyticsManager16from utils.openai_helper import ContentGenerator17from utils.config_loader import load_config18from utils.db_manager import DatabaseManager19 20# Must be the first Streamlit command21st.set_page_config(22 page_title="Smart Personalised Automation for Remarkable Campaigns (S.P.A.R.C)",23 page_icon="🎯",24 layout="wide"25)26 27class CampaignCraftAI:28 def __init__(self):29 load_dotenv()30 self.config = load_config()31 32 # Initialize session state for content history33 if 'content_history' not in st.session_state:34 st.session_state.content_history = []35 36 # Initialize state for Twitter posting37 if 'twitter_post_status' not in st.session_state:38 st.session_state.twitter_post_status = None39 if 'current_content' not in st.session_state:40 st.session_state.current_content = None41 if 'current_campaign_data' not in st.session_state:42 st.session_state.current_campaign_data = None43 if 'twitter_preview' not in st.session_state:44 st.session_state.twitter_preview = None45 46 # Initialize Azure OpenAI47 self.content_generator = ContentGenerator(48 api_key=os.getenv("AZURE_OPENAI_API_KEY"),49 azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),50 deployment_name=self.config["azure_openai"]["deployment_name"],51 api_version=self.config["azure_openai"]["api_version"]52 )53 54 # Initialize components55 self.persona_manager = PersonaManager()56 self.campaign_manager = CampaignManager()57 self.social_media_manager = SocialMediaManager(self.content_generator)58 self.email_service = EmailService(59 self.config["email"]["smtp_server"],60 self.config["email"]["smtp_port"]61 )62 self.analytics_manager = AnalyticsManager()63 64 def main(self):65 st.title("Smart Personalised Automation for Remarkable Campaigns (S.P.A.R.C)")66 st.subheader("AI-Powered Marketing Campaign Content Generator")67 68 # Sidebar navigation69 with st.sidebar:70 st.header("Navigation")71 page = st.radio("Go to", ["Create Persona", "Generate Content", "Content History", "Analytics"])72 73 if page == "Create Persona":74 self.persona_manager.create_persona_form()75 76 # Display saved personas77 personas = self.persona_manager.get_personas()78 if personas:79 st.subheader("Saved Personas")80 for persona in personas:81 st.write(f"📋 {persona['role']} ({persona['experience']})")82 83 elif page == "Generate Content":84 campaign_data = self.campaign_manager.create_campaign_form(85 self.persona_manager.get_personas()86 )87 88 if campaign_data:89 with st.spinner("Generating content..."):90 content = self.content_generator.generate_content(91 campaign_goal=campaign_data["campaign_goal"],92 persona=campaign_data["personas"][0], # Use first persona for content generation93 content_type=campaign_data["content_type"],94 tone=campaign_data["tone"]95 )96 97 # Store current content and campaign data in session state98 st.session_state.current_content = content99 st.session_state.current_campaign_data = campaign_data100 101 # Save to content history102 content_data = {103 "id": len(st.session_state.content_history) + 1,104 "campaign_goal": campaign_data["campaign_goal"],105 "content_type": campaign_data["content_type"],106 "tone": campaign_data["tone"],107 "content": content,108 "persona_roles": [p["role"] for p in campaign_data["personas"]],109 "hashtags": campaign_data["hashtags"],110 "tweet_url": None, # Will be updated when posted111 "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")112 }113 st.session_state.content_history.append(content_data)114 115 st.subheader("Generated Content")116 st.write(content)117 118 # Display metadata119 with st.expander("Campaign Details"):120 st.write(f"**Content Type:** {campaign_data['content_type']}")121 st.write(f"**Tone:** {campaign_data['tone']}")122 st.write("**Target Personas:**")123 for persona in campaign_data['personas']:124 st.write(f"- {persona['role']}")125 if campaign_data['hashtags']:126 st.write(f"**Hashtags:** {', '.join(campaign_data['hashtags'])}")127 128 # Show distribution options129 st.markdown("---")130 self.show_distribution_options(content, campaign_data)131 132 # Add preview of Twitter content133 with st.expander("Preview Twitter Content", expanded=True):134 twitter_content = self.social_media_manager.optimize_for_twitter(135 content, 136 campaign_data["hashtags"]137 )138 st.text(twitter_content)139 st.write(f"Character count: {len(twitter_content)}/280")140 141 elif page == "Analytics":142 self.analytics_manager.show_analytics_dashboard()143 144 else: # Content History145 self.show_content_history()146 147 def show_distribution_options(self, content: str, campaign_data: Dict):148 st.subheader("📤 Share Content")149 150 col1, col2 = st.columns(2)151 152 with col1:153 st.subheader("🐦 Twitter")154 st.info("Share this content on Twitter")155 156 # Generate Twitter preview once and store in session state157 if st.session_state.twitter_preview is None:158 st.session_state.twitter_preview = self.social_media_manager.optimize_for_twitter(159 content, 160 campaign_data["hashtags"]161 )162 163 # Show Twitter preview164 with st.expander("Preview Twitter Content", expanded=True):165 st.text(st.session_state.twitter_preview)166 st.write(f"Character count: {len(st.session_state.twitter_preview)}/280")167 168 # Initialize Twitter client if not already done169 twitter_ready = self.social_media_manager.setup_twitter_auth(170 api_key=os.getenv("TWITTER_API_KEY"),171 api_secret=os.getenv("TWITTER_API_SECRET"),172 access_token=os.getenv("TWITTER_ACCESS_TOKEN"),173 access_token_secret=os.getenv("TWITTER_ACCESS_TOKEN_SECRET")174 )175 176 # Handle Twitter posting177 if twitter_ready and st.button("🚀 Post to Twitter", type="primary"):178 st.session_state.twitter_post_status = "posting"179 180 # Show posting status181 if st.session_state.twitter_post_status == "posting":182 result = self.social_media_manager.post_to_twitter(183 st.session_state.twitter_preview,184 campaign_data["hashtags"]185 )186 if result["success"]:187 st.session_state.twitter_post_status = "success"188 st.success("Posted to Twitter successfully!")189 st.markdown(f"[View Tweet]({result['url']})")190 # Update content history with tweet URL191 for item in st.session_state.content_history:192 if item["content"] == content:193 item["tweet_url"] = result["url"]194 st.balloons()195 else:196 st.session_state.twitter_post_status = "failed"197 st.error(f"Failed to post to Twitter: {result.get('error', 'Unknown error')}")198 199 with col2:200 st.subheader("📧 Email Campaign")201 st.info("Send this content via email")202 203 # Show selected personas204 st.write("**Selected Personas:**")205 for persona in campaign_data['personas']:206 st.write(f"- {persona['role']}")207 208 recipients = st.text_area(209 "Email Recipients (one per line)",210 help="Enter email addresses, one per line"211 )212 213 if st.button("📨 Send Email Campaign", type="primary"):214 email_content = self.social_media_manager.format_email(content)215 216 recipient_list = [r.strip() for r in recipients.split('\n') if r.strip()]217 218 if not recipient_list:219 st.error("Please enter at least one recipient email address.")220 return221 222 if self.email_service.send_email(223 recipient_list,224 email_content["subject"],225 email_content["body"]226 ):227 st.success("Email campaign sent successfully!")228 st.write(f"Sent to {len(recipient_list)} recipients")229 st.balloons()230 231 def show_content_history(self):232 """Display content generation history"""233 st.header("📚 Content History")234 235 if not st.session_state.content_history:236 st.info("No content has been generated yet.")237 return238 239 for item in reversed(st.session_state.content_history):240 with st.expander(f"{item['content_type']} - {item['created_at']}"):241 st.write(f"**Campaign Goal:** {item['campaign_goal']}")242 st.write("**Target Personas:**")243 for role in item['persona_roles']:244 st.write(f"- {role}")245 st.write(f"**Tone:** {item['tone']}")246 st.text(item['content'])247 if item['hashtags']:248 st.write(f"**Hashtags:** {', '.join(item['hashtags'])}")249 if item.get('tweet_url'):250 st.markdown(f"**[View Tweet]({item['tweet_url']})**")251 252 # Add repost option253 if st.button("🔄 Repost to Twitter", key=f"repost_{item['id']}"):254 st.session_state.current_content = item['content']255 st.session_state.current_campaign_data = {256 "hashtags": item['hashtags'],257 "content_type": item['content_type'],258 "tone": item['tone']259 }260 st.session_state.twitter_post_status = "posting"261 st.experimental_rerun()262 263if __name__ == "__main__":264 app = CampaignCraftAI()265 app.main()