| from fastapi import FastAPI |
| from playwright.async_api import async_playwright, TimeoutError |
| import re |
|
|
| app = FastAPI() |
|
|
| async def scrape_google(query: str): |
| url = f"https://www.google.com/search?q={query}" |
| async with async_playwright() as pw: |
| browser = await pw.chromium.launch(headless=True) |
| context = await browser.new_context() |
| page = await context.new_page() |
|
|
| await page.goto(url, wait_until="domcontentloaded", timeout=60000) |
| try: |
| await page.wait_for_selector("div#search", timeout=10000) |
| except TimeoutError: |
| pass |
|
|
| links = [] |
| for h in await page.query_selector_all("h3"): |
| try: |
| a = await h.evaluate_handle("e => e.closest('a')") |
| href = await a.get_attribute("href") |
| title = await h.inner_text() |
| links.append({"title": title, "link": href}) |
| except: |
| continue |
|
|
| results = [] |
| for item in links[:5]: |
| await page.goto(item["link"], wait_until="domcontentloaded", timeout=30000) |
| html = await page.content() |
| emails = re.findall(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", html) |
| phones = re.findall(r"\+?\d[\d\s\-/]{7,}\d", html) |
| results.append({ |
| **item, |
| "emails": list(set(emails))[:2], |
| "phones": list(set(phones))[:2] |
| }) |
|
|
| await browser.close() |
| return results |
|
|
| @app.get("/search") |
| async def search(query: str): |
| data = await scrape_google(query.replace(" ", "+")) |
| return {"query": query, "results": data} |
|
|
