Tool for scraping contacts

contacts.py

@@ -1,19 +1,18 @@
-from fastapi import FastAPI, Query, HTTPException
+from fastapi import FastAPI, HTTPException, Query
 from pydantic import BaseModel
 from playwright.async_api import async_playwright
-from urllib.parse import urlparse
+import asyncio
 import base64
 import logging
-from typing import List, Optional, Set
-import re
-
-app = FastAPI(title="Lead Generation Web Scraper API")
+from typing import List, Optional
+from urllib.parse import urlparse
 
-# -------------------- Logging Setup --------------------
+# Set up logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# -------------------- Pydantic Models --------------------
+app = FastAPI(title="Playwright Web Scraper", description="A simple web scraper using Playwright")
+
 class LinkInfo(BaseModel):
     text: str
     href: str
@@ -24,6 +23,11 @@ class ContactInfo(BaseModel):
     social_media: List[str] = []
     contact_forms: List[str] = []
 
+class ScriptInfo(BaseModel):
+    src: str
+    script_type: Optional[str] = None
+    is_external: bool = False
+
 class BusinessInfo(BaseModel):
     company_name: Optional[str] = None
     address: Optional[str] = None
@@ -36,146 +40,184 @@ class LeadData(BaseModel):
     lead_score: int = 0
     technologies: List[str] = []
 
