Update app.py

app.py

@@ -3,6 +3,7 @@ import requests
 import json
 from typing import List, Dict, Any
 import pandas as pd
+from urllib.parse import urlparse
 
 # Hardcoded API endpoint - Updated to correct endpoint
 API_ENDPOINT = "https://jaybene1-testapicontacts.hf.space/contacts"
@@ -34,6 +35,107 @@ theme = gr.themes.Soft(
     button_secondary_text_color="#334155"
 )
 
+def normalize_url(url: str) -> str:
+    """
+    Normalize URL for consistent comparison.
+    
+    Args:
+        url: The URL to normalize
+        
+    Returns:
+        Normalized URL string
+    """
+    if not url:
+        return ""
+    
+    # Remove protocol and www
+    normalized = url.lower().replace('https://', '').replace('http://', '').replace('www.', '')
+    
+    # Remove trailing slash and path
+    normalized = normalized.split('/')[0]
+    
+    # Remove port numbers
+    normalized = normalized.split(':')[0]
+    
+    return normalized
+
+def extract_domain_variations(url: str) -> List[str]:
+    """
+    Extract various domain variations for matching.
+    
+    Args:
+        url: The URL to extract domains from
+        
+    Returns:
+        List of domain variations
+    """
+    normalized = normalize_url(url)
+    variations = [normalized]
+    
+    # Add with www
+    variations.append(f"www.{normalized}")
+    
+    # Add subdomains if any
+    if '.' in normalized:
+        parts = normalized.split('.')
+        if len(parts) > 2:
+            # Add without subdomain
+            variations.append('.'.join(parts[-2:]))
+    
+    return variations
+
+def filter_contacts_by_url(contacts, search_url):
+    """
+    Enhanced filter contacts based on the search URL.
+    This is a client-side filter in case the API doesn't filter properly.
+    """
+    if not isinstance(contacts, list):
+        return contacts
+    
+    # Get domain variations for the search URL
+    search_domains = extract_domain_variations(search_url)
+    
+    filtered_contacts = []
+    for contact in contacts:
+        if isinstance(contact, dict):
+            # Check various fields that might contain the domain
+            fields_to_check = [
+                contact.get('website', ''),
+                contact.get('company_website', ''),
+                contact.get('domain', ''),
+                contact.get('email', ''),
+                contact.get('company', ''),
+                str(contact.get('url', '')),
+                str(contact.get('source', '')),
+                str(contact.get('company_url', '')),
+                str(contact.get('origin_url', ''))
+            ]
+            
+            # Check if any field contains any of the search domains
+            contact_matched = False
+            for field in fields_to_check:
+                if field:
+                    field_normalized = normalize_url(str(field))
+                    for search_domain in search_domains:
+                        if search_domain in field_normalized or field_normalized in search_domain:
+                            filtered_contacts.append(contact)
+                            contact_matched = True
+                            break
+                    if contact_matched:
+                        break
+            
+            # Also check email domains
+            if not contact_matched:
+                email = contact.get('email', '')
+                if email and '@' in email:
+                    email_domain = email.split('@')[1].lower()
+                    for search_domain in search_domains:
+                        if search_domain in email_domain or email_domain in search_domain:
+                            filtered_contacts.append(contact)
+                            break
+    
+    return filtered_contacts
+
 def search_contacts(url: str) -> tuple[str, str]:
     """
     Search for contacts associated with a given URL using the Hugging Face API.
@@ -50,6 +152,7 @@ def search_contacts(url: str) -> tuple[str, str]:
     
     # Clean and validate URL
     url = url.strip()
+    original_url = url
     if not url.startswith(('http://', 'https://')):
         url = 'https://' + url
     
@@ -70,9 +173,6 @@ def search_contacts(url: str) -> tuple[str, str]:
             try:
                 result = response.json()
                 
-                # Add debugging info
-                request_url = f"{API_ENDPOINT}?url={url}"
-                
                 # Hugging Face Spaces API returns data in a different format
                 if isinstance(result, dict) and 'data' in result:
                     actual_result = result['data'][0] if result['data'] else {}
@@ -81,19 +181,51 @@ def search_contacts(url: str) -> tuple[str, str]:
                 
                 # Apply client-side filtering
                 if isinstance(actual_result, list):
-                    actual_result = filter_contacts_by_url(actual_result, url)
+                    original_count = len(actual_result)
+                    filtered_result = filter_contacts_by_url(actual_result, url)
+                    
+                    # If we filtered out everything or got very few results compared to original,
+                    # it's likely the API returned all contacts instead of URL-specific ones
+                    if len(filtered_result) == 0:
+                        return f"❌ No contacts found for: {original_url}\n\nThis URL may not be in our database.", json.dumps(result, indent=2)
+                    elif original_count > 50 and len(filtered_result) < 5:
+                        # Likely got all contacts, filtered to very few - show warning
+                        formatted_output = f"⚠️ **Warning:** API returned {original_count} total contacts. After filtering for '{original_url}', found {len(filtered_result)} matches.\n\n"
+                        formatted_output += "These results may not be accurate. The URL might not be in our database.\n\n"
+                        formatted_output += format_contact_results(filtered_result, url)
+                    else:
+                        formatted_output = format_contact_results(filtered_result, url)
+                    
+                    actual_result = filtered_result
+                    
                 elif isinstance(actual_result, dict) and 'contacts' in actual_result:
-                    actual_result['contacts'] = filter_contacts_by_url(actual_result['contacts'], url)
-                
-                # Format the results for display
-                formatted_output = format_contact_results(actual_result, url)
+                    original_count = len(actual_result['contacts'])
+                    filtered_contacts = filter_contacts_by_url(actual_result['contacts'], url)
+                    
+                    if len(filtered_contacts) == 0:
+                        return f"❌ No contacts found for: {original_url}\n\nThis URL may not be in our database.", json.dumps(result, indent=2)
+                    elif original_count > 50 and len(filtered_contacts) < 5:
+                        formatted_output = f"⚠️ **Warning:** API returned {original_count} total contacts. After filtering for '{original_url}', found {len(filtered_contacts)} matches.\n\n"
+                        formatted_output += "These results may not be accurate. The URL might not be in our database.\n\n"
+                        formatted_output += format_contact_results({'contacts': filtered_contacts}, url)
+                    else:
+                        formatted_output = format_contact_results({'contacts': filtered_contacts}, url)
+                    
+                    actual_result['contacts'] = filtered_contacts
+                else:
+                    # Single contact or unknown format
+                    formatted_output = format_contact_results(actual_result, url)
                 
-                # Add debugging info to the formatted output
+                # Add debugging info
                 debug_info = f"\n\n**Debug Info:**\n"
                 debug_info += f"- Request URL: {API_ENDPOINT}\n"
                 debug_info += f"- Search URL: {url}\n"
-                debug_info += f"- Response contains {len(actual_result) if isinstance(actual_result, list) else 'N/A'} items\n"
-                debug_info += f"- Client-side filtering applied\n"
+                debug_info += f"- Original response size: {len(result.get('data', [result])) if isinstance(result, dict) else len(result) if isinstance(result, list) else 'N/A'}\n"
+                debug_info += f"- Filtered results: {len(actual_result) if isinstance(actual_result, list) else len(actual_result.get('contacts', [])) if isinstance(actual_result, dict) else 'N/A'}\n"
+                debug_info += f"- Client-side filtering applied: Yes\n"
+                
+                if 'formatted_output' not in locals():
+                    formatted_output = format_contact_results(actual_result, url)
                 
                 formatted_output += debug_info
                 raw_json = json.dumps(result, indent=2)
@@ -113,40 +245,6 @@ def search_contacts(url: str) -> tuple[str, str]:
     except Exception as e:
         return f"❌ Error: {str(e)}", ""
 
-def filter_contacts_by_url(contacts, search_url):
-    """
-    Filter contacts based on the search URL.
-    This is a client-side filter in case the API doesn't filter properly.
-    """
-    if not isinstance(contacts, list):
-        return contacts
-    
-    # Extract domain from search URL
-    search_domain = search_url.replace('https://', '').replace('http://', '').replace('www.', '').split('/')[0].lower()
-    
-    filtered_contacts = []
-    for contact in contacts:
-        # Check if contact is associated with the search domain
-        if isinstance(contact, dict):
-            # Check various fields that might contain the domain
-            fields_to_check = [
-                contact.get('website', ''),
-                contact.get('company_website', ''),
-                contact.get('domain', ''),
-                contact.get('email', ''),
-                contact.get('company', ''),
-                str(contact.get('url', '')),
-                str(contact.get('source', ''))
-            ]
-            
-            # Check if any field contains the search domain
-            for field in fields_to_check:
-                if field and search_domain in str(field).lower():
-                    filtered_contacts.append(contact)
-                    break
-    
-    return filtered_contacts if filtered_contacts else contacts
-
 def format_contact_results(results: Dict[Any, Any], url: str) -> str:
     """
     Format the API results into a readable format.
@@ -312,6 +410,7 @@ with gr.Blocks(theme=theme, title="Contact Search - Kwekel Companies", css="""
             <li>View formatted results and raw JSON response</li>
         </ol>
         <p><strong>API Endpoint:</strong> {API_ENDPOINT}</p>
+        <p><strong>Note:</strong> The system will filter results to match your specific URL and warn you if the URL might not be in the database.</p>
     </div>
     """)