Spaces:
Runtime error
Runtime error
| """ | |
| Web Scraping Utilities | |
| Handles JavaScript-rendered pages using Playwright | |
| """ | |
| import asyncio | |
| from playwright.async_api import async_playwright, Page, Browser | |
| from bs4 import BeautifulSoup | |
| import pandas as pd | |
| from typing import List, Dict, Optional | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| async def scrape_page(url: str, wait_time: int = 5) -> Dict[str, str]: | |
| """ | |
| Scrape a JavaScript-rendered page using Playwright | |
| Args: | |
| url: URL to scrape | |
| wait_time: Time to wait for page load (seconds) | |
| Returns: | |
| Dictionary with 'html' and 'text' content | |
| """ | |
| try: | |
| logger.info(f"Scraping page: {url}") | |
| async with async_playwright() as p: | |
| # Launch browser in headless mode | |
| browser = await p.chromium.launch(headless=True) | |
| context = await browser.new_context() | |
| page = await context.new_page() | |
| # Navigate to URL and wait for network to be idle | |
| await page.goto(url, wait_until="networkidle", timeout=60000) | |
| # Additional wait for dynamic content | |
| await asyncio.sleep(wait_time) | |
| # Extract HTML and text content | |
| html = await page.content() | |
| text = await page.inner_text("body") | |
| await browser.close() | |
| logger.info(f"Page scraped successfully: {len(html)} chars HTML, {len(text)} chars text") | |
| return { | |
| "html": html, | |
| "text": text | |
| } | |
| except Exception as e: | |
| logger.error(f"Error scraping page: {e}") | |
| raise | |
| def extract_links(html: str) -> List[str]: | |
| """ | |
| Extract all href links from HTML | |
| Args: | |
| html: HTML content | |
| Returns: | |
| List of URLs | |
| """ | |
| try: | |
| soup = BeautifulSoup(html, 'html.parser') | |
| links = [] | |
| for link in soup.find_all('a', href=True): | |
| href = link['href'] | |
| links.append(href) | |
| logger.info(f"Extracted {len(links)} links") | |
| return links | |
| except Exception as e: | |
| logger.error(f"Error extracting links: {e}") | |
| return [] | |
| def extract_tables(html: str) -> List[pd.DataFrame]: | |
| """ | |
| Extract HTML tables and convert to pandas DataFrames | |
| Args: | |
| html: HTML content | |
| Returns: | |
| List of pandas DataFrames | |
| """ | |
| try: | |
| # Use pandas to read HTML tables | |
| tables = pd.read_html(html) | |
| logger.info(f"Extracted {len(tables)} HTML tables") | |
| return tables | |
| except Exception as e: | |
| logger.error(f"Error extracting tables: {e}") | |
| return [] | |
| async def get_page_with_browser(url: str, browser: Browser, wait_time: int = 5) -> Dict[str, any]: | |
| """ | |
| Scrape page using existing browser instance | |
| Args: | |
| url: URL to scrape | |
| browser: Playwright browser instance | |
| wait_time: Time to wait for page load | |
| Returns: | |
| Dictionary with page content and page object | |
| """ | |
| try: | |
| context = await browser.new_context() | |
| page = await context.new_page() | |
| await page.goto(url, wait_until="networkidle", timeout=60000) | |
| await asyncio.sleep(wait_time) | |
| html = await page.content() | |
| text = await page.inner_text("body") | |
| return { | |
| "html": html, | |
| "text": text, | |
| "page": page, | |
| "context": context | |
| } | |
| except Exception as e: | |
| logger.error(f"Error getting page: {e}") | |
| raise | |
| async def scrape_with_actions(url: str, actions: List[Dict]) -> Dict[str, str]: | |
| """ | |
| Scrape page and perform actions (click, type, etc.) | |
| Args: | |
| url: URL to scrape | |
| actions: List of actions to perform | |
| Each action is a dict with 'type' and parameters | |
| Returns: | |
| Dictionary with final page content | |
| """ | |
| try: | |
| async with async_playwright() as p: | |
| browser = await p.chromium.launch(headless=True) | |
| context = await browser.new_context() | |
| page = await context.new_page() | |
| await page.goto(url, wait_until="networkidle", timeout=60000) | |
| # Perform actions | |
| for action in actions: | |
| action_type = action.get('type') | |
| if action_type == 'click': | |
| selector = action.get('selector') | |
| await page.click(selector) | |
| await asyncio.sleep(1) | |
| elif action_type == 'type': | |
| selector = action.get('selector') | |
| text = action.get('text') | |
| await page.type(selector, text) | |
| elif action_type == 'wait': | |
| time = action.get('time', 1) | |
| await asyncio.sleep(time) | |
| # Get final content | |
| html = await page.content() | |
| text = await page.inner_text("body") | |
| await browser.close() | |
| return { | |
| "html": html, | |
| "text": text | |
| } | |
| except Exception as e: | |
| logger.error(f"Error scraping with actions: {e}") | |
| raise | |