devsbarn/fastapi-crawler

1import asyncio, httpx2from bs4 import BeautifulSoup3from urllib.parse import urlparse, urljoin4from urllib.robotparser import RobotFileParser5 6def get_domain(url: str) -> str:7    parsed = urlparse(url)8    return f"{parsed.scheme}://{parsed.netloc}"9 10async def can_crawl(url: str, user_agent="SiteCrawlerBot") -> bool:11    parsed = urlparse(url)12    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"13    rp = RobotFileParser()14    try:15        async with httpx.AsyncClient(timeout=5.0) as client:16            r = await client.get(robots_url)17            if r.status_code == 200:18                rp.parse(r.text.splitlines())19            else:20                return True21        return rp.can_fetch(user_agent, url)22    except: return True23 24def extract_clean_text(html: str) -> str:25    soup = BeautifulSoup(html, "html.parser")26    for tag in soup(["script","style","nav","footer","header","aside","noscript"]):27        tag.extract()28    raw_text = "\n".join([chunk.strip() for chunk in soup.stripped_strings])29    seen, lines = set(), []30    for line in raw_text.splitlines():31        if len(line.split()) > 3 and line not in seen:32            lines.append(line); seen.add(line)33    return "\n".join(lines)34 35async def crawl_site_async(start_url: str, max_pages=30, delay=0.5):36    domain = get_domain(start_url)37    visited, to_visit = set(), [start_url]38    async with httpx.AsyncClient(timeout=10.0, headers={"User-Agent": "SiteCrawlerBot"}) as client:39        while to_visit and len(visited) < max_pages:40            url = to_visit.pop(0)41            if url in visited or not url.startswith(domain):42                continue43            if not await can_crawl(url): continue44            try:45                resp = await client.get(url)46                if resp.status_code == 200 and "text/html" in resp.headers.get("Content-Type",""):47                    text = extract_clean_text(resp.text)48                    if text:49                        depth = url.replace(domain, "").count("/")50                        yield {"url": url, "depth": depth, "content": text}51                    soup = BeautifulSoup(resp.text, "html.parser")52                    for link in soup.find_all("a", href=True):53                        abs_link = urljoin(url, link["href"].split("#")[0])54                        if domain in abs_link and abs_link not in visited:55                            to_visit.append(abs_link)56                visited.add(url)57                await asyncio.sleep(delay)58            except Exception: continue