ALwrity version 0.5.4
This commit is contained in:
ajaysi
@@ -1,10 +1,18 @@
|
||||
"""
|
||||
AI Analysis Module
|
||||
AI recommendation generation and analysis services.
|
||||
AI recommendation generation and analysis.
|
||||
"""
|
||||
|
||||
from .ai_recommendations import AIRecommendationsService
|
||||
from .prompt_engineering import PromptEngineeringService
|
||||
from .quality_validation import QualityValidationService
|
||||
from .prompt_engineering import PromptEngineeringService
|
||||
from .strategic_intelligence_analyzer import StrategicIntelligenceAnalyzer
|
||||
from .content_distribution_analyzer import ContentDistributionAnalyzer
|
||||
|
||||
__all__ = ['AIRecommendationsService', 'PromptEngineeringService', 'QualityValidationService']
|
||||
__all__ = [
|
||||
'AIRecommendationsService',
|
||||
'QualityValidationService',
|
||||
'PromptEngineeringService',
|
||||
'StrategicIntelligenceAnalyzer',
|
||||
'ContentDistributionAnalyzer'
|
||||
]
|
||||
@@ -14,6 +14,7 @@ from models.enhanced_strategy_models import EnhancedContentStrategy, EnhancedAIA
|
||||
# Import modular components
|
||||
from .prompt_engineering import PromptEngineeringService
|
||||
from .quality_validation import QualityValidationService
|
||||
from .strategic_intelligence_analyzer import StrategicIntelligenceAnalyzer
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -23,6 +24,7 @@ class AIRecommendationsService:
|
||||
def __init__(self):
|
||||
self.prompt_engineering_service = PromptEngineeringService()
|
||||
self.quality_validation_service = QualityValidationService()
|
||||
self.strategic_intelligence_analyzer = StrategicIntelligenceAnalyzer()
|
||||
|
||||
# Analysis types for comprehensive recommendations
|
||||
self.analysis_types = [
|
||||
@@ -33,62 +35,82 @@ class AIRecommendationsService:
|
||||
'content_calendar_optimization'
|
||||
]
|
||||
|
||||
async def generate_comprehensive_recommendations(self, strategy: EnhancedContentStrategy, db: Session) -> None:
|
||||
"""Generate comprehensive AI recommendations using 5 specialized prompts."""
|
||||
async def _call_ai_service(self, prompt: str, analysis_type: str) -> Dict[str, Any]:
|
||||
"""Call AI service to generate recommendations."""
|
||||
try:
|
||||
logger.info(f"Generating comprehensive AI recommendations for strategy: {strategy.id}")
|
||||
# Import AI service manager
|
||||
from services.ai_service_manager import AIServiceManager
|
||||
|
||||
start_time = datetime.utcnow()
|
||||
# Initialize AI service
|
||||
ai_service = AIServiceManager()
|
||||
|
||||
# Generate recommendations for each analysis type
|
||||
ai_recommendations = {}
|
||||
# Generate AI response based on analysis type
|
||||
if analysis_type == "strategic_intelligence":
|
||||
response = await ai_service.generate_strategic_intelligence({
|
||||
"prompt": prompt,
|
||||
"analysis_type": analysis_type
|
||||
})
|
||||
elif analysis_type == "content_recommendations":
|
||||
response = await ai_service.generate_content_recommendations({
|
||||
"prompt": prompt,
|
||||
"analysis_type": analysis_type
|
||||
})
|
||||
elif analysis_type == "market_analysis":
|
||||
response = await ai_service.generate_market_position_analysis({
|
||||
"prompt": prompt,
|
||||
"analysis_type": analysis_type
|
||||
})
|
||||
else:
|
||||
# Default to strategic intelligence
|
||||
response = await ai_service.generate_strategic_intelligence({
|
||||
"prompt": prompt,
|
||||
"analysis_type": analysis_type
|
||||
})
|
||||
|
||||
for analysis_type in self.analysis_types:
|
||||
try:
|
||||
recommendations = await self._generate_specialized_recommendations(
|
||||
strategy, analysis_type, db
|
||||
)
|
||||
ai_recommendations[analysis_type] = recommendations
|
||||
|
||||
# Store individual analysis result
|
||||
analysis_result = EnhancedAIAnalysisResult(
|
||||
user_id=strategy.user_id,
|
||||
strategy_id=strategy.id,
|
||||
analysis_type=analysis_type,
|
||||
comprehensive_insights=recommendations.get('comprehensive_insights'),
|
||||
audience_intelligence=recommendations.get('audience_intelligence'),
|
||||
competitive_intelligence=recommendations.get('competitive_intelligence'),
|
||||
performance_optimization=recommendations.get('performance_optimization'),
|
||||
content_calendar_optimization=recommendations.get('content_calendar_optimization'),
|
||||
onboarding_data_used=strategy.onboarding_data_used,
|
||||
processing_time=(datetime.utcnow() - start_time).total_seconds(),
|
||||
ai_service_status="operational"
|
||||
)
|
||||
|
||||
db.add(analysis_result)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating {analysis_type} recommendations: {str(e)}")
|
||||
# Continue with other analysis types
|
||||
|
||||
db.commit()
|
||||
|
||||
# Update strategy with comprehensive AI analysis
|
||||
strategy.comprehensive_ai_analysis = ai_recommendations
|
||||
strategy.strategic_scores = self.quality_validation_service.calculate_strategic_scores(ai_recommendations)
|
||||
strategy.market_positioning = self.quality_validation_service.extract_market_positioning(ai_recommendations)
|
||||
strategy.competitive_advantages = self.quality_validation_service.extract_competitive_advantages(ai_recommendations)
|
||||
strategy.strategic_risks = self.quality_validation_service.extract_strategic_risks(ai_recommendations)
|
||||
strategy.opportunity_analysis = self.quality_validation_service.extract_opportunity_analysis(ai_recommendations)
|
||||
|
||||
db.commit()
|
||||
|
||||
processing_time = (datetime.utcnow() - start_time).total_seconds()
|
||||
logger.info(f"Comprehensive AI recommendations generated in {processing_time:.2f} seconds")
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating comprehensive AI recommendations: {str(e)}")
|
||||
# Don't raise error, just log it as this is enhancement, not core functionality
|
||||
logger.error(f"Error calling AI service: {str(e)}")
|
||||
raise Exception(f"Failed to generate AI recommendations: {str(e)}")
|
||||
|
||||
def _parse_ai_response(self, ai_response: Dict[str, Any], analysis_type: str) -> Dict[str, Any]:
|
||||
return ai_response # parsing now handled downstream
|
||||
|
||||
def get_output_schema(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"required": ["strategy_brief", "channels", "pillars", "plan_30_60_90", "kpis"],
|
||||
"properties": {
|
||||
"strategy_brief": {"type": "object"},
|
||||
"channels": {"type": "array", "items": {"type": "object"}},
|
||||
"pillars": {"type": "array", "items": {"type": "object"}},
|
||||
"plan_30_60_90": {"type": "object"},
|
||||
"kpis": {"type": "object"},
|
||||
"citations": {"type": "array", "items": {"type": "object"}}
|
||||
}
|
||||
}
|
||||
|
||||
async def generate_comprehensive_ai_recommendations(self, strategy: EnhancedContentStrategy, db: Session) -> None:
|
||||
try:
|
||||
# Build centralized prompts per analysis type
|
||||
prompt = self.prompt_engineering_service.create_specialized_prompt(strategy, "comprehensive_strategy")
|
||||
raw = await self._call_ai_service(prompt, "strategic_intelligence")
|
||||
# Validate against schema
|
||||
schema = self.get_output_schema()
|
||||
self.quality_validation_service.validate_against_schema(raw, schema)
|
||||
# Persist
|
||||
result = EnhancedAIAnalysisResult(
|
||||
strategy_id=strategy.id,
|
||||
analysis_type="comprehensive_strategy",
|
||||
result_json=raw,
|
||||
created_at=datetime.utcnow()
|
||||
)
|
||||
db.add(result)
|
||||
db.commit()
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
logger.error(f"Comprehensive recommendation generation failed: {str(e)}")
|
||||
raise
|
||||
|
||||
async def _generate_specialized_recommendations(self, strategy: EnhancedContentStrategy, analysis_type: str, db: Session) -> Dict[str, Any]:
|
||||
"""Generate specialized recommendations using specific AI prompts."""
|
||||
@@ -109,64 +131,8 @@ class AIRecommendationsService:
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating {analysis_type} recommendations: {str(e)}")
|
||||
return self._get_fallback_recommendations(analysis_type)
|
||||
|
||||
async def _call_ai_service(self, prompt: str, analysis_type: str) -> Dict[str, Any]:
|
||||
"""Call AI service to generate recommendations."""
|
||||
# Placeholder implementation - integrate with actual AI service
|
||||
# For now, return structured mock data
|
||||
return {
|
||||
'analysis_type': analysis_type,
|
||||
'recommendations': f"AI recommendations for {analysis_type}",
|
||||
'insights': f"Key insights for {analysis_type}",
|
||||
'metrics': {'score': 85, 'confidence': 0.9}
|
||||
}
|
||||
|
||||
def _parse_ai_response(self, ai_response: Dict[str, Any], analysis_type: str) -> Dict[str, Any]:
|
||||
"""Parse and structure AI response."""
|
||||
return {
|
||||
'analysis_type': analysis_type,
|
||||
'recommendations': ai_response.get('recommendations', []),
|
||||
'insights': ai_response.get('insights', []),
|
||||
'metrics': ai_response.get('metrics', {}),
|
||||
'confidence_score': ai_response.get('metrics', {}).get('confidence', 0.8)
|
||||
}
|
||||
|
||||
def _get_fallback_recommendations(self, analysis_type: str) -> Dict[str, Any]:
|
||||
"""Get fallback recommendations when AI service fails."""
|
||||
fallback_data = {
|
||||
'comprehensive_strategy': {
|
||||
'recommendations': ['Focus on core content pillars', 'Develop audience personas'],
|
||||
'insights': ['Strategy needs more specific objectives', 'Consider expanding content mix'],
|
||||
'metrics': {'score': 70, 'confidence': 0.6}
|
||||
},
|
||||
'audience_intelligence': {
|
||||
'recommendations': ['Conduct audience research', 'Analyze content preferences'],
|
||||
'insights': ['Limited audience data available', 'Need more engagement metrics'],
|
||||
'metrics': {'score': 65, 'confidence': 0.5}
|
||||
},
|
||||
'competitive_intelligence': {
|
||||
'recommendations': ['Analyze competitor content', 'Identify market gaps'],
|
||||
'insights': ['Competitive analysis needed', 'Market positioning unclear'],
|
||||
'metrics': {'score': 60, 'confidence': 0.4}
|
||||
},
|
||||
'performance_optimization': {
|
||||
'recommendations': ['Set up analytics tracking', 'Implement A/B testing'],
|
||||
'insights': ['Performance data limited', 'Need baseline metrics'],
|
||||
'metrics': {'score': 55, 'confidence': 0.3}
|
||||
},
|
||||
'content_calendar_optimization': {
|
||||
'recommendations': ['Create publishing schedule', 'Optimize content mix'],
|
||||
'insights': ['Calendar optimization needed', 'Frequency planning required'],
|
||||
'metrics': {'score': 50, 'confidence': 0.2}
|
||||
}
|
||||
}
|
||||
|
||||
return fallback_data.get(analysis_type, {
|
||||
'recommendations': ['General strategy improvement needed'],
|
||||
'insights': ['Limited data available for analysis'],
|
||||
'metrics': {'score': 50, 'confidence': 0.3}
|
||||
})
|
||||
# Raise exception instead of returning fallback data
|
||||
raise Exception(f"Failed to generate {analysis_type} recommendations: {str(e)}")
|
||||
|
||||
async def get_latest_ai_analysis(self, strategy_id: int, db: Session) -> Optional[Dict[str, Any]]:
|
||||
"""Get latest AI analysis for a strategy."""
|
||||
|
||||
@@ -0,0 +1,261 @@
|
||||
"""
|
||||
Content Distribution Analyzer
|
||||
Handles content distribution strategy analysis and optimization.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Dict, List, Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class ContentDistributionAnalyzer:
|
||||
"""Analyzes and generates content distribution strategies."""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def analyze_content_distribution(self, preferred_formats: list, content_frequency: str, industry: str, team_size: int) -> Dict[str, Any]:
|
||||
"""Analyze content distribution strategy for personalized insights."""
|
||||
distribution_channels = []
|
||||
|
||||
# Social media platforms
|
||||
if 'video' in preferred_formats:
|
||||
distribution_channels.extend([
|
||||
{
|
||||
"platform": "TikTok",
|
||||
"priority": "High",
|
||||
"content_type": "Short-form video",
|
||||
"posting_frequency": "Daily",
|
||||
"best_practices": ["Use trending sounds", "Create educational content", "Engage with comments"],
|
||||
"free_tools": ["TikTok Creator Studio", "CapCut"],
|
||||
"expected_reach": "10K-100K views per video"
|
||||
},
|
||||
{
|
||||
"platform": "Instagram Reels",
|
||||
"priority": "High",
|
||||
"content_type": "Short-form video",
|
||||
"posting_frequency": "Daily",
|
||||
"best_practices": ["Use trending hashtags", "Create behind-the-scenes content", "Cross-promote"],
|
||||
"free_tools": ["Instagram Insights", "Canva"],
|
||||
"expected_reach": "5K-50K views per reel"
|
||||
}
|
||||
])
|
||||
|
||||
# Blog and written content
|
||||
if 'blog' in preferred_formats or 'article' in preferred_formats:
|
||||
distribution_channels.append({
|
||||
"platform": "Personal Blog/Website",
|
||||
"priority": "High",
|
||||
"content_type": "Long-form articles",
|
||||
"posting_frequency": "Weekly",
|
||||
"best_practices": ["SEO optimization", "Email list building", "Social sharing"],
|
||||
"free_tools": ["WordPress.com", "Medium", "Substack"],
|
||||
"expected_reach": "1K-10K monthly readers"
|
||||
})
|
||||
|
||||
# Podcast distribution
|
||||
distribution_channels.append({
|
||||
"platform": "Podcast",
|
||||
"priority": "Medium",
|
||||
"content_type": "Audio content",
|
||||
"posting_frequency": "Weekly",
|
||||
"best_practices": ["Consistent publishing", "Guest interviews", "Cross-promotion"],
|
||||
"free_tools": ["Anchor", "Spotify for Podcasters", "Riverside"],
|
||||
"expected_reach": "500-5K monthly listeners"
|
||||
})
|
||||
|
||||
# Email newsletter
|
||||
distribution_channels.append({
|
||||
"platform": "Email Newsletter",
|
||||
"priority": "High",
|
||||
"content_type": "Personal updates and insights",
|
||||
"posting_frequency": "Weekly",
|
||||
"best_practices": ["Personal storytelling", "Exclusive content", "Call-to-action"],
|
||||
"free_tools": ["Mailchimp", "ConvertKit", "Substack"],
|
||||
"expected_reach": "100-1K subscribers"
|
||||
})
|
||||
|
||||
return {
|
||||
"distribution_channels": distribution_channels,
|
||||
"optimal_posting_schedule": self._generate_posting_schedule(content_frequency, team_size),
|
||||
"cross_promotion_strategy": self._generate_cross_promotion_strategy(preferred_formats),
|
||||
"content_repurposing_plan": self._generate_repurposing_plan(preferred_formats),
|
||||
"audience_growth_tactics": [
|
||||
"Collaborate with other creators in your niche",
|
||||
"Participate in industry hashtags and challenges",
|
||||
"Create shareable content that provides value",
|
||||
"Engage with your audience in comments and DMs",
|
||||
"Use trending topics to create relevant content"
|
||||
]
|
||||
}
|
||||
|
||||
def _generate_posting_schedule(self, content_frequency: str, team_size: int) -> Dict[str, Any]:
|
||||
"""Generate optimal posting schedule for personalized insights."""
|
||||
if team_size == 1:
|
||||
return {
|
||||
"monday": "Educational content or industry insights",
|
||||
"tuesday": "Behind-the-scenes or personal story",
|
||||
"wednesday": "Problem-solving content or tips",
|
||||
"thursday": "Community engagement or Q&A",
|
||||
"friday": "Weekend inspiration or fun content",
|
||||
"saturday": "Repurpose best-performing content",
|
||||
"sunday": "Planning and content creation"
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"monday": "Weekly theme announcement",
|
||||
"tuesday": "Educational content",
|
||||
"wednesday": "Interactive content",
|
||||
"thursday": "Behind-the-scenes",
|
||||
"friday": "Community highlights",
|
||||
"saturday": "Repurposed content",
|
||||
"sunday": "Planning and creation"
|
||||
}
|
||||
|
||||
def _generate_cross_promotion_strategy(self, preferred_formats: list) -> List[str]:
|
||||
"""Generate cross-promotion strategy for personalized insights."""
|
||||
strategies = []
|
||||
|
||||
if 'video' in preferred_formats:
|
||||
strategies.extend([
|
||||
"Share video snippets on Instagram Stories",
|
||||
"Create YouTube Shorts from longer videos",
|
||||
"Cross-post video content to TikTok and Instagram Reels"
|
||||
])
|
||||
|
||||
if 'blog' in preferred_formats or 'article' in preferred_formats:
|
||||
strategies.extend([
|
||||
"Share blog excerpts on LinkedIn",
|
||||
"Create Twitter threads from blog posts",
|
||||
"Turn blog posts into video content"
|
||||
])
|
||||
|
||||
strategies.extend([
|
||||
"Use consistent hashtags across platforms",
|
||||
"Cross-promote content on different platforms",
|
||||
"Create platform-specific content variations",
|
||||
"Share behind-the-scenes content across all platforms"
|
||||
])
|
||||
|
||||
return strategies
|
||||
|
||||
def _generate_repurposing_plan(self, preferred_formats: list) -> Dict[str, List[str]]:
|
||||
"""Generate content repurposing plan for personalized insights."""
|
||||
repurposing_plan = {}
|
||||
|
||||
if 'video' in preferred_formats:
|
||||
repurposing_plan['video_content'] = [
|
||||
"Extract key quotes for social media posts",
|
||||
"Create blog posts from video transcripts",
|
||||
"Turn video clips into GIFs for social media",
|
||||
"Create podcast episodes from video content",
|
||||
"Extract audio for podcast distribution"
|
||||
]
|
||||
|
||||
if 'blog' in preferred_formats or 'article' in preferred_formats:
|
||||
repurposing_plan['written_content'] = [
|
||||
"Create social media posts from blog highlights",
|
||||
"Turn blog posts into video scripts",
|
||||
"Extract quotes for Twitter threads",
|
||||
"Create infographics from blog data",
|
||||
"Turn blog series into email courses"
|
||||
]
|
||||
|
||||
repurposing_plan['general'] = [
|
||||
"Repurpose top-performing content across platforms",
|
||||
"Create different formats for different audiences",
|
||||
"Update and republish evergreen content",
|
||||
"Combine multiple pieces into comprehensive guides",
|
||||
"Extract tips and insights for social media"
|
||||
]
|
||||
|
||||
return repurposing_plan
|
||||
|
||||
def analyze_performance_optimization(self, target_metrics: Dict, content_preferences: Dict, preferred_formats: list, team_size: int) -> Dict[str, Any]:
|
||||
"""Analyze content performance optimization for personalized insights."""
|
||||
optimization_strategies = []
|
||||
|
||||
# Content quality optimization
|
||||
optimization_strategies.append({
|
||||
"strategy": "Content Quality Optimization",
|
||||
"focus_area": "Engagement and retention",
|
||||
"tactics": [
|
||||
"Create content that solves specific problems",
|
||||
"Use storytelling to make content memorable",
|
||||
"Include clear calls-to-action in every piece",
|
||||
"Optimize content length for each platform",
|
||||
"Use data to identify top-performing content types"
|
||||
],
|
||||
"free_tools": ["Google Analytics", "Platform Insights", "A/B Testing"],
|
||||
"expected_improvement": "50% increase in engagement"
|
||||
})
|
||||
|
||||
# SEO optimization
|
||||
optimization_strategies.append({
|
||||
"strategy": "SEO and Discoverability",
|
||||
"focus_area": "Organic reach and traffic",
|
||||
"tactics": [
|
||||
"Research and target relevant keywords",
|
||||
"Optimize titles and descriptions",
|
||||
"Create evergreen content that ranks",
|
||||
"Build backlinks through guest posting",
|
||||
"Improve page load speed and mobile experience"
|
||||
],
|
||||
"free_tools": ["Google Keyword Planner", "Google Search Console", "Yoast SEO"],
|
||||
"expected_improvement": "100% increase in organic traffic"
|
||||
})
|
||||
|
||||
# Audience engagement optimization
|
||||
optimization_strategies.append({
|
||||
"strategy": "Audience Engagement",
|
||||
"focus_area": "Community building and loyalty",
|
||||
"tactics": [
|
||||
"Respond to every comment within 24 hours",
|
||||
"Create interactive content (polls, questions)",
|
||||
"Host live sessions and Q&As",
|
||||
"Share behind-the-scenes content",
|
||||
"Create exclusive content for engaged followers"
|
||||
],
|
||||
"free_tools": ["Instagram Stories", "Twitter Spaces", "YouTube Live"],
|
||||
"expected_improvement": "75% increase in community engagement"
|
||||
})
|
||||
|
||||
# Content distribution optimization
|
||||
optimization_strategies.append({
|
||||
"strategy": "Distribution Optimization",
|
||||
"focus_area": "Reach and visibility",
|
||||
"tactics": [
|
||||
"Post at optimal times for your audience",
|
||||
"Use platform-specific features (Stories, Reels, etc.)",
|
||||
"Cross-promote content across platforms",
|
||||
"Collaborate with other creators",
|
||||
"Participate in trending conversations"
|
||||
],
|
||||
"free_tools": ["Later", "Buffer", "Hootsuite"],
|
||||
"expected_improvement": "200% increase in reach"
|
||||
})
|
||||
|
||||
return {
|
||||
"optimization_strategies": optimization_strategies,
|
||||
"performance_tracking_metrics": [
|
||||
"Engagement rate (likes, comments, shares)",
|
||||
"Reach and impressions",
|
||||
"Click-through rates",
|
||||
"Time spent on content",
|
||||
"Follower growth rate",
|
||||
"Conversion rates (email signups, sales)"
|
||||
],
|
||||
"free_analytics_tools": [
|
||||
"Google Analytics (website traffic)",
|
||||
"Platform Insights (social media)",
|
||||
"Google Search Console (SEO)",
|
||||
"Email marketing analytics",
|
||||
"YouTube Analytics (video performance)"
|
||||
],
|
||||
"optimization_timeline": {
|
||||
"immediate": "Set up tracking and identify baseline metrics",
|
||||
"week_1": "Implement one optimization strategy",
|
||||
"month_1": "Analyze results and adjust strategy",
|
||||
"month_3": "Scale successful tactics and experiment with new ones"
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,45 @@ class QualityValidationService:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def validate_against_schema(self, data: Dict[str, Any], schema: Dict[str, Any]) -> None:
|
||||
"""Validate data against a minimal JSON-like schema definition.
|
||||
Raises ValueError on failure.
|
||||
Schema format example:
|
||||
{"type": "object", "required": ["strategy_brief", "channels"], "properties": {"strategy_brief": {"type": "object"}, "channels": {"type": "array"}}}
|
||||
"""
|
||||
def _check(node, sch, path="$"):
|
||||
t = sch.get("type")
|
||||
if t == "object":
|
||||
if not isinstance(node, dict):
|
||||
raise ValueError(f"Schema error at {path}: expected object")
|
||||
for req in sch.get("required", []):
|
||||
if req not in node or node[req] in (None, ""):
|
||||
raise ValueError(f"Schema error at {path}.{req}: required field missing")
|
||||
for key, sub in sch.get("properties", {}).items():
|
||||
if key in node:
|
||||
_check(node[key], sub, f"{path}.{key}")
|
||||
elif t == "array":
|
||||
if not isinstance(node, list):
|
||||
raise ValueError(f"Schema error at {path}: expected array")
|
||||
item_s = sch.get("items")
|
||||
if item_s:
|
||||
for i, item in enumerate(node):
|
||||
_check(item, item_s, f"{path}[{i}]")
|
||||
elif t == "string":
|
||||
if not isinstance(node, str) or not node.strip():
|
||||
raise ValueError(f"Schema error at {path}: expected non-empty string")
|
||||
elif t == "number":
|
||||
if not isinstance(node, (int, float)):
|
||||
raise ValueError(f"Schema error at {path}: expected number")
|
||||
elif t == "boolean":
|
||||
if not isinstance(node, bool):
|
||||
raise ValueError(f"Schema error at {path}: expected boolean")
|
||||
elif t == "any":
|
||||
return
|
||||
else:
|
||||
return
|
||||
_check(data, schema)
|
||||
|
||||
def calculate_strategic_scores(self, ai_recommendations: Dict[str, Any]) -> Dict[str, float]:
|
||||
"""Calculate strategic performance scores from AI recommendations."""
|
||||
scores = {
|
||||
|
||||
@@ -0,0 +1,408 @@
|
||||
"""
|
||||
Strategic Intelligence Analyzer
|
||||
Handles comprehensive strategic intelligence analysis and generation.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Dict, List, Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class StrategicIntelligenceAnalyzer:
|
||||
"""Analyzes and generates comprehensive strategic intelligence."""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def analyze_market_positioning(self, business_objectives: Dict, industry: str, content_preferences: Dict, team_size: int) -> Dict[str, Any]:
|
||||
"""Analyze market positioning for personalized insights."""
|
||||
# Calculate positioning score based on multiple factors
|
||||
score = 75 # Base score
|
||||
|
||||
# Adjust based on business objectives
|
||||
if business_objectives.get('brand_awareness'):
|
||||
score += 10
|
||||
if business_objectives.get('lead_generation'):
|
||||
score += 8
|
||||
if business_objectives.get('thought_leadership'):
|
||||
score += 12
|
||||
|
||||
# Adjust based on team size (solopreneurs get bonus for agility)
|
||||
if team_size <= 3:
|
||||
score += 8 # Solopreneurs are more agile
|
||||
elif team_size <= 10:
|
||||
score += 3
|
||||
|
||||
# Adjust based on content preferences
|
||||
if content_preferences.get('video_content'):
|
||||
score += 8
|
||||
if content_preferences.get('interactive_content'):
|
||||
score += 6
|
||||
|
||||
score = min(100, max(0, score))
|
||||
|
||||
return {
|
||||
"score": score,
|
||||
"strengths": [
|
||||
"Agile content production and quick pivots",
|
||||
"Direct connection with audience",
|
||||
"Authentic personal brand voice",
|
||||
"Cost-effective content creation",
|
||||
"Rapid experimentation capabilities"
|
||||
],
|
||||
"weaknesses": [
|
||||
"Limited content production capacity",
|
||||
"Time constraints for content creation",
|
||||
"Limited access to professional tools",
|
||||
"Need for content automation",
|
||||
"Limited reach without paid promotion"
|
||||
],
|
||||
"opportunities": [
|
||||
"Leverage personal brand authenticity",
|
||||
"Focus on niche content areas",
|
||||
"Build community-driven content",
|
||||
"Utilize free content creation tools",
|
||||
"Partner with other creators"
|
||||
],
|
||||
"threats": [
|
||||
"Content saturation in market",
|
||||
"Algorithm changes affecting reach",
|
||||
"Time constraints limiting output",
|
||||
"Competition from larger brands",
|
||||
"Platform dependency risks"
|
||||
]
|
||||
}
|
||||
|
||||
def identify_competitive_advantages(self, business_objectives: Dict, content_preferences: Dict, preferred_formats: list, team_size: int) -> List[Dict[str, Any]]:
|
||||
"""Identify competitive advantages for personalized insights."""
|
||||
try:
|
||||
advantages = []
|
||||
|
||||
# Analyze business objectives for competitive advantages
|
||||
if business_objectives.get('lead_generation'):
|
||||
advantages.append({
|
||||
"advantage": "Direct lead generation capabilities",
|
||||
"description": "Ability to create content that directly converts visitors to leads",
|
||||
"impact": "High",
|
||||
"implementation": "Focus on lead magnets and conversion-optimized content",
|
||||
"roi_potential": "300% return on investment",
|
||||
"differentiation": "Personal connection vs corporate approach"
|
||||
})
|
||||
|
||||
if business_objectives.get('brand_awareness'):
|
||||
advantages.append({
|
||||
"advantage": "Authentic personal brand voice",
|
||||
"description": "Unique personal perspective that builds trust and connection",
|
||||
"impact": "High",
|
||||
"implementation": "Share personal stories and behind-the-scenes content",
|
||||
"roi_potential": "250% return on investment",
|
||||
"differentiation": "Authenticity vs polished corporate messaging"
|
||||
})
|
||||
|
||||
if business_objectives.get('thought_leadership'):
|
||||
advantages.append({
|
||||
"advantage": "Niche expertise and authority",
|
||||
"description": "Deep knowledge in specific areas that positions you as the go-to expert",
|
||||
"impact": "Very High",
|
||||
"implementation": "Create comprehensive, educational content in your niche",
|
||||
"roi_potential": "400% return on investment",
|
||||
"differentiation": "Specialized expertise vs generalist approach"
|
||||
})
|
||||
|
||||
# Analyze content preferences for advantages
|
||||
if content_preferences.get('video_content'):
|
||||
advantages.append({
|
||||
"advantage": "Video content expertise",
|
||||
"description": "Ability to create engaging video content that drives higher engagement",
|
||||
"impact": "High",
|
||||
"implementation": "Focus on short-form video platforms (TikTok, Instagram Reels)",
|
||||
"roi_potential": "400% return on investment",
|
||||
"differentiation": "Visual storytelling vs text-only content"
|
||||
})
|
||||
|
||||
if content_preferences.get('interactive_content'):
|
||||
advantages.append({
|
||||
"advantage": "Interactive content capabilities",
|
||||
"description": "Ability to create content that engages and involves the audience",
|
||||
"impact": "Medium",
|
||||
"implementation": "Use polls, questions, and interactive elements",
|
||||
"roi_potential": "200% return on investment",
|
||||
"differentiation": "Two-way communication vs one-way broadcasting"
|
||||
})
|
||||
|
||||
# Analyze team size advantages
|
||||
if team_size == 1:
|
||||
advantages.append({
|
||||
"advantage": "Agility and quick pivots",
|
||||
"description": "Ability to respond quickly to trends and opportunities",
|
||||
"impact": "High",
|
||||
"implementation": "Stay current with trends and adapt content quickly",
|
||||
"roi_potential": "150% return on investment",
|
||||
"differentiation": "Speed vs corporate approval processes"
|
||||
})
|
||||
|
||||
# Analyze preferred formats for advantages
|
||||
if 'video' in preferred_formats:
|
||||
advantages.append({
|
||||
"advantage": "Multi-platform video presence",
|
||||
"description": "Ability to create video content for multiple platforms",
|
||||
"impact": "High",
|
||||
"implementation": "Repurpose video content across TikTok, Instagram, YouTube",
|
||||
"roi_potential": "350% return on investment",
|
||||
"differentiation": "Visual engagement vs static content"
|
||||
})
|
||||
|
||||
if 'blog' in preferred_formats or 'article' in preferred_formats:
|
||||
advantages.append({
|
||||
"advantage": "SEO-optimized content creation",
|
||||
"description": "Ability to create content that ranks well in search engines",
|
||||
"impact": "High",
|
||||
"implementation": "Focus on keyword research and SEO best practices",
|
||||
"roi_potential": "300% return on investment",
|
||||
"differentiation": "Organic reach vs paid advertising"
|
||||
})
|
||||
|
||||
# If no specific advantages found, provide general ones
|
||||
if not advantages:
|
||||
advantages = [
|
||||
{
|
||||
"advantage": "Personal connection and authenticity",
|
||||
"description": "Ability to build genuine relationships with your audience",
|
||||
"impact": "High",
|
||||
"implementation": "Share personal stories and be transparent",
|
||||
"roi_potential": "250% return on investment",
|
||||
"differentiation": "Authentic voice vs corporate messaging"
|
||||
},
|
||||
{
|
||||
"advantage": "Niche expertise",
|
||||
"description": "Deep knowledge in your specific area of expertise",
|
||||
"impact": "High",
|
||||
"implementation": "Focus on your unique knowledge and experience",
|
||||
"roi_potential": "300% return on investment",
|
||||
"differentiation": "Specialized knowledge vs generalist approach"
|
||||
}
|
||||
]
|
||||
|
||||
return advantages
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating competitive advantages: {str(e)}")
|
||||
raise Exception(f"Failed to generate competitive advantages: {str(e)}")
|
||||
|
||||
def assess_strategic_risks(self, industry: str, market_gaps: list, team_size: int, content_frequency: str) -> List[Dict[str, Any]]:
|
||||
"""Assess strategic risks for personalized insights."""
|
||||
risks = []
|
||||
|
||||
# Content saturation risk
|
||||
risks.append({
|
||||
"risk": "Content saturation in market",
|
||||
"probability": "Medium",
|
||||
"impact": "High",
|
||||
"mitigation": "Focus on unique personal perspective and niche topics",
|
||||
"monitoring": "Track content performance vs competitors, monitor engagement rates",
|
||||
"timeline": "Ongoing",
|
||||
"resources_needed": "Free competitive analysis tools"
|
||||
})
|
||||
|
||||
# Algorithm changes risk
|
||||
risks.append({
|
||||
"risk": "Algorithm changes affecting reach",
|
||||
"probability": "High",
|
||||
"impact": "Medium",
|
||||
"mitigation": "Diversify content formats and platforms, build owned audience",
|
||||
"monitoring": "Monitor platform algorithm updates, track reach changes",
|
||||
"timeline": "Ongoing",
|
||||
"resources_needed": "Free multi-platform strategy"
|
||||
})
|
||||
|
||||
# Time constraints risk
|
||||
if team_size == 1:
|
||||
risks.append({
|
||||
"risk": "Time constraints limiting content output",
|
||||
"probability": "High",
|
||||
"impact": "High",
|
||||
"mitigation": "Implement content batching, repurposing, and automation",
|
||||
"monitoring": "Track content creation time, monitor output consistency",
|
||||
"timeline": "1-2 months",
|
||||
"resources_needed": "Free content planning tools"
|
||||
})
|
||||
|
||||
# Platform dependency risk
|
||||
risks.append({
|
||||
"risk": "Platform dependency risks",
|
||||
"probability": "Medium",
|
||||
"impact": "Medium",
|
||||
"mitigation": "Build owned audience through email lists and personal websites",
|
||||
"monitoring": "Track platform-specific vs owned audience growth",
|
||||
"timeline": "3-6 months",
|
||||
"resources_needed": "Free email marketing tools"
|
||||
})
|
||||
|
||||
return risks
|
||||
|
||||
def analyze_opportunities(self, business_objectives: Dict, market_gaps: list, preferred_formats: list) -> List[Dict[str, Any]]:
|
||||
"""Analyze opportunities for personalized insights."""
|
||||
opportunities = []
|
||||
|
||||
# Video content opportunity
|
||||
if 'video' not in preferred_formats:
|
||||
opportunities.append({
|
||||
"opportunity": "Video content expansion",
|
||||
"potential_impact": "High",
|
||||
"implementation_ease": "Medium",
|
||||
"timeline": "1-2 months",
|
||||
"resource_requirements": "Free video tools (TikTok, Instagram Reels, YouTube Shorts)",
|
||||
"roi_potential": "400% return on investment",
|
||||
"description": "Video content generates 4x more engagement than text-only content"
|
||||
})
|
||||
|
||||
# Podcast opportunity
|
||||
opportunities.append({
|
||||
"opportunity": "Start a podcast",
|
||||
"potential_impact": "High",
|
||||
"implementation_ease": "Medium",
|
||||
"timeline": "2-3 months",
|
||||
"resource_requirements": "Free podcast hosting platforms",
|
||||
"roi_potential": "500% return on investment",
|
||||
"description": "Podcasts build deep audience relationships and establish thought leadership"
|
||||
})
|
||||
|
||||
# Newsletter opportunity
|
||||
opportunities.append({
|
||||
"opportunity": "Email newsletter",
|
||||
"potential_impact": "High",
|
||||
"implementation_ease": "High",
|
||||
"timeline": "1 month",
|
||||
"resource_requirements": "Free email marketing tools",
|
||||
"roi_potential": "600% return on investment",
|
||||
"description": "Direct email communication builds owned audience and drives conversions"
|
||||
})
|
||||
|
||||
# Market gap opportunities
|
||||
for gap in market_gaps[:3]: # Top 3 gaps
|
||||
opportunities.append({
|
||||
"opportunity": f"Address market gap: {gap}",
|
||||
"potential_impact": "High",
|
||||
"implementation_ease": "Medium",
|
||||
"timeline": "2-4 months",
|
||||
"resource_requirements": "Free content research and creation",
|
||||
"roi_potential": "300% return on investment",
|
||||
"description": f"Filling the {gap} gap positions you as the go-to expert"
|
||||
})
|
||||
|
||||
return opportunities
|
||||
|
||||
def calculate_performance_metrics(self, target_metrics: Dict, team_size: int) -> Dict[str, Any]:
|
||||
"""Calculate performance metrics for personalized insights."""
|
||||
# Base metrics
|
||||
content_quality_score = 8.5
|
||||
engagement_rate = 4.2
|
||||
conversion_rate = 2.8
|
||||
roi_per_content = 320
|
||||
brand_awareness_score = 7.8
|
||||
|
||||
# Adjust based on team size (solopreneurs get bonus for authenticity)
|
||||
if team_size == 1:
|
||||
content_quality_score += 0.5 # Authenticity bonus
|
||||
engagement_rate += 0.3 # Personal connection
|
||||
elif team_size <= 3:
|
||||
content_quality_score += 0.2
|
||||
engagement_rate += 0.1
|
||||
|
||||
return {
|
||||
"content_quality_score": round(content_quality_score, 1),
|
||||
"engagement_rate": round(engagement_rate, 1),
|
||||
"conversion_rate": round(conversion_rate, 1),
|
||||
"roi_per_content": round(roi_per_content, 0),
|
||||
"brand_awareness_score": round(brand_awareness_score, 1),
|
||||
"content_efficiency": round(roi_per_content / 100 * 100, 1), # Normalized for solopreneurs
|
||||
"personal_brand_strength": round(brand_awareness_score * 1.2, 1) # Personal brand metric
|
||||
}
|
||||
|
||||
def generate_solopreneur_recommendations(self, business_objectives: Dict, team_size: int, preferred_formats: list, industry: str) -> List[Dict[str, Any]]:
|
||||
"""Generate personalized recommendations based on user data."""
|
||||
recommendations = []
|
||||
|
||||
# High priority recommendations
|
||||
if 'video' not in preferred_formats:
|
||||
recommendations.append({
|
||||
"priority": "High",
|
||||
"action": "Start creating short-form video content",
|
||||
"impact": "Increase engagement by 400% and reach by 300%",
|
||||
"timeline": "1 month",
|
||||
"resources_needed": "Free - use TikTok, Instagram Reels, YouTube Shorts",
|
||||
"roi_estimate": "400% return on investment",
|
||||
"implementation_steps": [
|
||||
"Download TikTok and Instagram apps",
|
||||
"Study trending content in your niche",
|
||||
"Create 3-5 short videos per week",
|
||||
"Engage with comments and build community"
|
||||
]
|
||||
})
|
||||
|
||||
# Email list building
|
||||
recommendations.append({
|
||||
"priority": "High",
|
||||
"action": "Build an email list",
|
||||
"impact": "Create owned audience, increase conversions by 200%",
|
||||
"timeline": "2 months",
|
||||
"resources_needed": "Free - use Mailchimp or ConvertKit free tier",
|
||||
"roi_estimate": "600% return on investment",
|
||||
"implementation_steps": [
|
||||
"Sign up for free email marketing tool",
|
||||
"Create lead magnet (free guide, checklist)",
|
||||
"Add signup forms to your content",
|
||||
"Send weekly valuable emails"
|
||||
]
|
||||
})
|
||||
|
||||
# Content batching
|
||||
if team_size == 1:
|
||||
recommendations.append({
|
||||
"priority": "High",
|
||||
"action": "Implement content batching",
|
||||
"impact": "Save 10 hours per week, increase output by 300%",
|
||||
"timeline": "2 weeks",
|
||||
"resources_needed": "Free - use Google Calendar and Notion",
|
||||
"roi_estimate": "300% return on investment",
|
||||
"implementation_steps": [
|
||||
"Block 4-hour content creation sessions",
|
||||
"Create content themes for each month",
|
||||
"Batch similar content types together",
|
||||
"Schedule content in advance"
|
||||
]
|
||||
})
|
||||
|
||||
# Medium priority recommendations
|
||||
recommendations.append({
|
||||
"priority": "Medium",
|
||||
"action": "Optimize for search engines",
|
||||
"impact": "Increase organic traffic by 200%",
|
||||
"timeline": "2 months",
|
||||
"resources_needed": "Free - use Google Keyword Planner",
|
||||
"roi_estimate": "200% return on investment",
|
||||
"implementation_steps": [
|
||||
"Research keywords in your niche",
|
||||
"Optimize existing content for target keywords",
|
||||
"Create SEO-optimized content calendar",
|
||||
"Monitor search rankings"
|
||||
]
|
||||
})
|
||||
|
||||
# Community building
|
||||
recommendations.append({
|
||||
"priority": "Medium",
|
||||
"action": "Build community engagement",
|
||||
"impact": "Increase loyalty and word-of-mouth by 150%",
|
||||
"timeline": "3 months",
|
||||
"resources_needed": "Free - use existing social platforms",
|
||||
"roi_estimate": "150% return on investment",
|
||||
"implementation_steps": [
|
||||
"Respond to every comment and message",
|
||||
"Create community challenges or contests",
|
||||
"Host live Q&A sessions",
|
||||
"Collaborate with other creators"
|
||||
]
|
||||
})
|
||||
|
||||
return recommendations
|
||||
Reference in New Issue
Block a user