import gradio as gr import requests import json import pandas as pd import os from typing import Dict, List, Optional import time import random # ========================================== # APOLLO.IO API CLIENT CLASS # ========================================== class ApolloDataFetcher: """ Apollo.io API client for fetching people and company data """ def __init__(self, api_key: str = None): # Get API key from environment (Hugging Face secrets) or parameter self.api_key = api_key or os.getenv("APOLLO_API_KEY") self.base_url = "https://api.apollo.io/v1" self.headers = { "Content-Type": "application/json", "Cache-Control": "no-cache" } if self.api_key and self.api_key != "demo": self.headers["X-Api-Key"] = self.api_key # ========================================== # PEOPLE SEARCH METHOD # ========================================== def search_people(self, query: str, limit: int = 10) -> Dict: """Search for people using Apollo.io API""" if not self.api_key or self.api_key == "demo": return self._generate_demo_people_data(query, limit) endpoint = f"{self.base_url}/people/search" payload = { "q_keywords": query, "page": 1, "per_page": min(limit, 25), "person_locations": ["United States"], } try: response = requests.post(endpoint, headers=self.headers, json=payload, timeout=10) return self._handle_api_response(response) except requests.exceptions.Timeout: return {"error": "Request timeout. Please try again."} except Exception as e: return {"error": f"Request failed: {str(e)}"} # ========================================== # COMPANY SEARCH METHOD # ========================================== def search_companies(self, query: str, limit: int = 10) -> Dict: """Search for companies using Apollo.io API""" if not self.api_key or self.api_key == "demo": return self._generate_demo_company_data(query, limit) endpoint = f"{self.base_url}/organizations/search" payload = { "q_keywords": query, "page": 1, "per_page": min(limit, 25), "organization_locations": ["United States"], } try: response = requests.post(endpoint, headers=self.headers, json=payload, timeout=10) return self._handle_api_response(response) except requests.exceptions.Timeout: return {"error": "Request timeout. Please try again."} except Exception as e: return {"error": f"Request failed: {str(e)}"} # ========================================== # CONTACT ENRICHMENT METHOD # ========================================== def enrich_person(self, email: str) -> Dict: """Enrich person data by email""" if not self.api_key or self.api_key == "demo": return self._generate_demo_enriched_person(email) endpoint = f"{self.base_url}/people/match" payload = {"email": email} try: response = requests.post(endpoint, headers=self.headers, json=payload, timeout=10) return self._handle_api_response(response) except requests.exceptions.Timeout: return {"error": "Request timeout. Please try again."} except Exception as e: return {"error": f"Request failed: {str(e)}"} # ========================================== # EMAIL FINDER METHOD # ========================================== def find_email(self, first_name: str, last_name: str, domain: str) -> Dict: """Find email for a person""" if not self.api_key or self.api_key == "demo": return {"email": f"{first_name.lower()}.{last_name.lower()}@{domain}"} endpoint = f"{self.base_url}/email_accounts" payload = { "first_name": first_name, "last_name": last_name, "domain": domain } try: response = requests.post(endpoint, headers=self.headers, json=payload, timeout=10) return self._handle_api_response(response) except requests.exceptions.Timeout: return {"error": "Request timeout. Please try again."} except Exception as e: return {"error": f"Request failed: {str(e)}"} # ========================================== # API RESPONSE HANDLER # ========================================== def _handle_api_response(self, response) -> Dict: """Handle API response with proper error codes""" if response.status_code == 200: return response.json() elif response.status_code == 401: return {"error": "Invalid API key. Please check your Apollo.io API key."} elif response.status_code == 403: return {"error": "Access denied. Your API key may not have permission for this endpoint. Try upgrading your Apollo.io plan."} elif response.status_code == 429: return {"error": "Rate limit exceeded. Please try again later."} else: return {"error": f"API Error: {response.status_code} - {response.text}"} # ========================================== # DEMO DATA GENERATORS # ========================================== def _generate_demo_people_data(self, query: str, limit: int) -> Dict: """Generate realistic demo data for people search""" demo_people = [] job_titles = [ "Software Engineer", "Marketing Manager", "Sales Director", "CEO", "CTO", "VP Sales", "Product Manager", "Data Scientist", "UX Designer", "DevOps Engineer" ] companies = [ "TechCorp", "InnovateLtd", "DataSystems", "CloudNine", "AIStartup", "FinTechPro", "HealthTech", "EduSolutions", "GreenTech", "CyberSec" ] domains = [ "techcorp.com", "innovate.ltd", "datasys.com", "cloudnine.io", "aistartup.ai", "fintech.pro", "healthtech.com", "edusol.com", "greentech.co", "cybersec.net" ] first_names = ["John", "Jane", "Michael", "Sarah", "David", "Emily", "Robert", "Lisa", "James", "Maria"] last_names = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis", "Rodriguez", "Martinez"] for i in range(min(limit, 10)): first_name = random.choice(first_names) last_name = random.choice(last_names) name = f"{first_name} {last_name}" company = random.choice(companies) domain = random.choice(domains) person = { "id": f"demo_{i+1}", "first_name": first_name, "last_name": last_name, "name": name, "title": random.choice(job_titles), "email": f"{first_name.lower()}.{last_name.lower()}@{domain}", "phone": f"+1 (555) {random.randint(100, 999)}-{random.randint(1000, 9999)}", "organization": { "name": company, "website_url": f"https://{domain}" }, "linkedin_url": f"https://linkedin.com/in/{first_name.lower()}-{last_name.lower()}", "city": random.choice(["San Francisco", "New York", "Austin", "Seattle", "Boston", "Los Angeles", "Chicago", "Miami"]), "state": random.choice(["CA", "NY", "TX", "WA", "MA", "IL", "FL"]), "country": "US" } demo_people.append(person) return { "people": demo_people, "pagination": {"page": 1, "per_page": limit, "total_entries": len(demo_people)} } def _generate_demo_company_data(self, query: str, limit: int) -> Dict: """Generate realistic demo data for company search""" demo_companies = [] industries = [ "Technology", "Healthcare", "Finance", "Education", "E-commerce", "Manufacturing", "Consulting", "Media", "Biotechnology", "Renewable Energy" ] company_suffixes = ["Corp", "Inc", "LLC", "Ltd", "Technologies", "Solutions", "Systems", "Labs", "Ventures", "Group"] for i in range(min(limit, 10)): if query and query.strip(): company_name = f"{query.title()} {random.choice(company_suffixes)}" else: company_name = f"Demo{random.choice(company_suffixes)} {i+1}" domain = f"{company_name.lower().replace(' ', '').replace('corp', '').replace('inc', '').replace('llc', '').replace('ltd', '')}.com" company = { "id": f"company_demo_{i+1}", "name": company_name, "website_url": f"https://{domain}", "industry": random.choice(industries), "employees_range": random.choice(["1-10", "11-50", "51-200", "201-500", "501-1000", "1000+", "2000+", "5000+"]), "estimated_num_employees": random.randint(10, 5000), "city": random.choice(["San Francisco", "New York", "Austin", "Seattle", "Boston", "Los Angeles", "Chicago", "Miami", "Denver", "Atlanta"]), "state": random.choice(["CA", "NY", "TX", "WA", "MA", "IL", "FL", "CO", "GA"]), "country": "US", "phone": f"+1 (555) {random.randint(100, 999)}-{random.randint(1000, 9999)}", "founded_year": random.randint(1990, 2023), "description": f"A leading {random.choice(industries).lower()} company focused on innovation and growth in the modern business landscape." } demo_companies.append(company) return { "organizations": demo_companies, "pagination": {"page": 1, "per_page": limit, "total_entries": len(demo_companies)} } def _generate_demo_enriched_person(self, email: str) -> Dict: """Generate demo enriched person data""" name_part = email.split('@')[0].replace('.', ' ').replace('_', ' ').title() domain = email.split('@')[1] first_name = name_part.split()[0] if name_part.split() else "John" last_name = name_part.split()[1] if len(name_part.split()) > 1 else "Doe" return { "person": { "id": "enriched_demo_1", "first_name": first_name, "last_name": last_name, "name": f"{first_name} {last_name}", "title": random.choice(["Senior Software Engineer", "Marketing Manager", "Sales Director", "Product Manager", "Data Scientist"]), "email": email, "phone": f"+1 (555) {random.randint(100, 999)}-{random.randint(1000, 9999)}", "organization": { "name": domain.split('.')[0].title() + " Corp", "website_url": f"https://{domain}" }, "linkedin_url": f"https://linkedin.com/in/{first_name.lower()}-{last_name.lower()}", "twitter_url": f"https://twitter.com/{first_name.lower()}{last_name.lower()}", "city": random.choice(["San Francisco", "New York", "Austin", "Seattle", "Boston"]), "state": random.choice(["CA", "NY", "TX", "WA", "MA"]), "country": "US", "employment_history": [ {"title": "Software Engineer", "organization_name": "Previous Corp", "start_date": "2020-01"}, {"title": "Junior Developer", "organization_name": "Startup Inc", "start_date": "2018-06"} ] } } # ========================================== # DATA FORMATTING FUNCTIONS # ========================================== def format_people_results(results: Dict) -> tuple: """Format people search results for display""" if "error" in results: return f"āŒ Error: {results['error']}", None # Handle both 'people' and 'data' keys (different API versions) people_data = results.get("people", results.get("data", [])) if not people_data: return "No results found.", None # Create formatted text output output_text = f"āœ… Found {len(people_data)} people:\n\n" # Create DataFrame for table table_data = [] for person in people_data: # Handle nested organization data org = person.get('organization', {}) or person.get('current_organization', {}) # Format text output name = person.get('name', f"{person.get('first_name', '')} {person.get('last_name', '')}").strip() output_text += f"šŸ‘¤ **{name or 'N/A'}**\n" output_text += f" šŸ“§ Email: {person.get('email', 'N/A')}\n" output_text += f" šŸ“ž Phone: {person.get('phone', person.get('personal_phone', 'N/A'))}\n" output_text += f" šŸ’¼ Title: {person.get('title', 'N/A')}\n" output_text += f" šŸ¢ Company: {org.get('name', 'N/A')}\n" output_text += f" šŸ“ Location: {person.get('city', 'N/A')}, {person.get('state', 'N/A')}\n" output_text += f" šŸ”— LinkedIn: {person.get('linkedin_url', 'N/A')}\n\n" # Add to table data table_data.append({ "Name": name or 'N/A', "Email": person.get('email', 'N/A'), "Phone": person.get('phone', person.get('personal_phone', 'N/A')), "Title": person.get('title', 'N/A'), "Company": org.get('name', 'N/A'), "Location": f"{person.get('city', 'N/A')}, {person.get('state', 'N/A')}" }) df = pd.DataFrame(table_data) return output_text, df def format_company_results(results: Dict) -> tuple: """Format company search results for display""" if "error" in results: return f"āŒ Error: {results['error']}", None # Handle both 'organizations' and 'data' keys companies_data = results.get("organizations", results.get("data", [])) if not companies_data: return "No results found.", None # Create formatted text output output_text = f"āœ… Found {len(companies_data)} companies:\n\n" # Create DataFrame for table table_data = [] for company in companies_data: # Format text output output_text += f"šŸ¢ **{company.get('name', 'N/A')}**\n" output_text += f" 🌐 Website: {company.get('website_url', 'N/A')}\n" output_text += f" šŸ­ Industry: {company.get('industry', company.get('primary_industry', 'N/A'))}\n" output_text += f" šŸ‘„ Employees: {company.get('employees_range', company.get('estimated_num_employees', 'N/A'))}\n" output_text += f" šŸ“ž Phone: {company.get('phone', 'N/A')}\n" output_text += f" šŸ“ Location: {company.get('city', 'N/A')}, {company.get('state', 'N/A')}\n" output_text += f" šŸ“… Founded: {company.get('founded_year', 'N/A')}\n\n" # Add to table data table_data.append({ "Company": company.get('name', 'N/A'), "Website": company.get('website_url', 'N/A'), "Industry": company.get('industry', company.get('primary_industry', 'N/A')), "Employees": str(company.get('employees_range', company.get('estimated_num_employees', 'N/A'))), "Location": f"{company.get('city', 'N/A')}, {company.get('state', 'N/A')}", "Founded": str(company.get('founded_year', 'N/A')) }) df = pd.DataFrame(table_data) return output_text, df # ========================================== # INTERFACE FUNCTIONS # ========================================== def search_people_interface(query: str, limit: int): """Interface function for people search""" if not query.strip(): return "Please enter a search query.", None results = apollo_client.search_people(query, limit) return format_people_results(results) def search_companies_interface(query: str, limit: int): """Interface function for company search""" if not query.strip(): return "Please enter a search query.", None results = apollo_client.search_companies(query, limit) return format_company_results(results) def find_email_interface(first_name: str, last_name: str, domain: str): """Interface function for email finding""" if not all([first_name.strip(), last_name.strip(), domain.strip()]): return "Please enter first name, last name, and company domain." results = apollo_client.find_email(first_name, last_name, domain) if "error" in results: return f"āŒ Error: {results['error']}" if "email" in results: output = f"šŸ“§ **Email Found**\n\n" output += f"šŸ‘¤ **Name:** {first_name} {last_name}\n" output += f"šŸ“§ **Email:** {results['email']}\n" output += f"šŸ¢ **Domain:** {domain}\n" if results.get('confidence'): output += f"šŸŽÆ **Confidence:** {results['confidence']}\n" return output return "No email found for this person and domain combination." def enrich_person_interface(email: str): """Interface function for person enrichment""" if not email.strip() or "@" not in email: return "Please enter a valid email address." results = apollo_client.enrich_person(email) if "error" in results: return f"āŒ Error: {results['error']}" # Handle both 'person' and 'data' keys person = results.get("person", results.get("data")) if not person: return "No person data found for this email." output = f"šŸ” **Enriched Person Data**\n\n" # Handle nested organization data org = person.get('organization', {}) or person.get('current_organization', {}) name = person.get('name', f"{person.get('first_name', '')} {person.get('last_name', '')}").strip() output += f"šŸ‘¤ **Name:** {name or 'N/A'}\n" output += f"šŸ“§ **Email:** {person.get('email', 'N/A')}\n" output += f"šŸ“ž **Phone:** {person.get('phone', person.get('personal_phone', 'N/A'))}\n" output += f"šŸ’¼ **Title:** {person.get('title', 'N/A')}\n" output += f"šŸ¢ **Company:** {org.get('name', 'N/A')}\n" output += f"🌐 **Company Website:** {org.get('website_url', 'N/A')}\n" output += f"šŸ“ **Location:** {person.get('city', 'N/A')}, {person.get('state', 'N/A')}, {person.get('country', 'N/A')}\n" output += f"šŸ”— **LinkedIn:** {person.get('linkedin_url', 'N/A')}\n" output += f"🐦 **Twitter:** {person.get('twitter_url', 'N/A')}\n" if person.get('employment_history'): output += f"\nšŸ“‹ **Employment History:**\n" for job in person['employment_history']: output += f" • {job.get('title', 'N/A')} at {job.get('organization_name', 'N/A')} (from {job.get('start_date', 'N/A')})\n" return output # ========================================== # INITIALIZE APOLLO CLIENT # ========================================== apollo_client = ApolloDataFetcher() # Check if API key is available api_key_status = "šŸ”‘ Live API Mode" if apollo_client.api_key and apollo_client.api_key != "demo" else "šŸŽ­ Demo Mode" # ========================================== # GRADIO INTERFACE # ========================================== with gr.Blocks(title="Apollo.io Data Fetcher", theme=gr.themes.Soft()) as demo: # ========================================== # HEADER SECTION # ========================================== gr.HTML(f"""

