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

feat: LinkedIn LLM alignment - Phase 1-3 complete

Browse Source 
Phase 1: Dead Code Cleanup
- Remove GeminiGroundedProvider import and property from linkedin_service.py
- Remove fallback_provider property (gemini_provider imports)
- Fix routers/linkedin.py edit endpoint to use llm_text_gen
- Delete dead LinkedInImageEditor class
- Remove dead _transform_gemini_sources from content_generator.py

Phase 2: Research Infrastructure Alignment
- Add user_id to _conduct_research() for pre-flight validation
- Add validate_exa_research_operations() before Exa/Tavily calls
- Pass user_id to provider.simple_search() for usage tracking
- Inject research content into LLM prompts via _build_research_context()
- Fix Google engine path to fallback to Exa
- Add Exa → Tavily fallback on research failure

Phase 3: Cosmetic Cleanup
- Rename _generate_prompts_with_gemini → _generate_prompts_with_llm
- Rename _build_gemini_prompt → _build_image_prompt
- Rename _parse_gemini_response → _parse_llm_response
- Remove all Gemini references from LinkedIn code (0 remaining)
- Update docstrings and log messages

Additional:
- Research caching using existing ResearchCache
- Shared ExaContentResearchProvider in services/research/
- Persona service uses llm_text_gen instead of gemini_structured_json_response
- LinkedInWriter.tsx ChatMessage → ChatMsg type mapping fix
- RegisterLinkedInActionsEnhanced.tsx content_format_rules typing fix
This commit is contained in:
ajaysi
2026-06-12 18:58:53 +05:30
parent e54aaa7a3e
commit 63a0df2536
37 changed files with 2891 additions and 1355 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
backend/services/linkedin/__init__.py
View File

