Tool for scraping contacts

scrapeAPI2.py

@@ -1,82 +1,93 @@
 from fastapi import FastAPI, HTTPException, Query
 from playwright.async_api import async_playwright
 from urllib.parse import urlparse
-from typing import List, Set, Dict
+from typing import List, Set
 import re
 
 app = FastAPI()
 
-contact_info_pattern = {
-    "email": re.compile(r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+'),
-    "phone": re.compile(r'(\+\d{1,3}[- ]?)?\(?\d{2,4}\)?[-.\s]?\d{3}[-.\s]?\d{4,6}'),
-    "social": re.compile(r'https?://(www\.)?(twitter|linkedin|facebook|instagram)\.com/[^\s"\'<>]+')
-}
+visited_links: Set[str] = set()
+
+# Improved regex patterns
+email_pattern = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+")
+phone_pattern = re.compile(r"\+?\d[\d\s().-]{7,}\d")
+social_pattern = re.compile(r"https?://(?:www\.)?(?:facebook|linkedin|twitter|instagram)\.com/[^\s\"'<>]+")
+
+def extract_matches(pattern, text):
+    return list(set(filter(lambda x: x and x.strip(), pattern.findall(text))))
 
 async def extract_internal_links(page, base_url: str) -> List[str]:
     anchors = await page.eval_on_selector_all('a[href]', 'els => els.map(el => el.href)')
     domain = urlparse(base_url).netloc
-    return list(set([
-        link for link in anchors if urlparse(link).netloc == domain
-    ]))
+    internal_links = [
+        link for link in anchors
+        if urlparse(link).netloc == domain and link not in visited_links
+    ]
+    return list(set(internal_links))
 
-def extract_contact_info(text: str) -> Dict[str, List[str]]:
-    return {
-        key: list(set(pattern.findall(text))) for key, pattern in contact_info_pattern.items()
-    }
+async def scrape_contacts_from_page(page, url: str):
+    contacts = {"emails": [], "phones": [], "socials": []}
 
-async def scrape_page_for_contacts(page, url: str) -> Dict[str, List[str]]:
     try:
-        await page.goto(url, timeout=60000)
-        await page.wait_for_timeout(2000)
+        await page.goto(url, timeout=30000)
+        await page.wait_for_timeout(1500)
+
         content = await page.content()
-        return extract_contact_info(content)
+
+        contacts["emails"] = extract_matches(email_pattern, content)
+        contacts["phones"] = extract_matches(phone_pattern, content)
+        contacts["socials"] = extract_matches(social_pattern, content)
+
     except Exception as e:
-        print(f"[!] Failed to scrape {url}: {e}")
-        return {key: [] for key in contact_info_pattern.keys()}
+        print(f"[!] Failed at {url}: {e}")
+
+    return contacts
 
-@app.get("/scrape-contact-info")
-async def scrape_contact_info(
-    website: str = Query(..., description="Website URL to crawl for contact info"),
-    max_depth: int = Query(1, description="Max link crawl depth (1 = homepage only)")
+@app.get("/scrape-contacts")
+async def scrape_contacts(
+    website: str = Query(..., description="Base website URL"),
+    max_depth: int = Query(1, description="How deep to crawl (recommended: 1 or 2)")
 ):
     try:
-        visited = set()
-        combined_info = {key: set() for key in contact_info_pattern.keys()}
+        all_emails, all_phones, all_socials = set(), set(), set()
+        visited_links.clear()
 
         async with async_playwright() as p:
             browser = await p.chromium.launch(headless=True)
             context = await browser.new_context()
             page = await context.new_page()
+
             queue = [(website, 0)]
 
             while queue:
                 current_url, depth = queue.pop(0)
-                if current_url in visited or depth > max_depth:
+                if current_url in visited_links or depth > max_depth:
                     continue
-                visited.add(current_url)
+                visited_links.add(current_url)
 
-                print(f"[+] Visiting: {current_url}")
-                info = await scrape_page_for_contacts(page, current_url)
-                for key in combined_info:
-                    combined_info[key].update(info.get(key, []))
+                print(f"[+] Crawling: {current_url}")
+                data = await scrape_contacts_from_page(page, current_url)
+                all_emails.update(data["emails"])
+                all_phones.update(data["phones"])
+                all_socials.update(data["socials"])
 
                 if depth < max_depth:
                     try:
-                        links = await extract_internal_links(page, website)
-                        for link in links:
-                            if link not in visited:
+                        internal_links = await extract_internal_links(page, website)
+                        for link in internal_links:
+                            if any(x in link.lower() for x in ["contact", "about", "support"]):
                                 queue.append((link, depth + 1))
                     except Exception as e:
-                        print(f"[!] Failed to get links: {e}")
+                        print(f"[!] Link extraction failed at {current_url}: {e}")
 
             await browser.close()
 
         return {
             "website": website,
-            "pages_visited": len(visited),
-            "emails": list(combined_info["email"]),
-            "phone_numbers": list(combined_info["phone"]),
-            "social_profiles": list(combined_info["social"])
+            "pages_visited": len(visited_links),
+            "emails": list(all_emails),
+            "phone_numbers": list(all_phones),
+            "social_profiles": list(all_socials)
         }
 
     except Exception as e: