From Static Reports to Living Intelligence: Building Real-Time Business Dashboards with AI Web Scraping

from scrapegraph_py import Client
from datetime import datetime
import time
import json
 
class LivingIntelligenceCollector:
    """
    Autonomous data collection agent for real-time dashboards
    Continuously monitors multiple sources and updates dashboard data
    """
    
    def __init__(self, api_key, data_sources):
        self.client = Client(api_key=api_key)
        self.data_sources = data_sources
        self.latest_data = {}
        self.change_history = []
        
    def collect_competitive_data(self):
        """Collect data from all competitive sources"""
        competitive_data = []
        
        for source in self.data_sources['competitors']:
            prompt = """
            Extract the following information:
            - All product prices and any discounts
            - New product launches or features
            - Stock availability status
            - Promotional campaigns or special offers
            - Customer review scores and sentiment
            - Estimated traffic or popularity metrics
            """
            
            try:
                response = self.client.smartscraper(
                    website_url=source['url'],
                    user_prompt=prompt
                )
                
                competitive_data.append({
                    'source': source['name'],
                    'url': source['url'],
                    'data': response,
                    'timestamp': datetime.now().isoformat(),
                    'status': 'success'
                })
                
            except Exception as e:
                competitive_data.append({
                    'source': source['name'],
                    'error': str(e),
                    'timestamp': datetime.now().isoformat(),
                    'status': 'failed'
                })
        
        return competitive_data
    
    def collect_market_intelligence(self):
        """Collect broader market intelligence"""
        market_data = []
        
        for source in self.data_sources['market']:
            prompt = """
            Extract relevant market intelligence:
            - Industry news and announcements
            - Market trends and emerging patterns
            - Regulatory changes or updates
            - Economic indicators affecting the industry
            - Expert opinions or analyst insights
            """
            
            response = self.client.smartscraper(
                website_url=source['url'],
                user_prompt=prompt
            )
            
            market_data.append({
                'source': source['name'],
                'data': response,
                'timestamp': datetime.now().isoformat()
            })
        
        return market_data
    
    def collect_customer_intelligence(self):
        """Collect customer sentiment and feedback"""
        customer_data = []
        
        for source in self.data_sources['customer_feedback']:
            prompt = """
            Extract customer feedback and sentiment:
            - Recent reviews and ratings
            - Common complaints or issues
            - Praised features or aspects
            - Overall sentiment (positive/negative/neutral)
            - Trending topics in customer discussions
            """
            
            response = self.client.smartscraper(
                website_url=source['url'],
                user_prompt=prompt
            )
            
            customer_data.append({
                'source': source['name'],
                'data': response,
                'timestamp': datetime.now().isoformat()
            })
        
        return customer_data
    
    def detect_significant_changes(self, new_data):
        """Identify significant changes requiring alerts"""
        changes = []
        
        # Compare with previous data
        if self.latest_data:
            # Price changes
            price_changes = self.compare_prices(
                self.latest_data.get('competitive'),
                new_data.get('competitive')
            )
            if price_changes:
                changes.extend(price_changes)
            
            # Market shifts
            market_changes = self.compare_market_data(
                self.latest_data.get('market'),
                new_data.get('market')
            )
            if market_changes:
                changes.extend(market_changes)
            
            # Sentiment shifts
            sentiment_changes = self.compare_sentiment(
                self.latest_data.get('customer'),
                new_data.get('customer')
            )
            if sentiment_changes:
                changes.extend(sentiment_changes)
        
        return changes
    
    def continuous_collection(self, interval_minutes=15):
        """
        Run continuous data collection for living dashboard
        This keeps your dashboard always up-to-date
        """
        print("🚀 Living Intelligence Collector Started")
        print(f"📊 Monitoring {len(self.data_sources)} data source categories")
        print(f"⚡ Update interval: {interval_minutes} minutes")
        print("-" * 60)
        
        cycle = 0
        
        while True:
            cycle += 1
            cycle_start = datetime.now()
            
            print(f"\n🔄 Collection Cycle #{cycle} - {cycle_start.strftime('%Y-%m-%d %H:%M:%S')}")
            
            try:
                # Collect all intelligence types
                new_data = {
                    'competitive': self.collect_competitive_data(),
                    'market': self.collect_market_intelligence(),
                    'customer': self.collect_customer_intelligence(),
                    'metadata': {
                        'cycle': cycle,
                        'timestamp': cycle_start.isoformat(),
                        'sources_monitored': self.count_sources()
                    }
                }
                
                # Detect significant changes
                changes = self.detect_significant_changes(new_data)
                
                if changes:
                    print(f"⚠️  {len(changes)} significant changes detected!")
                    for change in changes:
                        print(f"   • {change['type']}: {change['description']}")
                        # In production: trigger alerts, webhooks, etc.
                else:
                    print("✓ No significant changes detected")
                
                # Update latest data
                self.latest_data = new_data
                
                # Store in change history
                self.change_history.append({
                    'cycle': cycle,
                    'timestamp': cycle_start.isoformat(),
                    'changes': changes,
                    'data_summary': self.summarize_data(new_data)
                })
                
                # In production: push to dashboard database
                self.push_to_dashboard(new_data)
                
                cycle_duration = (datetime.now() - cycle_start).total_seconds()
                print(f"✓ Cycle completed in {cycle_duration:.1f} seconds")
                
            except Exception as e:
                print(f"✗ Error in collection cycle: {e}")
            
            # Wait for next cycle
            time.sleep(interval_minutes * 60)
    
    def count_sources(self):
        """Count total sources being monitored"""
        return sum(len(sources) for sources in self.data_sources.values())
    
    def summarize_data(self, data):
        """Create summary of collected data for storage"""
        return {
            'competitive_sources': len(data.get('competitive', [])),
            'market_sources': len(data.get('market', [])),
            'customer_sources': len(data.get('customer', [])),
            'total_data_points': self.count_data_points(data)
        }
    
    def push_to_dashboard(self, data):
        """Push data to dashboard database (implement with your DB)"""
        # Example: PostgreSQL, MongoDB, InfluxDB, etc.
        # db.dashboards.update_one(
        #     {'dashboard_id': 'main'},
        #     {'$set': {'data': data, 'updated_at': datetime.now()}}
        # )
        pass
 
