File size: 15,949 Bytes
05c92cc d382ddb 02554df d382ddb 02554df 05c92cc d382ddb 35b24cc f9275f7 35b24cc d382ddb 9e4b598 d382ddb f9275f7 9e4b598 35b24cc d382ddb 05c92cc 35b24cc 05c92cc 35b24cc 05c92cc 35b24cc f9275f7 35b24cc 9e4b598 05c92cc 35b24cc 9e4b598 35b24cc 05c92cc 02554df b55a15f 02554df 9e4b598 02554df 9e4b598 02554df 9e4b598 02554df 9e4b598 02554df 9e4b598 02554df 9e4b598 02554df 35b24cc 02554df d382ddb c2f18d6 a8c57d0 d382ddb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 |
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from playwright.async_api import async_playwright
import asyncio
import base64
import logging
from typing import List, Optional
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="Playwright Web Scraper", description="A simple web scraper using Playwright")
class LinkInfo(BaseModel):
text: str
href: str
class ContactInfo(BaseModel):
emails: List[str] = []
phones: List[str] = []
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
description: Optional[str] = None
industry_keywords: List[str] = []
class LeadData(BaseModel):
contact_info: ContactInfo
business_info: BusinessInfo
lead_score: int = 0
technologies: List[str] = []
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
@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"
},
"example": "/scrape?url=https://example.com&lead_generation=true&screenshot=true",
"lead_generation_features": [
"π§ Extract email addresses and contact forms",
"π Find phone numbers and contact info",
"π’ Identify company names and addresses",
"π Discover social media profiles",
"β‘ Detect technologies and tools used",
"π Calculate lead quality scores",
"π― Industry keyword extraction"
],
"basic_features": [
"π Clean body text extraction",
"π Smart link filtering",
"οΏ½ Script and JavaScript file extraction",
"οΏ½πΈ Full page screenshots",
"π Page metadata extraction"
],
"use_cases": [
"B2B lead generation",
"Sales prospecting",
"Market research",
"Competitor analysis",
"Contact discovery"
]
}
@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)")
):
logger.info(f"Starting scrape for URL: {url}")
try:
async with async_playwright() as p:
logger.info("Launching browser...")
browser = await p.chromium.launch(
headless=True,
args=[
'--no-sandbox',
'--disable-setuid-sandbox',
'--disable-dev-shm-usage',
'--disable-accelerated-2d-canvas',
'--no-first-run',
'--no-zygote',
'--disable-gpu'
]
)
page = await browser.new_page()
try:
logger.info(f"Navigating to {url}...")
# await page.goto(url, wait_until="networkidle")
await page.goto(url, wait_until="domcontentloaded", timeout=60000)
response = ScrapeResponse()
# Always get page title and meta description
logger.info("Getting page metadata...")
response.page_title = await page.title()
meta_desc = await page.evaluate("""
() => {
const meta = document.querySelector('meta[name="description"]');
return meta ? meta.getAttribute('content') : null;
}
""")
response.meta_description = meta_desc
# Get body content (clean text)
if get_body:
logger.info("Extracting body content...")
body_content = await page.evaluate("""
() => {
const body = document.querySelector('body');
if (!body) return null;
// Remove script and style elements
const scripts = body.querySelectorAll('script, style, noscript');
scripts.forEach(el => el.remove());
// Get clean text content
return body.innerText.trim();
}
""")
response.body_content = body_content
# Get screenshot (full page)
if screenshot:
logger.info("Taking full page screenshot...")
screenshot_bytes = await page.screenshot(full_page=True)
response.screenshot = base64.b64encode(screenshot_bytes).decode('utf-8')
# Get links with better filtering
if get_links:
logger.info("Extracting links...")
links = await page.evaluate("""
() => {
return Array.from(document.querySelectorAll('a[href]')).map(a => {
const text = a.innerText.trim();
const href = a.href;
// Only include links with meaningful text and valid URLs
if (text && href && href.startsWith('http')) {
return {
text: text.substring(0, 200), // Limit text length
href: href
}
}
return null;
}).filter(link => link !== null);
}
""")
response.links = [LinkInfo(**link) for link in links]
# Lead Generation Extraction
if lead_generation:
logger.info("Extracting lead generation data...")
lead_data_raw = await page.evaluate("""
() => {
const result = {
emails: [],
phones: [],
social_media: [],
contact_forms: [],
company_name: null,
address: null,
technologies: [],
industry_keywords: []
};
// Extract emails
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) || [];
result.emails = [...new Set(emails)].slice(0, 10); // Unique emails, max 10
// Extract phone numbers
const phoneRegex = /(\+?1?[-.\s]?)?\(?([0-9]{3})\)?[-.\s]?([0-9]{3})[-.\s]?([0-9]{4})/g;
const phones = pageText.match(phoneRegex) || [];
result.phones = [...new Set(phones)].slice(0, 5); // Unique phones, max 5
// Extract social media links
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);
// Find contact forms
const forms = Array.from(document.querySelectorAll('form')).map(form => {
const action = form.action || window.location.href;
return action;
});
result.contact_forms = [...new Set(forms)].slice(0, 5);
// Extract company name (try multiple methods)
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();
// Extract address
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;
// Detect technologies
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));
// Industry keywords
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;
}
""")
# Calculate lead score
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 += 20
if lead_data_raw['social_media']: lead_score += 15
if lead_data_raw['company_name']: lead_score += 10
if lead_data_raw['address']: lead_score += 15
if lead_data_raw['technologies']: lead_score += 10
if lead_data_raw['industry_keywords']: lead_score += 5
# Create lead data object
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), # Cap at 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:
logger.error(f"Error launching browser: {str(e)}")
raise HTTPException(status_code=500, detail=f"Browser launch error: {str(e)}")
# @app.get("/search_leads")
# async def search_leads(
# query: str = Query(..., description="Search term for business leads")
# ):
# logger.info(f"Searching Google Maps for: {query}")
# async with async_playwright() as p:
# browser = await p.chromium.launch(headless=True)
# page = await browser.new_page()
# try:
# # Go to Google Maps
# await page.goto("https://www.google.com/maps", wait_until="networkidle")
# # Accept cookies if present (optional, depends on region)
# try:
# await page.click('button[aria-label="Accept all"]', timeout=180000)
# except:
# pass
# # Type the query in the search box and press Enter
# await page.fill('input#searchboxinput', query)
# await page.click('button#searchbox-searchbutton')
# # Wait for search results to load - selector for listings container
# await page.wait_for_selector('div[role="article"]', timeout=180000)
# # Scroll results container to load more items (optional)
# # For now, scrape the visible ones
# # Extract data from listings
# results = await page.evaluate("""
# () => {
# const listings = [];
# const elements = document.querySelectorAll('div[role="article"]');
# elements.forEach(el => {
# const nameEl = el.querySelector('h3 span');
# const name = nameEl ? nameEl.innerText : null;
# const addressEl = el.querySelector('[data-tooltip="Address"]');
# const address = addressEl ? addressEl.innerText : null;
# const phoneEl = el.querySelector('button[data-tooltip="Copy phone number"]');
# const phone = phoneEl ? phoneEl.getAttribute('aria-label')?.replace('Copy phone number ', '') : null;
# const websiteEl = el.querySelector('a[aria-label*="Website"]');
# const website = websiteEl ? websiteEl.href : null;
# listings.push({name, address, phone, website});
# });
# return listings;
# }
# """)
# await browser.close()
# # Filter out empty entries
# filtered = [r for r in results if r['name']]
# return {"query": query, "results_count": len(filtered), "results": filtered}
# except Exception as e:
# await browser.close()
# logger.error(f"Error during Google Maps search scraping: {str(e)}")
# raise HTTPException(status_code=500, detail=f"Search scraping error: {str(e)}")
|