Automate Competitive Analysis 🚀

# Production: Real-Time Monitoring (Continuous with alerts)
from langgraph.graph import Graph, END
from typing import TypedDict, List, Dict, Any
import asyncio
import aiohttp
from bs4 import BeautifulSoup
import openai
from datetime import datetime, timedelta
import hashlib
import json
import redis
from perplexity import Perplexity

class MonitoringState(TypedDict):
    competitors: List[Dict[str, Any]]
    scraped_data: List[Dict[str, Any]]
    previous_data: Dict[str, Dict[str, Any]]
    changes: List[Dict[str, Any]]
    news_mentions: List[Dict[str, Any]]
    analysis: Dict[str, Any]
    alerts_sent: bool

class CompetitiveMonitoringSystem:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.perplexity = Perplexity(api_key=os.getenv('PERPLEXITY_API_KEY'))
        
    async def scrape_node(self, state: MonitoringState) -> MonitoringState:
        """Scrape competitor pages concurrently"""
        async with aiohttp.ClientSession() as session:
            tasks = []
            for comp in state['competitors']:
                tasks.append(self._scrape_competitor(session, comp))
            
            state['scraped_data'] = await asyncio.gather(*tasks)
        
        return state
    
    async def _scrape_competitor(self, session: aiohttp.ClientSession, 
                                 competitor: Dict[str, Any]) -> Dict[str, Any]:
        """Scrape single competitor with error handling"""
        try:
            async with session.get(
                competitor['url'],
                headers={'User-Agent': 'Mozilla/5.0'},
                timeout=aiohttp.ClientTimeout(total=15)
            ) as response:
                html = await response.text()
                soup = BeautifulSoup(html, 'html.parser')
                
                # Remove noise
                for tag in soup(['script', 'style', 'nav', 'footer']):
                    tag.decompose()
                
                content = soup.get_text(separator=' ', strip=True)
                
                # Generate content hash
                content_hash = hashlib.sha256(
                    content.encode('utf-8')
                ).hexdigest()
                
                return {
                    'name': competitor['name'],
                    'url': competitor['url'],
                    'content': content[:10000],  # Limit size
                    'content_hash': content_hash,
                    'scraped_at': datetime.now().isoformat()
                }
        except Exception as e:
            print(f"Error scraping {competitor['name']}: {e}")
            return {
                'name': competitor['name'],
                'error': str(e)
            }
    
    def extract_node(self, state: MonitoringState) -> MonitoringState:
        """Extract structured data using GPT-4"""
        for data in state['scraped_data']:
            if 'error' in data:
                continue
            
            extraction_prompt = f"""Extract competitive intelligence from this page.
Competitor: {data['name']}

Content:
{data['content']}

Extract as JSON:
{{
  "pricing": {{"rates": {{}}, "tiers": [], "discounts": ""}},
  "features": ["feature1", "feature2"],
  "target_market": "description",
  "unique_value_props": ["prop1", "prop2"],
  "recent_updates": ["update1"]
}}"""

            response = openai.chat.completions.create(
                model="gpt-4-turbo-preview",
                messages=[{"role": "user", "content": extraction_prompt}],
                temperature=0.3,
                response_format={"type": "json_object"}
            )
            
            extracted = json.loads(response.choices[0].message.content)
            data['extracted'] = extracted
            data['extraction_hash'] = hashlib.sha256(
                json.dumps(extracted, sort_keys=True).encode()
            ).hexdigest()
        
        return state
    
    def detect_changes_node(self, state: MonitoringState) -> MonitoringState:
        """Compare with previous data to detect changes"""
        changes = []
        
        for data in state['scraped_data']:
            if 'error' in data or 'extracted' not in data:
                continue
            
            # Get previous data from Redis
            prev_key = f"competitor:{data['name']}:latest"
            prev_data = self.redis_client.get(prev_key)
            
            if prev_data:
                prev_data = json.loads(prev_data)
                
                # Check if content changed
                if prev_data.get('extraction_hash') != data['extraction_hash']:
                    change = self._analyze_change(prev_data, data)
                    if change:
                        changes.append(change)
            
            # Store current data
            self.redis_client.setex(
                prev_key,
                timedelta(days=30),
                json.dumps(data)
            )
        
        state['changes'] = changes
        return state
    
    def _analyze_change(self, old: Dict, new: Dict) -> Dict[str, Any]:
        """Analyze what changed and assess significance"""
        changes_detected = []
        
        old_extracted = old.get('extracted', {})
        new_extracted = new.get('extracted', {})
        
        # Pricing changes (high priority)
        if old_extracted.get('pricing') != new_extracted.get('pricing'):
            changes_detected.append({
                'type': 'pricing',
                'old': old_extracted.get('pricing'),
                'new': new_extracted.get('pricing'),
                'significance': 'high'
            })
        
        # Feature changes (medium priority)
        old_features = set(old_extracted.get('features', []))
        new_features = set(new_extracted.get('features', []))
        
        added_features = new_features - old_features
        removed_features = old_features - new_features
        
        if added_features or removed_features:
            changes_detected.append({
                'type': 'features',
                'added': list(added_features),
                'removed': list(removed_features),
                'significance': 'high' if len(added_features) > 2 else 'medium'
            })
        
        if changes_detected:
            return {
                'competitor': new['name'],
                'timestamp': new['scraped_at'],
                'changes': changes_detected
            }
        
        return None
    
    async def news_monitoring_node(self, state: MonitoringState) -> MonitoringState:
        """Monitor news/social mentions using Perplexity"""
        mentions = []
        
        for comp in state['competitors']:
            query = f"Latest news and announcements from {comp['name']} in the last 7 days"
            
            response = await self.perplexity.search(
                query=query,
                search_recency_filter="week"
            )
            
            mentions.append({
                'competitor': comp['name'],
                'articles': response.get('results', [])[:5],
                'summary': response.get('answer', '')
            })
        
        state['news_mentions'] = mentions
        return state
    
    def analysis_node(self, state: MonitoringState) -> MonitoringState:
        """Generate strategic analysis of changes"""
        if not state['changes'] and not state['news_mentions']:
            state['analysis'] = {'status': 'no_changes'}
            return state
        
        analysis_prompt = f"""Analyze these competitive intelligence updates:

Website Changes:
{json.dumps(state['changes'], indent=2)}

News Mentions:
{json.dumps(state['news_mentions'], indent=2)}

Provide strategic analysis:
1. What do these changes mean for market positioning?
2. What threats/opportunities do they present?
3. What should we do in response?

Output as JSON with: market_implications, threats, opportunities, recommended_actions"""

        response = openai.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=[{"role": "user", "content": analysis_prompt}],
            temperature=0.7,
            response_format={"type": "json_object"}
        )
        
        state['analysis'] = json.loads(response.choices[0].message.content)
        return state
    
    async def alert_node(self, state: MonitoringState) -> MonitoringState:
        """Send alerts for high-priority changes"""
        high_priority = [
            change for change in state['changes']
            if any(c['significance'] == 'high' for c in change['changes'])
        ]
        
        if high_priority:
            alert_message = self._format_alert(high_priority, state['analysis'])
            
            # Send to Slack
            await self._send_slack_alert(alert_message)
            
            # Send email to strategy team
            await self._send_email_alert(alert_message)
            
            state['alerts_sent'] = True
        else:
            state['alerts_sent'] = False
        
        return state
    
    def _format_alert(self, changes: List[Dict], analysis: Dict) -> str:
        """Format alert message"""
        message = "🚨 **HIGH PRIORITY COMPETITIVE CHANGES DETECTED**\n\n"
        
        for change in changes:
            message += f"**{change['competitor']}**\n"
            for c in change['changes']:
                if c['significance'] == 'high':
                    message += f"  • {c['type'].upper()}: {json.dumps(c.get('new', c.get('added')))}\n"
            message += "\n"
        
        if analysis.get('recommended_actions'):
            message += "\n**Recommended Actions:**\n"
            for action in analysis['recommended_actions']:
                message += f"  • {action}\n"
        
        return message
    
    async def _send_slack_alert(self, message: str):
        """Send alert to Slack"""
        webhook_url = os.getenv('SLACK_WEBHOOK_URL')
        if webhook_url:
            async with aiohttp.ClientSession() as session:
                await session.post(webhook_url, json={'text': message})
    
    async def _send_email_alert(self, message: str):
        """Send email alert"""
        # Implement email sending (SendGrid, AWS SES, etc.)
        pass
    
    def build_monitoring_graph(self) -> Graph:
        """Build LangGraph monitoring workflow"""
        graph = Graph()
        
        # Add nodes
        graph.add_node("scrape", self.scrape_node)
        graph.add_node("extract", self.extract_node)
        graph.add_node("detect_changes", self.detect_changes_node)
        graph.add_node("monitor_news", self.news_monitoring_node)
        graph.add_node("analyze", self.analysis_node)
        graph.add_node("alert", self.alert_node)
        
        # Add edges
        graph.set_entry_point("scrape")
        graph.add_edge("scrape", "extract")
        graph.add_edge("extract", "detect_changes")
        graph.add_edge("detect_changes", "monitor_news")
        graph.add_edge("monitor_news", "analyze")
        graph.add_edge("analyze", "alert")
        graph.add_edge("alert", END)
        
        return graph.compile()

# Usage
async def run_monitoring():
    system = CompetitiveMonitoringSystem()
    graph = system.build_monitoring_graph()
    
    initial_state: MonitoringState = {
        'competitors': [
            {'name': 'Stripe', 'url': 'https://stripe.com/pricing'},
            {'name': 'Square', 'url': 'https://squareup.com/pricing'},
            {'name': 'PayPal', 'url': 'https://paypal.com/pricing'}
        ],
        'scraped_data': [],
        'previous_data': {},
        'changes': [],
        'news_mentions': [],
        'analysis': {},
        'alerts_sent': False
    }
    
    result = await graph.ainvoke(initial_state)
    print(f"Monitoring complete. Alerts sent: {result['alerts_sent']}")

# Run continuously
if __name__ == '__main__':
    while True:
        asyncio.run(run_monitoring())
        # Wait 6 hours between runs
        time.sleep(6 * 60 * 60)