# Configuration for living dashboard
data_sources = {
    'competitors': [
        {'name': 'Competitor A', 'url': 'https://competitor-a.com'},
        {'name': 'Competitor B', 'url': 'https://competitor-b.com'},
        {'name': 'Competitor C', 'url': 'https://competitor-c.com'}
    ],
    'market': [
        {'name': 'Industry News', 'url': 'https://industry-news.com'},
        {'name': 'Market Analysis', 'url': 'https://market-research.com'}
    ],
    'customer_feedback': [
        {'name': 'Review Site A', 'url': 'https://reviews-a.com'},
        {'name': 'Review Site B', 'url': 'https://reviews-b.com'}
    ]
}
 
# Deploy living intelligence collector
if __name__ == "__main__":
    collector = LivingIntelligenceCollector(
        api_key="your-scrapegraphai-api-key",
        data_sources=data_sources
    )
    
    # Start continuous collection (updates every 15 minutes)
    collector.continuous_collection(interval_minutes=15)

from datetime import datetime, timedelta
import statistics
 
class IntelligentProcessor:
    """
    Process collected data into dashboard-ready insights
    Performs real-time analysis, pattern recognition, and alerting
    """
    
    def __init__(self):
        self.historical_data = []
        self.alert_thresholds = {
            'price_change': 0.05,  # 5% price change triggers alert
            'sentiment_drop': 0.3,  # 0.3 point sentiment drop
            'traffic_spike': 0.25,  # 25% traffic change
            'stock_out': True       # Any stock-out triggers alert
        }
    
    def process_competitive_data(self, competitive_data):
        """
        Transform competitive data into dashboard metrics
        """
        processed = {
            'timestamp': datetime.now().isoformat(),
            'metrics': {},
            'alerts': [],
            'trends': {}
        }
        
        # Calculate competitive metrics
        for competitor in competitive_data:
            if competitor['status'] == 'success':
                data = competitor['data']
                
                # Price analysis
                prices = self.extract_prices(data)
                processed['metrics'][competitor['source']] = {
                    'average_price': statistics.mean(prices) if prices else 0,
                    'min_price': min(prices) if prices else 0,
                    'max_price': max(prices) if prices else 0,
                    'products_tracked': len(prices),
                    'price_trend': self.calculate_trend(competitor['source'], prices)
                }
                
                # Check for alerts
                alerts = self.check_price_alerts(competitor['source'], prices)
                processed['alerts'].extend(alerts)
        
        return processed
    
    def process_market_intelligence(self, market_data):
        """
        Extract and analyze market trends
        """
        processed = {
            'timestamp': datetime.now().isoformat(),
            'trends': [],
            'sentiment': {},
            'key_events': []
        }
        
        for source in market_data:
            data = source['data']
            
            # Extract trends
            trends = self.extract_trends(data)
            processed['trends'].extend(trends)
            
            # Analyze sentiment
            sentiment = self.analyze_sentiment(data)
            processed['sentiment'][source['source']] = sentiment
            
            # Identify key events
            events = self.identify_key_events(data)
            processed['key_events'].extend(events)
        
        return processed
    
    def process_customer_intelligence(self, customer_data):
        """
        Analyze customer feedback and sentiment
        """
        processed = {
            'timestamp': datetime.now().isoformat(),
            'overall_sentiment': 0,
            'common_issues': [],
            'satisfaction_score': 0,
            'trending_topics': []
        }
        
        all_sentiments = []
        all_ratings = []
        
        for source in customer_data:
            data = source['data']
            
            # Extract sentiment
            sentiment = self.extract_sentiment_score(data)
            all_sentiments.append(sentiment)
            
            # Extract ratings
            ratings = self.extract_ratings(data)
            all_ratings.extend(ratings)
            
            # Extract issues
            issues = self.extract_issues(data)
            processed['common_issues'].extend(issues)
        
        # Calculate aggregates
        if all_sentiments:
            processed['overall_sentiment'] = statistics.mean(all_sentiments)
        
        if all_ratings:
            processed['satisfaction_score'] = statistics.mean(all_ratings)
        
        # Check for sentiment alerts
        if self.detect_sentiment_drop(processed['overall_sentiment']):
            processed['alert'] = {
                'type': 'sentiment_drop',
                'severity': 'high',
                'message': f"Customer sentiment dropped to {processed['overall_sentiment']:.2f}"
            }
        
        return processed
    
    def calculate_trend(self, source, current_prices):
        """Calculate price trend direction"""
        if not self.historical_data:
            return 'stable'
        
        # Get historical prices for this source
        historical_prices = self.get_historical_prices(source)
        
        if not historical_prices or not current_prices:
            return 'stable'
        
        current_avg = statistics.mean(current_prices)
        historical_avg = statistics.mean(historical_prices)
        
        change = (current_avg - historical_avg) / historical_avg
        
        if change > 0.03:
            return 'increasing'
        elif change < -0.03:
            return 'decreasing'
        else:
            return 'stable'
    
    def check_price_alerts(self, source, current_prices):
        """Check if price changes warrant alerts"""
        alerts = []
        
        historical_prices = self.get_historical_prices(source)
        
        if historical_prices and current_prices:
            current_avg = statistics.mean(current_prices)
            historical_avg = statistics.mean(historical_prices)
            
            change_pct = abs(current_avg - historical_avg) / historical_avg
            
            if change_pct > self.alert_thresholds['price_change']:
                direction = "increased" if current_avg > historical_avg else "decreased"
                alerts.append({
                    'type': 'price_change',
                    'source': source,
                    'severity': 'high' if change_pct > 0.10 else 'medium',
                    'message': f"{source} prices {direction} by {change_pct:.1%}",
                    'current': current_avg,
                    'previous': historical_avg,
                    'change_pct': change_pct
                })
        
        return alerts
    
    def detect_sentiment_drop(self, current_sentiment):
        """Detect significant sentiment drops"""
        if not self.historical_data:
            return False
        
        # Get recent historical sentiment
        recent_sentiment = self.get_recent_sentiment()
        
        if recent_sentiment:
            drop = recent_sentiment - current_sentiment
            return drop > self.alert_thresholds['sentiment_drop']
        
        return False
    
    def generate_dashboard_update(self, all_processed_data):
        """
        Generate complete dashboard update with all metrics
        """
        dashboard_update = {
            'timestamp': datetime.now().isoformat(),
            'competitive_intelligence': all_processed_data['competitive'],
            'market_intelligence': all_processed_data['market'],
            'customer_intelligence': all_processed_data['customer'],
            'alerts': self.aggregate_alerts(all_processed_data),
            'kpis': self.calculate_kpis(all_processed_data),
            'recommendations': self.generate_recommendations(all_processed_data)
        }
        
        return dashboard_update
    
    def calculate_kpis(self, data):
        """Calculate key performance indicators"""
        return {
            'competitive_position': self.calculate_competitive_position(data),
            'market_sentiment': self.calculate_market_sentiment(data),
            'customer_satisfaction': data['customer']['satisfaction_score'],
            'alert_count': len(self.aggregate_alerts(data)),
            'data_freshness': self.calculate_data_freshness(data)
        }

