Automate Crisis Detection 🚀

# Production Pattern with Real-Time Streaming (500+ sources)
import asyncio
import aiohttp
import tweepy
from anthropic import AsyncAnthropic
from dataclasses import dataclass
from datetime import datetime
import json
import redis
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class CrisisSignal:
    source: str
    content: str
    sentiment: float
    reach: int
    timestamp: str
    url: str = None
    keywords: List[str] = None

class RealTimeCrisisMonitor:
    def __init__(self):
        self.anthropic = AsyncAnthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            decode_responses=True
        )
        self.signal_buffer = []
        self.buffer_lock = asyncio.Lock()
        
    async def stream_twitter(self, keywords: List[str]):
        """Real-time Twitter stream using filtered stream API"""
        class StreamListener(tweepy.StreamingClient):
            def __init__(self, monitor, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self.monitor = monitor
            
            def on_tweet(self, tweet):
                signal = CrisisSignal(
                    source='twitter',
                    content=tweet.text,
                    sentiment=0,  # Will be analyzed
                    reach=tweet.public_metrics.get('impression_count', 0),
                    timestamp=datetime.now().isoformat(),
                    url=f'https://twitter.com/user/status/{tweet.id}'
                )
                asyncio.create_task(self.monitor.process_signal(signal))
        
        stream = StreamListener(
            self,
            bearer_token=os.getenv('TWITTER_BEARER_TOKEN')
        )
        
        # Add rules for keywords
        for keyword in keywords:
            stream.add_rules(tweepy.StreamRule(f'"{keyword}"'))
        
        stream.filter(tweet_fields=['public_metrics', 'created_at'])
    
    async def stream_reddit(self, subreddits: List[str]):
        """Poll Reddit in real-time (no native streaming)"""
        while True:
            for sub in subreddits:
                try:
                    async with aiohttp.ClientSession() as session:
                        async with session.get(
                            f'https://www.reddit.com/r/{sub}/new.json',
                            headers={'User-Agent': 'CrisisMonitor/2.0'}
                        ) as response:
                            data = await response.json()
                            
                            for post in data['data']['children']:
                                post_data = post['data']
                                
                                # Check if we've seen this post
                                post_id = f'reddit:{post_data["id"]}'
                                if self.redis_client.exists(post_id):
                                    continue
                                
                                self.redis_client.setex(post_id, 3600, '1')
                                
                                signal = CrisisSignal(
                                    source='reddit',
                                    content=f"{post_data['title']}. {post_data.get('selftext', '')}",
                                    sentiment=0,
                                    reach=post_data['ups'] * 10,
                                    timestamp=datetime.fromtimestamp(post_data['created_utc']).isoformat(),
                                    url=f"https://reddit.com{post_data['permalink']}"
                                )
                                
                                await self.process_signal(signal)
                
                except Exception as e:
                    logger.error(f'Reddit streaming error: {e}')
            
            await asyncio.sleep(60)  # Poll every minute
    
    async def stream_news(self, keywords: List[str]):
        """Poll news APIs for updates"""
        while True:
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(
                        'https://newsapi.org/v2/everything',
                        params={
                            'q': ' OR '.join(keywords),
                            'apiKey': os.getenv('NEWS_API_KEY'),
                            'sortBy': 'publishedAt',
                            'pageSize': 20
                        }
                    ) as response:
                        data = await response.json()
                        
                        for article in data.get('articles', []):
                            article_id = f"news:{article['url']}"
                            if self.redis_client.exists(article_id):
                                continue
                            
                            self.redis_client.setex(article_id, 3600, '1')
                            
                            signal = CrisisSignal(
                                source='news',
                                content=f"{article['title']}. {article['description']}",
                                sentiment=0,
                                reach=self._estimate_news_reach(article['source']['name']),
                                timestamp=article['publishedAt'],
                                url=article['url']
                            )
                            
                            await self.process_signal(signal)
            
            except Exception as e:
                logger.error(f'News streaming error: {e}')
            
            await asyncio.sleep(300)  # Poll every 5 minutes
    
    async def process_signal(self, signal: CrisisSignal):
        """Process individual signal and add to buffer"""
        async with self.buffer_lock:
            self.signal_buffer.append(signal)
            
            # Analyze in batches of 10 or every 2 minutes
            if len(self.signal_buffer) >= 10:
                await self.analyze_batch()
    
    async def analyze_batch(self):
        """Analyze batch of signals for crisis patterns"""
        async with self.buffer_lock:
            if not self.signal_buffer:
                return
            
            batch = self.signal_buffer.copy()
            self.signal_buffer.clear()
        
        logger.info(f'Analyzing batch of {len(batch)} signals')
        
        # Combine signals for analysis
        content = '\n\n'.join([
            f"[{s.source}] {s.content}" for s in batch
        ])
        
        prompt = f"""Analyze these crisis signals in real-time.

Signals:
{content}

Return JSON with:
- crisis_level: LOW/MEDIUM/HIGH/CRITICAL
- crisis_score: 0-10
- sentiment: -1 to +1
- key_issues: array
- recommended_actions: array
- affected_areas: array
- escalation_needed: boolean"""
        
        try:
            response = await self.anthropic.messages.create(
                model='claude-3-5-sonnet-20241022',
                max_tokens=2048,
                messages=[{'role': 'user', 'content': prompt}]
            )
            
            content = response.content[0]
            if content.type != 'text':
                raise ValueError('Invalid response')
            
            analysis = json.loads(content.text)
            
            # Store in Redis for dashboard
            self.redis_client.setex(
                f'analysis:{datetime.now().isoformat()}',
                3600,
                json.dumps(analysis)
            )
            
            # Send alerts if needed
            if analysis.get('escalation_needed') or analysis['crisis_level'] in ['HIGH', 'CRITICAL']:
                await self.send_alerts(analysis, batch)
            
            logger.info(f"Analysis complete: {analysis['crisis_level']} (score: {analysis['crisis_score']})")
        
        except Exception as e:
            logger.error(f'Analysis error: {e}')
    
    async def send_alerts(self, analysis: Dict, signals: List[CrisisSignal]):
        """Send alerts through multiple channels"""
        tasks = []
        
        # Always send Slack
        tasks.append(self._send_slack(analysis, signals))
        
        # PagerDuty for CRITICAL
        if analysis['crisis_level'] == 'CRITICAL':
            tasks.append(self._send_pagerduty(analysis))
            tasks.append(self._send_sms(analysis))
        
        # Email for HIGH or CRITICAL
        if analysis['crisis_level'] in ['HIGH', 'CRITICAL']:
            tasks.append(self._send_email(analysis))
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _send_slack(self, analysis: Dict, signals: List[CrisisSignal]):
        emoji_map = {
            'CRITICAL': '🚨',
            'HIGH': '⚠️',
            'MEDIUM': '⚡',
            'LOW': 'ℹ️'
        }
        
        total_reach = sum(s.reach for s in signals)
        
        message = {
            'text': f"{emoji_map[analysis['crisis_level']]} Crisis Alert: {analysis['crisis_level']}",
            'blocks': [
                {
                    'type': 'header',
                    'text': {
                        'type': 'plain_text',
                        'text': f"{emoji_map[analysis['crisis_level']]} Crisis Detected: {analysis['crisis_level']}"
                    }
                },
                {
                    'type': 'section',
                    'fields': [
                        {'type': 'mrkdwn', 'text': f"*Score:* {analysis['crisis_score']}/10"},
                        {'type': 'mrkdwn', 'text': f"*Sentiment:* {analysis['sentiment']}"},
                        {'type': 'mrkdwn', 'text': f"*Signals:* {len(signals)}"},
                        {'type': 'mrkdwn', 'text': f"*Reach:* {total_reach:,}"}
                    ]
                },
                {
                    'type': 'section',
                    'text': {
                        'type': 'mrkdwn',
                        'text': f"*Key Issues:*\n" + '\n'.join([f"• {i}" for i in analysis['key_issues']])
                    }
                },
                {
                    'type': 'section',
                    'text': {
                        'type': 'mrkdwn',
                        'text': f"*Immediate Actions:*\n" + '\n'.join([f"• {a}" for a in analysis['recommended_actions']])
                    }
                }
            ]
        }
        
        async with aiohttp.ClientSession() as session:
            await session.post(
                os.getenv('SLACK_WEBHOOK_URL'),
                json=message
            )
    
    async def _send_pagerduty(self, analysis: Dict):
        payload = {
            'incident': {
                'type': 'incident',
                'title': f"CRITICAL Crisis: {', '.join(analysis['key_issues'][:2])}",
                'service': {
                    'id': os.getenv('PAGERDUTY_SERVICE_ID'),
                    'type': 'service_reference'
                },
                'urgency': 'high',
                'body': {
                    'type': 'incident_body',
                    'details': f"Crisis score: {analysis['crisis_score']}\nActions: {', '.join(analysis['recommended_actions'])}"
                }
            }
        }
        
        async with aiohttp.ClientSession() as session:
            await session.post(
                'https://api.pagerduty.com/incidents',
                json=payload,
                headers={
                    'Authorization': f"Token token={os.getenv('PAGERDUTY_API_KEY')}",
                    'Content-Type': 'application/json'
                }
            )
    
    async def _send_email(self, analysis: Dict):
        # Implement email via SendGrid, AWS SES, etc.
        logger.info(f'Email alert sent: {analysis["crisis_level"]}')
    
    async def _send_sms(self, analysis: Dict):
        # Implement SMS via Twilio, AWS SNS, etc.
        logger.info(f'SMS alert sent: {analysis["crisis_level"]}')
    
    def _estimate_news_reach(self, source: str) -> int:
        reach_map = {
            'TechCrunch': 5000000,
            'The Verge': 3000000,
            'Reuters': 10000000,
            'Bloomberg': 8000000
        }
        return reach_map.get(source, 100000)
    
    async def run(self):
        """Start all monitoring streams"""
        keywords = ['@YourCompany', 'YourProduct defect', 'YourCompany recall']
        subreddits = ['YourIndustry', 'ProductReviews']
        
        logger.info('Starting real-time crisis monitoring...')
        
        # Start all streams concurrently
        await asyncio.gather(
            self.stream_twitter(keywords),
            self.stream_reddit(subreddits),
            self.stream_news(keywords),
            self._periodic_batch_analysis(),  # Backup timer
            return_exceptions=True
        )
    
    async def _periodic_batch_analysis(self):
        """Backup: analyze every 2 minutes even if buffer not full"""
        while True:
            await asyncio.sleep(120)
            await self.analyze_batch()

# Usage
if __name__ == '__main__':
    monitor = RealTimeCrisisMonitor()
    asyncio.run(monitor.run())