This commit is contained in:
ajaysi
2025-04-15 12:37:25 +05:30
parent d4f1fc77a1
commit a2b92e27bc
6 changed files with 500 additions and 134 deletions

View File

@@ -28,6 +28,7 @@ from .modules.post_generator.linkedin_post_generator import linkedin_post_genera
from .modules.article_generator.linkedin_article_generator import linkedin_article_generator_ui
from .modules.carousel_generator.linkedin_carousel_generator import linkedin_carousel_generator_ui
from .modules.video_script_generator.linkedin_video_script_generator import linkedin_video_script_generator_ui
from .modules.comment_response_generator.linkedin_comment_response_generator_ui import linkedin_comment_response_generator_ui
# Import image generation
from ...gpt_providers.text_to_image_generation.main_generate_image_from_prompt import generate_image
@@ -131,6 +132,27 @@ def linkedin_main_menu():
"Hashtag strategy"
]
},
{
"name": "LinkedIn Comment Response Generator",
"icon": "💬",
"description": "Generate professional and engaging responses to LinkedIn comments with AI-powered analysis and optimization.",
"color": "#0A66C2",
"category": "Engagement",
"function": linkedin_comment_response_generator_ui,
"status": "active",
"features": [
"Comment analysis and categorization",
"Multiple response types (general, disagreement, value-add)",
"Brand voice customization",
"Engagement goal targeting",
"Resource suggestion generation",
"Follow-up question generation",
"Tone optimization",
"Response strategy recommendations",
"Context-aware responses",
"Professional formatting"
]
},
# Profile & Personal Branding Tools
{

View File

@@ -9,37 +9,70 @@ import streamlit as st
from typing import Dict, List, Optional
from loguru import logger
import json
from pydantic import BaseModel
from .....gpt_providers.text_generation.main_text_generation import llm_text_gen
from .....gpt_providers.text_generation.gemini_pro_text import gemini_structured_json_response
from .....gpt_providers.text_to_image_generation.main_generate_image_from_prompt import generate_image
from .....ai_web_researcher.gpt_online_researcher import do_google_serp_search
from .....ai_web_researcher.metaphor_basic_neural_web_search import metaphor_search_articles
from .....ai_web_researcher.tavily_ai_search import do_tavily_ai_search
class CarouselSlide:
class CarouselSlide(BaseModel):
"""Represents a single slide in the carousel."""
def __init__(self, index: int, slide_type: str = "text_and_image"):
self.index = index
self.slide_type = slide_type
self.content = ""
self.image_prompt = ""
self.image_path = None
self.heading = ""
self.subheading = ""
def to_dict(self) -> Dict:
"""Convert slide to dictionary format."""
return {
"index": self.index,
"type": self.slide_type,
"content": self.content,
"image_prompt": self.image_prompt,
"image_path": self.image_path,
"heading": self.heading,
"subheading": self.subheading
index: int
heading: str
subheading: str
content: str
image_prompt: str
image_path: Optional[str] = None
class CarouselStructure(BaseModel):
"""Represents the complete carousel structure."""
slides: List[CarouselSlide]
def generate_structured_content(prompt: str, model: str = 'gemini-pro') -> Optional[Dict]:
"""Generate structured content using Gemini's structured output feature."""
try:
# Define the schema for the carousel structure
schema = {
"type": "object",
"properties": {
"slides": {
"type": "array",
"items": {
"type": "object",
"properties": {
"index": {"type": "integer"},
"heading": {"type": "string"},
"subheading": {"type": "string"},
"content": {"type": "string"},
"image_prompt": {"type": "string"}
},
"required": ["index", "heading", "subheading", "content", "image_prompt"]
}
}
},
"required": ["slides"]
}
# Use the new function to generate structured content
result = gemini_structured_json_response(prompt, schema)
# Check if there was an error
if "error" in result:
logger.error(f"Error generating structured content: {result['error']}")
return None
logger.debug(f"Structured response: {json.dumps(result, indent=2)}")
return result
except Exception as e:
logger.error(f"Error generating structured content: {e}")
logger.exception("Full traceback:")
return None
class LinkedInCarouselGenerator:
@@ -86,75 +119,83 @@ class LinkedInCarouselGenerator:
logger.error(f"Error researching topic: {e}")
return {}
def generate_slide_content(self, topic: str, industry: str, tone: str, content_type: str, num_slides: int) -> List[CarouselSlide]:
def generate_slide_content(self, topic: str, num_slides: int = 5) -> bool:
"""Generate content for carousel slides."""
try:
# Store parameters
self.topic = topic
self.industry = industry
self.tone = tone
self.content_type = content_type
self.num_slides = num_slides
logger.info(f"Generating carousel outline for topic: {topic}")
# Generate content structure
structure_prompt = f"""
Create a {num_slides}-slide carousel post about '{topic}' for {industry} industry.
Style: {content_type} (e.g., How-to, List, Story)
Tone: {tone}
For each slide, provide:
1. Heading (short, attention-grabbing)
2. Subheading (supporting context)
3. Main content (clear, concise points)
4. Image prompt (visual representation)
Format as JSON:
{{
"slides": [
{{
"index": 1,
"heading": "Attention-grabbing title",
"subheading": "Supporting context",
"content": "Main slide content",
"image_prompt": "Detailed image generation prompt"
}}
]
}}
Requirements:
- First slide should hook the audience
- Maintain clear progression of ideas
- Each slide should be self-contained but connected
- Last slide should have clear call-to-action
- Keep text concise and impactful
- Ensure all content is professional and LinkedIn-appropriate
# Step 1: Generate detailed outline
outline_prompt = f"""
Create a detailed outline for a LinkedIn carousel about {topic}.
The outline should include:
1. Main topic and key message
2. {num_slides} main points to cover
3. Supporting details for each point
4. Key takeaways and call-to-action
Format the outline in a clear, structured way.
"""
# Generate structure using LLM
carousel_structure = llm_text_gen(structure_prompt)
outline = llm_text_gen(outline_prompt)
logger.debug(f"Generated outline: {outline}")
try:
structure_data = json.loads(carousel_structure)
# Step 2: Generate structured carousel content
carousel_prompt = f"""
Based on this outline:
{outline}
Create a LinkedIn carousel with {num_slides} slides.
Each slide should have:
- A compelling heading
- A brief subheading
- Concise, engaging content
- A detailed image prompt for visual content
The content should be informative, engaging, and valuable for LinkedIn professionals.
"""
logger.info("Generating structured carousel content")
carousel_structure = generate_structured_content(carousel_prompt)
if not carousel_structure:
logger.error("Failed to generate structured carousel content")
return False
# Create slides from structure
self.slides = []
for slide_data in structure_data["slides"]:
slide = CarouselSlide(slide_data["index"])
slide.heading = slide_data["heading"]
slide.subheading = slide_data["subheading"]
slide.content = slide_data["content"]
slide.image_prompt = slide_data["image_prompt"]
logger.debug(f"Generated carousel structure: {json.dumps(carousel_structure, indent=2)}")
# Validate and process the structure
if "slides" not in carousel_structure:
logger.error("Invalid carousel structure: missing 'slides' key")
return False
self.slides = []
for slide_data in carousel_structure["slides"]:
try:
slide = CarouselSlide(
index=slide_data["index"],
heading=slide_data["heading"],
subheading=slide_data["subheading"],
content=slide_data["content"],
image_prompt=slide_data["image_prompt"]
)
self.slides.append(slide)
logger.debug(f"Created slide {slide.index}: {slide.heading}")
except Exception as e:
logger.error(f"Error processing slide data: {e}")
logger.debug(f"Problematic slide data: {slide_data}")
continue
if not self.slides:
logger.error("No valid slides were created")
return False
return self.slides
except json.JSONDecodeError:
logger.error("Failed to parse carousel structure")
return []
logger.info(f"Successfully generated {len(self.slides)} slides")
return True
except Exception as e:
logger.error(f"Error generating slide content: {e}")
return []
logger.exception("Full traceback:")
return False
def generate_slide_image(self, slide: CarouselSlide) -> Optional[str]:
"""Generate an image for a carousel slide."""
@@ -277,8 +318,20 @@ def linkedin_carousel_generator_ui():
# Content generation phase
status_container.text("✍️ Phase 2: Creating carousel content...")
slides = generator.generate_slide_content(
topic, industry, tone.lower(), content_type.lower(), num_slides
topic, num_slides
)
if not slides:
status_container.error("❌ Failed to generate carousel content. Please try again with different parameters.")
st.error("""
Tips to resolve this issue:
1. Try a different topic or industry
2. Adjust the number of slides
3. Change the content type or tone
4. Try a different research source
""")
return
st.session_state.carousel_data["slides"] = slides
# Image generation phase
@@ -291,6 +344,13 @@ def linkedin_carousel_generator_ui():
status_container.text("✅ Carousel generation complete!")
else:
status_container.error("❌ No research results found. Please try a different topic or research source.")
st.error("""
Tips to resolve this issue:
1. Try a more specific topic
2. Use a different research source
3. Check if your topic is too niche or too broad
4. Ensure your industry selection is accurate
""")
# Display carousel if we have slides
if st.session_state.carousel_data["slides"]:

View File

@@ -0,0 +1,108 @@
import streamlit as st
import json
from typing import Optional, List
from .linkedin_carousel_generator import LinkedInCarouselGenerator, CarouselSlide
def linkedin_carousel_generator_ui():
"""Streamlit UI for LinkedIn Carousel Generator."""
st.title("LinkedIn Carousel Generator")
st.write("Create engaging carousel posts for LinkedIn with AI-powered content generation.")
# Initialize session state
if 'generator' not in st.session_state:
st.session_state.generator = LinkedInCarouselGenerator()
if 'slides' not in st.session_state:
st.session_state.slides = []
if 'current_slide' not in st.session_state:
st.session_state.current_slide = 0
# Sidebar for input parameters
with st.sidebar:
st.header("Carousel Parameters")
topic = st.text_input("Topic", help="Enter the main topic for your carousel")
num_slides = st.slider("Number of Slides", min_value=3, max_value=10, value=5,
help="Choose how many slides you want in your carousel")
if st.button("Generate Carousel"):
if not topic:
st.error("Please enter a topic")
return
with st.spinner("Generating carousel content..."):
success = st.session_state.generator.generate_slide_content(topic, num_slides)
if success:
st.session_state.slides = st.session_state.generator.slides
st.session_state.current_slide = 0
st.success("Carousel content generated successfully!")
else:
st.error("Failed to generate carousel content. Please try again.")
# Main content area
if st.session_state.slides:
# Display current slide
current_slide = st.session_state.slides[st.session_state.current_slide]
col1, col2 = st.columns([2, 1])
with col1:
st.subheader(f"Slide {current_slide.index}")
st.write("**Heading:**")
st.write(current_slide.heading)
st.write("**Subheading:**")
st.write(current_slide.subheading)
st.write("**Content:**")
st.write(current_slide.content)
st.write("**Image Prompt:**")
st.write(current_slide.image_prompt)
# Navigation buttons
col_prev, col_next = st.columns(2)
with col_prev:
if st.session_state.current_slide > 0:
if st.button("← Previous"):
st.session_state.current_slide -= 1
st.experimental_rerun()
with col_next:
if st.session_state.current_slide < len(st.session_state.slides) - 1:
if st.button("Next →"):
st.session_state.current_slide += 1
st.experimental_rerun()
with col2:
# Display structured output
st.subheader("Carousel Structure")
carousel_data = {
"slides": [
{
"index": slide.index,
"heading": slide.heading,
"subheading": slide.subheading,
"content": slide.content,
"image_prompt": slide.image_prompt
}
for slide in st.session_state.slides
]
}
st.json(carousel_data)
# Export options
st.download_button(
"Download as JSON",
data=json.dumps(carousel_data, indent=2),
file_name="carousel_content.json",
mime="application/json"
)
if st.button("Generate Images"):
with st.spinner("Generating images for slides..."):
for slide in st.session_state.slides:
if not slide.image_path:
image_path = st.session_state.generator.generate_slide_image(slide)
if image_path:
slide.image_path = image_path
st.success(f"Generated image for slide {slide.index}")
else:
st.error(f"Failed to generate image for slide {slide.index}")
else:
st.info("Enter a topic and click 'Generate Carousel' to create your carousel content.")

View File

@@ -11,6 +11,7 @@ from typing import Dict, List, Optional, Union
from loguru import logger
from .....gpt_providers.text_generation.main_text_generation import llm_text_gen
from .....gpt_providers.text_generation.gemini_pro_text import gemini_structured_json_response
from .....ai_web_researcher.gpt_online_researcher import do_google_serp_search
from .....ai_web_researcher.metaphor_basic_neural_web_search import metaphor_search_articles
from .....ai_web_researcher.tavily_ai_search import do_tavily_ai_search
@@ -103,19 +104,38 @@ class LinkedInVideoScriptGenerator:
"""Extract key insights and trends from research articles."""
try:
prompt = f"""
Based on the following articles, extract:
1. Key insights relevant for a video script
2. Current trends in the industry
Analyze these articles and extract key insights and trends:
Articles:
{json.dumps(articles, indent=2)}
Return a JSON with two lists: 'insights' and 'trends'
Identify the most valuable insights and emerging trends from these articles.
"""
response = llm_text_gen(prompt)
result = json.loads(response)
# Define the schema for insights and trends
schema = {
"type": "object",
"properties": {
"insights": {
"type": "array",
"items": {"type": "string"}
},
"trends": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["insights", "trends"]
}
# Use the structured JSON response function
result = gemini_structured_json_response(prompt, schema)
# Check if there was an error
if "error" in result:
logger.error(f"Error extracting insights and trends: {result['error']}")
return [], []
return result.get("insights", []), result.get("trends", [])
except Exception as e:
@@ -159,35 +179,65 @@ class LinkedInVideoScriptGenerator:
return ""
def generate_story_structure(self, topic: str, video_type: str, length: str, insights: List[str]) -> Dict:
"""
Generate a structured outline for the video.
Args:
topic: Main topic of the video
video_type: Type of video content
length: Target video length
insights: Research insights to incorporate
Returns:
Dict containing the story structure
"""
"""Generate a structured story for the video."""
try:
prompt = f"""
Create a professional video script structure for LinkedIn with:
Create a story structure for a LinkedIn video:
- Topic: {topic}
- Video Type: {self.video_types[video_type]}
- Length: {self.video_lengths[length]}
- Key Insights: {json.dumps(insights)}
Return a JSON with:
1. sections: List of sections with timing and content
2. transitions: List of smooth transitions
3. key_points: Main points to emphasize
4. pacing_notes: Guidance on pacing and delivery
Create a well-structured story with clear sections, transitions, and key points.
Include guidance on pacing and delivery.
"""
response = llm_text_gen(prompt)
return json.loads(response)
# Define the schema for the story structure
schema = {
"type": "object",
"properties": {
"sections": {
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string"},
"duration": {"type": "string"}
},
"required": ["title", "content", "duration"]
}
},
"transitions": {
"type": "array",
"items": {"type": "string"}
},
"key_points": {
"type": "array",
"items": {"type": "string"}
},
"pacing_notes": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["sections", "transitions", "key_points", "pacing_notes"]
}
# Use the structured JSON response function
result = gemini_structured_json_response(prompt, schema)
# Check if there was an error
if "error" in result:
logger.error(f"Error generating story structure: {result['error']}")
return {
"sections": [],
"transitions": [],
"key_points": [],
"pacing_notes": []
}
return result
except Exception as e:
logger.error(f"Error generating story structure: {str(e)}")
@@ -199,33 +249,46 @@ class LinkedInVideoScriptGenerator:
}
def generate_visual_cues(self, script_sections: List[Dict]) -> List[Dict]:
"""
Generate visual suggestions for each section of the script.
Args:
script_sections: List of script sections
Returns:
List of visual cue suggestions
"""
"""Generate visual cues for the video script."""
try:
prompt = f"""
For each section of the LinkedIn video script, suggest visual elements:
Create visual cues for this video script:
Script Sections:
{json.dumps(script_sections, indent=2)}
For each section, provide:
For each section, suggest:
1. Visual type (b-roll, graphics, text overlay, etc.)
2. Description of visual content
3. Timing and duration
4. Visual transitions
Return a JSON array of visual suggestions.
"""
response = llm_text_gen(prompt)
return json.loads(response)
# Define the schema for visual cues
schema = {
"type": "array",
"items": {
"type": "object",
"properties": {
"section_title": {"type": "string"},
"visual_type": {"type": "string"},
"description": {"type": "string"},
"timing": {"type": "string"},
"transitions": {"type": "string"}
},
"required": ["section_title", "visual_type", "description", "timing", "transitions"]
}
}
# Use the structured JSON response function
result = gemini_structured_json_response(prompt, schema)
# Check if there was an error
if "error" in result:
logger.error(f"Error generating visual cues: {result['error']}")
return []
return result
except Exception as e:
logger.error(f"Error generating visual cues: {str(e)}")
@@ -282,26 +345,66 @@ class LinkedInVideoScriptGenerator:
return {}
def generate_cta(self, topic: str, video_type: str, target_audience: str) -> Dict:
"""Generate an effective call-to-action."""
"""Generate a compelling call-to-action for the video."""
try:
prompt = f"""
Create an effective call-to-action for a LinkedIn video:
Create a compelling call-to-action for a LinkedIn video:
- Topic: {topic}
- Video Type: {self.video_types[video_type]}
- Target Audience: {self.target_audiences[target_audience]}
- Target Audience: {target_audience}
Return a JSON with:
1. primary_cta: Main call-to-action text
2. secondary_cta: Optional follow-up action
3. engagement_prompts: Questions or prompts for comments
Generate a clear, engaging call-to-action that encourages viewer engagement.
"""
response = llm_text_gen(prompt)
return json.loads(response)
# Define the schema for the CTA
schema = {
"type": "object",
"properties": {
"primary_cta": {
"type": "string",
"description": "Main call-to-action text"
},
"secondary_cta": {
"type": "string",
"description": "Optional secondary call-to-action"
},
"engagement_hooks": {
"type": "array",
"items": {"type": "string"},
"description": "Additional engagement prompts"
},
"hashtag_suggestions": {
"type": "array",
"items": {"type": "string"},
"description": "Relevant hashtags for the CTA"
}
},
"required": ["primary_cta", "secondary_cta", "engagement_hooks", "hashtag_suggestions"]
}
# Use the structured JSON response function
result = gemini_structured_json_response(prompt, schema)
# Check if there was an error
if "error" in result:
logger.error(f"Error generating CTA: {result['error']}")
return {
"primary_cta": "Thanks for watching!",
"secondary_cta": "Let me know your thoughts in the comments.",
"engagement_hooks": ["What did you think about this topic?"],
"hashtag_suggestions": ["#LinkedInVideo"]
}
return result
except Exception as e:
logger.error(f"Error generating CTA: {str(e)}")
return {}
return {
"primary_cta": "Thanks for watching!",
"secondary_cta": "Let me know your thoughts in the comments.",
"engagement_hooks": ["What did you think about this topic?"],
"hashtag_suggestions": ["#LinkedInVideo"]
}
def linkedin_video_script_generator_ui():
"""Streamlit UI for the LinkedIn Video Script Generator."""
@@ -447,11 +550,17 @@ def linkedin_video_script_generator_ui():
st.write(f"**Secondary CTA:** {script['cta']['secondary_cta']}")
# Display engagement prompts
if script['cta'].get('engagement_prompts'):
if script['cta'].get('engagement_hooks'):
st.subheader("Engagement Prompts")
for prompt in script['cta']['engagement_prompts']:
for prompt in script['cta']['engagement_hooks']:
st.markdown(f"- {prompt}")
# Display hashtag suggestions
if script['cta'].get('hashtag_suggestions'):
st.subheader("Hashtag Suggestions")
for hashtag in script['cta']['hashtag_suggestions']:
st.markdown(f"- {hashtag}")
# Display pacing notes
st.subheader("Pacing & Delivery Notes")
for note in script["structure"]["pacing_notes"]:

View File

@@ -21,6 +21,8 @@ from tenacity import (
)
import asyncio
import json
import re
# Configure standard logging
import logging
@@ -167,3 +169,64 @@ def gemini_pro_text_gen(prompt, temperature=0.7, top_p=0.9, top_k=40, max_tokens
except Exception as e:
logger.error(f"Error in Gemini Pro text generation: {e}")
return str(e)
def gemini_structured_json_response(prompt, schema, temperature=0.7, top_p=0.9, top_k=40, max_tokens=2048, system_prompt=None):
"""
Generate structured JSON response using Google's Gemini Pro model.
Args:
prompt (str): The input text to generate completion for
schema (dict): The JSON schema to follow for the response
temperature (float, optional): Controls randomness. Defaults to 0.7
top_p (float, optional): Controls diversity. Defaults to 0.9
top_k (int, optional): Controls vocabulary size. Defaults to 40
max_tokens (int, optional): Maximum number of tokens to generate. Defaults to 2048
system_prompt (str, optional): System instructions for the model
Returns:
dict: The generated structured JSON response
"""
try:
# Configure the model
client = genai.Client(api_key=os.getenv('GEMINI_API_KEY'))
# Set up generation config
generation_config = {
"temperature": temperature,
"top_p": top_p,
"top_k": top_k,
"max_output_tokens": max_tokens,
}
# Generate content with structured response
response = client.models.generate_content(
model='gemini-2.0-flash',
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
max_output_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
response_mime_type='application/json',
response_schema=schema
),
)
# Parse the response
try:
# First try to get the parsed response
if hasattr(response, 'parsed'):
return response.parsed
# If parsed is not available, try to parse the text
response_text = response.text
return json.loads(response_text)
except json.JSONDecodeError as e:
logger.error(f"Error parsing JSON response: {e}")
return {"error": f"Failed to parse JSON response: {e}", "raw_response": response_text}
except Exception as e:
logger.error(f"Error in Gemini Pro structured JSON generation: {e}")
return {"error": str(e)}

View File

@@ -19,12 +19,13 @@ from .deepseek_text_gen import deepseek_text_response
from ...utils.read_main_config_params import read_return_config_section
def llm_text_gen(prompt, system_prompt=None):
def llm_text_gen(prompt, system_prompt=None, json_struct=None):
"""
Generate text using Language Model (LLM) based on the provided prompt.
Args:
prompt (str): The prompt to generate text from.
system_prompt (str, optional): Custom system prompt to use instead of the default one.
json_struct (dict, optional): JSON schema structure for structured responses.
Returns:
str: Generated text based on the prompt.
"""
@@ -77,7 +78,10 @@ def llm_text_gen(prompt, system_prompt=None):
if 'google' in gpt_provider.lower():
try:
logger.info("Using Google Gemini Pro text generation model.")
response = gemini_text_response(prompt, temperature, top_p, n, max_tokens, system_instructions)
if json_struct:
response = gemini_structured_json_response(prompt, json_struct, temperature, top_p, n, max_tokens, system_instructions)
else:
response = gemini_text_response(prompt, temperature, top_p, n, max_tokens, system_instructions)
return response
except Exception as err:
logger.error(f"Failed to get response from gemini: {err}")