// Example: React + WebSockets for real-time updates
// dashboard-frontend.jsx
 
import React, { useState, useEffect } from 'react';
import { Line, Bar } from 'react-chartjs-2';
import io from 'socket.io-client';
 
const LivingDashboard = () => {
  const [dashboardData, setDashboardData] = useState(null);
  const [alerts, setAlerts] = useState([]);
  const [lastUpdate, setLastUpdate] = useState(null);
  
  useEffect(() => {
    // Connect to real-time data stream
    const socket = io('your-backend-url');
    
    // Listen for dashboard updates
    socket.on('dashboard_update', (data) => {
      setDashboardData(data);
      setLastUpdate(new Date());
      
      // Handle new alerts
      if (data.alerts && data.alerts.length > 0) {
        setAlerts(prev => [...data.alerts, ...prev].slice(0, 10));
        // Trigger notification
        showNotification(data.alerts[0]);
      }
    });
    
    // Cleanup
    return () => socket.disconnect();
  }, []);
  
  return (
    <div className="living-dashboard">
      <header>
        <h1>Living Intelligence Dashboard</h1>
        <div className="status">
          <span className="live-indicator">● LIVE</span>
          <span>Last Update: {lastUpdate?.toLocaleTimeString()}</span>
        </div>
      </header>
      
      <AlertsPanel alerts={alerts} />
      <KPIsPanel kpis={dashboardData?.kpis} />
      <CompetitiveIntelligence data={dashboardData?.competitive_intelligence} />
      <MarketTrends data={dashboardData?.market_intelligence} />
      <CustomerSentiment data={dashboardData?.customer_intelligence} />
    </div>
  );
};

