feat: concurrent crawl with anti-bot, RSS fallback, and depth-by-depth processing

- Refactor crawler from sequential FIFO loop to depth-by-depth concurrent
- Use crawl4ai arun_many() for parallel HTML crawling
- Use asyncio.gather() for parallel PDF/text file fetching
- Add User-Agent rotation, full browser headers, rate limiting
- Add retry with exponential backoff on 403/429/503
- Add anti-bot challenge detection (Cloudflare, Akamai, etc.)
- Add RSS/Atom feed detection and fallback before crawling
- Add crawl4ai stealth mode (simulate_user, override_navigator, magic)
- Add PyMuPDF to requirements for PDF extraction
- Mark URLs visited before batch execution to prevent race conditions

app.py

@@ -106,6 +106,7 @@ async def run_crawl(url, depth, max_pages, progress=gr.Progress()):
         progress(0.05, desc=f"Crawling {url} (depth={depth}, max={max_pages})...")
         pages_df = await crawler.crawl(url, depth, max_pages, progress)
         pdf_count = crawler.pdf_count
+        rss_count = crawler.rss_count
         n_pages = len(pages_df)
 
         if n_pages == 0:
@@ -150,7 +151,7 @@ async def run_crawl(url, depth, max_pages, progress=gr.Progress()):
         progress(1.0, desc="Done!")
         return (
             f"**Crawl Complete!**\n"
-            + f"- Pages crawled: {n_pages} ({pdf_count} PDF, {crawler.html_count} HTML, {crawler.text_count} text)\n"
+            + f"- Pages crawled: {n_pages} ({pdf_count} PDF, {crawler.html_count} HTML, {crawler.text_count} text, {rss_count} RSS)\n"
             + f"- Chunks created: {n_chunks}\n"
             + f"- Facts extracted: {n_facts}\n"
             + f"- Indexes: {idx_status}\n"

requirements.txt

@@ -13,3 +13,4 @@ pydantic>=2.0.0
 httpx>=0.27.0
 beautifulsoup4>=4.12.0
 lxml>=5.0.0
+PyMuPDF>=1.24.0

src/config.py

@@ -11,6 +11,14 @@ class Config:
     max_depth: int = 10
     max_pages_limit: int = 1000
 
+    # Anti-bot settings
+    crawl_delay_min: float = 0.5
+    crawl_delay_max: float = 2.0
+    crawl_max_retries: int = 3
+    crawl_retry_base_delay: float = 5.0
+    crawl_min_content_bytes: int = 200
+    crawl_rss_fallback: bool = True
+
     # Chunking settings
     chunk_max_chars: int = 800
     chunk_overlap: int = 100

src/crawler.py

@@ -1,4 +1,5 @@
 import asyncio
+import random
 import time
 from pathlib import Path
 from urllib.parse import urljoin, urlparse
@@ -6,7 +7,7 @@ from urllib.parse import urljoin, urlparse
 import httpx
 import polars as pl
 from bs4 import BeautifulSoup
-from crawl4ai import AsyncWebCrawler
+from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
 
 from src.config import Config
 from src.utils import content_hash, normalize_url
@@ -14,6 +15,29 @@ from src.utils import content_hash, normalize_url
 PDF_TYPES = {".pdf"}
 TEXT_TYPES = {".txt", ".csv", ".xml", ".md", ".json"}
 
+USER_AGENTS = [
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_7_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:133.0) Gecko/20100101 Firefox/133.0",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14.7; rv:133.0) Gecko/20100101 Firefox/133.0",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36 Edg/131.0.0.0",
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (X11; Linux x86_64; rv:133.0) Gecko/20100101 Firefox/133.0",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36 Vivaldi/7.0.3495.11",
+]
+
+RSS_PATHS = [
+    "/feed", "/rss", "/rss.xml", "/atom.xml", "/feed.xml",
+    "/index.xml", "/blog/feed", "/news/feed", "/feeds/posts/default",
+]
+
+CHALLENGE_INDICATORS = [
+    "access denied", "blocked", "captcha", "verify you are human",
+    "ray id", "cloudflare", "incapsula", "akamai", "distil networks",
+    "press & hold", "checking your browser", "just a moment",
+    "attention required", "security check", "ddos protection",
+]
+
 
 class Crawler:
     def __init__(self, config: Config):
@@ -23,8 +47,10 @@ class Crawler:
         self.pdf_count = 0
         self.html_count = 0
         self.text_count = 0
+        self.rss_count = 0
         self.domain: str = ""
         self.http_client: httpx.AsyncClient | None = None
+        self._last_request_time: float = 0.0
 
     def _get_domain(self, url: str) -> str:
         return urlparse(url).netloc