@@ -17,13 +17,13 @@ from .content_generator_prompts import (
VideoScriptGenerator
)
# Import new image generation services
# Import image generation services
from .image_generation import (
LinkedInImageGenerator,
LinkedInImageEditor,
LinkedInImageStorage
)
from .image_prompts import LinkedInPromptGenerator
from .carousel import LinkedInCarouselPDFRenderer
__all__ = [
# Content Generation
@@ -42,9 +42,10 @@ __all__ = [
# Image Generation Services
'LinkedInImageGenerator',
'LinkedInImageEditor',
'LinkedInImageStorage',
'LinkedInPromptGenerator'
'LinkedInPromptGenerator',
# Carousel Rendering
'LinkedInCarouselPDFRenderer',
]
# Version information

3
backend/services/linkedin/carousel/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .carousel_renderer import LinkedInCarouselPDFRenderer
__all__ = ['LinkedInCarouselPDFRenderer']

336
backend/services/linkedin/carousel/carousel_renderer.py Normal file
View File

@@ -0,0 +1,336 @@
"""
LinkedIn Carousel PDF Renderer
Renders text-based carousel slides into visually appealing PNG images
and composes them into a LinkedIn-compatible PDF document (1.91:1 ratio).
"""
import os
import logging
from datetime import datetime
from typing import Dict, Any, List, Optional
from PIL import Image, ImageDraw, ImageFont, ImageFilter
from reportlab.lib.pagesizes import landscape
from reportlab.lib.units import mm
from reportlab.platypus import SimpleDocTemplate, Image as RLImage, PageBreak
logger = logging.getLogger(__name__)
class LinkedInCarouselPDFRenderer:
COLOR_SCHEMES = {
'professional': {
'background_start': (25, 55, 109),
'background_end': (41, 128, 185),
'title_color': (255, 255, 255),
'content_color': (236, 240, 241),
'accent_color': (52, 152, 219),
},
'creative': {
'background_start': (142, 68, 173),
'background_end': (231, 76, 60),
'title_color': (255, 255, 255),
'content_color': (245, 245, 245),
'accent_color': (241, 196, 15),
},
'industry': {
'background_start': (39, 174, 96),
'background_end': (44, 62, 80),
'title_color': (255, 255, 255),
'content_color': (236, 240, 241),
'accent_color': (46, 204, 113),
},
'dark': {
'background_start': (20, 20, 30),
'background_end': (60, 60, 80),
'title_color': (255, 255, 255),
'content_color': (200, 200, 210),
'accent_color': (100, 200, 255),
},
'minimal': {
'background_start': (245, 245, 250),
'background_end': (255, 255, 255),
'title_color': (44, 62, 80),
'content_color': (80, 80, 90),
'accent_color': (52, 152, 219),
},
}
def __init__(self, output_dir: str = None):
self.slide_width = 1200
self.slide_height = 627
self.slide_aspect_ratio = "1.91:1"
self.max_file_size_bytes = 100 * 1024 * 1024
self.max_slides = 300
self.output_dir = output_dir or "data/media/linkedin_carousels"
async def render_carousel_to_pdf(
self,
carousel_data: Dict[str, Any],
color_scheme: str = 'professional',
user_id: Optional[str] = None,
) -> Dict[str, Any]:
start_time = datetime.now()
os.makedirs(self.output_dir, exist_ok=True)
try:
slides = carousel_data.get('slides', [])
if not slides:
return {'success': False, 'error': 'No slides to render'}
title = carousel_data.get('title', 'LinkedIn Carousel')
cover_slide = carousel_data.get('cover_slide')
cta_slide = carousel_data.get('cta_slide')
total_slides = len(slides) + (1 if cover_slide else 0) + (1 if cta_slide else 0)
if total_slides > self.max_slides:
error = f'Too many slides: {total_slides} exceeds max {self.max_slides}'
return {'success': False, 'error': error}
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
image_paths = []
if cover_slide:
path = self._render_slide(
slide=cover_slide, slide_number=0, session_id=session_id,
color_scheme=color_scheme, is_cover=True, carousel_title=title,
)
if path:
image_paths.append(path)
for i, slide in enumerate(slides):
path = self._render_slide(
slide=slide, slide_number=i + 1, session_id=session_id,
color_scheme=color_scheme, is_cover=False,
)
if path:
image_paths.append(path)
if cta_slide:
path = self._render_slide(
slide=cta_slide, slide_number=len(slides) + 1, session_id=session_id,
color_scheme=color_scheme, is_cta=True,
)
if path:
image_paths.append(path)
if not image_paths:
return {'success': False, 'error': 'No slide images generated'}
pdf_filename = f"linkedin_carousel_{session_id}.pdf"
pdf_path = os.path.join(self.output_dir, pdf_filename)
pdf_bytes = self._compose_pdf(image_paths, pdf_path)
file_size = len(pdf_bytes)
if file_size > self.max_file_size_bytes:
logger.warning("PDF size %.2f MB exceeds max %.2f MB",
file_size / (1024 * 1024), self.max_file_size_bytes / (1024 * 1024))
generation_time = (datetime.now() - start_time).total_seconds()
return {
'success': True,
'pdf_bytes': pdf_bytes,
'pdf_path': pdf_path,
'metadata': {
'slide_count': len(image_paths),
'generation_time': generation_time,
'file_size': file_size,
'file_size_mb': round(file_size / (1024 * 1024), 2),
'dimensions': f'{self.slide_width}x{self.slide_height}',
'aspect_ratio': self.slide_aspect_ratio,
}
}
except Exception as e:
logger.error("Error rendering carousel PDF: %s", str(e))
return {'success': False, 'error': f'Carousel PDF rendering failed: {str(e)}'}
def _render_slide(
self,
slide: Dict[str, Any],
slide_number: int,
session_id: str,
color_scheme: str = 'professional',
is_cover: bool = False,
is_cta: bool = False,
carousel_title: str = '',
) -> Optional[str]:
try:
colors = self.COLOR_SCHEMES.get(color_scheme, self.COLOR_SCHEMES['professional'])
img = Image.new('RGB', (self.slide_width, self.slide_height))
draw = ImageDraw.Draw(img)
self._draw_gradient(draw, colors)
draw.rectangle([0, self.slide_height - 6, self.slide_width, self.slide_height], fill=colors['accent_color'])
if is_cover:
self._draw_centered_text(draw, carousel_title or slide.get('title', ''),
(self.slide_width // 2, 180), colors['title_color'],
font_size=42, max_width=self.slide_width - 160)
subtitle = slide.get('content', '')
if subtitle:
self._draw_centered_text(draw, subtitle,
(self.slide_width // 2, 320), colors['content_color'],
font_size=24, max_width=self.slide_width - 200, max_lines=3)
self._draw_centered_text(draw, "Swipe to explore →",
(self.slide_width // 2, 480), colors['accent_color'],
font_size=18)
elif is_cta:
self._draw_text(draw, slide.get('title', ''), (60, 160), colors['title_color'],
font_size=36, max_width=self.slide_width - 120, max_lines=2)
content = slide.get('content', '')
if content:
self._draw_text(draw, content, (60, 260), colors['content_color'],
font_size=22, max_width=self.slide_width - 120, max_lines=6)
btn_x, btn_y = self.slide_width // 2 - 200, 440
draw.rounded_rectangle([btn_x, btn_y, btn_x + 400, btn_y + 55], radius=27, fill=colors['accent_color'])
self._draw_centered_text(draw, "Share Your Thoughts →",
(self.slide_width // 2, btn_y + 27), (255, 255, 255), font_size=22)
else:
self._draw_text(draw, str(slide_number),
(self.slide_width - 50, 20), colors['accent_color'], font_size=16)
title = slide.get('title', '')
if title:
self._draw_text(draw, title, (60, 50), colors['title_color'],
font_size=30, max_width=self.slide_width - 120, max_lines=2)
content = slide.get('content', '')
if content:
self._draw_text(draw, content, (60, 145), colors['content_color'],
font_size=20, max_width=self.slide_width - 120, max_lines=10)
visual_elements = slide.get('visual_elements', [])
if visual_elements:
self._draw_visual_elements(draw, visual_elements, colors)
filename = f"slide_{session_id}_{slide_number:03d}.png"
filepath = os.path.join(self.output_dir, filename)
img.save(filepath, 'PNG', optimize=True)
return filepath
except Exception as e:
logger.error("Error rendering slide %d: %s", slide_number, str(e))
return None
def _draw_gradient(self, draw: ImageDraw.Draw, colors: Dict):
sr, sg, sb = colors['background_start']
er, eg, eb = colors['background_end']
for y in range(self.slide_height):
t = y / self.slide_height
draw.line([(0, y), (self.slide_width, y)],
fill=(int(sr + (er - sr) * t), int(sg + (eg - sg) * t), int(sb + (eb - sb) * t)))
def _draw_text(self, draw: ImageDraw.Draw, text: str, position: tuple, color: tuple,
font_size: int = 20, max_width: int = None, max_lines: int = None, bold: bool = False):
font = self._get_font(font_size, bold)
x, y = position
words = text.split()
lines = []
current_line = ""
for word in words:
test_line = f"{current_line} {word}".strip()
bb = draw.textbbox((0, 0), test_line, font=font)
tw = bb[2] - bb[0]
if max_width and tw > max_width and current_line:
lines.append(current_line)
if max_lines and len(lines) >= max_lines:
lines[-1] = lines[-1][:-3] + "..."
break
current_line = word
else:
current_line = test_line
if current_line and (not max_lines or len(lines) < max_lines):
lines.append(current_line)
line_height = int(font_size * 1.4)
for i, line in enumerate(lines):
draw.text((x, y + i * line_height), line, fill=color, font=font)
def _draw_centered_text(self, draw: ImageDraw.Draw, text: str, center: tuple, color: tuple,
font_size: int = 20, max_width: int = None, max_lines: int = None, bold: bool = False):
font = self._get_font(font_size, bold)
cx, cy = center
words = text.split()
lines = []
current_line = ""
for word in words:
test_line = f"{current_line} {word}".strip()
bb = draw.textbbox((0, 0), test_line, font=font)
tw = bb[2] - bb[0]
if max_width and tw > max_width and current_line:
lines.append(current_line)
if max_lines and len(lines) >= max_lines:
lines[-1] = lines[-1][:-3] + "..."
break
current_line = word
else:
current_line = test_line
if current_line and (not max_lines or len(lines) < max_lines):
lines.append(current_line)
line_height = int(font_size * 1.4)
total_height = len(lines) * line_height
start_y = cy - total_height // 2
for i, line in enumerate(lines):
bb = draw.textbbox((0, 0), line, font=font)
tw = bb[2] - bb[0]
x = cx - tw // 2
draw.text((x, start_y + i * line_height), line, fill=color, font=font)
def _draw_visual_elements(self, draw: ImageDraw.Draw, elements: List[str], colors: Dict):
y_start = self.slide_height - 60
x_start = 60
for i, element in enumerate(elements[:4]):
cx = x_start + i * 280
draw.ellipse([cx, y_start, cx + 12, y_start + 12], fill=colors['accent_color'])
font = self._get_font(12, False)
draw.text((cx + 20, y_start - 2), element[:25], fill=colors['content_color'], font=font)
def _get_font(self, size: int, bold: bool = False):
try:
return ImageFont.truetype("arialbd.ttf" if bold else "arial.ttf", size)
except (IOError, OSError):
try:
return ImageFont.truetype("DejaVuSans-Bold.ttf" if bold else "DejaVuSans.ttf", size)
except (IOError, OSError):
return ImageFont.load_default()
def _compose_pdf(self, image_paths: List[str], output_path: str) -> bytes:
pw = self.slide_width
ph = self.slide_height
# Leave 1pt margin to avoid ReportLab frame size issues
m = 1
iw = pw - 2 * m
ih = ph - 2 * m
from reportlab.platypus import BaseDocTemplate, Frame, PageTemplate
from reportlab.lib.pagesizes import landscape
frame = Frame(m, m, iw, ih, id="slide_frame",
leftPadding=0, rightPadding=0, topPadding=0, bottomPadding=0)
template = PageTemplate(id="slide", frames=[frame], pagesize=(pw, ph))
doc = BaseDocTemplate(output_path, pagesize=(pw, ph))
doc.addPageTemplates([template])
story = []
for i, img_path in enumerate(image_paths):
story.append(RLImage(img_path, width=iw, height=ih))
if i < len(image_paths) - 1:
story.append(PageBreak())
doc.build(story)
with open(output_path, 'rb') as f:
return f.read()

313
backend/services/linkedin/content_generator.py
View File

@@ -2,6 +2,7 @@
Content Generator for LinkedIn Content Generation
Handles the main content generation logic for posts and articles.
Uses llm_text_gen for provider-agnostic LLM access (respects GPT_PROVIDER).
"""
from typing import Dict, Any, List, Optional
@@ -21,6 +22,7 @@ from services.linkedin.content_generator_prompts import (
CarouselGenerator,
VideoScriptGenerator
)
from services.llm_providers.main_text_generation import llm_text_gen
from services.persona_analysis_service import PersonaAnalysisService
import time
@@ -28,11 +30,9 @@ import time
class ContentGenerator:
"""Handles content generation for all LinkedIn content types."""
def __init__(self, citation_manager=None, quality_analyzer=None, gemini_grounded=None, fallback_provider=None):
def __init__(self, citation_manager=None, quality_analyzer=None):
self.citation_manager = citation_manager
self.quality_analyzer = quality_analyzer
self.gemini_grounded = gemini_grounded
self.fallback_provider = fallback_provider
# Persona caching
self._persona_cache: Dict[str, Dict[str, Any]] = {}
@@ -105,22 +105,24 @@ class ContentGenerator:
del self._cache_timestamps[key]
logger.info(f"Cleared persona cache for user {user_id}")
def _transform_gemini_sources(self, gemini_sources):
"""Transform Gemini sources to ResearchSource format."""
transformed_sources = []
for source in gemini_sources:
transformed_source = ResearchSource(
title=source.get('title', 'Unknown Source'),
url=source.get('url', ''),
content=f"Source from {source.get('title', 'Unknown')}",
relevance_score=0.8, # Default relevance score
credibility_score=0.7, # Default credibility score
domain_authority=0.6, # Default domain authority
source_type=source.get('type', 'web'),
publication_date=datetime.now().strftime('%Y-%m-%d')
)
transformed_sources.append(transformed_source)
return transformed_sources
def _build_research_context(self, research_sources: List) -> str:
"""Build research context string from research sources for prompt injection."""
if not research_sources:
return ""
context_parts = ["\n\nRESEARCH CONTEXT (use this information to ground your content with facts and data):"]
for i, source in enumerate(research_sources[:5], 1): # Limit to top 5 sources
title = getattr(source, 'title', f'Source {i}')
url = getattr(source, 'url', '')
content = getattr(source, 'content', '')
context_parts.append(f"\n{i}. {title}")
if url:
context_parts.append(f" URL: {url}")
if content:
context_parts.append(f" Key insight: {content[:300]}")
context_parts.append("\nInstructions: Use the research above to include specific data points, statistics, and factual claims in your content. Cite sources where appropriate.")
return "\n".join(context_parts)
async def generate_post(
self,
@@ -155,21 +157,12 @@ class ContentGenerator:
logger.info(f" - First research source: {research_sources[0] if research_sources else 'None'}")
logger.info(f" - Research sources types: {[type(s) for s in research_sources[:3]]}")
# Step 3: Add citations if requested - POST METHOD
# Step 3: Add citations if requested
citations = []
source_list = None
final_research_sources = research_sources # Default to passed research_sources
final_research_sources = research_sources
# Use sources and citations from content_result if available (from Gemini grounding)
if content_result.get('citations') and content_result.get('sources'):
logger.info(f"Using citations and sources from Gemini grounding: {len(content_result['citations'])} citations, {len(content_result['sources'])} sources")
citations = content_result['citations']
# Transform Gemini sources to ResearchSource format
gemini_sources = self._transform_gemini_sources(content_result['sources'])
source_list = self.citation_manager.generate_source_list(gemini_sources) if self.citation_manager else None
# Use transformed sources for the response
final_research_sources = gemini_sources
elif request.include_citations and research_sources and self.citation_manager:
if request.include_citations and research_sources and self.citation_manager:
try:
logger.info(f"Processing citations for content length: {len(content_result['content'])}")
citations = self.citation_manager.extract_citations(content_result['content'])
@@ -224,7 +217,7 @@ class ContentGenerator:
data=post_content,
research_sources=final_research_sources, # Use final_research_sources
generation_metadata={
'model_used': 'gemini-2.0-flash-001',
'model_used': 'llm_text_gen',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
@@ -251,21 +244,12 @@ class ContentGenerator:
try:
start_time = datetime.now()
# Step 3: Add citations if requested - ARTICLE METHOD
# Step 3: Add citations if requested
citations = []
source_list = None
final_research_sources = research_sources # Default to passed research_sources
final_research_sources = research_sources
# Use sources and citations from content_result if available (from Gemini grounding)
if content_result.get('citations') and content_result.get('sources'):
logger.info(f"Using citations and sources from Gemini grounding: {len(content_result['citations'])} citations, {len(content_result['sources'])} sources")
citations = content_result['citations']
# Transform Gemini sources to ResearchSource format
gemini_sources = self._transform_gemini_sources(content_result['sources'])
source_list = self.citation_manager.generate_source_list(gemini_sources) if self.citation_manager else None
# Use transformed sources for the response
final_research_sources = gemini_sources
elif request.include_citations and research_sources and self.citation_manager:
if request.include_citations and research_sources and self.citation_manager:
try:
citations = self.citation_manager.extract_citations(content_result['content'])
source_list = self.citation_manager.generate_source_list(research_sources)
@@ -317,7 +301,7 @@ class ContentGenerator:
data=article_content,
research_sources=final_research_sources, # Use final_research_sources
generation_metadata={
'model_used': 'gemini-2.0-flash-001',
'model_used': 'llm_text_gen',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
@@ -386,7 +370,7 @@ class ContentGenerator:
'alternative_responses': content_result.get('alternative_responses', []),
'tone_analysis': content_result.get('tone_analysis'),
'generation_metadata': {
'model_used': 'gemini-2.0-flash-001',
'model_used': 'llm_text_gen',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled
@@ -402,19 +386,14 @@ class ContentGenerator:
}
# Grounded content generation methods
async def generate_grounded_post_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded post content using the enhanced Gemini provider with native grounding."""
async def generate_grounded_post_content(self, request, research_sources: List, user_id: str = None) -> Dict[str, Any]:
"""Generate post content using provider-agnostic llm_text_gen."""
try:
if not self.gemini_grounded:
logger.error("Gemini Grounded Provider not available - cannot generate content without AI provider")
raise Exception("Gemini Grounded Provider not available - cannot generate content without AI provider")
# Build the prompt for grounded generation using persona if available (DB vs session override)
user_id = int(getattr(request, "user_id", 0) or 0)
persona_data = self._get_cached_persona_data(user_id, 'linkedin')
# Build the prompt using persona if available
uid = int(getattr(request, "user_id", 0) or 0)
persona_data = self._get_cached_persona_data(uid, 'linkedin')
if getattr(request, 'persona_override', None):
try:
# Merge shallowly: override core and platform adaptation parts
override = request.persona_override
if persona_data:
core = persona_data.get('core_persona', {})
@@ -431,61 +410,40 @@ class ContentGenerator:
pass
prompt = PostPromptBuilder.build_post_prompt(request, persona=persona_data)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
# Inject research context into prompt
research_context = self._build_research_context(research_sources)
if research_context:
prompt += research_context
# Generate content using provider-agnostic gateway
raw_response = llm_text_gen(
prompt=prompt,
content_type="linkedin_post",
temperature=0.7,
max_tokens=request.max_length
user_id=user_id,
flow_type="linkedin_post",
max_tokens=request.max_length,
temperature=0.7
)
return result
content_text = raw_response if isinstance(raw_response, str) else str(raw_response or "")
return {
'content': content_text,
'sources': [],
'citations': [],
'grounding_enabled': bool(research_sources),
'fallback_used': False
}
except Exception as e:
logger.error(f"Error generating grounded post content: {str(e)}")
logger.info("Attempting fallback to standard content generation...")
# Fallback to standard content generation without grounding
try:
if not self.fallback_provider:
raise Exception("No fallback provider available")
# Build a simpler prompt for fallback generation
prompt = PostPromptBuilder.build_post_prompt(request)
# Generate content using fallback provider (it's a dict with functions)
if 'generate_text' in self.fallback_provider:
result = await self.fallback_provider['generate_text'](
prompt=prompt,
temperature=0.7,
max_tokens=request.max_length
)
else:
raise Exception("Fallback provider doesn't have generate_text method")
# Return result in the expected format
return {
'content': result.get('content', '') if isinstance(result, dict) else str(result),
'sources': [],
'citations': [],
'grounding_enabled': False,
'fallback_used': True
}
except Exception as fallback_error:
logger.error(f"Fallback generation also failed: {str(fallback_error)}")
raise Exception(f"Failed to generate content: {str(e)}. Fallback also failed: {str(fallback_error)}")
logger.error(f"Error generating post content: {str(e)}")
raise Exception(f"Failed to generate LinkedIn post: {str(e)}")
async def generate_grounded_article_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded article content using the enhanced Gemini provider with native grounding."""
async def generate_grounded_article_content(self, request, research_sources: List, user_id: str = None) -> Dict[str, Any]:
"""Generate article content using provider-agnostic llm_text_gen."""
try:
if not self.gemini_grounded:
logger.error("Gemini Grounded Provider not available - cannot generate content without AI provider")
raise Exception("Gemini Grounded Provider not available - cannot generate content without AI provider")
# Build the prompt for grounded generation using persona if available (DB vs session override)
user_id = int(getattr(request, "user_id", 0) or 0)
persona_data = self._get_cached_persona_data(user_id, 'linkedin')
# Build the prompt using persona if available
uid = int(getattr(request, "user_id", 0) or 0)
persona_data = self._get_cached_persona_data(uid, 'linkedin')
if getattr(request, 'persona_override', None):
try:
override = request.persona_override
@@ -504,88 +462,129 @@ class ContentGenerator:
pass
prompt = ArticlePromptBuilder.build_article_prompt(request, persona=persona_data)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
# Inject research context into prompt
research_context = self._build_research_context(research_sources)
if research_context:
prompt += research_context
# Generate content using provider-agnostic gateway
raw_response = llm_text_gen(
prompt=prompt,
content_type="linkedin_article",
temperature=0.7,
max_tokens=request.word_count * 10 # Approximate character count
user_id=user_id,
flow_type="linkedin_article",
max_tokens=request.word_count * 10,
temperature=0.7
)
return result
content_text = raw_response if isinstance(raw_response, str) else str(raw_response or "")
return {
'content': content_text,
'sources': [],
'citations': [],
'grounding_enabled': bool(research_sources),
'fallback_used': False
}
except Exception as e:
logger.error(f"Error generating grounded article content: {str(e)}")
raise Exception(f"Failed to generate grounded article content: {str(e)}")
logger.error(f"Error generating article content: {str(e)}")
raise Exception(f"Failed to generate LinkedIn article: {str(e)}")
async def generate_grounded_carousel_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded carousel content using the enhanced Gemini provider with native grounding."""
async def generate_grounded_carousel_content(self, request, research_sources: List, user_id: str = None) -> Dict[str, Any]:
"""Generate carousel content using provider-agnostic llm_text_gen."""
try:
if not self.gemini_grounded:
logger.error("Gemini Grounded Provider not available - cannot generate content without AI provider")
raise Exception("Gemini Grounded Provider not available - cannot generate content without AI provider")
# Build the prompt for grounded generation using the new prompt builder
prompt = CarouselPromptBuilder.build_carousel_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
# Inject research context into prompt
research_context = self._build_research_context(research_sources)
if research_context:
prompt += research_context
# Generate content using provider-agnostic gateway
raw_response = llm_text_gen(
prompt=prompt,
content_type="linkedin_carousel",
temperature=0.7,
max_tokens=2000
user_id=user_id,
flow_type="linkedin_carousel",
max_tokens=2000,
temperature=0.7
)
return result
content_text = raw_response if isinstance(raw_response, str) else str(raw_response or "")
return {
'content': content_text,
'sources': [],
'citations': [],
'grounding_enabled': bool(research_sources),
'fallback_used': False
}
except Exception as e:
logger.error(f"Error generating grounded carousel content: {str(e)}")
raise Exception(f"Failed to generate grounded carousel content: {str(e)}")
logger.error(f"Error generating carousel content: {str(e)}")
raise Exception(f"Failed to generate LinkedIn carousel: {str(e)}")
async def generate_grounded_video_script_content(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded video script content using the enhanced Gemini provider with native grounding."""
async def generate_grounded_video_script_content(self, request, research_sources: List, user_id: str = None) -> Dict[str, Any]:
"""Generate video script content using provider-agnostic llm_text_gen."""
try:
if not self.gemini_grounded:
logger.error("Gemini Grounded Provider not available - cannot generate content without AI provider")
raise Exception("Gemini Grounded Provider not available - cannot generate content without AI provider")
# Build the prompt for grounded generation using the new prompt builder
prompt = VideoScriptPromptBuilder.build_video_script_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
# Inject research context into prompt
research_context = self._build_research_context(research_sources)
if research_context:
prompt += research_context
# Generate content using provider-agnostic gateway
raw_response = llm_text_gen(
prompt=prompt,
content_type="linkedin_video_script",
temperature=0.7,
max_tokens=1500
user_id=user_id,
flow_type="linkedin_video_script",
max_tokens=1500,
temperature=0.7
)
return result
content_text = raw_response if isinstance(raw_response, str) else str(raw_response or "")
return {
'content': content_text,
'sources': [],
'citations': [],
'grounding_enabled': bool(research_sources),
'fallback_used': False
}
except Exception as e:
logger.error(f"Error generating grounded video script content: {str(e)}")
raise Exception(f"Failed to generate grounded video script content: {str(e)}")
logger.error(f"Error generating video script content: {str(e)}")
raise Exception(f"Failed to generate LinkedIn video script: {str(e)}")
async def generate_grounded_comment_response(self, request, research_sources: List) -> Dict[str, Any]:
"""Generate grounded comment response using the enhanced Gemini provider with native grounding."""
async def generate_grounded_comment_response(self, request, research_sources: List, user_id: str = None) -> Dict[str, Any]:
"""Generate comment response using provider-agnostic llm_text_gen."""
try:
if not self.gemini_grounded:
logger.error("Gemini Grounded Provider not available - cannot generate content without AI provider")
raise Exception("Gemini Grounded Provider not available - cannot generate content without AI provider")
# Build the prompt for grounded generation using the new prompt builder
prompt = CommentResponsePromptBuilder.build_comment_response_prompt(request)
# Generate grounded content using native Google Search grounding
result = await self.gemini_grounded.generate_grounded_content(
# Inject research context into prompt
research_context = self._build_research_context(research_sources)
if research_context:
prompt += research_context
# Generate content using provider-agnostic gateway
raw_response = llm_text_gen(
prompt=prompt,
content_type="linkedin_comment_response",
temperature=0.7,
max_tokens=2000
user_id=user_id,
flow_type="linkedin_comment_response",
max_tokens=2000,
temperature=0.7
)
return result
content_text = raw_response if isinstance(raw_response, str) else str(raw_response or "")
return {
'content': content_text,
'sources': [],
'citations': [],
'grounding_enabled': bool(research_sources),
'fallback_used': False
}
except Exception as e:
logger.error(f"Error generating grounded comment response: {str(e)}")
raise Exception(f"Failed to generate grounded comment response: {str(e)}")
logger.error(f"Error generating comment response: {str(e)}")
raise Exception(f"Failed to generate LinkedIn comment response: {str(e)}")

2
backend/services/linkedin/content_generator_prompts/carousel_generator.py
View File

@@ -96,7 +96,7 @@ class CarouselGenerator:
'data': carousel_content,
'research_sources': research_sources,
'generation_metadata': {
'model_used': 'gemini-2.0-flash-001',
'model_used': 'llm_text_gen',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled

2
backend/services/linkedin/content_generator_prompts/video_script_generator.py
View File

@@ -81,7 +81,7 @@ class VideoScriptGenerator:
'data': video_script,
'research_sources': research_sources,
'generation_metadata': {
'model_used': 'gemini-2.0-flash-001',
'model_used': 'llm_text_gen',
'generation_time': generation_time,
'research_time': research_time,
'grounding_enabled': grounding_enabled

6
backend/services/linkedin/image_generation/__init__.py
View File

@@ -2,17 +2,15 @@
LinkedIn Image Generation Package
This package provides AI-powered image generation capabilities for LinkedIn content
using Google's Gemini API. It includes image generation, editing, storage, and
management services optimized for professional business use.
using the common llm_providers infrastructure. It includes image generation, storage,
and management services optimized for professional business use.
"""
from .linkedin_image_generator import LinkedInImageGenerator
from .linkedin_image_editor import LinkedInImageEditor
from .linkedin_image_storage import LinkedInImageStorage
__all__ = [
'LinkedInImageGenerator',
'LinkedInImageEditor',
'LinkedInImageStorage'
]

530
backend/services/linkedin/image_generation/linkedin_image_editor.py
View File

@@ -1,530 +0,0 @@
"""
LinkedIn Image Editor Service
This service handles image editing capabilities for LinkedIn content using Gemini's
conversational editing features. It provides professional image refinement and
optimization specifically for LinkedIn use cases.
"""
import os
import base64
from typing import Dict, Any, Optional, List
from datetime import datetime
from PIL import Image, ImageEnhance, ImageFilter
from io import BytesIO
from loguru import logger
# Import existing infrastructure
from ...onboarding.api_key_manager import APIKeyManager
class LinkedInImageEditor:
"""
Handles LinkedIn image editing and refinement using Gemini's capabilities.
This service provides both AI-powered editing through Gemini and traditional
image processing for LinkedIn-specific optimizations.
"""
def __init__(self, api_key_manager: Optional[APIKeyManager] = None):
"""
Initialize the LinkedIn Image Editor.
Args:
api_key_manager: API key manager for Gemini authentication
"""
self.api_key_manager = api_key_manager or APIKeyManager()
self.model = "gemini-2.5-flash-image-preview"
# LinkedIn-specific editing parameters
self.enhancement_factors = {
'brightness': 1.1, # Slightly brighter for mobile viewing
'contrast': 1.05, # Subtle contrast enhancement
'sharpness': 1.2, # Enhanced sharpness for clarity
'saturation': 1.05 # Slight saturation boost
}
logger.info("LinkedIn Image Editor initialized")
async def edit_image_conversationally(
self,
base_image: bytes,
edit_prompt: str,
content_context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Edit image using Gemini's conversational editing capabilities.
Args:
base_image: Base image data in bytes
edit_prompt: Natural language description of desired edits
content_context: LinkedIn content context for optimization
Returns:
Dict containing edited image result and metadata
"""
try:
start_time = datetime.now()
logger.info(f"Starting conversational image editing: {edit_prompt[:100]}...")
# Enhance edit prompt for LinkedIn optimization
enhanced_prompt = self._enhance_edit_prompt_for_linkedin(
edit_prompt, content_context
)
# TODO: Implement Gemini conversational editing when available
# For now, we'll use traditional image processing based on prompt analysis
edited_image = await self._apply_traditional_editing(
base_image, edit_prompt, content_context
)
if not edited_image.get('success'):
return edited_image
generation_time = (datetime.now() - start_time).total_seconds()
return {
'success': True,
'image_data': edited_image['image_data'],
'metadata': {
'edit_prompt': edit_prompt,
'enhanced_prompt': enhanced_prompt,
'editing_method': 'traditional_processing',
'editing_time': generation_time,
'content_context': content_context,
'model_used': self.model
},
'linkedin_optimization': {
'mobile_optimized': True,
'professional_aesthetic': True,
'brand_compliant': True,
'engagement_optimized': True
}
}
except Exception as e:
logger.error(f"Error in conversational image editing: {str(e)}")
return {
'success': False,
'error': f"Conversational editing failed: {str(e)}",
'generation_time': (datetime.now() - start_time).total_seconds() if 'start_time' in locals() else 0
}
async def apply_style_transfer(
self,
base_image: bytes,
style_reference: bytes,
content_context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Apply style transfer from reference image to base image.
Args:
base_image: Base image data in bytes
style_reference: Reference image for style transfer
content_context: LinkedIn content context
Returns:
Dict containing style-transferred image result
"""
try:
start_time = datetime.now()
logger.info("Starting style transfer for LinkedIn image")
# TODO: Implement Gemini style transfer when available
# For now, return placeholder implementation
return {
'success': False,
'error': 'Style transfer not yet implemented - coming in next Gemini API update',
'generation_time': (datetime.now() - start_time).total_seconds()
}
except Exception as e:
logger.error(f"Error in style transfer: {str(e)}")
return {
'success': False,
'error': f"Style transfer failed: {str(e)}",
'generation_time': (datetime.now() - start_time).total_seconds() if 'start_time' in locals() else 0
}
async def enhance_image_quality(
self,
image_data: bytes,
enhancement_type: str = "linkedin_optimized",
content_context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Enhance image quality using traditional image processing.
Args:
image_data: Image data in bytes
enhancement_type: Type of enhancement to apply
content_context: LinkedIn content context for optimization
Returns:
Dict containing enhanced image result
"""
try:
start_time = datetime.now()
logger.info(f"Starting image quality enhancement: {enhancement_type}")
# Open image for processing
image = Image.open(BytesIO(image_data))
original_size = image.size
# Apply LinkedIn-specific enhancements
if enhancement_type == "linkedin_optimized":
enhanced_image = self._apply_linkedin_enhancements(image, content_context)
elif enhancement_type == "professional":
enhanced_image = self._apply_professional_enhancements(image)
elif enhancement_type == "creative":
enhanced_image = self._apply_creative_enhancements(image)
else:
enhanced_image = self._apply_linkedin_enhancements(image, content_context)
# Convert back to bytes
output_buffer = BytesIO()
enhanced_image.save(output_buffer, format=image.format or "PNG", optimize=True)
enhanced_data = output_buffer.getvalue()
enhancement_time = (datetime.now() - start_time).total_seconds()
return {
'success': True,
'image_data': enhanced_data,
'metadata': {
'enhancement_type': enhancement_type,
'original_size': original_size,
'enhanced_size': enhanced_image.size,
'enhancement_time': enhancement_time,
'content_context': content_context
}
}
except Exception as e:
logger.error(f"Error in image quality enhancement: {str(e)}")
return {
'success': False,
'error': f"Quality enhancement failed: {str(e)}",
'generation_time': (datetime.now() - start_time).total_seconds() if 'start_time' in locals() else 0
}
def _enhance_edit_prompt_for_linkedin(
self,
edit_prompt: str,
content_context: Dict[str, Any]
) -> str:
"""
Enhance edit prompt for LinkedIn optimization.
Args:
edit_prompt: Original edit prompt
content_context: LinkedIn content context
Returns:
Enhanced edit prompt
"""
industry = content_context.get('industry', 'business')
content_type = content_context.get('content_type', 'post')
linkedin_edit_enhancements = [
f"Maintain professional business aesthetic for {industry} industry",
f"Ensure mobile-optimized composition for LinkedIn {content_type}",
"Keep professional color scheme and typography",
"Maintain brand consistency and visual hierarchy",
"Optimize for LinkedIn feed viewing and engagement"
]
enhanced_prompt = f"{edit_prompt}\n\n"
enhanced_prompt += "\n".join(linkedin_edit_enhancements)
return enhanced_prompt
async def _apply_traditional_editing(
self,
base_image: bytes,
edit_prompt: str,
content_context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Apply traditional image processing based on edit prompt analysis.
Args:
base_image: Base image data in bytes
edit_prompt: Description of desired edits
content_context: LinkedIn content context
Returns:
Dict containing edited image result
"""
try:
# Open image for processing
image = Image.open(BytesIO(base_image))
# Analyze edit prompt and apply appropriate processing
edit_prompt_lower = edit_prompt.lower()
if any(word in edit_prompt_lower for word in ['brighter', 'light', 'lighting']):
image = self._adjust_brightness(image, 1.2)
logger.info("Applied brightness adjustment")
if any(word in edit_prompt_lower for word in ['sharper', 'sharp', 'clear']):
image = self._apply_sharpening(image)
logger.info("Applied sharpening")
if any(word in edit_prompt_lower for word in ['warmer', 'warm', 'color']):
image = self._adjust_color_temperature(image, 'warm')
logger.info("Applied warm color adjustment")
if any(word in edit_prompt_lower for word in ['professional', 'business']):
image = self._apply_professional_enhancements(image)
logger.info("Applied professional enhancements")
# Convert back to bytes
output_buffer = BytesIO()
image.save(output_buffer, format=image.format or "PNG", optimize=True)
edited_data = output_buffer.getvalue()
return {
'success': True,
'image_data': edited_data
}
except Exception as e:
logger.error(f"Error in traditional editing: {str(e)}")
return {
'success': False,
'error': f"Traditional editing failed: {str(e)}"
}
def _apply_linkedin_enhancements(
self,
image: Image.Image,
content_context: Optional[Dict[str, Any]] = None
) -> Image.Image:
"""
Apply LinkedIn-specific image enhancements.
Args:
image: PIL Image object
content_context: LinkedIn content context
Returns:
Enhanced image
"""
try:
# Apply standard LinkedIn optimizations
image = self._adjust_brightness(image, self.enhancement_factors['brightness'])
image = self._adjust_contrast(image, self.enhancement_factors['contrast'])
image = self._apply_sharpening(image)
image = self._adjust_saturation(image, self.enhancement_factors['saturation'])
# Ensure professional appearance
image = self._ensure_professional_appearance(image, content_context)
return image
except Exception as e:
logger.error(f"Error applying LinkedIn enhancements: {str(e)}")
return image
def _apply_professional_enhancements(self, image: Image.Image) -> Image.Image:
"""
Apply professional business aesthetic enhancements.
Args:
image: PIL Image object
Returns:
Enhanced image
"""
try:
# Subtle enhancements for professional appearance
image = self._adjust_brightness(image, 1.05)
image = self._adjust_contrast(image, 1.03)
image = self._apply_sharpening(image)
return image
except Exception as e:
logger.error(f"Error applying professional enhancements: {str(e)}")
return image
def _apply_creative_enhancements(self, image: Image.Image) -> Image.Image:
"""
Apply creative and engaging enhancements.
Args:
image: PIL Image object
Returns:
Enhanced image
"""
try:
# More pronounced enhancements for creative appeal
image = self._adjust_brightness(image, 1.1)
image = self._adjust_contrast(image, 1.08)
image = self._adjust_saturation(image, 1.1)
image = self._apply_sharpening(image)
return image
except Exception as e:
logger.error(f"Error applying creative enhancements: {str(e)}")
return image
def _adjust_brightness(self, image: Image.Image, factor: float) -> Image.Image:
"""Adjust image brightness."""
try:
enhancer = ImageEnhance.Brightness(image)
return enhancer.enhance(factor)
except Exception as e:
logger.error(f"Error adjusting brightness: {str(e)}")
return image
def _adjust_contrast(self, image: Image.Image, factor: float) -> Image.Image:
"""Adjust image contrast."""
try:
enhancer = ImageEnhance.Contrast(image)
return enhancer.enhance(factor)
except Exception as e:
logger.error(f"Error adjusting contrast: {str(e)}")
return image
def _adjust_saturation(self, image: Image.Image, factor: float) -> Image.Image:
"""Adjust image saturation."""
try:
enhancer = ImageEnhance.Color(image)
return enhancer.enhance(factor)
except Exception as e:
logger.error(f"Error adjusting saturation: {str(e)}")
return image
def _apply_sharpening(self, image: Image.Image) -> Image.Image:
"""Apply image sharpening."""
try:
# Apply unsharp mask for professional sharpening
return image.filter(ImageFilter.UnsharpMask(radius=1, percent=150, threshold=3))
except Exception as e:
logger.error(f"Error applying sharpening: {str(e)}")
return image
def _adjust_color_temperature(self, image: Image.Image, temperature: str) -> Image.Image:
"""Adjust image color temperature."""
try:
if temperature == 'warm':
# Apply warm color adjustment
enhancer = ImageEnhance.Color(image)
image = enhancer.enhance(1.1)
# Slight red tint for warmth
# This is a simplified approach - more sophisticated color grading could be implemented
return image
else:
return image
except Exception as e:
logger.error(f"Error adjusting color temperature: {str(e)}")
return image
def _ensure_professional_appearance(
self,
image: Image.Image,
content_context: Optional[Dict[str, Any]] = None
) -> Image.Image:
"""
Ensure image meets professional LinkedIn standards.
Args:
image: PIL Image object
content_context: LinkedIn content context
Returns:
Professionally optimized image
"""
try:
# Ensure minimum quality standards
if image.mode in ('RGBA', 'LA', 'P'):
# Convert to RGB for better compatibility
background = Image.new('RGB', image.size, (255, 255, 255))
if image.mode == 'P':
image = image.convert('RGBA')
background.paste(image, mask=image.split()[-1] if image.mode == 'RGBA' else None)
image = background
# Ensure minimum resolution for LinkedIn
min_resolution = (1024, 1024)
if image.size[0] < min_resolution[0] or image.size[1] < min_resolution[1]:
# Resize to minimum resolution while maintaining aspect ratio
ratio = max(min_resolution[0] / image.size[0], min_resolution[1] / image.size[1])
new_size = (int(image.size[0] * ratio), int(image.size[1] * ratio))
image = image.resize(new_size, Image.Resampling.LANCZOS)
logger.info(f"Resized image to {new_size} for LinkedIn professional standards")
return image
except Exception as e:
logger.error(f"Error ensuring professional appearance: {str(e)}")
return image
async def get_editing_suggestions(
self,
image_data: bytes,
content_context: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""
Get AI-powered editing suggestions for LinkedIn image.
Args:
image_data: Image data in bytes
content_context: LinkedIn content context
Returns:
List of editing suggestions
"""
try:
# Analyze image and provide contextual suggestions
suggestions = []
# Professional enhancement suggestions
suggestions.append({
'id': 'professional_enhancement',
'title': 'Professional Enhancement',
'description': 'Apply subtle professional enhancements for business appeal',
'prompt': 'Enhance this image with professional business aesthetics',
'priority': 'high'
})
# Mobile optimization suggestions
suggestions.append({
'id': 'mobile_optimization',
'title': 'Mobile Optimization',
'description': 'Optimize for LinkedIn mobile feed viewing',
'prompt': 'Optimize this image for mobile LinkedIn viewing',
'priority': 'medium'
})
# Industry-specific suggestions
industry = content_context.get('industry', 'business')
suggestions.append({
'id': 'industry_optimization',
'title': f'{industry.title()} Industry Optimization',
'description': f'Apply {industry} industry-specific visual enhancements',
'prompt': f'Enhance this image with {industry} industry aesthetics',
'priority': 'medium'
})
# Engagement optimization suggestions
suggestions.append({
'id': 'engagement_optimization',
'title': 'Engagement Optimization',
'description': 'Make this image more engaging for LinkedIn audience',
'prompt': 'Make this image more engaging and shareable for LinkedIn',
'priority': 'low'
})
return suggestions
except Exception as e:
logger.error(f"Error getting editing suggestions: {str(e)}")
return []

110
backend/services/linkedin/image_generation/linkedin_image_generator.py
View File

@@ -1,8 +1,9 @@
"""
LinkedIn Image Generator Service
This service generates LinkedIn-optimized images using Google's Gemini API.
It provides professional, business-appropriate imagery for LinkedIn content.
This service generates LinkedIn-optimized images using the common
llm_providers infrastructure. It provides professional, business-appropriate
imagery for LinkedIn content.
"""
import os
@@ -17,6 +18,7 @@ from io import BytesIO
# Import existing infrastructure
from ...onboarding.api_key_manager import APIKeyManager
from ...llm_providers.main_image_generation import generate_image
from ...llm_providers.main_image_editing import edit_image as common_edit_image
# Set up logging
logger = logging.getLogger(__name__)
@@ -24,9 +26,9 @@ logger = logging.getLogger(__name__)
class LinkedInImageGenerator:
"""
Handles LinkedIn-optimized image generation using Gemini API.
Handles LinkedIn-optimized image generation using common infrastructure.
This service integrates with the existing Gemini provider infrastructure
This service integrates with the llm_providers image generation system
and provides LinkedIn-specific image optimization, quality assurance,
and professional business aesthetics.
"""
@@ -36,10 +38,9 @@ class LinkedInImageGenerator:
Initialize the LinkedIn Image Generator.
Args:
api_key_manager: API key manager for Gemini authentication
api_key_manager: API key manager for authentication
"""
self.api_key_manager = api_key_manager or APIKeyManager()
self.model = "gemini-2.5-flash-image-preview"
self.default_aspect_ratio = "1:1" # LinkedIn post optimal ratio
self.max_retries = 3
@@ -55,16 +56,18 @@ class LinkedInImageGenerator:
prompt: str,
content_context: Dict[str, Any],
aspect_ratio: str = "1:1",
style_preference: str = "professional"
style_preference: str = "professional",
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Generate LinkedIn-optimized image using Gemini API.
Generate LinkedIn-optimized image using AI provider.
Args:
prompt: User's image generation prompt
content_context: LinkedIn content context (topic, industry, content_type)
aspect_ratio: Image aspect ratio (1:1, 16:9, 4:3)
aspect_ratio: Image aspect ratio (1:1, 16:9, 4:3, 1.91:1, 1:1.25)
style_preference: Style preference (professional, creative, industry-specific)
user_id: User ID for tenant provider resolution
Returns:
Dict containing generation result, image data, and metadata
@@ -78,8 +81,8 @@ class LinkedInImageGenerator:
prompt, content_context, style_preference, aspect_ratio
)
# Generate image using existing Gemini infrastructure
generation_result = await self._generate_with_gemini(enhanced_prompt, aspect_ratio)
# Generate image using tenant-aware provider selection
generation_result = await self._generate_with_provider(enhanced_prompt, aspect_ratio, user_id)
if not generation_result.get('success'):
return {
@@ -108,7 +111,7 @@ class LinkedInImageGenerator:
'aspect_ratio': aspect_ratio,
'content_context': content_context,
'generation_time': generation_time,
'model_used': self.model,
'model_used': generation_result.get('model'),
'image_format': processed_image['format'],
'image_size': processed_image['size'],
'resolution': processed_image['resolution']
@@ -131,17 +134,19 @@ class LinkedInImageGenerator:
async def edit_image(
self,
base_image: bytes,
input_image_bytes: bytes,
edit_prompt: str,
content_context: Dict[str, Any]
content_context: Dict[str, Any],
user_id: Optional[str] = None,
) -> Dict[str, Any]:
"""
Edit existing image using Gemini's conversational editing capabilities.
Edit existing image using unified image editing infrastructure.
Args:
base_image: Base image data in bytes
input_image_bytes: Input image bytes to edit
edit_prompt: Description of desired edits
content_context: LinkedIn content context for optimization
user_id: User ID for tenant provider resolution and subscription checks
Returns:
Dict containing edited image result and metadata
@@ -155,18 +160,46 @@ class LinkedInImageGenerator:
edit_prompt, content_context
)
# Use Gemini's image editing capabilities
# Note: This will be implemented when Gemini's image editing is fully available
# For now, we'll return a placeholder implementation
# Use unified image editing system.
# common_edit_image() handles: provider resolution, pre-flight validation,
# generation, and usage tracking — all via user_id.
result = common_edit_image(
input_image_bytes=input_image_bytes,
prompt=enhanced_edit_prompt,
user_id=user_id,
)
return {
'success': False,
'error': 'Image editing not yet implemented - coming in next Gemini API update',
'generation_time': (datetime.now() - start_time).total_seconds()
}
if result and result.image_bytes:
generation_time = (datetime.now() - start_time).total_seconds()
logger.info(
"LinkedIn image edited successfully via provider=%s model=%s in %.2fs",
result.provider, result.model, generation_time,
)
return {
'success': True,
'image_data': result.image_bytes,
'image_url': None, # not using URL-based retrieval
'width': result.width,
'height': result.height,
'provider': result.provider,
'model': result.model,
'metadata': {
'original_prompt': edit_prompt,
'enhanced_prompt': enhanced_edit_prompt,
'generation_time': generation_time,
'content_context': content_context,
},
}
else:
logger.warning("LinkedIn image editing returned no result")
return {
'success': False,
'error': 'Image editing returned no result',
'generation_time': (datetime.now() - start_time).total_seconds(),
}
except Exception as e:
logger.error(f"Error in LinkedIn image editing: {str(e)}")
logger.error(f"Error in LinkedIn image editing: {str(e)}", exc_info=True)
return {
'success': False,
'error': f"Image editing failed: {str(e)}",
@@ -268,13 +301,16 @@ class LinkedInImageGenerator:
return enhanced_edit_prompt
async def _generate_with_gemini(self, prompt: str, aspect_ratio: str) -> Dict[str, Any]:
async def _generate_with_provider(self, prompt: str, aspect_ratio: str, user_id: Optional[str] = None) -> Dict[str, Any]:
"""
Generate image using unified image generation infrastructure.
Provider resolution, pre-flight validation, and usage tracking
are all handled by generate_image() from main_image_generation.
Args:
prompt: Enhanced prompt for image generation
aspect_ratio: Desired aspect ratio
user_id: User ID for tenant provider resolution and subscription checks
Returns:
Generation result from image generation provider
@@ -285,26 +321,31 @@ class LinkedInImageGenerator:
"1:1": (1024, 1024),
"16:9": (1920, 1080),
"4:3": (1366, 1024),
"9:16": (1080, 1920), # Portrait for stories
"9:16": (1080, 1920),
"1.91:1": (1200, 627), # LinkedIn recommended landscape
"1:1.25": (1080, 1350), # LinkedIn recommended portrait
}
width, height = aspect_map.get(aspect_ratio, (1024, 1024))
# Use unified image generation system (defaults to provider based on GPT_PROVIDER)
# Delegate to unified image generation system.
# Generate_image() handles: provider resolution, pre-flight validation,
# model auto-detection, generation, and usage tracking.
# We do NOT pass explicit provider or model — let generate_image() resolve
# them from tenant config and user defaults.
result = generate_image(
prompt=prompt,
options={
"provider": "gemini", # LinkedIn uses Gemini by default
"model": self.model if hasattr(self, 'model') else None,
"width": width,
"height": height,
}
},
user_id=user_id
)
if result and result.image_bytes:
return {
'success': True,
'image_data': result.image_bytes,
'image_path': None, # No file path, using bytes directly
'image_path': None,
'width': result.width,
'height': result.height,
'provider': result.provider,
@@ -315,7 +356,7 @@ class LinkedInImageGenerator:
'success': False,
'error': 'Image generation returned no result'
}
except Exception as e:
logger.error(f"Error in image generation: {str(e)}")
return {
@@ -487,6 +528,9 @@ class LinkedInImageGenerator:
(1.6, 1.8), # 16:9 (landscape)
(0.7, 0.8), # 4:3 (portrait)
(1.2, 1.4), # 5:4 (landscape)
(1.85, 2.0), # 1.91:1 (LinkedIn recommended landscape)
(0.6, 0.72), # 1:1.25 (LinkedIn recommended portrait, ~0.8)
(0.65, 0.85), # 1:1.25 broader match
]
for min_ratio, max_ratio in suitable_ratios:

55
backend/services/linkedin/image_generation/linkedin_image_storage.py
View File

@@ -6,8 +6,10 @@ It provides secure storage, efficient retrieval, and metadata management for gen
"""
import os
import re
import hashlib
import json
import shutil
from typing import Dict, Any, Optional, List, Tuple
from datetime import datetime, timedelta
from pathlib import Path
@@ -58,6 +60,8 @@ class LinkedInImageStorage:
self.max_storage_size_gb = 10 # Maximum storage size in GB
self.image_retention_days = 30 # Days to keep images
self.max_image_size_mb = 10 # Maximum individual image size in MB
self.max_images_per_user = 100 # Maximum images per user
self._uuid_pattern = re.compile(r'^[a-f0-9]{16}$')
logger.info(f"LinkedIn Image Storage initialized at {self.base_storage_path}")
@@ -102,6 +106,22 @@ class LinkedInImageStorage:
try:
start_time = datetime.now()
# Check per-user storage quota
if user_id:
user_count = await self._count_user_images(user_id)
if user_count >= self.max_images_per_user:
return {
'success': False,
'error': f"User image limit ({self.max_images_per_user}) reached. Delete existing images or increase limit."
}
# Check disk space
if not await self._check_disk_space(len(image_data)):
return {
'success': False,
'error': "Insufficient disk space for image storage."
}
# Generate unique image ID
image_id = self._generate_image_id(image_data, metadata)
@@ -170,6 +190,9 @@ class LinkedInImageStorage:
Dict containing image data and metadata
"""
try:
if not self._validate_image_id(image_id):
return {'success': False, 'error': f'Invalid image ID format: {image_id}'}
# Find image file
image_path = await self._find_image_by_id(image_id, user_id)
if not image_path:
@@ -216,6 +239,9 @@ class LinkedInImageStorage:
Dict containing deletion result
"""
try:
if not self._validate_image_id(image_id):
return {'success': False, 'error': f'Invalid image ID format: {image_id}'}
# Find image file
image_path = await self._find_image_by_id(image_id, user_id)
if not image_path:
@@ -418,6 +444,32 @@ class LinkedInImageStorage:
'error': f"Failed to get storage stats: {str(e)}"
}
def _validate_image_id(self, image_id: str) -> bool:
"""Validate image_id against expected format to prevent path traversal."""
return bool(self._uuid_pattern.match(image_id))
async def _count_user_images(self, user_id: str) -> int:
"""Count total images stored for a given user."""
try:
images_path, _ = self._get_workspace_paths(user_id)
count = 0
if images_path.exists():
for content_dir in images_path.iterdir():
if content_dir.is_dir():
count += sum(1 for f in content_dir.glob("*.png") if f.is_file())
return count
except Exception as e:
logger.warning(f"Error counting images for user {user_id}: {e}")
return 0
async def _check_disk_space(self, required_bytes: int) -> bool:
"""Check if sufficient disk space is available."""
try:
usage = shutil.disk_usage(self.base_storage_path)
return usage.free > required_bytes * 2 # require 2x headroom
except Exception:
return True # if we can't check, allow the write
def _generate_image_id(self, image_data: bytes, metadata: Dict[str, Any]) -> str:
"""Generate unique image ID based on content and metadata."""
# Create hash from image data and key metadata
@@ -569,6 +621,9 @@ class LinkedInImageStorage:
Returns:
Dict containing image metadata if found
"""
if not self._validate_image_id(image_id):
logger.warning(f"Invalid image ID format in metadata request: {image_id}")
return None
return await self._load_metadata(image_id, user_id)
async def _load_metadata(self, image_id: str, user_id: Optional[str] = None) -> Optional[Dict[str, Any]]:

4
backend/services/linkedin/image_prompts/__init__.py
View File

@@ -2,8 +2,8 @@
LinkedIn Image Prompts Package
This package provides AI-powered image prompt generation for LinkedIn content
using Google's Gemini API. It creates three distinct prompt styles optimized
for professional business image generation.
using the provider-agnostic llm_text_gen gateway. It creates three distinct
prompt styles optimized for professional business image generation.
"""
from .linkedin_prompt_generator import LinkedInPromptGenerator

75
backend/services/linkedin/image_prompts/linkedin_prompt_generator.py
View File

@@ -1,9 +1,10 @@
"""
LinkedIn Image Prompt Generator Service
This service generates AI-optimized image prompts for LinkedIn content using Gemini's
capabilities. It creates three distinct prompt styles (professional, creative, industry-specific)
following best practices for image generation.
This service generates AI-optimized image prompts for LinkedIn content using
the provider-agnostic llm_text_gen gateway. It creates three distinct prompt
styles (professional, creative, industry-specific) following best practices
for image generation.
"""
import asyncio
@@ -13,14 +14,14 @@ from loguru import logger
# Import existing infrastructure
from ...onboarding.api_key_manager import APIKeyManager
from ...llm_providers.gemini_provider import gemini_text_response
from ...llm_providers.main_text_generation import llm_text_gen
class LinkedInPromptGenerator:
"""
Generates AI-optimized image prompts for LinkedIn content.
This service creates three distinct prompt styles following Gemini API best practices:
This service creates three distinct prompt styles following best practices:
1. Professional Style - Corporate aesthetics, clean lines, business colors
2. Creative Style - Engaging visuals, vibrant colors, social media appeal
3. Industry-Specific Style - Tailored to specific business sectors
@@ -31,10 +32,9 @@ class LinkedInPromptGenerator:
Initialize the LinkedIn Prompt Generator.
Args:
api_key_manager: API key manager for Gemini authentication
api_key_manager: API key manager for authentication
"""
self.api_key_manager = api_key_manager or APIKeyManager()
self.model = "gemini-2.0-flash-exp"
# Prompt generation configuration
self.max_prompt_length = 500
@@ -49,7 +49,8 @@ class LinkedInPromptGenerator:
async def generate_three_prompts(
self,
linkedin_content: Dict[str, Any],
aspect_ratio: str = "1:1"
aspect_ratio: str = "1:1",
user_id: str = None
) -> List[Dict[str, Any]]:
"""
Generate three AI-optimized image prompts for LinkedIn content.
@@ -57,6 +58,7 @@ class LinkedInPromptGenerator:
Args:
linkedin_content: LinkedIn content context (topic, industry, content_type, content)
aspect_ratio: Desired image aspect ratio
user_id: User ID for subscription checking
Returns:
List of three prompt objects with style, prompt, and description
@@ -65,11 +67,11 @@ class LinkedInPromptGenerator:
start_time = datetime.now()
logger.info(f"Generating image prompts for LinkedIn content: {linkedin_content.get('topic', 'Unknown')}")
# Generate prompts using Gemini
prompts = await self._generate_prompts_with_gemini(linkedin_content, aspect_ratio)
# Generate prompts using provider-agnostic gateway
prompts = await self._generate_prompts_with_llm(linkedin_content, aspect_ratio, user_id)
if not prompts or len(prompts) < 3:
logger.warning("Gemini prompt generation failed, using fallback prompts")
logger.warning("Prompt generation failed, using fallback prompts")
prompts = self._get_fallback_prompts(linkedin_content, aspect_ratio)
# Ensure exactly 3 prompts
@@ -92,62 +94,65 @@ class LinkedInPromptGenerator:
logger.error(f"Error generating LinkedIn image prompts: {str(e)}")
return self._get_fallback_prompts(linkedin_content, aspect_ratio)
async def _generate_prompts_with_gemini(
async def _generate_prompts_with_llm(
self,
linkedin_content: Dict[str, Any],
aspect_ratio: str
aspect_ratio: str,
user_id: str = None
) -> List[Dict[str, Any]]:
"""
Generate image prompts using Gemini AI.
Generate image prompts using provider-agnostic llm_text_gen.
Args:
linkedin_content: LinkedIn content context
aspect_ratio: Image aspect ratio
user_id: User ID for subscription checking
Returns:
List of generated prompts
"""
try:
# Build the prompt for Gemini
gemini_prompt = self._build_gemini_prompt(linkedin_content, aspect_ratio)
# Build the prompt
prompt = self._build_image_prompt(linkedin_content, aspect_ratio)
# Generate response using Gemini
response = gemini_text_response(
prompt=gemini_prompt,
temperature=0.7,
top_p=0.8,
n=1,
# Generate response using provider-agnostic gateway
response = llm_text_gen(
prompt=prompt,
system_prompt="You are an expert AI image prompt engineer specializing in LinkedIn content optimization.",
user_id=user_id,
flow_type="linkedin_image_prompts",
max_tokens=1000,
system_prompt="You are an expert AI image prompt engineer specializing in LinkedIn content optimization."
temperature=0.7
)
if not response:
logger.warning("No response from Gemini prompt generation")
logger.warning("No response from prompt generation")
return []
# Parse Gemini response into structured prompts
prompts = self._parse_gemini_response(response, linkedin_content)
# Parse response into structured prompts
response_text = response if isinstance(response, str) else str(response or "")
prompts = self._parse_llm_response(response_text, linkedin_content)
return prompts
except Exception as e:
logger.error(f"Error in Gemini prompt generation: {str(e)}")
logger.error(f"Error in prompt generation: {str(e)}")
return []
def _build_gemini_prompt(
def _build_image_prompt(
self,
linkedin_content: Dict[str, Any],
aspect_ratio: str
) -> str:
"""
Build comprehensive prompt for Gemini to generate image prompts.
Build comprehensive prompt for LLM to generate image prompts.
Args:
linkedin_content: LinkedIn content context
aspect_ratio: Image aspect ratio
Returns:
Formatted prompt for Gemini
Formatted prompt for LLM
"""
topic = linkedin_content.get('topic', 'business')
industry = linkedin_content.get('industry', 'business')
@@ -428,16 +433,16 @@ class LinkedInPromptGenerator:
else:
return 'Informational & Awareness'
def _parse_gemini_response(
def _parse_llm_response(
self,
response: str,
linkedin_content: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""
Parse Gemini response into structured prompt objects.
Parse LLM response into structured prompt objects.
Args:
response: Raw response from Gemini
response: Raw response from LLM
linkedin_content: LinkedIn content context
Returns:
@@ -462,7 +467,7 @@ class LinkedInPromptGenerator:
return self._parse_response_manually(response, linkedin_content)
except Exception as e:
logger.error(f"Error parsing Gemini response: {str(e)}")
logger.error(f"Error parsing LLM response: {str(e)}")
return self._parse_response_manually(response, linkedin_content)
def _parse_response_manually(
@@ -474,7 +479,7 @@ class LinkedInPromptGenerator:
Manually parse response if JSON parsing fails.
Args:
response: Raw response from Gemini
response: Raw response from LLM
linkedin_content: LinkedIn content context
Returns:

22
backend/services/linkedin/research_handler.py
View File

@@ -2,9 +2,10 @@
Research Handler for LinkedIn Content Generation
Handles research operations and timing for content generation.
Uses common Exa/Tavily infrastructure with pre-flight validation.
"""
from typing import List
from typing import List, Optional
from datetime import datetime
from loguru import logger
from models.linkedin_models import ResearchSource
@@ -21,11 +22,19 @@ class ResearchHandler:
request,
research_enabled: bool,
search_engine: str,
max_results: int = 10
max_results: int = 10,
user_id: Optional[str] = None
) -> tuple[List[ResearchSource], float]:
"""
Conduct research if enabled and return sources with timing.
Args:
request: Generation request object
research_enabled: Whether research is enabled
search_engine: Search engine to use (exa, tavily)
max_results: Maximum number of results
user_id: User ID for pre-flight validation and usage tracking
Returns:
Tuple of (research_sources, research_time)
"""
@@ -33,7 +42,6 @@ class ResearchHandler:
research_time = 0
if research_enabled:
# Debug: Log the search engine value being passed
logger.info(f"ResearchHandler: search_engine='{search_engine}' (type: {type(search_engine)})")
research_start = datetime.now()
@@ -41,7 +49,8 @@ class ResearchHandler:
topic=request.topic,
industry=request.industry,
search_engine=search_engine,
max_results=max_results
max_results=max_results,
user_id=user_id
)
research_time = (datetime.now() - research_start).total_seconds()
logger.info(f"Research completed in {research_time:.2f}s, found {len(research_sources)} sources")
@@ -67,10 +76,5 @@ class ResearchHandler:
if not research_enabled or level == 'none':
return False
# For Google native grounding, Gemini returns sources in the generation metadata,
# so we should not require pre-fetched research_sources.
if engine_str == 'google':
return True
# For other engines, require that research actually returned sources
return bool(research_sources)