Spaces:
Sleeping
Sleeping
| """ | |
| StrategyStack Multi-API Data Enrichment Module | |
| Integrates multiple RapidAPI sources for comprehensive contact/lead enrichment: | |
| - LinkedIn Data APIs (3 providers) | |
| - Apollo.io | |
| - Social Media (Twitter, Facebook) | |
| - Mediafy API | |
| - Google Search | |
| - LetScrape (Local Business Data + Website Contacts) | |
| Development version with fallback support and multi-provider redundancy. | |
| """ | |
| import os | |
| import requests | |
| import json | |
| import time | |
| from datetime import datetime, timedelta | |
| from pathlib import Path | |
| from typing import Dict, Optional, List, Any | |
| import logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class DataEnrichmentManager: | |
| """ | |
| Manages data enrichment across multiple API providers. | |
| Implements fallback logic, caching, and rate limiting. | |
| """ | |
| def __init__(self, username: str): | |
| self.username = username | |
| self.cache_dir = Path(f"/data/users/{username}/enrichment_cache") | |
| self.cache_dir.mkdir(parents=True, exist_ok=True) | |
| # API Keys from environment | |
| self.api_keys = { | |
| 'linkedin_fresh': os.getenv('RAPIDAPI_KEY_LINKEDIN_FRESH'), | |
| 'linkedin_rockapis': os.getenv('RAPIDAPI_KEY_LINKEDIN_ROCKAPIS'), | |
| 'linkedin_liscraper': os.getenv('RAPIDAPI_KEY_LISCRAPER'), | |
| 'linkedin_scraper': os.getenv('RAPIDAPI_KEY_LINKEDIN_SCRAPER'), | |
| 'apollo': os.getenv('RAPIDAPI_KEY_APOLLO'), | |
| 'mediafy': os.getenv('RAPIDAPI_KEY_MEDIAFY'), | |
| 'twitter': os.getenv('RAPIDAPI_KEY_TWITTER'), | |
| 'facebook': os.getenv('RAPIDAPI_KEY_FACEBOOK'), | |
| 'google': os.getenv('RAPIDAPI_KEY_GOOGLE'), | |
| 'letscrape': os.getenv('RAPIDAPI_KEY_LETSCRAPE') | |
| } | |
| # RapidAPI host configurations | |
| self.api_hosts = { | |
| 'linkedin_fresh': 'fresh-linkedin-profile-data.p.rapidapi.com', | |
| 'linkedin_rockapis': 'linkedin-data-api.p.rapidapi.com', | |
| 'linkedin_liscraper': 'li-data-scraper.p.rapidapi.com', | |
| 'linkedin_scraper': 'linkedin-data-scraper.p.rapidapi.com', | |
| 'apollo': 'apollo-io-no-cookies-required.p.rapidapi.com', | |
| 'mediafy': 'mediafy-api.p.rapidapi.com', | |
| 'twitter': 'twitter241.p.rapidapi.com', | |
| 'facebook': 'facebook-scraper3.p.rapidapi.com', | |
| 'google': 'google-search74.p.rapidapi.com', | |
| 'letscrape_business': 'local-business-data.p.rapidapi.com', | |
| 'letscrape_contacts': 'website-contacts-scraper.p.rapidapi.com' | |
| } | |
| self.cache_ttl = timedelta(days=30) | |
| def _make_request(self, api_name: str, url: str, params: Dict = None, headers: Dict = None) -> Optional[Dict]: | |
| """Make API request with error handling""" | |
| try: | |
| api_key = self.api_keys.get(api_name) or self.api_keys.get('letscrape') | |
| if not api_key: | |
| logger.warning(f"API key not found for {api_name}") | |
| return None | |
| # Determine host | |
| host = self.api_hosts.get(api_name, '') | |
| request_headers = { | |
| 'x-rapidapi-key': api_key, | |
| 'x-rapidapi-host': host | |
| } | |
| if headers: | |
| request_headers.update(headers) | |
| response = requests.get(url, headers=request_headers, params=params, timeout=30) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| logger.error(f"API request failed for {api_name}: {str(e)}") | |
| return None | |
| def _get_cache_path(self, identifier: str, data_type: str) -> Path: | |
| """Get cache file path for identifier""" | |
| safe_id = identifier.replace('/', '_').replace(':', '_') | |
| return self.cache_dir / f"{data_type}_{safe_id}.json" | |
| def _load_from_cache(self, identifier: str, data_type: str) -> Optional[Dict]: | |
| """Load data from cache if valid""" | |
| cache_path = self._get_cache_path(identifier, data_type) | |
| if not cache_path.exists(): | |
| return None | |
| try: | |
| with open(cache_path, 'r') as f: | |
| cached = json.load(f) | |
| cached_time = datetime.fromisoformat(cached.get('cached_at', '2020-01-01')) | |
| if datetime.now() - cached_time < self.cache_ttl: | |
| logger.info(f"Cache hit for {data_type}: {identifier}") | |
| return cached.get('data') | |
| except Exception as e: | |
| logger.error(f"Cache read error: {str(e)}") | |
| return None | |
| def _save_to_cache(self, identifier: str, data_type: str, data: Dict): | |
| """Save data to cache""" | |
| cache_path = self._get_cache_path(identifier, data_type) | |
| try: | |
| cache_data = { | |
| 'cached_at': datetime.now().isoformat(), | |
| 'identifier': identifier, | |
| 'data': data | |
| } | |
| with open(cache_path, 'w') as f: | |
| json.dump(cache_data, f, indent=2) | |
| logger.info(f"Cached {data_type}: {identifier}") | |
| except Exception as e: | |
| logger.error(f"Cache write error: {str(e)}") | |
| # LinkedIn Enrichment Methods | |
| def enrich_linkedin_profile(self, linkedin_url: str) -> Optional[Dict]: | |
| """Enrich LinkedIn profile with fallback across multiple providers""" | |
| cached = self._load_from_cache(linkedin_url, 'linkedin') | |
| if cached: | |
| return cached | |
| providers = ['linkedin_fresh', 'linkedin_rockapis', 'linkedin_liscraper', 'linkedin_scraper'] | |
| for provider in providers: | |
| try: | |
| if provider == 'linkedin_fresh': | |
| url = 'https://fresh-linkedin-profile-data.p.rapidapi.com/enrich-lead' | |
| params = {'linkedin_url': linkedin_url} | |
| elif provider == 'linkedin_rockapis': | |
| url = 'https://linkedin-data-api.p.rapidapi.com/get-profile-data-by-url' | |
| params = {'url': linkedin_url} | |
| elif provider == 'linkedin_liscraper': | |
| url = 'https://li-data-scraper.p.rapidapi.com/get-profile-data-by-url' | |
| params = {'url': linkedin_url} | |
| else: | |
| url = 'https://linkedin-data-scraper.p.rapidapi.com/person' | |
| params = {'url': linkedin_url} | |
| data = self._make_request(provider, url, params) | |
| if data: | |
| self._save_to_cache(linkedin_url, 'linkedin', data) | |
| logger.info(f"LinkedIn enrichment successful via {provider}") | |
| return data | |
| except Exception as e: | |
| logger.warning(f"Provider {provider} failed: {str(e)}") | |
| continue | |
| return None | |
| def enrich_company(self, domain: str) -> Optional[Dict]: | |
| """Enrich company data by domain""" | |
| cached = self._load_from_cache(domain, 'company') | |
| if cached: | |
| return cached | |
| url = 'https://linkedin-data-api.p.rapidapi.com/get-company-by-domain' | |
| params = {'domain': domain} | |
| data = self._make_request('linkedin_rockapis', url, params) | |
| if data: | |
| self._save_to_cache(domain, 'company', data) | |
| return data | |
| # Apollo.io Enrichment | |
| def enrich_with_apollo(self, email: str = None, domain: str = None) -> Optional[Dict]: | |
| """Enrich contact using Apollo.io API""" | |
| cache_key = email or domain | |
| cached = self._load_from_cache(cache_key, 'apollo') | |
| if cached: | |
| return cached | |
| url = 'https://apollo-io-no-cookies-required.p.rapidapi.com/enrich' | |
| params = {} | |
| if email: | |
| params['email'] = email | |
| if domain: | |
| params['domain'] = domain | |
| data = self._make_request('apollo', url, params) | |
| if data: | |
| self._save_to_cache(cache_key, 'apollo', data) | |
| return data | |
| # Social Media Enrichment | |
| def enrich_twitter_profile(self, username: str) -> Optional[Dict]: | |
| """Enrich Twitter profile data""" | |
| cached = self._load_from_cache(username, 'twitter') | |
| if cached: | |
| return cached | |
| url = 'https://twitter241.p.rapidapi.com/user' | |
| params = {'username': username} | |
| data = self._make_request('twitter', url, params) | |
| if data: | |
| self._save_to_cache(username, 'twitter', data) | |
| return data | |
| def enrich_facebook_profile(self, profile_id: str) -> Optional[Dict]: | |
| """Enrich Facebook profile data""" | |
| cached = self._load_from_cache(profile_id, 'facebook') | |
| if cached: | |
| return cached | |
| url = 'https://facebook-scraper3.p.rapidapi.com/profile' | |
| params = {'profile_id': profile_id} | |
| data = self._make_request('facebook', url, params) | |
| if data: | |
| self._save_to_cache(profile_id, 'facebook', data) | |
| return data | |
| # Mediafy API | |
| def enrich_with_mediafy(self, query: str) -> Optional[Dict]: | |
| """Enrich with Mediafy media/social data""" | |
| cached = self._load_from_cache(query, 'mediafy') | |
| if cached: | |
| return cached | |
| url = 'https://mediafy-api.p.rapidapi.com/search' | |
| params = {'query': query} | |
| data = self._make_request('mediafy', url, params) | |
| if data: | |
| self._save_to_cache(query, 'mediafy', data) | |
| return data | |
| # Google Search | |
| def search_google(self, query: str, num_results: int = 10) -> Optional[Dict]: | |
| """Perform Google search for contact information""" | |
| cached = self._load_from_cache(query, 'google_search') | |
| if cached: | |
| return cached | |
| url = 'https://google-search74.p.rapidapi.com/' | |
| params = {'query': query, 'limit': str(num_results)} | |
| data = self._make_request('google', url, params) | |
| if data: | |
| self._save_to_cache(query, 'google_search', data) | |
| return data | |
| # LetScrape APIs - Local Business & Website Contacts | |
| def scrape_local_business(self, query: str, location: str = None, limit: int = 20) -> Optional[Dict]: | |
| """ | |
| Scrape local business data from Google Maps. | |
| Returns business information including address, phone, email, ratings, reviews. | |
| """ | |
| cache_key = f"{query}_{location or 'any'}" | |
| cached = self._load_from_cache(cache_key, 'local_business') | |
| if cached: | |
| return cached | |
| url = 'https://local-business-data.p.rapidapi.com/search' | |
| params = {'query': query, 'limit': limit} | |
| if location: | |
| params['subtypes'] = location | |
| data = self._make_request('letscrape_business', url, params) | |
| if data: | |
| self._save_to_cache(cache_key, 'local_business', data) | |
| logger.info(f"Local business data retrieved for: {query}") | |
| return data | |
| def scrape_website_contacts(self, domain: str) -> Optional[Dict]: | |
| """ | |
| Scrape and extract email addresses, phone numbers, and social profile links from a website. | |
| Supports Facebook, Instagram, TikTok, LinkedIn, Twitter, YouTube, Pinterest, Snapchat, GitHub. | |
| """ | |
| cached = self._load_from_cache(domain, 'website_contacts') | |
| if cached: | |
| return cached | |
| url = 'https://website-contacts-scraper.p.rapidapi.com/scrape-contacts' | |
| params = {'query': domain} | |
| data = self._make_request('letscrape_contacts', url, params) | |
| if data: | |
| self._save_to_cache(domain, 'website_contacts', data) | |
| logger.info(f"Website contacts scraped for: {domain}") | |
| return data | |
| # Comprehensive Enrichment | |
| def enrich_contact_full(self, contact_data: Dict) -> Dict: | |
| """ | |
| Perform comprehensive enrichment using all available data sources. | |
| Input contact_data can include: email, linkedin_url, twitter, facebook, domain, name, company | |
| """ | |
| enriched = {'original': contact_data, 'enrichments': {}, 'sources': []} | |
| # LinkedIn enrichment | |
| if contact_data.get('linkedin_url'): | |
| linkedin_data = self.enrich_linkedin_profile(contact_data['linkedin_url']) | |
| if linkedin_data: | |
| enriched['enrichments']['linkedin'] = linkedin_data | |
| enriched['sources'].append('linkedin') | |
| # Apollo enrichment | |
| if contact_data.get('email') or contact_data.get('domain'): | |
| apollo_data = self.enrich_with_apollo( | |
| email=contact_data.get('email'), | |
| domain=contact_data.get('domain') | |
| ) | |
| if apollo_data: | |
| enriched['enrichments']['apollo'] = apollo_data | |
| enriched['sources'].append('apollo') | |
| # Company enrichment | |
| if contact_data.get('domain'): | |
| company_data = self.enrich_company(contact_data['domain']) | |
| if company_data: | |
| enriched['enrichments']['company'] = company_data | |
| enriched['sources'].append('company') | |
| # Website contacts scraping | |
| if contact_data.get('domain'): | |
| website_contacts = self.scrape_website_contacts(contact_data['domain']) | |
| if website_contacts: | |
| enriched['enrichments']['website_contacts'] = website_contacts | |
| enriched['sources'].append('website_contacts') | |
| # Local business data (if company name and location provided) | |
| if contact_data.get('company') and contact_data.get('location'): | |
| business_data = self.scrape_local_business( | |
| query=contact_data['company'], | |
| location=contact_data['location'] | |
| ) | |
| if business_data: | |
| enriched['enrichments']['local_business'] = business_data | |
| enriched['sources'].append('local_business') | |
| # Twitter enrichment | |
| if contact_data.get('twitter_username'): | |
| twitter_data = self.enrich_twitter_profile(contact_data['twitter_username']) | |
| if twitter_data: | |
| enriched['enrichments']['twitter'] = twitter_data | |
| enriched['sources'].append('twitter') | |
| # Facebook enrichment | |
| if contact_data.get('facebook_id'): | |
| facebook_data = self.enrich_facebook_profile(contact_data['facebook_id']) | |
| if facebook_data: | |
| enriched['enrichments']['facebook'] = facebook_data | |
| enriched['sources'].append('facebook') | |
| # Google search for additional info | |
| if contact_data.get('name'): | |
| search_query = f"{contact_data['name']}" | |
| if contact_data.get('company'): | |
| search_query += f" {contact_data['company']}" | |
| google_results = self.search_google(search_query, num_results=5) | |
| if google_results: | |
| enriched['enrichments']['google_search'] = google_results | |
| enriched['sources'].append('google') | |
| enriched['enriched_at'] = datetime.now().isoformat() | |
| enriched['total_sources'] = len(enriched['sources']) | |
| return enriched | |
| def get_enrichment_stats(self) -> Dict: | |
| """Get cache statistics and API usage""" | |
| stats = { | |
| 'cache_directory': str(self.cache_dir), | |
| 'cached_profiles': 0, | |
| 'cache_types': {}, | |
| 'total_api_calls': 0, | |
| 'cache_hits': 0, | |
| 'cache_misses': 0, | |
| 'total_enrichments': 0, | |
| 'successful_enrichments': 0, | |
| 'failed_enrichments': 0 | |
| } | |
| if self.cache_dir.exists(): | |
| cache_files = list(self.cache_dir.glob('*.json')) | |
| stats['cached_profiles'] = len(cache_files) | |
| for cache_file in cache_files: | |
| cache_type = cache_file.stem.split('_')[0] | |
| stats['cache_types'][cache_type] = stats['cache_types'].get(cache_type, 0) + 1 | |
| stats['configured_apis'] = [k for k, v in self.api_keys.items() if v] | |
| stats['total_apis'] = len(stats['configured_apis']) | |
| return stats | |
| # Convenience functions | |
| def create_enrichment_manager(username: str) -> DataEnrichmentManager: | |
| """Create a new enrichment manager instance""" | |
| return DataEnrichmentManager(username) | |
| def enrich_contact(username: str, contact_data: Dict) -> Dict: | |
| """Quick enrichment for a single contact""" | |
| manager = DataEnrichmentManager(username) | |
| return manager.enrich_contact_full(contact_data) | |