Facebook Smart Scraper

from scrapegraph_py import Client
 
# Initialize the client
client = Client(api_key="your-scrapegraph-api-key-here")
 
# Extract Facebook profile data
response = client.smartscraper(
    website_url="https://facebook.com/profile-url",
    user_prompt="Extract profile name, bio, location, work history, and contact information"
)
 
print(response.result)

# Define a schema for structured extraction
schema = {
    "profile_name": "string",
    "bio": "string", 
    "location": "string",
    "work": ["string"],
    "education": ["string"],
    "contact_info": "string",
    "verified": "boolean"
}
 
response = client.smartscraper(
    website_url="https://facebook.com/business-page",
    user_prompt="Extract profile information according to the schema",
    output_schema=schema
)

# Extract business-specific information
business_response = client.smartscraper(
    website_url="https://facebook.com/business-page",
    user_prompt="""
    Extract:
    - Business name and category
    - Contact information (phone, email, website)
    - Business hours
    - Location and address
    - Page description
    - Number of likes/followers
    - Recent posts engagement
    """
)

import json
import time
 
class FacebookCompetitorAnalysis:
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
    
    def analyze_competitor_pages(self, competitor_urls):
        """Analyze multiple competitor Facebook pages"""
        results = []
        
        for url in competitor_urls:
            try:
                data = self.client.smartscraper(
                    website_url=url,
                    user_prompt="""
                    Extract business information:
                    - Company name and description
                    - Industry/category
                    - Contact details
                    - Location
                    - Page engagement metrics
                    - Recent post content and engagement
                    """
                )
                
                results.append({
                    'url': url,
                    'data': data.result,
                    'timestamp': time.time()
                })
                
                # Respectful delay between requests
                time.sleep(2)
                
            except Exception as e:
                print(f"Error scraping {url}: {e}")
                continue
        
        return results
    
    def generate_competitor_report(self, analysis_results):
        """Generate competitive analysis report"""
        report = {
            'total_competitors': len(analysis_results),
            'summary': {},
            'detailed_analysis': analysis_results
        }
        
        # Add summary statistics
        industries = []
        locations = []
        
        for result in analysis_results:
            data = result.get('data', {})
            if 'category' in data:
                industries.append(data['category'])
            if 'location' in data:
                locations.append(data['location'])
        
        report['summary'] = {
            'industries': list(set(industries)),
            'locations': list(set(locations))
        }
        
        return report
 
# Usage
analyzer = FacebookCompetitorAnalysis(api_key="your-key")
competitors = [
    "https://facebook.com/competitor1",
    "https://facebook.com/competitor2", 
    "https://facebook.com/competitor3"
]
 
analysis = analyzer.analyze_competitor_pages(competitors)
report = analyzer.generate_competitor_report(analysis)

class FacebookLeadGenerator:
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
    
    def extract_business_leads(self, search_urls):
        """Extract potential business leads from Facebook pages"""
        leads = []
        
        for url in search_urls:
            lead_data = self.client.smartscraper(
                website_url=url,
                user_prompt="""
                Extract business lead information:
                - Business name
                - Industry/services offered
                - Contact email and phone
                - Website URL
                - Business size indicators
                - Location and address
                """
            )
            
            # Filter and qualify leads
            if self.qualify_lead(lead_data.result):
                leads.append({
                    'source_url': url,
                    'contact_info': lead_data.result,
                    'lead_score': self.score_lead(lead_data.result)
                })
        
        return sorted(leads, key=lambda x: x['lead_score'], reverse=True)
    
    def qualify_lead(self, lead_data):
        """Basic lead qualification logic"""
        required_fields = ['business_name', 'contact_email']
        return all(field in lead_data for field in required_fields)
    
    def score_lead(self, lead_data):
        """Score leads based on available information"""
        score = 0
        
        if 'website' in lead_data:
            score += 2
        if 'phone' in lead_data:
            score += 2
        if 'address' in lead_data:
            score += 1
        if 'industry' in lead_data:
            score += 1
            
        return score