# Starter implementation for your specific use case
 
from scrapegraph_py import Client
import json
 
class YourCompanyDashboardCollector:
    """
    Customize this for your specific business needs
    """
    
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
        
        # Define YOUR specific data sources
        self.competitors = [
            # Add your actual competitors here
        ]
        
        self.market_sources = [
            # Add industry news sites, analyst sites, etc.
        ]
        
        self.review_sites = [
            # Add review platforms relevant to your business
        ]
    
    def collect_your_critical_metrics(self):
        """
        Customize this to collect YOUR most important metrics
        """
        
        # Example: E-commerce competitive pricing
        pricing_prompt = """
        Extract:
        - Product names
        - Current prices
        - Original prices (if on sale)
        - Stock status
        - Shipping cost and time
        """
        
        # Example: SaaS competitive features
        features_prompt = """
        Extract:
        - Plan names and pricing
        - Features included in each plan
        - New feature announcements
        - Integration offerings
        - Customer testimonials
        """
        
        # Example: B2B competitive intelligence
        business_prompt = """
        Extract:
        - Case studies and customer wins
        - Partnership announcements
        - Product updates
        - Pricing and packaging changes
        - Target market signals
        """
        
        # Implement YOUR collection logic
        results = {}
        
        for competitor in self.competitors:
            # Choose the right prompt for your business
            response = self.client.smartscraper(
                website_url=competitor,
                user_prompt=pricing_prompt  # or features_prompt, or business_prompt
            )
            
            results[competitor] = response
        
        return results
 