@@ -36,6 +62,63 @@ class Crawler:
         path = urlparse(url).path.lower()
         return Path(path).suffix
 
+    def _is_static_asset(self, url: str) -> bool:
+        ext = self._get_ext(url)
+        return ext in PDF_TYPES or ext in TEXT_TYPES
+
+    def _random_ua(self) -> str:
+        return random.choice(USER_AGENTS)
+
+    def _build_headers(self) -> dict[str, str]:
+        return {
+            "User-Agent": self._random_ua(),
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
+            "Accept-Language": "en-US,en;q=0.9",
+            "Accept-Encoding": "gzip, deflate, br",
+            "Connection": "keep-alive",
+            "Upgrade-Insecure-Requests": "1",
+            "Sec-Fetch-Dest": "document",
+            "Sec-Fetch-Mode": "navigate",
+            "Sec-Fetch-Site": "none",
+            "Sec-Fetch-User": "?1",
+            "Cache-Control": "max-age=0",
+        }
+
+    async def _rate_limit(self):
+        elapsed = time.time() - self._last_request_time
+        delay = random.uniform(self.config.crawl_delay_min, self.config.crawl_delay_max)
+        if elapsed < delay:
+            await asyncio.sleep(delay - elapsed)
+        self._last_request_time = time.time()
+
+    def _is_blocked_content(self, html: str, content_length: int) -> bool:
+        if content_length < self.config.crawl_min_content_bytes:
+            return True
+        html_lower = html.lower()
+        for indicator in CHALLENGE_INDICATORS:
+            if indicator in html_lower:
+                return True
+        return False
+
+    async def _fetch_with_retry(self, url: str) -> httpx.Response | None:
+        for attempt in range(self.config.crawl_max_retries):
+            await self._rate_limit()
+            headers = self._build_headers()
+            try:
+                resp = await self.http_client.get(url, headers=headers)
+                content_len = len(resp.content)
+                if resp.status_code == 200 and not self._is_blocked_content(resp.text, content_len):
+                    return resp
+                if resp.status_code in (403, 429, 503):
+                    backoff = self.config.crawl_retry_base_delay * (2 ** attempt) + random.uniform(0, 2)
+                    await asyncio.sleep(backoff)
+                    continue
+                return resp
+            except (httpx.TimeoutException, httpx.ConnectError, httpx.ReadError):
+                backoff = self.config.crawl_retry_base_delay * (2 ** attempt) + random.uniform(0, 2)
+                await asyncio.sleep(backoff)
+        return None
+
     async def crawl(self, url: str, depth: int, max_pages: int, progress=None) -> pl.DataFrame:
         url = normalize_url(url)
         self.domain = self._get_domain(url)
@@ -44,52 +127,100 @@ class Crawler:
         self.pdf_count = 0
         self.html_count = 0
         self.text_count = 0
-
-        queue: list[tuple[str, int]] = [(url, 0)]
+        self.rss_count = 0
+
+        if self.config.crawl_rss_fallback:
+            await self._try_rss_feed(url)
+
+        browser_config = BrowserConfig(
+            headless=True,
+            browser_type="chromium",
+            user_agent=self._random_ua(),
+            verbose=False,
+            enable_stealth=True,
+        )
+        crawl_config = CrawlerRunConfig(
+            word_count_threshold=10,
+            bypass_cache=True,
+            simulate_user=True,
+            override_navigator=True,
+            magic=True,
+            remove_overlay_elements=True,
+            remove_consent_popups=True,
+            semaphore_count=8,
+        )
 
         async with httpx.AsyncClient(
-            timeout=30.0, follow_redirects=True,
-            headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
+            timeout=30.0,
+            follow_redirects=True,
         ) as self.http_client:
-            async with AsyncWebCrawler() as crawler:
-                while queue and len(self.pages) < max_pages:
-                    current_url, current_depth = queue.pop(0)
-
-                    if current_url in self.visited:
-                        continue
-                    if current_depth > depth:
-                        continue
-                    domain = self._get_domain(current_url)
-                    if domain and domain != self.domain:
-                        continue
-
-                    self.visited.add(current_url)
-
-                    ext = self._get_ext(current_url)
-                    page = None
-
-                    if ext == ".pdf":
-                        page = await self._process_pdf(current_url, current_depth)
-                    elif ext in TEXT_TYPES:
-                        page = await self._process_text_file(current_url, ext, current_depth)
-                    else:
-                        page = await self._process_html(crawler, current_url, current_depth)
-
-                    if page:
-                        links = page.pop("_links", [])
-                        self.pages.append(page)
-
-                        if progress:
-                            pct = min(len(self.pages) / max_pages, 1.0)
-                            details = f"{self.html_count} HTML"
-                            if self.pdf_count:
-                                details += f", {self.pdf_count} PDF"
-                            progress(pct, desc=f"Crawled {len(self.pages)} pages ({details})")
-
-                        if current_depth < depth:
-                            for link in links:
-                                if link not in self.visited:
-                                    queue.append((link, current_depth + 1))
+            async with AsyncWebCrawler(config=browser_config) as crawler:
+                next_layer = [url]
+
+                for current_depth in range(depth + 1):
+                    if not next_layer or len(self.pages) >= max_pages:
+                        break
+
+                    remaining = max_pages - len(self.pages)
+                    current_layer = next_layer[:remaining]
+                    next_layer = []
+
+                    for u in current_layer:
+                        self.visited.add(u)
+
+                    html_urls: list[str] = []
+                    static_tasks: list[tuple[str, str]] = []
+
+                    for u in current_layer:
+                        if self._is_static_asset(u):
+                            ext = self._get_ext(u)
+                            static_tasks.append((u, ext))
+                        else:
+                            html_urls.append(u)
+
+                    if static_tasks:
+                        coros = [
+                            self._process_pdf(u, current_depth) if e in PDF_TYPES
+                            else self._process_text_file(u, e, current_depth)
+                            for u, e in static_tasks
+                        ]
+                        static_results = await asyncio.gather(*coros, return_exceptions=True)
+                        for r in static_results:
+                            if isinstance(r, dict) and r:
+                                links = r.pop("_links", [])
+                                self.pages.append(r)
+                                if current_depth < depth:
+                                    for link in links:
+                                        if link not in self.visited:
+                                            self.visited.add(link)
+                                            next_layer.append(link)
+
+                    if html_urls:
+                        html_results = await crawler.arun_many(
+                            urls=html_urls,
+                            config=crawl_config,
+                        )
+                        for result in html_results:
+                            page = self._parse_arun_result(result, current_depth)
+                            if page:
+                                links = page.pop("_links", [])
+                                self.pages.append(page)
+                                if current_depth < depth:
+                                    for link in links:
+                                        if link not in self.visited:
+                                            self.visited.add(link)
+                                            next_layer.append(link)
+
+                    if progress:
+                        pct = min(len(self.pages) / max_pages, 1.0)
+                        details = f"{self.html_count} HTML"
+                        if self.pdf_count:
+                            details += f", {self.pdf_count} PDF"
+                        if self.text_count:
+                            details += f", {self.text_count} text"
+                        if self.rss_count:
+                            details += f", {self.rss_count} RSS"
+                        progress(pct, desc=f"Depth {current_depth}: {len(self.pages)} pages ({details})")
 
         if not self.pages:
             return pl.DataFrame(schema={
@@ -102,11 +233,134 @@ class Crawler:
         df = df.unique(subset=["content_hash"], keep="first")
         return df
 
+    def _parse_arun_result(self, result, depth: int) -> dict | None:
+        if not result.success:
+            return None
+
+        html = result.html or ""
+        markdown = result.markdown or ""
+        if hasattr(markdown, "raw_markdown"):
+            markdown = markdown.raw_markdown or ""
+
+        if not markdown.strip():
+            return None
+
+        if self._is_blocked_content(markdown, len(markdown)):
+            return None
+
+        links = self._extract_links(html, result.url) if depth < self.config.max_depth else []
+
+        self.html_count += 1
+        return {
+            "url": result.url,
+            "title": (
+                result.metadata.get("title", "") if result.metadata
+                else Path(urlparse(result.url).path).stem or result.url
+            ),
+            "markdown": markdown,
+            "html": html,
+            "depth": depth,
+            "timestamp": time.time(),
+            "content_hash": content_hash(markdown),
+            "file_type": "html",
+            "_links": links,
+        }
+
+    async def _try_rss_feed(self, base_url: str):
+        parsed = urlparse(base_url)
+        base = f"{parsed.scheme}://{parsed.netloc}"
+
+        feed_urls = [normalize_url(base + path) for path in RSS_PATHS]
+
+        for feed_url in feed_urls:
+            if feed_url in self.visited:
+                continue
+            try:
+                headers = self._build_headers()
+                resp = await self.http_client.get(feed_url, headers=headers)
+                if resp.status_code != 200:
+                    continue
+                content_type = resp.headers.get("content-type", "").lower()
+                text = resp.text
+                if len(text) < 100:
+                    continue
+                is_rss = (
+                    "xml" in content_type
+                    or "<rss" in text[:500]
+                    or "<feed" in text[:500]
+                    or "<rdf" in text[:500]
+                )
+                if not is_rss:
+                    continue
+
+                self.visited.add(feed_url)
+                items = self._parse_rss(text, feed_url)
+                for item in items:
+                    if len(self.pages) >= 50:
+                        break
+                    item["depth"] = 0
+                    if not any(p["content_hash"] == item["content_hash"] for p in self.pages):
+                        self.pages.append(item)
+                        self.rss_count += 1
+                if items:
+                    break
+            except Exception:
+                continue
+
+    def _parse_rss(self, xml_text: str, feed_url: str) -> list[dict]:
+        soup = BeautifulSoup(xml_text, "lxml-xml")
+        items = []
+
+        for item in soup.find_all(["item", "entry"]):
+            title_tag = item.find("title")
+            title = title_tag.get_text(strip=True) if title_tag else ""
+
+            link_tag = item.find("link")
+            if link_tag:
+                link = link_tag.get("href", "") or link_tag.get_text(strip=True)
+            else:
+                link = ""
+
+            desc_tag = item.find(["description", "summary", "content"])
+            description = desc_tag.get_text(strip=True) if desc_tag else ""
+
+            pub_tag = item.find(["pubDate", "published", "updated"])
+            pub_date = pub_tag.get_text(strip=True) if pub_tag else ""
+
+            if not description and not link:
+                continue
+
+            md = f"## {title}\n\n" if title else ""
+            if pub_date:
+                md += f"*Published: {pub_date}*\n\n"
+            md += description
+
+            if link:
+                link = urljoin(feed_url, link)
+                link = normalize_url(link)
+
+            items.append({
+                "url": link or feed_url,
+                "title": title or "RSS Item",
+                "markdown": md,
+                "html": f"<pre>{description}</pre>",
+                "depth": 0,
+                "timestamp": time.time(),
+                "content_hash": content_hash(md),
+                "file_type": "rss",
+                "_links": [link] if link else [],
+            })
+
+        return items
+
     async def _process_pdf(self, url: str, depth: int) -> dict | None:
         try:
-            resp = await self.http_client.get(url)
-            resp.raise_for_status()
+            resp = await self._fetch_with_retry(url)
+            if resp is None or resp.status_code != 200:
+                return None
             content = resp.content
+            if len(content) < 100:
+                return None
 
             import fitz
 
@@ -121,7 +375,7 @@ class Crawler:
             doc.close()
 
             markdown = "\n\n".join(md_parts)
-            if not markdown.strip():
+            if not markdown.strip() or len(markdown) < self.config.crawl_min_content_bytes:
                 return None
 
             self.pdf_count += 1
@@ -141,10 +395,11 @@ class Crawler:
 
     async def _process_text_file(self, url: str, ext: str, depth: int) -> dict | None:
         try:
-            resp = await self.http_client.get(url)
-            resp.raise_for_status()
+            resp = await self._fetch_with_retry(url)
+            if resp is None or resp.status_code != 200:
+                return None
             text = resp.text
-            if not text.strip():
+            if not text.strip() or len(text) < self.config.crawl_min_content_bytes:
                 return None
 
             title = Path(urlparse(url).path).stem or "file"
@@ -163,40 +418,6 @@ class Crawler:
         except Exception:
             return None
 
-    async def _process_html(self, crawler: AsyncWebCrawler, url: str, depth: int) -> dict | None:
-        try:
-            result = await crawler.arun(
-                url=url,
-                word_count_threshold=10,
-                bypass_cache=True,
-            )
-
-            if not result.success:
-                return None
-
-            html = result.html or ""
-            markdown = result.markdown or ""
-            if not markdown.strip():
-                return None
-
-            links = self._extract_links(html, url) if depth < self.config.max_depth else []
-
-            self.html_count += 1
-            return {
-                "url": url,
-                "title": (result.metadata.get("title", "") if result.metadata
-                          else Path(urlparse(url).path).stem or url),
-                "markdown": markdown,
-                "html": html,
-                "depth": depth,
-                "timestamp": time.time(),
-                "content_hash": content_hash(markdown),
-                "file_type": "html",
-                "_links": links,
-            }
-        except Exception:
-            return None
-
     def _extract_links(self, html: str, base_url: str) -> list[str]:
         soup = BeautifulSoup(html, "lxml")
         links = []
@@ -204,6 +425,8 @@ class Crawler:
             href = a_tag["href"].strip()
             if not href or href.startswith("#") or href.startswith("javascript:"):
                 continue
+            if href.startswith("mailto:") or href.startswith("tel:"):
+                continue
             absolute_url = urljoin(base_url, href)
             absolute_url = absolute_url.rstrip("/")
             if absolute_url.startswith("http://"):