test1.py

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
from urllib.parse import urlparse, parse_qs

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Playwright Web Scraper", description="Scrape websites based on search queries")

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
    source_url: Optional[str] = None

@app.get("/")
async def root():
    return {
        "message": "🚀 Query-Based Web Scraper API",
        "tagline": "Search and scrape websites based on queries",
        "endpoints": {
            "/scrape": "Search Google for the query and scrape the top result",
            "/docs": "API documentation"
        },
        "example": "/scrape?query=plumbers+near+me&lead_generation=true&screenshot=true",
        "note": "Now accepts search queries instead of direct URLs"
    }

async def get_top_search_result(query: str):
    """Perform Google search and return top result URL"""
    user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            user_agent=user_agent,
            locale='en-US',
            viewport={'width': 1920, 'height': 1080}
        )
        page = await context.new_page()
        
        try:
            logger.info(f"Searching Google for: {query}")
            await page.goto("https://www.google.com", timeout=60000)
            
            # Handle consent form if it appears
            try:
                consent_button = await page.wait_for_selector('button:has-text("Accept all")', timeout=5000)
                if consent_button:
                    await consent_button.click()
                    logger.info("Accepted Google consent form")
                    await page.wait_for_load_state("networkidle")
            except:
                pass  # Consent form didn't appear
            
            # Perform search
            await page.fill('textarea[name="q"]', query)
            await page.keyboard.press("Enter")
            await page.wait_for_selector("#search", timeout=30000)
            
            # Extract top results
            results = await page.query_selector_all('.g')
            if not results:
                raise Exception("No search results found")
            
            urls = []
            for result in results[:3]:  # Check top 3 results
                try:
                    link = await result.query_selector('a')
                    if not link:
                        continue
                    
                    # Extract both data-href and href attributes
                    data_href = await link.get_attribute('data-href')
                    href = await link.get_attribute('href')
                    target_url = data_href or href
                    
                    if target_url and target_url.startswith('/url?q='):
                        target_url = f"https://www.google.com{target_url}"
                    
                    if target_url and target_url.startswith('https://www.google.com/url?'):
                        parsed = urlparse(target_url)
                        qs = parse_qs(parsed.query)
                        target_url = qs.get('q', [target_url])[0]
                    
                    if target_url and target_url.startswith('http'):
                        urls.append(target_url)
                        logger.info(f"Found search result: {target_url}")
                except Exception as e:
                    logger.warning(f"Error processing result: {str(e)}")
            
            if not urls:
                raise Exception("No valid URLs found in search results")
            
            await browser.close()
            return urls[0]  # Return top result
        
        except Exception as e:
            logger.error(f"Search failed: {str(e)}")
            await browser.close()
            raise

@app.get("/scrape")
async def scrape_page(
    query: str = Query(..., description="Search query to find a website"),
    lead_generation: bool = Query(True, description="Extract lead generation data"),
    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")
):
    logger.info(f"Starting scrape for query: {query}")
    
    try:
        # Get top search result URL
        target_url = await get_top_search_result(query)
        logger.info(f"Scraping top result: {target_url}")
    except Exception as e:
        logger.error(f"Search error: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}")

    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 {target_url}...")
                await page.goto(target_url, wait_until="networkidle")
                response = ScrapeResponse(source_url=target_url)

                # 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)}")