Create linkedin_data.py

LinkedIn data fetching module with RapidAPI integration.

Features:
- Fetches LinkedIn profile data via RapidAPI Fresh LinkedIn Profile Data API
- Automatic caching per user (30-day cache expiration)
- Rate limit handling and error management
- User-isolated cache directories
- Convenience functions for easy integration

Requires RAPIDAPI_KEY environment variable to be set in Space Secrets.

linkedin_data.py

@@ -0,0 +1,268 @@
+"""LinkedIn Data Fetching Module
+
+Provides functions to fetch and enrich LinkedIn profile data using RapidAPI.
+Includes caching to minimize API usage and respect rate limits.
+
+Usage:
+    from linkedin_data import fetch_linkedin_profile, enrich_lead
+    
+    profile = fetch_linkedin_profile("https://linkedin.com/in/username")
+    enriched = enrich_lead(lead_data)
+"""
+
+import os
+import json
+import time
+import requests
+from pathlib import Path
+from typing import Dict, Optional, Any
+
+
+class LinkedInDataFetcher:
+    """Handles LinkedIn profile data fetching via RapidAPI"""
+    
+    def __init__(self, api_key: Optional[str] = None, username: Optional[str] = None):
+        """Initialize LinkedIn data fetcher
+        
+        Args:
+            api_key: RapidAPI key (defaults to RAPIDAPI_KEY env var)
+            username: Current user's username for cache directory
+        """
+        self.api_key = api_key or os.environ.get('RAPIDAPI_KEY', '')
+        self.username = username
+        self.base_url = "https://fresh-linkedin-profile-data.p.rapidapi.com"
+        self.headers = {
+            "x-rapidapi-key": self.api_key,
+            "x-rapidapi-host": "fresh-linkedin-profile-data.p.rapidapi.com"
+        }
+        
+        # Set up cache directory for current user
+        if self.username:
+            self.cache_dir = Path(f"/data/users/{self.username}/linkedin_cache")
+            self.cache_dir.mkdir(parents=True, exist_ok=True)
+        else:
+            self.cache_dir = None
+    
+    def _get_cache_path(self, linkedin_url: str) -> Optional[Path]:
+        """Get cache file path for a LinkedIn URL"""
+        if not self.cache_dir:
+            return None
+        
+        # Create a safe filename from the LinkedIn URL
+        safe_name = linkedin_url.replace('https://', '').replace('/', '_').replace(':', '_')
+        return self.cache_dir / f"{safe_name}.json"
+    
+    def _load_from_cache(self, linkedin_url: str) -> Optional[Dict]:
+        """Load profile data from cache if available and not expired"""
+        cache_path = self._get_cache_path(linkedin_url)
+        if not cache_path or not cache_path.exists():
+            return None
+        
+        try:
+            with open(cache_path, 'r') as f:
+                cached_data = json.load(f)
+            
+            # Check if cache is still valid (30 days)
+            cache_time = cached_data.get('cached_at', 0)
+            if time.time() - cache_time < (30 * 24 * 60 * 60):  # 30 days in seconds
+                return cached_data.get('data')
+        except Exception as e:
+            print(f"Error loading cache: {e}")
+        
+        return None
+    
+    def _save_to_cache(self, linkedin_url: str, data: Dict) -> None:
+        """Save profile data to cache"""
+        cache_path = self._get_cache_path(linkedin_url)
+        if not cache_path:
+            return
+        
+        try:
+            cached_data = {
+                'cached_at': time.time(),
+                'url': linkedin_url,
+                'data': data
+            }
+            with open(cache_path, 'w') as f:
+                json.dump(cached_data, f, indent=2)
+        except Exception as e:
+            print(f"Error saving cache: {e}")
+    
+    def fetch_profile(self, linkedin_url: str, use_cache: bool = True) -> Dict[str, Any]:
+        """Fetch LinkedIn profile data
+        
+        Args:
+            linkedin_url: Full LinkedIn profile URL
+            use_cache: Whether to use cached data if available
+        
+        Returns:
+            Dictionary containing profile data or error information
+        """
+        # Check cache first
+        if use_cache:
+            cached = self._load_from_cache(linkedin_url)
+            if cached:
+                return {
+                    'success': True,
+                    'data': cached,
+                    'from_cache': True
+                }
+        
+        # Check if API key is configured
+        if not self.api_key:
+            return {
+                'success': False,
+                'error': 'RapidAPI key not configured',
+                'message': 'Please add RAPIDAPI_KEY to Space Secrets'
+            }
+        
+        # Fetch from API
+        try:
+            querystring = {"linkedin_url": linkedin_url}
+            response = requests.get(
+                f"{self.base_url}/get-linkedin-profile",
+                headers=self.headers,
+                params=querystring,
+                timeout=30
+            )
+            
+            if response.status_code == 200:
+                data = response.json()
+                
+                # Save to cache
+                if use_cache:
+                    self._save_to_cache(linkedin_url, data)
+                
+                return {
+                    'success': True,
+                    'data': data,
+                    'from_cache': False
+                }
+            elif response.status_code == 429:
+                return {
+                    'success': False,
+                    'error': 'Rate limit exceeded',
+                    'message': 'You have reached your API rate limit. Consider upgrading or waiting.'
+                }
+            else:
+                return {
+                    'success': False,
+                    'error': f'API error: {response.status_code}',
+                    'message': response.text
+                }
+        
+        except requests.exceptions.Timeout:
+            return {
+                'success': False,
+                'error': 'Request timeout',
+                'message': 'The API request timed out. Please try again.'
+            }
+        except Exception as e:
+            return {
+                'success': False,
+                'error': 'Fetch error',
+                'message': str(e)
+            }
+    
+    def enrich_lead(self, lead_data: Dict) -> Dict:
+        """Enrich lead data with LinkedIn profile information
+        
+        Args:
+            lead_data: Dictionary containing lead information with 'linkedin_url' key
+        
+        Returns:
+            Enriched lead data dictionary
+        """
+        linkedin_url = lead_data.get('linkedin_url', '')
+        if not linkedin_url:
+            return lead_data
+        
+        # Fetch LinkedIn profile
+        result = self.fetch_profile(linkedin_url)
+        
+        if result['success']:
+            profile = result['data']
+            
+            # Add LinkedIn data to lead
+            enriched = lead_data.copy()
+            enriched['linkedin_profile'] = {
+                'full_name': profile.get('full_name', ''),
+                'headline': profile.get('headline', ''),
+                'location': profile.get('location', ''),
+                'summary': profile.get('summary', ''),
+                'experiences': profile.get('experiences', []),
+                'education': profile.get('education', []),
+                'skills': profile.get('skills', []),
+                'from_cache': result.get('from_cache', False)
+            }
+            enriched['linkedin_enriched'] = True
+            enriched['linkedin_enriched_at'] = time.time()
+        else:
+            # Add error info
+            enriched = lead_data.copy()
+            enriched['linkedin_enriched'] = False
+            enriched['linkedin_error'] = result.get('error', 'Unknown error')
+        
+        return enriched
+    
+    def get_cache_stats(self) -> Dict:
+        """Get cache statistics
+        
+        Returns:
+            Dictionary with cache stats
+        """
+        if not self.cache_dir or not self.cache_dir.exists():
+            return {'cached_profiles': 0}
+        
+        cached_files = list(self.cache_dir.glob('*.json'))
+        return {
+            'cached_profiles': len(cached_files),
+            'cache_directory': str(self.cache_dir)
+        }
+
+
+# Convenience functions for easy import
+def fetch_linkedin_profile(linkedin_url: str, api_key: Optional[str] = None, 
+                          username: Optional[str] = None, use_cache: bool = True) -> Dict:
+    """Fetch LinkedIn profile data (convenience function)
+    
+    Args:
+        linkedin_url: Full LinkedIn profile URL
+        api_key: RapidAPI key (optional, uses env var if not provided)
+        username: Current user's username for caching
+        use_cache: Whether to use cached data
+    
+    Returns:
+        Dictionary with profile data or error information
+    """
+    fetcher = LinkedInDataFetcher(api_key=api_key, username=username)
+    return fetcher.fetch_profile(linkedin_url, use_cache=use_cache)
+
+
+def enrich_lead(lead_data: Dict, api_key: Optional[str] = None, 
+               username: Optional[str] = None) -> Dict:
+    """Enrich lead data with LinkedIn information (convenience function)
+    
+    Args:
+        lead_data: Lead dictionary with 'linkedin_url' key
+        api_key: RapidAPI key (optional)
+        username: Current user's username for caching
+    
+    Returns:
+        Enriched lead data dictionary
+    """
+    fetcher = LinkedInDataFetcher(api_key=api_key, username=username)
+    return fetcher.enrich_lead(lead_data)
+
+
+def get_cache_stats(username: str) -> Dict:
+    """Get LinkedIn cache statistics (convenience function)
+    
+    Args:
+        username: Username to check cache for
+    
+    Returns:
+        Dictionary with cache statistics
+    """
+    fetcher = LinkedInDataFetcher(username=username)
+    return fetcher.get_cache_stats()