ALwrity Version 0.5.0 (Fastapi + React )
This commit is contained in:
ajaysi
127
lib/content_calendar/integrations/gap_analyzer.py
Normal file
127
lib/content_calendar/integrations/gap_analyzer.py
Normal file
@@ -0,0 +1,127 @@
|
||||
"""
|
||||
Gap analyzer integration for content calendar.
|
||||
"""
|
||||
|
||||
import streamlit as st
|
||||
from typing import Dict, Any, List, Optional
|
||||
from loguru import logger
|
||||
from lib.utils.website_analyzer.analyzer import WebsiteAnalyzer
|
||||
from lib.ai_seo_tools.content_gap_analysis.main import ContentGapAnalysis
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
# Configure logger for content calendar debugging
|
||||
logger.remove() # Remove default handler
|
||||
logger.add(
|
||||
sys.stdout,
|
||||
level="DEBUG",
|
||||
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan> | <yellow>{function}</yellow> | {message}",
|
||||
filter=lambda record: "content_calendar" in record["name"].lower()
|
||||
)
|
||||
|
||||
class GapAnalyzerIntegration:
|
||||
"""Integrates content gap analysis with content calendar."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the gap analyzer integration."""
|
||||
self.gap_analyzer = ContentGapAnalysis()
|
||||
logger.debug("GapAnalyzerIntegration initialized for content calendar")
|
||||
|
||||
def analyze_gaps(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Analyze content gaps.
|
||||
|
||||
Args:
|
||||
data: Dictionary containing content data
|
||||
|
||||
Returns:
|
||||
Dictionary containing gap analysis results
|
||||
"""
|
||||
try:
|
||||
logger.debug(f"Starting gap analysis with data: {json.dumps(data, indent=2)}")
|
||||
# Run gap analysis
|
||||
results = self.gap_analyzer.analyze(data)
|
||||
logger.debug(f"Gap analysis completed with results: {json.dumps(results, indent=2)}")
|
||||
return results
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Error analyzing content gaps: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return {
|
||||
'error': error_msg,
|
||||
'gaps': [],
|
||||
'recommendations': []
|
||||
}
|
||||
|
||||
def get_topic_suggestions(
|
||||
self,
|
||||
gap_analysis: Dict[str, Any],
|
||||
platform: str,
|
||||
count: int = 5
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get topic suggestions for a specific platform based on gap analysis.
|
||||
|
||||
Args:
|
||||
gap_analysis: Results from gap analysis
|
||||
platform: Target platform for content
|
||||
count: Number of suggestions to generate
|
||||
|
||||
Returns:
|
||||
List of topic suggestions
|
||||
"""
|
||||
try:
|
||||
logger.debug(f"Generating topic suggestions for platform: {platform}, count: {count}")
|
||||
suggestions = []
|
||||
|
||||
for gap in gap_analysis.get('processed_gaps', []):
|
||||
# Generate platform-specific topics
|
||||
platform_topics = self.ai_processor.generate_platform_topics(
|
||||
gap=gap,
|
||||
platform=platform,
|
||||
count=count
|
||||
)
|
||||
logger.debug(f"Generated topics for gap: {json.dumps(platform_topics, indent=2)}")
|
||||
suggestions.extend(platform_topics)
|
||||
|
||||
logger.debug(f"Total suggestions generated: {len(suggestions)}")
|
||||
return suggestions
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating topic suggestions: {str(e)}")
|
||||
return []
|
||||
|
||||
def analyze_topic_relevance(
|
||||
self,
|
||||
topic: Dict[str, Any],
|
||||
gap_analysis: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Analyze how well a topic addresses content gaps.
|
||||
|
||||
Args:
|
||||
topic: Topic to analyze
|
||||
gap_analysis: Results from gap analysis
|
||||
|
||||
Returns:
|
||||
Dictionary containing relevance analysis
|
||||
"""
|
||||
try:
|
||||
logger.debug(f"Analyzing topic relevance: {json.dumps(topic, indent=2)}")
|
||||
relevance = self.ai_processor.analyze_topic_relevance(
|
||||
topic=topic,
|
||||
gaps=gap_analysis.get('gaps', [])
|
||||
)
|
||||
|
||||
logger.debug(f"Topic relevance analysis completed: {json.dumps(relevance, indent=2)}")
|
||||
return relevance
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error analyzing topic relevance: {str(e)}")
|
||||
return {
|
||||
'error': str(e),
|
||||
'score': 0
|
||||
}
|
||||
196
lib/content_calendar/integrations/integration_manager.py
Normal file
196
lib/content_calendar/integrations/integration_manager.py
Normal file
@@ -0,0 +1,196 @@
|
||||
import logging
|
||||
from typing import Dict, List, Any, Optional
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from ..core.calendar_manager import CalendarManager
|
||||
from ..core.content_brief import ContentBriefGenerator
|
||||
from .platform_adapters import UnifiedPlatformAdapter
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class IntegrationManager:
|
||||
"""Manages integration between content calendar and platform adapters."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the integration manager."""
|
||||
self.calendar_manager = CalendarManager()
|
||||
self.content_brief_generator = ContentBriefGenerator()
|
||||
self.platform_adapter = UnifiedPlatformAdapter()
|
||||
|
||||
def create_cross_platform_calendar(
|
||||
self,
|
||||
start_date: datetime,
|
||||
end_date: datetime,
|
||||
platforms: List[str],
|
||||
content_types: List[str],
|
||||
target_audience: Optional[Dict[str, Any]] = None,
|
||||
industry: Optional[str] = None,
|
||||
keywords: Optional[List[str]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a cross-platform content calendar."""
|
||||
try:
|
||||
# Generate base calendar
|
||||
calendar = self.calendar_manager.create_calendar(
|
||||
start_date=start_date,
|
||||
end_date=end_date,
|
||||
content_types=content_types,
|
||||
target_audience=target_audience,
|
||||
industry=industry,
|
||||
keywords=keywords
|
||||
)
|
||||
|
||||
# Adapt content for each platform
|
||||
platform_calendars = {}
|
||||
for platform in platforms:
|
||||
platform_calendars[platform] = self._adapt_calendar_for_platform(
|
||||
calendar=calendar,
|
||||
platform=platform
|
||||
)
|
||||
|
||||
return {
|
||||
'base_calendar': calendar,
|
||||
'platform_calendars': platform_calendars,
|
||||
'metadata': {
|
||||
'start_date': start_date,
|
||||
'end_date': end_date,
|
||||
'platforms': platforms,
|
||||
'content_types': content_types,
|
||||
'industry': industry,
|
||||
'keywords': keywords
|
||||
}
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating cross-platform calendar: {str(e)}")
|
||||
raise
|
||||
|
||||
def _adapt_calendar_for_platform(
|
||||
self,
|
||||
calendar: Dict[str, Any],
|
||||
platform: str
|
||||
) -> Dict[str, Any]:
|
||||
"""Adapt calendar content for a specific platform."""
|
||||
try:
|
||||
adapted_calendar = {
|
||||
'platform': platform,
|
||||
'content_items': [],
|
||||
'metadata': calendar.get('metadata', {})
|
||||
}
|
||||
|
||||
# Adapt each content item
|
||||
for item in calendar.get('content_items', []):
|
||||
adapted_item = self._adapt_content_item(item, platform)
|
||||
if adapted_item:
|
||||
adapted_calendar['content_items'].append(adapted_item)
|
||||
|
||||
return adapted_calendar
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error adapting calendar for platform {platform}: {str(e)}")
|
||||
return {
|
||||
'platform': platform,
|
||||
'content_items': [],
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _adapt_content_item(
|
||||
self,
|
||||
item: Dict[str, Any],
|
||||
platform: str
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Adapt a content item for a specific platform."""
|
||||
try:
|
||||
# Generate content brief if not exists
|
||||
if 'brief' not in item:
|
||||
item['brief'] = self.content_brief_generator.generate_brief(item)
|
||||
|
||||
# Adapt content for platform
|
||||
adapted_content = self.platform_adapter.adapt_content(
|
||||
content=item,
|
||||
platform=platform
|
||||
)
|
||||
|
||||
if adapted_content:
|
||||
return {
|
||||
'original_item': item,
|
||||
'adapted_content': adapted_content,
|
||||
'platform_specifics': self.platform_adapter.get_platform_specs(platform)
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error adapting content item for platform {platform}: {str(e)}")
|
||||
return None
|
||||
|
||||
def get_platform_suggestions(
|
||||
self,
|
||||
content: Dict[str, Any],
|
||||
platforms: List[str]
|
||||
) -> Dict[str, Any]:
|
||||
"""Get platform-specific suggestions for content."""
|
||||
try:
|
||||
suggestions = {}
|
||||
|
||||
for platform in platforms:
|
||||
platform_suggestions = self.platform_adapter.get_platform_suggestions(
|
||||
content=content,
|
||||
platform=platform
|
||||
)
|
||||
if platform_suggestions:
|
||||
suggestions[platform] = platform_suggestions
|
||||
|
||||
return suggestions
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting platform suggestions: {str(e)}")
|
||||
return {}
|
||||
|
||||
def validate_platform_content(
|
||||
self,
|
||||
content: Dict[str, Any],
|
||||
platform: str
|
||||
) -> Dict[str, Any]:
|
||||
"""Validate content for a specific platform."""
|
||||
try:
|
||||
validation_result = self.platform_adapter.validate_content(
|
||||
content=content,
|
||||
platform=platform
|
||||
)
|
||||
|
||||
return {
|
||||
'platform': platform,
|
||||
'is_valid': validation_result,
|
||||
'specifications': self.platform_adapter.get_platform_specs(platform)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error validating platform content: {str(e)}")
|
||||
return {
|
||||
'platform': platform,
|
||||
'is_valid': False,
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def optimize_cross_platform_content(
|
||||
self,
|
||||
content: Dict[str, Any],
|
||||
platforms: List[str]
|
||||
) -> Dict[str, Any]:
|
||||
"""Optimize content for multiple platforms."""
|
||||
try:
|
||||
optimized_content = {}
|
||||
|
||||
for platform in platforms:
|
||||
platform_optimized = self.platform_adapter.optimize_content(
|
||||
content=content,
|
||||
platform=platform
|
||||
)
|
||||
if platform_optimized:
|
||||
optimized_content[platform] = platform_optimized
|
||||
|
||||
return optimized_content
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error optimizing cross-platform content: {str(e)}")
|
||||
return {}
|
||||
307
lib/content_calendar/integrations/platform_adapters.py
Normal file
307
lib/content_calendar/integrations/platform_adapters.py
Normal file
@@ -0,0 +1,307 @@
|
||||
"""
|
||||
Unified platform adapter for content adaptation across different platforms.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Dict, Any, List, Optional, TypedDict
|
||||
from datetime import datetime
|
||||
from loguru import logger
|
||||
|
||||
from lib.utils.website_analyzer.analyzer import WebsiteAnalyzer
|
||||
from lib.ai_seo_tools.content_gap_analysis.main import ContentGapAnalysis
|
||||
from lib.ai_seo_tools.content_title_generator import ai_title_generator
|
||||
from lib.ai_seo_tools.meta_desc_generator import metadesc_generator_main
|
||||
from lib.ai_seo_tools.seo_structured_data import ai_structured_data
|
||||
|
||||
class ContentItem(TypedDict):
|
||||
"""Type definition for content items."""
|
||||
id: str
|
||||
title: str
|
||||
content: str
|
||||
platforms: List[str]
|
||||
status: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
published_at: Optional[datetime]
|
||||
metadata: Dict[str, Any]
|
||||
analytics: Optional[Dict[str, Any]]
|
||||
|
||||
class UnifiedPlatformAdapter:
|
||||
"""Unified adapter for different social media platforms."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the platform adapter."""
|
||||
self.platform_handlers = {
|
||||
'instagram': self._handle_instagram,
|
||||
'linkedin': self._handle_linkedin,
|
||||
'twitter': self._handle_twitter,
|
||||
'facebook': self._handle_facebook
|
||||
}
|
||||
logger.info("UnifiedPlatformAdapter initialized")
|
||||
|
||||
def generate_content(self, platform: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Generate content for a specific platform.
|
||||
|
||||
Args:
|
||||
platform: Target platform
|
||||
data: Content data
|
||||
|
||||
Returns:
|
||||
Dictionary containing generated content
|
||||
"""
|
||||
try:
|
||||
handler = self.platform_handlers.get(platform.lower())
|
||||
if not handler:
|
||||
raise ValueError(f"Unsupported platform: {platform}")
|
||||
|
||||
return handler(data)
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Error generating content for {platform}: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return {
|
||||
'error': error_msg,
|
||||
'content': None
|
||||
}
|
||||
|
||||
def get_content_performance(self, content_item: ContentItem) -> Dict[str, Any]:
|
||||
"""Get performance metrics for content across platforms."""
|
||||
try:
|
||||
logger.info(f"Getting performance metrics for content: {getattr(content_item, 'title', 'Untitled')}")
|
||||
|
||||
# Get platform from content item
|
||||
platforms = getattr(content_item, 'platforms', None)
|
||||
if platforms and len(platforms) > 0:
|
||||
platform = platforms[0].name if hasattr(platforms[0], 'name') else str(platforms[0])
|
||||
else:
|
||||
platform = 'Unknown'
|
||||
|
||||
# Initialize performance metrics
|
||||
performance = {
|
||||
'engagement_metrics': {
|
||||
'likes': 0,
|
||||
'comments': 0,
|
||||
'shares': 0,
|
||||
'reach': 0
|
||||
},
|
||||
'seo_metrics': {
|
||||
'impressions': 0,
|
||||
'clicks': 0,
|
||||
'ctr': 0,
|
||||
'position': 0
|
||||
},
|
||||
'conversion_metrics': {
|
||||
'conversions': 0,
|
||||
'conversion_rate': 0,
|
||||
'revenue': 0
|
||||
},
|
||||
'platform_specific': {},
|
||||
'performance_trends': [],
|
||||
'recommendations': []
|
||||
}
|
||||
|
||||
# Add platform-specific metrics
|
||||
if platform.upper() == 'WEBSITE':
|
||||
performance['platform_specific'] = {
|
||||
'bounce_rate': 0,
|
||||
'time_on_page': 0,
|
||||
'page_views': 0
|
||||
}
|
||||
|
||||
return performance
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Error getting content performance: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return {
|
||||
'error': error_msg,
|
||||
'metrics': {},
|
||||
'trends': {},
|
||||
'recommendations': []
|
||||
}
|
||||
|
||||
def _handle_instagram(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Handle Instagram content generation."""
|
||||
try:
|
||||
# Generate Instagram-specific content
|
||||
caption = metadesc_generator_main(data)
|
||||
hashtags = self._generate_hashtags(data)
|
||||
|
||||
return {
|
||||
'platform': 'instagram',
|
||||
'content': {
|
||||
'caption': caption,
|
||||
'hashtags': hashtags,
|
||||
'media_suggestions': self._get_media_suggestions(data)
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating Instagram content: {str(e)}")
|
||||
return {
|
||||
'platform': 'instagram',
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _handle_linkedin(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Handle LinkedIn content generation."""
|
||||
try:
|
||||
# Generate LinkedIn-specific content
|
||||
post = metadesc_generator_main(data)
|
||||
|
||||
return {
|
||||
'platform': 'linkedin',
|
||||
'content': {
|
||||
'post': post,
|
||||
'engagement_optimization': self._get_engagement_suggestions(data),
|
||||
'media_suggestions': self._get_media_suggestions(data)
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating LinkedIn content: {str(e)}")
|
||||
return {
|
||||
'platform': 'linkedin',
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _handle_twitter(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Handle Twitter content generation."""
|
||||
try:
|
||||
# Generate Twitter-specific content
|
||||
tweet = metadesc_generator_main(data)
|
||||
hashtags = self._generate_hashtags(data)
|
||||
|
||||
return {
|
||||
'platform': 'twitter',
|
||||
'content': {
|
||||
'tweet': tweet,
|
||||
'hashtags': hashtags,
|
||||
'thread_structure': self._get_thread_structure(data),
|
||||
'media_suggestions': self._get_media_suggestions(data)
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating Twitter content: {str(e)}")
|
||||
return {
|
||||
'platform': 'twitter',
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _handle_facebook(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Handle Facebook content generation."""
|
||||
try:
|
||||
# Generate Facebook-specific content
|
||||
post = metadesc_generator_main(data)
|
||||
|
||||
return {
|
||||
'platform': 'facebook',
|
||||
'content': {
|
||||
'post': post,
|
||||
'engagement_optimization': self._get_engagement_suggestions(data),
|
||||
'media_suggestions': self._get_media_suggestions(data)
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating Facebook content: {str(e)}")
|
||||
return {
|
||||
'platform': 'facebook',
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _generate_hashtags(self, data: Dict[str, Any]) -> List[str]:
|
||||
"""Generate relevant hashtags for content."""
|
||||
try:
|
||||
# Extract keywords from content
|
||||
keywords = data.get('keywords', [])
|
||||
|
||||
# Add platform-specific hashtags
|
||||
platform = data.get('platform', '').lower()
|
||||
platform_hashtags = {
|
||||
'instagram': ['#instagood', '#photooftheday'],
|
||||
'twitter': ['#trending', '#followme'],
|
||||
'linkedin': ['#business', '#professional'],
|
||||
'facebook': ['#social', '#community']
|
||||
}.get(platform, [])
|
||||
|
||||
return keywords + platform_hashtags
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating hashtags: {str(e)}")
|
||||
return []
|
||||
|
||||
def _get_media_suggestions(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""Get media suggestions for content."""
|
||||
try:
|
||||
# Generate media suggestions based on content type
|
||||
content_type = data.get('type', 'post')
|
||||
|
||||
suggestions = []
|
||||
if content_type == 'blog':
|
||||
suggestions.append({
|
||||
'type': 'featured_image',
|
||||
'description': 'Main blog post image',
|
||||
'dimensions': '1200x630'
|
||||
})
|
||||
elif content_type == 'social':
|
||||
suggestions.append({
|
||||
'type': 'post_image',
|
||||
'description': 'Social media post image',
|
||||
'dimensions': '1080x1080'
|
||||
})
|
||||
|
||||
return suggestions
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting media suggestions: {str(e)}")
|
||||
return []
|
||||
|
||||
def _get_engagement_suggestions(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get engagement optimization suggestions."""
|
||||
try:
|
||||
return {
|
||||
'best_posting_times': ['9:00 AM', '5:00 PM'],
|
||||
'engagement_tips': [
|
||||
'Ask questions to encourage comments',
|
||||
'Use relevant hashtags',
|
||||
'Include a clear call-to-action'
|
||||
],
|
||||
'content_length': {
|
||||
'optimal': '150-200 characters',
|
||||
'maximum': '300 characters'
|
||||
}
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting engagement suggestions: {str(e)}")
|
||||
return {}
|
||||
|
||||
def _get_thread_structure(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""Get thread structure for Twitter threads."""
|
||||
try:
|
||||
content = data.get('content', '')
|
||||
sentences = content.split('.')
|
||||
|
||||
thread = []
|
||||
current_tweet = ''
|
||||
|
||||
for sentence in sentences:
|
||||
if len(current_tweet + sentence) <= 280:
|
||||
current_tweet += sentence + '.'
|
||||
else:
|
||||
if current_tweet:
|
||||
thread.append({
|
||||
'content': current_tweet.strip(),
|
||||
'type': 'tweet'
|
||||
})
|
||||
current_tweet = sentence + '.'
|
||||
|
||||
if current_tweet:
|
||||
thread.append({
|
||||
'content': current_tweet.strip(),
|
||||
'type': 'tweet'
|
||||
})
|
||||
|
||||
return thread
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating thread structure: {str(e)}")
|
||||
return []
|
||||
219
lib/content_calendar/integrations/seo_optimizer.py
Normal file
219
lib/content_calendar/integrations/seo_optimizer.py
Normal file
@@ -0,0 +1,219 @@
|
||||
import logging
|
||||
from typing import Dict, Any, List, Optional
|
||||
from datetime import datetime
|
||||
|
||||
from ...meta_desc_generator import generate_blog_metadesc
|
||||
from ...content_title_generator import generate_blog_titles
|
||||
from ...seo_structured_data import generate_json_data
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class SEOOptimizer:
|
||||
"""Integrates SEO tools with content calendar system."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the SEO optimizer."""
|
||||
self._setup_logging()
|
||||
|
||||
def _setup_logging(self):
|
||||
"""Configure logging for SEO optimizer."""
|
||||
logger.setLevel(logging.INFO)
|
||||
handler = logging.StreamHandler()
|
||||
formatter = logging.Formatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
handler.setFormatter(formatter)
|
||||
logger.addHandler(handler)
|
||||
|
||||
def optimize_content(
|
||||
self,
|
||||
content: Dict[str, Any],
|
||||
content_type: str = 'article',
|
||||
language: str = 'English',
|
||||
search_intent: str = 'Informational Intent'
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Optimize content for SEO using existing tools.
|
||||
|
||||
Args:
|
||||
content: Content to optimize
|
||||
content_type: Type of content (article, product, etc.)
|
||||
language: Content language
|
||||
search_intent: Search intent type
|
||||
|
||||
Returns:
|
||||
Optimized content with SEO elements
|
||||
"""
|
||||
try:
|
||||
# Extract content details
|
||||
title = content.get('title', '')
|
||||
keywords = content.get('keywords', [])
|
||||
content_text = content.get('content', '')
|
||||
|
||||
# Generate SEO elements
|
||||
optimized_title = self._optimize_title(
|
||||
title=title,
|
||||
keywords=keywords,
|
||||
content_type=content_type,
|
||||
language=language,
|
||||
search_intent=search_intent
|
||||
)
|
||||
|
||||
meta_description = self._generate_meta_description(
|
||||
keywords=keywords,
|
||||
content_type=content_type,
|
||||
language=language,
|
||||
search_intent=search_intent
|
||||
)
|
||||
|
||||
structured_data = self._generate_structured_data(
|
||||
content=content,
|
||||
content_type=content_type
|
||||
)
|
||||
|
||||
return {
|
||||
'original_content': content,
|
||||
'seo_optimized': {
|
||||
'title': optimized_title,
|
||||
'meta_description': meta_description,
|
||||
'structured_data': structured_data,
|
||||
'keywords': keywords,
|
||||
'content_type': content_type,
|
||||
'language': language,
|
||||
'search_intent': search_intent
|
||||
}
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error optimizing content: {str(e)}")
|
||||
return {
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _optimize_title(
|
||||
self,
|
||||
title: str,
|
||||
keywords: List[str],
|
||||
content_type: str,
|
||||
language: str,
|
||||
search_intent: str
|
||||
) -> List[str]:
|
||||
"""Generate SEO-optimized titles."""
|
||||
try:
|
||||
# Convert keywords list to comma-separated string
|
||||
keywords_str = ', '.join(keywords)
|
||||
|
||||
# Generate titles using existing tool
|
||||
titles = generate_blog_titles(
|
||||
input_blog_keywords=keywords_str,
|
||||
input_blog_content=title,
|
||||
input_title_type=content_type,
|
||||
input_title_intent=search_intent,
|
||||
input_language=language
|
||||
)
|
||||
|
||||
return titles.split('\n') if titles else []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error optimizing title: {str(e)}")
|
||||
return []
|
||||
|
||||
def _generate_meta_description(
|
||||
self,
|
||||
keywords: List[str],
|
||||
content_type: str,
|
||||
language: str,
|
||||
search_intent: str
|
||||
) -> List[str]:
|
||||
"""Generate SEO-optimized meta descriptions."""
|
||||
try:
|
||||
# Convert keywords list to comma-separated string
|
||||
keywords_str = ', '.join(keywords)
|
||||
|
||||
# Generate meta descriptions using existing tool
|
||||
descriptions = generate_blog_metadesc(
|
||||
keywords=keywords_str,
|
||||
tone='Informative',
|
||||
search_type=search_intent,
|
||||
language=language
|
||||
)
|
||||
|
||||
return descriptions.split('\n') if descriptions else []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating meta description: {str(e)}")
|
||||
return []
|
||||
|
||||
def _generate_structured_data(
|
||||
self,
|
||||
content: Dict[str, Any],
|
||||
content_type: str
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Generate structured data for content."""
|
||||
try:
|
||||
# Prepare content details for structured data
|
||||
details = {
|
||||
'Headline': content.get('title', ''),
|
||||
'Author': content.get('author', ''),
|
||||
'Date Published': content.get('publish_date', datetime.now().isoformat()),
|
||||
'Keywords': ', '.join(content.get('keywords', [])),
|
||||
'Description': content.get('description', ''),
|
||||
'Image URL': content.get('image_url', '')
|
||||
}
|
||||
|
||||
# Generate structured data using existing tool
|
||||
structured_data = generate_json_data(
|
||||
content_type=content_type,
|
||||
details=details,
|
||||
url=content.get('url', '')
|
||||
)
|
||||
|
||||
return structured_data
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating structured data: {str(e)}")
|
||||
return None
|
||||
|
||||
def optimize_calendar_content(
|
||||
self,
|
||||
calendar: Dict[str, Any],
|
||||
content_type: str = 'article',
|
||||
language: str = 'English',
|
||||
search_intent: str = 'Informational Intent'
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Optimize all content in calendar for SEO.
|
||||
|
||||
Args:
|
||||
calendar: Content calendar to optimize
|
||||
content_type: Type of content
|
||||
language: Content language
|
||||
search_intent: Search intent type
|
||||
|
||||
Returns:
|
||||
Calendar with SEO-optimized content
|
||||
"""
|
||||
try:
|
||||
optimized_calendar = {
|
||||
'metadata': calendar.get('metadata', {}),
|
||||
'content_items': []
|
||||
}
|
||||
|
||||
# Optimize each content item
|
||||
for item in calendar.get('content_items', []):
|
||||
optimized_item = self.optimize_content(
|
||||
content=item,
|
||||
content_type=content_type,
|
||||
language=language,
|
||||
search_intent=search_intent
|
||||
)
|
||||
if optimized_item:
|
||||
optimized_calendar['content_items'].append(optimized_item)
|
||||
|
||||
return optimized_calendar
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error optimizing calendar content: {str(e)}")
|
||||
return {
|
||||
'error': str(e)
|
||||
}
|
||||
143
lib/content_calendar/integrations/seo_tools.py
Normal file
143
lib/content_calendar/integrations/seo_tools.py
Normal file
@@ -0,0 +1,143 @@
|
||||
"""SEO tools integration for content calendar."""
|
||||
|
||||
import streamlit as st
|
||||
from loguru import logger
|
||||
from typing import Dict, Any, List, Optional
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
from lib.ai_seo_tools.content_title_generator import ai_title_generator
|
||||
from lib.utils.website_analyzer.analyzer import WebsiteAnalyzer
|
||||
from lib.gpt_providers.text_generation.main_text_generation import llm_text_gen
|
||||
|
||||
# Configure logger
|
||||
logger.remove() # Remove default handler
|
||||
logger.add(
|
||||
"logs/seo_tools_integration.log",
|
||||
rotation="50 MB",
|
||||
retention="10 days",
|
||||
level="DEBUG",
|
||||
format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}"
|
||||
)
|
||||
logger.add(
|
||||
sys.stdout,
|
||||
level="INFO",
|
||||
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{message}</cyan>"
|
||||
)
|
||||
|
||||
# Ensure logs directory exists
|
||||
os.makedirs("logs", exist_ok=True)
|
||||
|
||||
class SEOToolsIntegration:
|
||||
"""Integration with SEO tools for content calendar."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the SEO tools integration."""
|
||||
self.website_analyzer = WebsiteAnalyzer()
|
||||
logger.info("SEOToolsIntegration initialized")
|
||||
|
||||
def analyze_content(self, url: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Analyze content for SEO optimization.
|
||||
|
||||
Args:
|
||||
url: The URL to analyze
|
||||
|
||||
Returns:
|
||||
Dictionary containing SEO analysis results
|
||||
"""
|
||||
try:
|
||||
# Analyze website
|
||||
analysis = self.website_analyzer.analyze_website(url)
|
||||
if not analysis.get('success', False):
|
||||
return {
|
||||
'error': analysis.get('error', 'Unknown error in analysis'),
|
||||
'seo_score': 0,
|
||||
'recommendations': []
|
||||
}
|
||||
|
||||
# Extract SEO information
|
||||
seo_info = analysis['data']['analysis']['seo_info']
|
||||
|
||||
return {
|
||||
'seo_score': seo_info.get('overall_score', 0),
|
||||
'meta_tags': seo_info.get('meta_tags', {}),
|
||||
'content': seo_info.get('content', {}),
|
||||
'recommendations': seo_info.get('recommendations', [])
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Error analyzing content: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return {
|
||||
'error': error_msg,
|
||||
'seo_score': 0,
|
||||
'recommendations': []
|
||||
}
|
||||
|
||||
def generate_title(self, url: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Generate SEO-optimized title.
|
||||
|
||||
Args:
|
||||
url: The URL to analyze
|
||||
|
||||
Returns:
|
||||
Dictionary containing title suggestions
|
||||
"""
|
||||
return ai_title_generator(url)
|
||||
|
||||
def optimize_content(self, content: str, keywords: List[str]) -> Dict[str, Any]:
|
||||
"""
|
||||
Optimize content for SEO.
|
||||
|
||||
Args:
|
||||
content: The content to optimize
|
||||
keywords: List of target keywords
|
||||
|
||||
Returns:
|
||||
Dictionary containing optimization suggestions
|
||||
"""
|
||||
try:
|
||||
# Prepare prompt for content optimization
|
||||
prompt = f"""Optimize the following content for SEO:
|
||||
|
||||
Content: {content}
|
||||
Target Keywords: {', '.join(keywords)}
|
||||
|
||||
Provide optimization suggestions for:
|
||||
1. Keyword usage and placement
|
||||
2. Content structure and readability
|
||||
3. Meta information
|
||||
4. Internal linking opportunities
|
||||
5. Content length and depth
|
||||
|
||||
Format the response as JSON with 'suggestions' and 'score' keys."""
|
||||
|
||||
# Get AI optimization suggestions
|
||||
suggestions = llm_text_gen(
|
||||
prompt=prompt,
|
||||
system_prompt="You are an SEO expert specializing in content optimization.",
|
||||
response_format="json_object"
|
||||
)
|
||||
|
||||
if not suggestions:
|
||||
return {
|
||||
'error': 'Failed to generate optimization suggestions',
|
||||
'suggestions': [],
|
||||
'score': 0
|
||||
}
|
||||
|
||||
return {
|
||||
'suggestions': suggestions.get('suggestions', []),
|
||||
'score': suggestions.get('score', 0)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Error optimizing content: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return {
|
||||
'error': error_msg,
|
||||
'suggestions': [],
|
||||
'score': 0
|
||||
}
|
||||
Reference in New Issue
Block a user