class FacebookMonitor:
    def __init__(self, api_key):
        self.client = Client(api_key=api_key)
        self.monitored_pages = []
    
    def add_page_to_monitor(self, page_url, keywords):
        """Add a page to monitoring list"""
        self.monitored_pages.append({
            'url': page_url,
            'keywords': keywords,
            'last_check': None
        })
    
    def monitor_brand_mentions(self):
        """Monitor pages for brand mentions"""
        mentions = []
        
        for page in self.monitored_pages:
            try:
                data = self.client.smartscraper(
                    website_url=page['url'],
                    user_prompt=f"""
                    Look for mentions of these keywords: {', '.join(page['keywords'])}
                    Extract:
                    - Any posts or comments mentioning these terms
                    - Context around the mentions
                    - Sentiment (positive/negative/neutral)
                    - Engagement metrics
                    """
                )
                
                mentions.append({
                    'page': page['url'],
                    'mentions': data.result,
                    'timestamp': time.time()
                })
                
                # Update last check time
                page['last_check'] = time.time()
                
            except Exception as e:
                print(f"Error monitoring {page['url']}: {e}")
        
        return mentions
    
    def generate_mention_alerts(self, mentions, threshold=5):
        """Generate alerts for significant mentions"""
        alerts = []
        
        for mention_data in mentions:
            mention_count = len(mention_data.get('mentions', []))
            
            if mention_count >= threshold:
                alerts.append({
                    'type': 'high_mention_volume',
                    'page': mention_data['page'],
                    'count': mention_count,
                    'urgency': 'high' if mention_count > 10 else 'medium'
                })
        
        return alerts

def enrich_profile_data(raw_facebook_data):
    """Enrich scraped Facebook data with additional insights"""
    
    enriched_data = raw_facebook_data.copy()
    
    # Extract additional insights
    if 'bio' in enriched_data:
        enriched_data['bio_sentiment'] = analyze_sentiment(enriched_data['bio'])
        enriched_data['bio_keywords'] = extract_keywords(enriched_data['bio'])
    
    if 'work' in enriched_data:
        enriched_data['industry_classification'] = classify_industry(enriched_data['work'])
    
    if 'education' in enriched_data:
        enriched_data['education_level'] = determine_education_level(enriched_data['education'])
    
    return enriched_data
 
def analyze_sentiment(text):
    """Simple sentiment analysis"""
    positive_words = ['great', 'excellent', 'amazing', 'love', 'best']
    negative_words = ['bad', 'terrible', 'hate', 'worst', 'awful']
    
    text_lower = text.lower()
    pos_count = sum(1 for word in positive_words if word in text_lower)
    neg_count = sum(1 for word in negative_words if word in text_lower)
    
    if pos_count > neg_count:
        return 'positive'
    elif neg_count > pos_count:
        return 'negative'
    else:
        return 'neutral'

import concurrent.futures
import threading
 
class BatchFacebookScraper:
    def __init__(self, api_key, max_workers=5):
        self.client = Client(api_key=api_key)
        self.max_workers = max_workers
        self.results = []
        self.lock = threading.Lock()
    
    def scrape_multiple_profiles(self, urls, prompt):
        """Scrape multiple Facebook profiles concurrently"""
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_url = {
                executor.submit(self._scrape_single_profile, url, prompt): url 
                for url in urls
            }
            
            for future in concurrent.futures.as_completed(future_to_url):
                url = future_to_url[future]
                try:
                    result = future.result()
                    with self.lock:
                        self.results.append({
                            'url': url,
                            'data': result,
                            'status': 'success'
                        })
                except Exception as e:
                    with self.lock:
                        self.results.append({
                            'url': url,
                            'error': str(e),
                            'status': 'error'
                        })
        
        return self.results
    
    def _scrape_single_profile(self, url, prompt):
        """Scrape a single profile with error handling"""
        try:
            response = self.client.smartscraper(
                website_url=url,
                user_prompt=prompt
            )
            return response.result
        except Exception as e:
            raise Exception(f"Failed to scrape {url}: {str(e)}")
 
# Usage
batch_scraper = BatchFacebookScraper(api_key="your-key")
urls = ["https://facebook.com/page1", "https://facebook.com/page2"]
results = batch_scraper.scrape_multiple_profiles(
    urls, 
    "Extract business name, contact info, and description"
)

def robust_facebook_scraper(url, prompt, max_retries=3):
    """Robust scraper with retry logic"""
    
    client = Client(api_key="your-key")
    
    for attempt in range(max_retries):
        try:
            response = client.smartscraper(
                website_url=url,
                user_prompt=prompt
            )
            return response.result
            
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                return {'error': f"Failed after {max_retries} attempts", 'last_error': str(e)}

def validate_facebook_data(scraped_data):
    """Validate scraped Facebook data"""
    
    validation_results = {
        'valid': True,
        'issues': []
    }
    
    # Check for required fields
    required_fields = ['profile_name']
    for field in required_fields:
        if field not in scraped_data or not scraped_data[field]:
            validation_results['valid'] = False
            validation_results['issues'].append(f"Missing required field: {field}")
    
    # Validate data types
    if 'contact_info' in scraped_data:
        if not isinstance(scraped_data['contact_info'], (str, dict)):
            validation_results['valid'] = False
            validation_results['issues'].append("Invalid contact_info format")
    
    return validation_results