Update crawler: incomplete shard upload + UI options

app.py

@@ -6,6 +6,7 @@ import contextlib
 import inspect
 import threading
 import traceback
+from collections import deque
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from html import escape
@@ -111,6 +112,61 @@ APP_CSS = """
   --border: #2a5d36;
 }
 
+:root[data-crawler-theme="sunset"] {
+  --bg-main: #1c0f0b;
+  --bg-surface: #2f1810;
+  --bg-panel: #422015;
+  --text-main: #ffeede;
+  --text-muted: #d9b59d;
+  --accent: #ff7f3f;
+  --accent-2: #ff4f81;
+  --border: #6e3525;
+}
+
+:root[data-crawler-theme="ocean"] {
+  --bg-main: #04131d;
+  --bg-surface: #092230;
+  --bg-panel: #0e3144;
+  --text-main: #e8fbff;
+  --text-muted: #9fc3cf;
+  --accent: #2cd9ff;
+  --accent-2: #38ffcb;
+  --border: #1b5062;
+}
+
+:root[data-crawler-theme="amber"] {
+  --bg-main: #171104;
+  --bg-surface: #2a1d07;
+  --bg-panel: #3a2a0a;
+  --text-main: #fff6dc;
+  --text-muted: #d2bd84;
+  --accent: #ffb300;
+  --accent-2: #ffd54f;
+  --border: #6b4d12;
+}
+
+:root[data-crawler-theme="graphite"] {
+  --bg-main: #0f1012;
+  --bg-surface: #1a1d21;
+  --bg-panel: #242a30;
+  --text-main: #f1f3f6;
+  --text-muted: #adb5bf;
+  --accent: #8fa0b7;
+  --accent-2: #d7dce2;
+  --border: #3a424c;
+}
+
+:root[data-crawler-theme="mint"] {
+  --bg-main: #06140f;
+  --bg-surface: #0b2319;
+  --bg-panel: #123125;
+  --text-main: #e8fff4;
+  --text-muted: #a4d8bf;
+  --accent: #3dffb2;
+  --accent-2: #7df9d1;
+  --border: #256347;
+}
+
 .gradio-container {
   background:
     radial-gradient(1200px 550px at 8% 0%, color-mix(in srgb, var(--accent) 18%, transparent), transparent),
@@ -404,6 +460,8 @@ def build_crawler_config(
     max_response_bytes: int,
     shard_size_rows: int,
     enable_hf_upload: bool,
+    upload_incomplete_shards: bool,
+    incomplete_shard_flush_seconds: float,
     hf_repo_id: str,
     hf_token: str,
     hf_private_repo: bool,
@@ -421,6 +479,8 @@ def build_crawler_config(
         shard_size_rows=int(shard_size_rows),
         output_dir=Path(__file__).resolve().parent / "shards",
         enable_hf_upload=bool(enable_hf_upload),
+        upload_incomplete_shards=bool(upload_incomplete_shards),
+        incomplete_shard_flush_seconds=float(incomplete_shard_flush_seconds),
         hf_repo_id=hf_repo_id.strip(),
         hf_token=hf_token.strip(),
         hf_private_repo=bool(hf_private_repo),
@@ -446,7 +506,7 @@ class CrawlerRunManager:
         self._loop: asyncio.AbstractEventLoop | None = None
         self._crawler: AsyncCrawler | None = None
         self._state = RunState()
-        self._logs: list[str] = []
+        self._logs: deque[str] = deque(maxlen=600)
         self._last_snapshot: dict[str, Any] | None = None
 
     def start(self, config: CrawlerConfig) -> str:
@@ -594,7 +654,7 @@ class CrawlerRunManager:
                 "record_queue": 0,
                 "stop_event": False,
             }
-            logs_text = "\n".join(self._logs[-500:])
+            logs_text = "\n".join(self._logs)
 
         status_lines = [
             "### Crawler Status",
@@ -638,6 +698,8 @@ def _start_crawl(
     max_response_bytes: int,
     shard_size_rows: int,
     enable_hf_upload: bool,
+    upload_incomplete_shards: bool,
+    incomplete_shard_flush_seconds: float,
     hf_repo_id: str,
     hf_token: str,
     hf_private_repo: bool,
@@ -651,6 +713,8 @@ def _start_crawl(
             max_response_bytes=max_response_bytes,
             shard_size_rows=shard_size_rows,
             enable_hf_upload=enable_hf_upload,
+            upload_incomplete_shards=upload_incomplete_shards,
+            incomplete_shard_flush_seconds=incomplete_shard_flush_seconds,
             hf_repo_id=hf_repo_id,
             hf_token=hf_token,
             hf_private_repo=hf_private_repo,
@@ -672,6 +736,8 @@ def start_crawl_standard(
     max_response_bytes: int,
     shard_size_rows: int,
     enable_hf_upload: bool,
+    upload_incomplete_shards: bool,
+    incomplete_shard_flush_seconds: float,
     hf_repo_id: str,
     hf_token: str,
     hf_private_repo: bool,
@@ -685,6 +751,8 @@ def start_crawl_standard(
         max_response_bytes=max_response_bytes,
         shard_size_rows=shard_size_rows,
         enable_hf_upload=enable_hf_upload,
+        upload_incomplete_shards=upload_incomplete_shards,
+        incomplete_shard_flush_seconds=incomplete_shard_flush_seconds,
         hf_repo_id=hf_repo_id,
         hf_token=hf_token,
         hf_private_repo=hf_private_repo,
@@ -699,6 +767,8 @@ def start_crawl_super(
     max_response_bytes: int,
     shard_size_rows: int,
     enable_hf_upload: bool,
+    upload_incomplete_shards: bool,
+    incomplete_shard_flush_seconds: float,
     hf_repo_id: str,
     hf_token: str,
     hf_private_repo: bool,
@@ -712,6 +782,8 @@ def start_crawl_super(
         max_response_bytes=max_response_bytes,
         shard_size_rows=shard_size_rows,
         enable_hf_upload=enable_hf_upload,
+        upload_incomplete_shards=upload_incomplete_shards,
+        incomplete_shard_flush_seconds=incomplete_shard_flush_seconds,
         hf_repo_id=hf_repo_id,
         hf_token=hf_token,
         hf_private_repo=hf_private_repo,
@@ -730,9 +802,16 @@ def poll_dashboard() -> tuple[str, str, str, str]:
     return _format_dashboard_response(status, snapshot, logs)
 
 
-def toggle_hf_fields(enable_hf_upload: bool) -> tuple[Any, Any, Any, Any]:
+def toggle_hf_fields(enable_hf_upload: bool) -> tuple[Any, Any, Any, Any, Any]:
     update = gr.update(visible=enable_hf_upload)
-    return update, update, update, update
+    return update, update, update, update, update
+
+
+def toggle_incomplete_flush_field(
+    enable_hf_upload: bool,
+    upload_incomplete_shards: bool,
+) -> Any:
+    return gr.update(visible=bool(enable_hf_upload and upload_incomplete_shards))
 
 
 def build_ui() -> gr.Blocks:
@@ -752,7 +831,18 @@ def build_ui() -> gr.Blocks:
 
         with gr.Row():
             theme_name = gr.Dropdown(
-                choices=["red", "blue", "light", "dark", "green"],
+                choices=[
+                    "red",
+                    "blue",
+                    "light",
+                    "dark",
+                    "green",
+                    "sunset",
+                    "ocean",
+                    "amber",
+                    "graphite",
+                    "mint",
+                ],
                 value="dark",
                 label="Theme",
                 interactive=True,
@@ -838,6 +928,19 @@ def build_ui() -> gr.Blocks:
                 value="crawl_shards",
                 visible=False,
             )
+            upload_incomplete_shards = gr.Checkbox(
+                label="Upload incomplete shard buffers",
+                value=False,
+                visible=False,
+            )
+            incomplete_shard_flush_seconds = gr.Slider(
+                label="Incomplete Upload Flush Interval (seconds)",
+                minimum=5,
+                maximum=300,
+                step=1,
+                value=int(defaults.incomplete_shard_flush_seconds),
+                visible=False,
+            )
 
         with gr.Row():
             start_button = gr.Button("Start Crawl (12 Threads)", variant="primary")
@@ -856,6 +959,8 @@ def build_ui() -> gr.Blocks:
             max_response_bytes,
             shard_size_rows,
             enable_hf_upload,
+            upload_incomplete_shards,
+            incomplete_shard_flush_seconds,
             hf_repo_id,
             hf_token,
             hf_private_repo,
@@ -871,7 +976,23 @@ def build_ui() -> gr.Blocks:
         enable_hf_upload.change(
             toggle_hf_fields,
             inputs=enable_hf_upload,
-            outputs=[hf_repo_id, hf_token, hf_private_repo, hf_path_prefix],
+            outputs=[
+                hf_repo_id,
+                hf_token,
+                hf_private_repo,
+                hf_path_prefix,
+                upload_incomplete_shards,
+            ],
+        )
+        enable_hf_upload.change(
+            toggle_incomplete_flush_field,
+            inputs=[enable_hf_upload, upload_incomplete_shards],
+            outputs=[incomplete_shard_flush_seconds],
+        )
+        upload_incomplete_shards.change(
+            toggle_incomplete_flush_field,
+            inputs=[enable_hf_upload, upload_incomplete_shards],
+            outputs=[incomplete_shard_flush_seconds],
         )
 
         seed_urls_table.change(

crawler/config.py

@@ -51,6 +51,8 @@ class CrawlerConfig:
     parquet_compression_level: int = 9
 
     enable_hf_upload: bool = False
+    upload_incomplete_shards: bool = False
+    incomplete_shard_flush_seconds: float = 30.0
     hf_repo_id: str = ""
     hf_token: str = ""
     hf_repo_type: str = "dataset"
@@ -59,7 +61,8 @@ class CrawlerConfig:
 
     total_workers: int = NORMAL_TOTAL_WORKERS
     request_delay_global_seconds: float = 0.02
-    request_delay_per_domain_seconds: float = 2.0
+    request_delay_per_domain_seconds: float = 0.0
+    same_site_delay_per_worker_seconds: float = 0.5
 
     robots_cache_ttl_seconds: float = 3600.0
     robots_fail_closed: bool = True
@@ -89,6 +92,9 @@ class CrawlerConfig:
         self.hf_repo_id = self.hf_repo_id.strip()
         self.hf_token = self.hf_token.strip()
         self.hf_path_prefix = self.hf_path_prefix.strip() or "crawl_shards"
+        self.incomplete_shard_flush_seconds = float(self.incomplete_shard_flush_seconds)
+        if self.incomplete_shard_flush_seconds <= 0:
+            raise ValueError("incomplete_shard_flush_seconds must be > 0.")
 
         if self.enable_hf_upload:
             if not self.hf_repo_id:

crawler/engine.py

@@ -4,6 +4,7 @@ import asyncio
 import contextlib
 from collections import deque
 from typing import Any
+from urllib.parse import urlsplit
 
 import aiohttp
 
@@ -185,6 +186,9 @@ class AsyncCrawler:
         del worker_id
         assert self.rate_limiter is not None
         assert self.robots_policy is not None
+        loop = asyncio.get_running_loop()
+        last_domain = ""
+        last_request_started = 0.0
 
         while True:
             url = await self.fetch_queue.get()
@@ -199,6 +203,16 @@ class AsyncCrawler:
 
             self.active_fetchers += 1
             try:
+                requested_domain = (urlsplit(url).hostname or "").lower().strip(".")
+                if requested_domain and requested_domain == last_domain:
+                    elapsed = loop.time() - last_request_started
+                    wait = self.config.same_site_delay_per_worker_seconds - elapsed
+                    if wait > 0:
+                        await asyncio.sleep(wait)
+                if requested_domain:
+                    last_domain = requested_domain
+                    last_request_started = loop.time()
+
                 outcome = await fetch_url(
                     session,
                     url,

crawler/shards.py

@@ -52,8 +52,33 @@ class ParquetShardWriter:
         await self.uploader.initialize()
 
     async def consume(self, record_queue: asyncio.Queue[dict[str, Any] | None]) -> None:
+        allow_incomplete_upload = (
+            self.config.enable_hf_upload and self.config.upload_incomplete_shards
+        )
+        loop = asyncio.get_running_loop()
+        last_flush_started = loop.time()
         while True:
-            item = await record_queue.get()
+            try:
+                if allow_incomplete_upload:
+                    timeout = self.config.incomplete_shard_flush_seconds
+                    if self.buffer:
+                        elapsed = loop.time() - last_flush_started
+                        timeout = max(
+                            0.0,
+                            self.config.incomplete_shard_flush_seconds - elapsed,
+                        )
+                    item = await asyncio.wait_for(
+                        record_queue.get(),
+                        timeout=timeout,
+                    )
+                else:
+                    item = await record_queue.get()
+            except asyncio.TimeoutError:
+                if self.buffer:
+                    await self.flush()
+                    last_flush_started = loop.time()
+                continue
+
             if item is None:
                 record_queue.task_done()
                 break
@@ -62,6 +87,12 @@ class ParquetShardWriter:
                 self.buffer.append(item)
                 if len(self.buffer) >= self.config.shard_size_rows:
                     await self.flush()
+                    last_flush_started = loop.time()
+                elif allow_incomplete_upload and self.buffer:
+                    elapsed = loop.time() - last_flush_started
+                    if elapsed >= self.config.incomplete_shard_flush_seconds:
+                        await self.flush()
+                        last_flush_started = loop.time()
             finally:
                 record_queue.task_done()