Spaces:
Sleeping
Sleeping
File size: 2,805 Bytes
ab4952c 1fee1c2 ab4952c 1fee1c2 ab4952c | 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 | 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]
|