Spaces:
Sleeping
Sleeping
| import asyncio | |
| from datetime import datetime, timezone | |
| from bs4 import BeautifulSoup | |
| import httpx | |
| AGENTRAX_URL = "https://agentrax.net/" | |
| # All public pages to scrape — add new pages here as the site grows | |
| AGENTRAX_PAGES = [ | |
| "https://agentrax.net/", | |
| "https://agentrax.net/pricing", | |
| "https://agentrax.net/about", | |
| "https://agentrax.net/help", | |
| "https://agentrax.net/blog", | |
| "https://agentrax.net/contact", | |
| ] | |
| _HEADERS = { | |
| "User-Agent": ( | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " | |
| "AppleWebKit/537.36 (KHTML, like Gecko) " | |
| "Chrome/124.0.0.0 Safari/537.36" | |
| ) | |
| } | |
| _NOISE_TAGS = {"nav", "footer", "header", "script", "style", "aside"} | |
| _NOISE_CLASSES = { | |
| "ad", "ads", "advert", "advertisement", "banner", "cookie", | |
| "popup", "modal", "sidebar", "widget", "promo", | |
| } | |
| async def fetch_page(url: str) -> str: | |
| async with httpx.AsyncClient(headers=_HEADERS, timeout=15, follow_redirects=True) as client: | |
| response = await client.get(url) | |
| response.raise_for_status() | |
| return response.text | |
| def extract_clean_text(html: str, url: str) -> dict: | |
| soup = BeautifulSoup(html, "lxml") | |
| for tag in soup.find_all(_NOISE_TAGS): | |
| tag.decompose() | |
| for tag in soup.find_all(True): | |
| classes = set(tag.get("class") or []) | |
| if classes & _NOISE_CLASSES: | |
| tag.decompose() | |
| title = soup.title.get_text(strip=True) if soup.title else "" | |
| description = "" | |
| meta = soup.find("meta", attrs={"name": "description"}) | |
| if meta: | |
| description = meta.get("content", "").strip() | |
| headings = [ | |
| tag.get_text(strip=True) | |
| for tag in soup.find_all(["h1", "h2", "h3"]) | |
| if tag.get_text(strip=True) | |
| ] | |
| paragraphs = [p.get_text(strip=True) for p in soup.find_all("p") if p.get_text(strip=True)] | |
| list_items = [li.get_text(strip=True) for li in soup.find_all("li") if li.get_text(strip=True)] | |
| body_parts = paragraphs + list_items | |
| body_text = "\n".join(body_parts) | |
| return { | |
| "title": title, | |
| "description": description, | |
| "headings": headings, | |
| "body_text": body_text, | |
| "url": url, | |
| "scraped_at": datetime.now(timezone.utc).isoformat(), | |
| } | |
| async def scrape_site(url: str = AGENTRAX_URL) -> dict: | |
| try: | |
| html = await fetch_page(url) | |
| return extract_clean_text(html, url) | |
| except Exception as exc: | |
| return {"error": str(exc), "url": url} | |
| async def scrape_all_pages() -> list[dict]: | |
| """Scrape all pages in AGENTRAX_PAGES concurrently. Returns only successful results.""" | |
| tasks = [scrape_site(url) for url in AGENTRAX_PAGES] | |
| results = await asyncio.gather(*tasks, return_exceptions=True) | |
| return [r for r in results if isinstance(r, dict) and "error" not in r] | |