Improve robustness and fetch tool

.flake8

@@ -1,2 +1,3 @@
 [flake8]
 ignore = E501,E203,W503,E704
+exclude = .git,__pycache__,docs,build,dist,.venv,.mypy_cache,.pytest_cache

.github/workflows/check.yml

@@ -1,3 +1,7 @@
+# SPDX-FileCopyrightText: 2026 Greg Brandt <brandt.greg@gmail.com>
+#
+# SPDX-License-Identifier: Apache-2.0
+
 name: Check
 
 on:

.github/workflows/release.yml

@@ -1,3 +1,7 @@
+# SPDX-FileCopyrightText: 2026 Greg Brandt <brandt.greg@gmail.com>
+#
+# SPDX-License-Identifier: Apache-2.0
+
 name: Publish to PyPI
 
 on:

.gitignore

@@ -15,3 +15,4 @@ downloaded_files/
 env
 tags
 tmp
+latest_logs/

pyproject.toml

@@ -4,7 +4,7 @@
 
 [tool.poetry]
 name = "rubbernecker"
-version = "0.0.3"
+version = "0.0.4"
 description = "Web scraping engine"
 authors = ["Greg Brandt <brandt.greg@gmail.com>"]
 license = "Apache-2.0"
@@ -20,9 +20,10 @@ classifiers = [
 
 [tool.poetry.dependencies]
 python = "^3.12"
-avrokit = "^0.0.3"
+avrokit = "^0.0.4"
 beautifulsoup4 = "^4.13.4"
 seleniumbase = "^4.46.4"
+requests = "^2.32.5"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^9.0.2"

rubbernecker/__main__.py

@@ -10,11 +10,13 @@
 from .parse import ParseTool
 from .crawl import CrawlTool
 from .browser import ProxyTool
+from .fetch import FetchTool
 
 TOOLS: list[Tool] = [
     CrawlTool(),
     ParseTool(),
     ProxyTool(),
+    FetchTool(),
 ]
 
 

rubbernecker/crawl/tool.py

@@ -92,6 +92,37 @@ def bloom_filter_key(self, url: str) -> str:
         parsed_url = urlparse(url.lower())
         return f"{parsed_url.netloc}:{parsed_url.path}:{parsed_url.query}"
 
+    def is_driver_alive(self, driver) -> bool:
+        """
+        Check if the Selenium driver is still alive.
+
+        :param driver: The Selenium WebDriver instance.
+        :return: True if the driver is alive, False otherwise.
+        """
+        try:
+            driver.current_url
+            return True
+        except Exception:
+            return False
+
+    def restart_browser(self, sb, headless: bool, proxy_server: str | None = None):
+        """
+        Restart the Selenium browser.
+
+        :param sb: The SeleniumBase instance.
+        :param headless: Whether to run in headless mode.
+        :param proxy_server: Optional proxy server URL.
+        """
+        sb.driver.quit()
+        sb.__init__(
+            uc=True,
+            test=True,
+            incognito=True,
+            headless=headless,
+            proxy=proxy_server,
+            locale="en-US",
+        )
+
     def load_bloom_filter(self, output_url: URL) -> BloomFilter | None:
         """
         Load a Bloom filter from the output URL.
@@ -144,6 +175,8 @@ def load_requests(
             with input_url.with_mode("r") as file:
                 for line in file:
                     url = line.strip()
+                    if isinstance(url, bytes):
+                        url = url.decode("utf-8")
                     if not bloom_filter or not bloom_filter.check(
                         self.bloom_filter_key(url)
                     ):
@@ -154,6 +187,8 @@ def load_requests(
                 for line in file:
                     data = json.loads(line)
                     url = data["url"]
+                    if isinstance(url, bytes):
+                        url = url.decode("utf-8")
                     if not bloom_filter or not bloom_filter.check(
                         self.bloom_filter_key(url)
                     ):
@@ -164,6 +199,8 @@ def load_requests(
                 for record in reader:
                     if isinstance(record, dict) and "url" in record:
                         url = record["url"]
+                        if isinstance(url, bytes):
+                            url = url.decode("utf-8")
                         if not bloom_filter or not bloom_filter.check(
                             self.bloom_filter_key(url)
                         ):
@@ -190,8 +227,11 @@ def crawl(
         # Load the Bloom filter if needed
         bloom_filter: BloomFilter | None = None
         if use_bloom_filter:
-            bloom_filter = self.load_bloom_filter(base_output_url)
-            logger.debug("Loaded Bloom filter: %s", bloom_filter)
+            try:
+                bloom_filter = self.load_bloom_filter(base_output_url)
+                logger.debug("Loaded Bloom filter: %s", bloom_filter)
+            except Exception as e:
+                logger.warning("Failed to load Bloom filter: %s", e)
 
         stats = CrawlToolStats()
         first_request = True
@@ -230,6 +270,14 @@ def crawl(
 
                         while True:
                             try:
+                                # Check if browser is still alive
+                                if not self.is_driver_alive(sb.driver):
+                                    logger.warning(
+                                        "Browser died, restarting for URL: %s", url
+                                    )
+                                    first_request = True
+                                    self.restart_browser(sb, headless, proxy_server)
+
                                 logger.info(
                                     "Crawling URL: %s (depth=%d, retries=%s)",
                                     url,
@@ -251,6 +299,9 @@ def crawl(
                                             "Page.navigate", {"url": url}
                                         )
 
+                                # Wait for page to be fully loaded before capturing
+                                sb.wait_for_ready_state_complete()
+
                                 # Perform load actions if provided
                                 if load_actions:
                                     for plan in load_actions:
@@ -277,9 +328,12 @@ def crawl(
                                     time.sleep(sleep_success)
 
                                 # Save the crawled data to the output URL
+                                current_url = sb.get_current_url()
+                                if isinstance(current_url, bytes):
+                                    current_url = current_url.decode("utf-8")
                                 writer.append(
                                     {
-                                        "url": sb.get_current_url(),
+                                        "url": current_url,
                                         "timestamp": timestamp,
                                         "body": sb.get_page_source(),
                                     }
@@ -309,6 +363,16 @@ def crawl(
                                     # No crawl actions provided, just break
                                     break
                             except Exception as e:
+                                # Check if browser died
+                                if not self.is_driver_alive(sb.driver):
+                                    logger.warning(
+                                        "Browser died during crawl, restarting: %s", e
+                                    )
+                                    first_request = True
+                                    self.restart_browser(sb, headless, proxy_server)
+                                    # Retry the same URL
+                                    continue
+
                                 # Sleep on error if not in interactive mode
                                 if not interactive:
                                     time.sleep(sleep_error)

rubbernecker/fetch/__init__.py

@@ -0,0 +1,7 @@
+# SPDX-FileCopyrightText: 2026 Greg Brandt <brandt.greg@gmail.com>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from .tool import FetchTool
+
+__all__ = ["FetchTool"]

rubbernecker/fetch/tool.py

@@ -0,0 +1,166 @@
+# SPDX-FileCopyrightText: 2026 Greg Brandt <brandt.greg@gmail.com>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import argparse
+import logging
+import requests
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass
+from urllib.parse import urlparse
+from avrokit import URL, parse_url
+from rubbernecker.crawl.bloomfilter import BloomFilter
+
+logger = logging.getLogger("fetchtool")
+
+
+@dataclass
+class FetchToolStats:
+    count_input: int = 0
+    count_output: int = 0
+    count_error: int = 0
+    count_skipped: int = 0
+
+
+class FetchTool:
+    def name(self) -> str:
+        return "fetch"
+
+    def configure(self, subparsers: argparse._SubParsersAction) -> None:
+        parser = subparsers.add_parser(self.name(), help="Fetch assets from URLs")
+        parser.add_argument(
+            "input_url",
+            help="URL to the text file containing URLs to fetch",
+        )
+        parser.add_argument(
+            "output_url",
+            help="URL to the output directory",
+        )
+        parser.add_argument(
+            "--parallelism",
+            type=int,
+            default=1,
+            help="Number of concurrent fetches (default: 1)",
+        )
+        parser.add_argument(
+            "--force",
+            action="store_true",
+            help="Redownload even if file already exists (default: skip existing)",
+        )
+
+    def load_bloom_filter(self, output_url: URL) -> BloomFilter | None:
+        if output_url.exists():
+            bloom_filter = BloomFilter()
+            count = 0
+            base_path = output_url.url.rstrip("/") + "/"
+            for url in output_url.expand():
+                if url.url.startswith(base_path):
+                    path = url.url[len(base_path) :]
+                else:
+                    parsed = urlparse(url.url)
+                    path = parsed.path.lstrip("/")
+                if path:
+                    bloom_filter.add(path)
+                    count += 1
+            if count > 0:
+                logger.debug(
+                    "Loaded %d paths into Bloom filter from %s", count, output_url
+                )
+                return bloom_filter
+        return None
+
+    def fetch_url(
+        self,
+        url: str,
+        output_base_url: URL,
+        force: bool,
+        bloom_filter: BloomFilter | None,
+    ) -> tuple[bool, bool]:
+        try:
+            response = requests.get(url, timeout=30)
+            response.raise_for_status()
+            content = response.content
+
+            parsed = urlparse(url)
+            path = parsed.path.lstrip("/")
+
+            if not path:
+                logger.warning("URL %s has no path, skipping", url)
+                return False, False
+
+            if not force and bloom_filter and bloom_filter.check(path):
+                logger.debug("File already exists (bloom filter), skipping: %s", path)
+                return True, True
+
+            output_url = parse_url(output_base_url.url + "/" + path)
+            with output_url.with_mode("wb") as f:
+                f.write(content)
+
+            if bloom_filter:
+                bloom_filter.add(path)
+
+            logger.debug("Fetched %s -> %s", url, output_url)
+            return True, False
+
+        except Exception as e:
+            logger.error("Error fetching %s: %s", url, e)
+            return False, False
+
+    def fetch(
+        self, input_url: URL, output_url: URL, parallelism: int, force: bool = False
+    ) -> FetchToolStats:
+        stats = FetchToolStats()
+
+        bloom_filter: BloomFilter | None = None
+        if not force:
+            try:
+                bloom_filter = self.load_bloom_filter(output_url)
+            except Exception as e:
+                logger.warning("Failed to load bloom filter: %s", e)
+
+        with input_url.with_mode("r") as f:
+            urls = [line.strip() for line in f if line.strip()]
+
+        logger.info("Fetching %d URLs with parallelism=%d", len(urls), parallelism)
+
+        if bloom_filter:
+            logger.info("Bloom filter loaded, will skip existing files")
+
+        with ThreadPoolExecutor(max_workers=parallelism) as executor:
+            futures = {
+                executor.submit(
+                    self.fetch_url, url, output_url, force, bloom_filter
+                ): url
+                for url in urls
+            }
+
+            for future in as_completed(futures):
+                url = futures[future]
+                stats.count_input += 1
+                try:
+                    success, skipped = future.result()
+                    if success:
+                        if skipped:
+                            stats.count_skipped += 1
+                        else:
+                            stats.count_output += 1
+                    else:
+                        stats.count_error += 1
+                except Exception as e:
+                    logger.error("Error processing %s: %s", url, e)
+                    stats.count_error += 1
+
+                if logging.DEBUG != logger.getEffectiveLevel():
+                    if stats.count_output > 0 and stats.count_output % 100 == 0:
+                        logger.info("%s", stats)
+
+        return stats
+
+    def run(self, args: argparse.Namespace) -> None:
+        stats = self.fetch(
+            parse_url(args.input_url),
+            parse_url(args.output_url),
+            args.parallelism,
+            args.force,
+        )
+        logger.info("%s (done)", stats)