# Start building YOUR living dashboard
collector = YourCompanyDashboardCollector(api_key="your-key")
data = collector.collect_your_critical_metrics()
print(json.dumps(data, indent=2))

class DashboardIntelligenceEngine:
    """
    Add intelligence to your raw data
    """
    
    def __init__(self):
        self.baseline_metrics = {}
        self.alert_rules = {}
    
    def analyze_and_alert(self, current_data):
        """
        Your custom analysis logic
        """
        insights = {
            'alerts': [],
            'trends': [],
            'recommendations': []
        }
        
        # Detect price changes (example)
        if self.baseline_metrics.get('prices'):
            for product, current_price in current_data['prices'].items():
                baseline = self.baseline_metrics['prices'].get(product)
                
                if baseline and abs(current_price - baseline) / baseline > 0.10:
                    insights['alerts'].append({
                        'type': 'price_alert',
                        'product': product,
                        'change': ((current_price - baseline) / baseline) * 100,
                        'action': 'Review pricing strategy'
                    })
        
        # Detect market opportunities
        # Add YOUR business logic here
        
        return insights

# E-commerce pricing dashboard implementation
 
from scrapegraph_py import Client
import streamlit as st
import pandas as pd
from datetime import datetime
 
class EcommercePricingDashboard:
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
        self.competitors = [
            'https://competitor1.com',
            'https://competitor2.com',
            'https://competitor3.com'
        ]
    
    def collect_pricing_data(self):
        """Collect pricing from all competitors"""
        prompt = """
        Extract for each product:
        - Product name
        - Current price
        - Original price
        - Discount percentage
        - Stock status
        - Shipping cost
        - Rating and review count
        """
        
        all_pricing = []
        
        for competitor_url in self.competitors:
            result = self.client.smartscraper(
                website_url=competitor_url,
                user_prompt=prompt
            )
            all_pricing.append({
                'competitor': competitor_url,
                'products': result,
                'timestamp': datetime.now()
            })
        
        return all_pricing
    
    def build_streamlit_dashboard(self):
        """Create real-time Streamlit dashboard"""
        st.title("🔴 LIVE E-Commerce Pricing Intelligence")
        st.caption(f"Last updated: {datetime.now().strftime('%H:%M:%S')}")
        
        # Auto-refresh every 15 minutes
        if st.button("Refresh Data") or time_to_refresh():
            pricing_data = self.collect_pricing_data()
            
            # Store in session state
            st.session_state['pricing_data'] = pricing_data
        
        # Display data
        if 'pricing_data' in st.session_state:
            data = st.session_state['pricing_data']
            
            # Show alerts
            alerts = self.detect_pricing_alerts(data)
            if alerts:
                for alert in alerts:
                    st.warning(f"⚠️ {alert['message']}")
            
            # Show competitive pricing table
            df = self.create_pricing_dataframe(data)
            st.dataframe(df, use_container_width=True)
            
            # Show pricing trends
            st.line_chart(self.create_trend_data(data))
 
# Deploy dashboard
dashboard = EcommercePricingDashboard(api_key="your-key")
dashboard.build_streamlit_dashboard()