-class ScrapedPage(BaseModel):
-    url: str
-    page_title: Optional[str]
-    meta_description: Optional[str]
-    screenshot: Optional[str]
-    links: Optional[List[LinkInfo]]
-    lead_data: Optional[LeadData]
-
-class CrawlResponse(BaseModel):
-    pages_visited: int
-    results: List[ScrapedPage]
-
-# -------------------- Utility Functions --------------------
-async def extract_page_data(page, url: str, take_screenshot: bool) -> ScrapedPage:
-    await page.goto(url, timeout=30000)
-    title = await page.title()
-
-    meta_desc = await page.evaluate("""
-        () => {
-            const meta = document.querySelector('meta[name="description"]');
-            return meta ? meta.getAttribute('content') : null;
+class ScrapeResponse(BaseModel):
+    body_content: Optional[str] = None
+    screenshot: Optional[str] = None
+    links: Optional[List[LinkInfo]] = None
+    scripts: Optional[List[ScriptInfo]] = None
+    page_title: Optional[str] = None
+    meta_description: Optional[str] = None
+    lead_data: Optional[LeadData] = None
+
+visited_urls = set()
+
+@app.get("/")
+async def root():
+    return {
+        "message": "🚀 Lead Generation Web Scraper API",
+        "tagline": "Turn any website into qualified leads",
+        "endpoints": {
+            "/scrape": "Extract leads, contacts, and business data from any website",
+            "/docs": "API documentation"
         }
-    """)
-
-    screenshot = None
-    if take_screenshot:
-        shot = await page.screenshot(full_page=True)
-        screenshot = base64.b64encode(shot).decode('utf-8')
-
-    links = await page.evaluate("""
-        () => Array.from(document.querySelectorAll('a[href]')).map(a => ({
-            text: a.innerText.trim().substring(0, 200),
-            href: a.href
-        }))
-    """)
-
-    text = await page.evaluate("document.body.innerText")
-
-    emails = list(set(re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}", text)))[:10]
-    phones = list(set(re.findall(r"(\+?1?[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}", text)))[:5]
-
-    social_links = list(set(
-        l['href'] for l in links if re.search(r"facebook|linkedin|twitter|instagram|youtube", l['href'], re.I)
-    ))[:10]
-
-    contact_forms = list(set(await page.evaluate("""
-        () => Array.from(document.querySelectorAll('form')).map(f => f.action || location.href)
-    """)))[:5]
-
-    company_name = await page.evaluate("""
-        () => document.querySelector('meta[property="og:site_name"]')?.content ||
-              document.querySelector('meta[name="application-name"]')?.content ||
-              document.querySelector('h1')?.innerText ||
-              document.title?.split('|')[0]?.split('-')[0]?.trim()
-    """)
-
-    address_matches = re.findall(r"\d+\s+[A-Za-z\s]+(?:Street|St|Avenue|Ave|Road|Rd|Boulevard|Blvd|Lane|Ln|Drive|Dr|Court|Ct|Place|Pl)[^\n,]*", text)
-    address = address_matches[0] if address_matches else None
-
-    html = await page.content()
-    tech_stack = [t for t in ['wordpress', 'shopify', 'react', 'angular', 'vue', 'bootstrap', 'jquery', 'google analytics'] if t in html.lower()]
-
-    industry_keywords = [k for k in ['consulting', 'marketing', 'software', 'healthcare', 'finance', 'education'] if k in text.lower()]
-
-    lead_score = min(100, sum([
-        30 if emails else 0,
-        25 if phones else 0,
-        15 if social_links else 0,
-        10 if company_name else 0,
-        15 if address else 0,
-        10 if tech_stack else 0,
-        5 if industry_keywords else 0
-    ]))
-
-    return ScrapedPage(
-        url=url,
-        page_title=title,
-        meta_description=meta_desc,
-        screenshot=screenshot,
-        links=[LinkInfo(**l) for l in links],
-        lead_data=LeadData(
-            contact_info=ContactInfo(
-                emails=emails,
-                phones=phones,
-                social_media=social_links,
-                contact_forms=contact_forms,
-            ),
-            business_info=BusinessInfo(
-                company_name=company_name,
-                address=address,
-                description=meta_desc,
-                industry_keywords=industry_keywords
-            ),
-            lead_score=lead_score,
-            technologies=tech_stack
-        )
-    )
-
-# -------------------- API Endpoint --------------------
-@app.get("/crawl-leads", response_model=CrawlResponse)
-async def crawl_leads(
-    website: str = Query(..., description="Base website URL to crawl"),
-    max_depth: int = Query(1, ge=1, le=3),
-    screenshot: bool = Query(False, description="Take full page screenshots")
+    }
+
+def normalize_url(url):
+    parsed = urlparse(url)
+    return parsed._replace(fragment='', query='').geturl().rstrip('/')
+
+@app.get("/scrape")
+async def scrape_page(
+    url: str = Query(..., description="URL to scrape"),
+    lead_generation: bool = Query(True, description="Extract lead generation data (emails, phones, business info)"),
+    screenshot: bool = Query(True, description="Take a full page screenshot"),
+    get_links: bool = Query(True, description="Extract all links from the page"),
+    get_body: bool = Query(False, description="Extract body tag content (can be large)")
 ):
-    visited: Set[str] = set()
-    results: List[ScrapedPage] = []
+    norm_url = normalize_url(url)
+    if norm_url in visited_urls:
+        raise HTTPException(status_code=400, detail="URL already scraped")
 
+    visited_urls.add(norm_url)
+    logger.info(f"Starting scrape for URL: {norm_url}")
     try:
         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:
-                url, depth = queue.pop(0)
-                if url in visited or depth > max_depth:
-                    continue
-                visited.add(url)
-
-                try:
-                    result = await extract_page_data(page, url, screenshot)
-                    results.append(result)
-
-                    if depth < max_depth:
-                        links = [link.href for link in result.links if urlparse(link.href).netloc == urlparse(website).netloc]
-                        for link in links:
-                            if link not in visited:
-                                queue.append((link, depth + 1))
-                except Exception as e:
-                    logger.warning(f"Failed to scrape {url}: {e}")
-
-            await browser.close()
-
-        return CrawlResponse(
-            pages_visited=len(visited),
-            results=results
-        )
+            page = await browser.new_page()
+
+            try:
+                await page.goto(norm_url, wait_until="domcontentloaded", timeout=60000)
+
+                response = ScrapeResponse()
+                response.page_title = await page.title()
+
+                response.meta_description = await page.evaluate("""
+                    () => {
+                        const meta = document.querySelector('meta[name="description"]');
+                        return meta ? meta.getAttribute('content') : null;
+                    }
+                """)
+
+                if get_body:
+                    response.body_content = await page.evaluate("""
+                        () => {
+                            const body = document.querySelector('body');
+                            if (!body) return null;
+                            const scripts = body.querySelectorAll('script, style, noscript');
+                            scripts.forEach(el => el.remove());
+                            return body.innerText.trim();
+                        }
+                    """)
+
+                if screenshot:
+                    screenshot_bytes = await page.screenshot(full_page=True)
+                    response.screenshot = base64.b64encode(screenshot_bytes).decode('utf-8')
+
+                if get_links:
+                    links = await page.evaluate("""
+                        () => {
+                            return Array.from(document.querySelectorAll('a[href]')).map(a => {
+                                const text = a.innerText.trim() || a.getAttribute('aria-label') || a.getAttribute('title') || a.href;
+                                const href = a.href;
+                                if (href && href.startsWith('http')) {
+                                    return { text: text.substring(0, 200), href: href };
+                                }
+                                return null;
+                            }).filter(link => link !== null);
+                        }
+                    """)
+                    response.links = [LinkInfo(**link) for link in links]
+
+                if lead_generation:
+                    lead_data_raw = await page.evaluate("""
+                        () => {
+                            const result = {
+                                emails: [], phones: [], social_media: [], contact_forms: [],
+                                company_name: null, address: null, technologies: [], industry_keywords: []
+                            };
+
+                            const emailRegex = /[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/g;
+                            const pageText = document.body.innerText;
+                            const emails = pageText.match(emailRegex) || [];
+                            const mailtoEmails = Array.from(document.querySelectorAll('a[href^="mailto:"]'))
+                                .map(a => a.href.replace(/^mailto:/, '').split('?')[0]);
+                            result.emails = [...new Set([...emails, ...mailtoEmails])].slice(0, 10);
+
+                            const phoneRegex = /(\+?1?[-.\s]?)?\(?([0-9]{3})\)?[-.\s]?([0-9]{3})[-.\s]?([0-9]{4})/g;
+                            const phones = pageText.match(phoneRegex) || [];
+                            const telPhones = Array.from(document.querySelectorAll('a[href^="tel:"]'))
+                                .map(a => a.href.replace(/^tel:/, '').split('?')[0]);
+                            result.phones = [...new Set([...phones, ...telPhones])].slice(0, 5);
+
+                            const socialLinks = Array.from(document.querySelectorAll('a[href]'))
+                                .map(a => a.href).filter(href => /facebook|twitter|linkedin|instagram|youtube|tiktok/i.test(href));
+                            result.social_media = [...new Set(socialLinks)].slice(0, 10);
+
+                            const forms = Array.from(document.querySelectorAll('form')).map(form => form.action || window.location.href);
+                            result.contact_forms = [...new Set(forms)].slice(0, 5);
+
+                            result.company_name =
+                                document.querySelector('meta[property="og:site_name"]')?.content ||
+                                document.querySelector('meta[name="application-name"]')?.content ||
+                                document.querySelector('h1')?.innerText?.trim() ||
+                                document.title?.split('|')[0]?.split('-')[0]?.trim();
+
+                            const addressRegex = /\d+\s+[A-Za-z\s]+(?:Street|St|Avenue|Ave|Road|Rd|Boulevard|Blvd|Lane|Ln|Drive|Dr|Court|Ct|Place|Pl)\s*,?\s*[A-Za-z\s]+,?\s*[A-Z]{2}\s*\d{5}/g;
+                            const addresses = pageText.match(addressRegex) || [];
+                            result.address = addresses[0] || null;
+
+                            const techKeywords = ['wordpress', 'shopify', 'react', 'angular', 'vue', 'bootstrap', 'jquery', 'google analytics', 'facebook pixel'];
+                            const htmlContent = document.documentElement.outerHTML.toLowerCase();
+                            result.technologies = techKeywords.filter(tech => htmlContent.includes(tech));
+
+                            const industryKeywords = ['consulting', 'marketing', 'software', 'healthcare', 'finance', 'real estate', 'education', 'retail', 'manufacturing', 'legal', 'restaurant', 'fitness', 'beauty', 'automotive'];
+                            const lowerPageText = pageText.toLowerCase();
+                            result.industry_keywords = industryKeywords.filter(keyword => lowerPageText.includes(keyword));
+
+                            return result;
+                        }
+                    """)
+
+                    lead_score = 0
+                    if lead_data_raw['emails']: lead_score += 30
+                    if lead_data_raw['phones']: lead_score += 25
+                    if lead_data_raw['contact_forms']: lead_score += 15
+                    if lead_data_raw['social_media']: lead_score += 10
+                    if lead_data_raw['company_name']: lead_score += 10
+                    if lead_data_raw['address']: lead_score += 10
+                    if lead_data_raw['technologies']: lead_score += 5
+                    if lead_data_raw['industry_keywords']: lead_score += 5
+
+                    contact_info = ContactInfo(
+                        emails=lead_data_raw['emails'],
+                        phones=lead_data_raw['phones'],
+                        social_media=lead_data_raw['social_media'],
+                        contact_forms=lead_data_raw['contact_forms']
+                    )
+
+                    business_info = BusinessInfo(
+                        company_name=lead_data_raw['company_name'],
+                        address=lead_data_raw['address'],
+                        description=response.meta_description,
+                        industry_keywords=lead_data_raw['industry_keywords']
+                    )
+
+                    response.lead_data = LeadData(
+                        contact_info=contact_info,
+                        business_info=business_info,
+                        lead_score=min(lead_score, 100),
+                        technologies=lead_data_raw['technologies']
+                    )
+
+                await browser.close()
+                logger.info("Scraping completed successfully")
+                return response
+
+            except Exception as e:
+                logger.error(f"Error during scraping: {str(e)}")
+                await browser.close()
+                raise HTTPException(status_code=500, detail=f"Scraping error: {str(e)}")
 
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        logger.error(f"Error launching browser: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Browser launch error: {str(e)}")