"""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()