Spaces:
Sleeping
Sleeping
| """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() | |