devsbarn/fastapi-crawler

1import time, sys, requests2from bs4 import BeautifulSoup3from urllib.parse import urlparse, urljoin4from collections import deque5from urllib.robotparser import RobotFileParser6 7def get_domain(url: str) -> str:8    parsed = urlparse(url)9    return f"{parsed.scheme}://{parsed.netloc}"10 11def can_crawl(url: str, user_agent="SiteCrawlerBot") -> bool:12    parsed = urlparse(url)13    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"14    rp = RobotFileParser()15    try:16        rp.set_url(robots_url)17        rp.read()18        return rp.can_fetch(user_agent, url)19    except:20        return True21 22def extract_clean_text(html: str) -> str:23    soup = BeautifulSoup(html, "html.parser")24    for tag in soup(["script","style","nav","footer","header","aside","noscript"]):25        tag.extract()26    raw_text = "\n".join([chunk.strip() for chunk in soup.stripped_strings])27    seen, lines = set(), []28    for line in raw_text.splitlines():29        if len(line.split()) > 3 and line not in seen:30            lines.append(line); seen.add(line)31    return "\n".join(lines)32 33def crawl_site(start_url: str, max_pages=30, delay=1.0):34    domain = get_domain(start_url)35    visited, to_visit = set(), deque([start_url])36    pages = []37    while to_visit and len(visited) < max_pages:38        url = to_visit.popleft()39        if url in visited or not url.startswith(domain):40            continue41        if not can_crawl(url): continue42        try:43            resp = requests.get(url, timeout=10, headers={"User-Agent":"SiteCrawlerBot"})44            if resp.status_code == 200 and "text/html" in resp.headers.get("Content-Type",""):45                text = extract_clean_text(resp.text)46                if text:47                    depth = url.replace(domain, "").count("/")48                    pages.append({"url": url, "depth": depth, "content": text})49                soup = BeautifulSoup(resp.text, "html.parser")50                for link in soup.find_all("a", href=True):51                    abs_link = urljoin(url, link["href"].split("#")[0])52                    if domain in abs_link and abs_link not in visited:53                        to_visit.append(abs_link)54                visited.add(url)55                time.sleep(delay)56        except Exception as e:57            print("Skipping:", e, file=sys.stderr)58    return pages