šŸš€ Apollo.io Data Fetcher

Search for people and companies, enrich contact data using Apollo.io's powerful database

Status: {api_key_status}
{"āœ… Using real Apollo.io API" if apollo_client.api_key and apollo_client.api_key != "demo" else "šŸŽ­ Showing demo data - Add APOLLO_API_KEY secret for live data"}
""") # ========================================== # MAIN TABS # ========================================== with gr.Tabs(): # ========================================== # PEOPLE SEARCH TAB # ========================================== with gr.TabItem("šŸ‘„ Search People"): with gr.Row(): with gr.Column(scale=2): people_query = gr.Textbox( label="Search Query", placeholder="e.g., 'software engineer', 'John Smith', 'marketing manager'", info="Search by name, job title, or keywords" ) with gr.Column(scale=1): people_limit = gr.Slider( minimum=1, maximum=25, value=10, step=1, label="Number of Results" ) people_search_btn = gr.Button("šŸ” Search People", variant="primary", size="lg") with gr.Row(): with gr.Column(): people_output = gr.Markdown(label="Search Results") with gr.Column(): people_table = gr.Dataframe(label="Results Table", wrap=True) people_search_btn.click( search_people_interface, inputs=[people_query, people_limit], outputs=[people_output, people_table] ) # ========================================== # COMPANIES SEARCH TAB # ========================================== with gr.TabItem("šŸ¢ Search Companies"): with gr.Row(): with gr.Column(scale=2): company_query = gr.Textbox( label="Search Query", placeholder="e.g., 'technology', 'Apple', 'healthcare startup'", info="Search by company name, industry, or keywords" ) with gr.Column(scale=1): company_limit = gr.Slider( minimum=1, maximum=25, value=10, step=1, label="Number of Results" ) company_search_btn = gr.Button("šŸ” Search Companies", variant="primary", size="lg") with gr.Row(): with gr.Column(): company_output = gr.Markdown(label="Search Results") with gr.Column(): company_table = gr.Dataframe(label="Results Table", wrap=True) company_search_btn.click( search_companies_interface, inputs=[company_query, company_limit], outputs=[company_output, company_table] ) # ========================================== # EMAIL FINDER TAB # ========================================== with gr.TabItem("šŸ“§ Find Email"): gr.Markdown("### Find email addresses for specific people") with gr.Row(): with gr.Column(): first_name_input = gr.Textbox( label="First Name", placeholder="e.g., John", info="Person's first name" ) with gr.Column(): last_name_input = gr.Textbox( label="Last Name", placeholder="e.g., Smith", info="Person's last name" ) with gr.Column(): domain_input = gr.Textbox( label="Company Domain", placeholder="e.g., apple.com", info="Company website domain (without https://)" ) find_email_btn = gr.Button("šŸ“§ Find Email", variant="primary", size="lg") email_finder_output = gr.Markdown(label="Email Results") find_email_btn.click( find_email_interface, inputs=[first_name_input, last_name_input, domain_input], outputs=[email_finder_output] ) # ========================================== # PERSON ENRICHMENT TAB # ========================================== with gr.TabItem("šŸ” Enrich Person"): email_input = gr.Textbox( label="Email Address", placeholder="e.g., john.doe@company.com", info="Enter an email address to get enriched contact information" ) enrich_btn = gr.Button("šŸ” Enrich Contact", variant="primary", size="lg") enrich_output = gr.Markdown(label="Enriched Data") enrich_btn.click( enrich_person_interface, inputs=[email_input], outputs=[enrich_output] ) # ========================================== # SETUP INSTRUCTIONS TAB # ========================================== with gr.TabItem("āš™ļø Setup & Troubleshooting"): gr.Markdown(""" ## šŸ”§ How to Set Up Apollo.io API on Hugging Face ### Step 1: Get Your Apollo.io API Key 1. Sign up at [Apollo.io](https://apollo.io) 2. Go to **Settings > Integrations** 3. Click **"Create API Key"** 4. Set appropriate permissions 5. Copy your API key ### Step 2: Add API Key to Hugging Face Space 1. Go to your Space settings 2. Click on **"Repository secrets"** 3. Add a new secret: - **Name**: `APOLLO_API_KEY` - **Value**: Your Apollo.io API key 4. Save the secret 5. Restart your Space ### Step 3: Verify Setup - If API key is configured correctly, you'll see "šŸ”‘ Live API Mode" at the top - If not configured, you'll see "šŸŽ­ Demo Mode" with sample data ## 🚨 Common API Access Issues & Solutions ### 403 Error - Access Denied **Problem**: Your API key doesn't have access to certain endpoints. **Solutions**: 1. **Try Email Finder**: Often more accessible than people/company search 2. **Check API Plan**: Some features require paid Apollo.io plans 3. **Verify Permissions**: Ensure your API key has the right permissions 4. **Use Contact Enrichment**: Usually available on all plans ### Available Features by Plan: #### šŸ†“ Free Plan - Contact enrichment (limited) - Email finder (limited) - Basic API access #### šŸ’° Paid Plans - Full people search - Company search - Unlimited contact enrichment - Higher rate limits ### Rate Limits - **Free**: 50 requests/month - **Starter**: 1,000+ requests/month - **Professional**: 5,000+ requests/month - **Organization**: Custom limits ## šŸ”’ Security Features - API keys with granular permissions - HTTPS encryption for all requests - SOC 2 Type II compliant - Secure secret management via Hugging Face ## šŸ“Š Apollo.io Database Stats - **275+ million contacts** worldwide - **70+ million companies** across industries - Global B2B coverage with detailed information - Regularly updated and verified data ## šŸ› ļø Technical Details - Built with Gradio for interactive web interface - Uses Apollo.io REST API for data retrieval - Pandas for data processing and table display - Error handling and rate limiting awareness ## 🚨 Troubleshooting Guide #### Issue: "API Error: 403 - Access Denied" **Solution**: Your API key lacks permissions. Try: - Upgrading your Apollo.io plan - Using Email Finder instead of People Search - Checking API key permissions in Apollo.io #### Issue: "Invalid API key" **Solution**: - Check if APOLLO_API_KEY secret is set correctly in Space settings - Verify the API key is copied correctly from Apollo.io - Restart your Space after adding the secret #### Issue: "Rate limit exceeded" **Solution**: - Wait for the rate limit to reset - Upgrade your Apollo.io plan for higher limits - Use demo mode for testing #### Issue: "No results found" **Solution**: - Try broader search terms - Check spelling and formatting - Use different keywords or filters #### Issue: Demo mode always active **Solution**: - Ensure APOLLO_API_KEY is added to repository secrets - Check the secret name is exactly: `APOLLO_API_KEY` - Restart the Space after adding secrets ## šŸ’” Best Practices 1. **Start with Demo Mode**: Test the interface before adding API key 2. **Monitor Usage**: Check Apollo.io dashboard for API usage 3. **Use Specific Searches**: More specific queries return better results 4. **Export Data**: Use table view to copy/export results 5. **Respect Rate Limits**: Don't exceed your plan's limits ## šŸ“ž Support Resources - **Apollo.io API Docs**: [apolloio.github.io/apollo-api-docs](https://apolloio.github.io/apollo-api-docs/) - **Apollo.io Support**: Contact through their platform - **Hugging Face Docs**: [huggingface.co/docs/hub/spaces-overview](https://huggingface.co/docs/hub/spaces-overview) """) # ========================================== # FOOTER # ========================================== gr.HTML("""

šŸš€ Built with Apollo.io API & Gradio | API Docs | HF Spaces

""") # ========================================== # LAUNCH APPLICATION # ========================================== if __name__ == "__main__": demo.launch()