class SaaSCompetitiveDashboard:
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
    
    def monitor_competitor_changes(self, competitor_url):
        """Comprehensive competitor monitoring"""
        
        # Pricing intelligence
        pricing_prompt = """
        Extract:
        - All pricing plans and costs
        - Features included in each tier
        - Annual vs monthly pricing
        - Enterprise/custom pricing mentions
        - Free trial details
        """
        
        # Product intelligence
        product_prompt = """
        Extract:
        - New features announced
        - Product updates or releases
        - Beta programs
        - Deprecated features
        - Technology stack mentions
        """
        
        # Market intelligence
        market_prompt = """
        Extract:
        - Customer testimonials and case studies
        - Industry awards or recognition
        - Partnership announcements
        - Integration marketplace additions
        - Event sponsorships or speaking
        """
        
        # Collect all intelligence types
        pricing_data = self.client.smartscraper(
            website_url=f"{competitor_url}/pricing",
            user_prompt=pricing_prompt
        )
        
        product_data = self.client.smartscraper(
            website_url=f"{competitor_url}/features",
            user_prompt=product_prompt
        )
        
        market_data = self.client.smartscraper(
            website_url=f"{competitor_url}/customers",
            user_prompt=market_prompt
        )
        
        return {
            'pricing': pricing_data,
            'product': product_data,
            'market': market_data,
            'timestamp': datetime.now().isoformat()
        }

class PredictiveDashboard:
    """
    Dashboard that predicts future trends
    """
    
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
        self.historical_data = []
    
    def generate_predictions(self, current_data):
        """
        Use historical patterns to predict future states
        """
        
        # Analyze patterns in collected data
        patterns = self.analyze_patterns(self.historical_data)
        
        # Generate predictions
        predictions = {
            'next_7_days': {
                'price_forecast': self.predict_pricing_trends(patterns),
                'demand_forecast': self.predict_demand_changes(patterns),
                'competitive_moves': self.predict_competitor_actions(patterns)
            },
            'confidence_scores': {
                'price': 0.87,
                'demand': 0.82,
                'competitive': 0.76
            },
            'recommended_actions': self.generate_proactive_recommendations(predictions)
        }
        
        return predictions

class MultiDimensionalDashboard:
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
    
    def collect_all_dimensions(self):
        """Collect data across all business dimensions"""
        
        return {
            'competitive': self.collect_competitive_intel(),
            'market': self.collect_market_intel(),
            'customer': self.collect_customer_intel(),
            'operational': self.collect_operational_metrics(),
            'financial': self.collect_financial_data()
        }
    
    def generate_cross_dimensional_insights(self, all_data):
        """
        Find insights that span multiple dimensions
        Example: Competitor price increase (competitive) + 
                 Rising commodity costs (market) + 
                 Customer price sensitivity (customer) =
                 Recommendation: Hold pricing, emphasize value
        """
        insights = []
        
        # Cross-reference competitive and market data
        if self.detect_market_cost_pressure(all_data):
            if self.detect_competitive_price_increases(all_data):
                insights.append({
                    'type': 'strategic_opportunity',
                    'message': 'Market cost pressure causing competitor price increases',
                    'recommendation': 'Consider selective price increases',
                    'expected_impact': '+5-8% margin improvement',
                    'risk_level': 'low'
                })
        
        return insights

from scrapegraph_py import Client
import time
from datetime import datetime
 
# 1. Initialize ScrapeGraphAI
client = Client(api_key="your-api-key")
 
# 2. Define what to monitor
competitors = [
    "https://competitor1.com",
    "https://competitor2.com"
]
 
# 3. Create collection function
def collect_intelligence():
    results = []
    for url in competitors:
        data = client.smartscraper(
            website_url=url,
            user_prompt="Extract all pricing and product information"
        )
        results.append({
            'competitor': url,
            'data': data,
            'time': datetime.now().isoformat()
        })
    return results
 
# 4. Run continuous monitoring
while True:
    print(f"🔴 LIVE UPDATE - {datetime.now().strftime('%H:%M:%S')}")
    intelligence = collect_intelligence()
    
    # Your dashboard update logic here
    print(f"Collected data from {len(intelligence)} sources")
    
    # Wait 15 minutes
    time.sleep(15 * 60)
 
# That's it! You now have living intelligence.