Upload AutoWS app files without plan/readme

README.md

@@ -1,15 +0,0 @@
----
-license: cc-by-nc-nd-4.0
-title: AutoW
-sdk: gradio
-emoji: 📚
-colorFrom: indigo
-colorTo: red
-short_description: Automated Web Scraping!
-sdk_version: 6.5.1
----
-# This is the only official space for this project.
-# Do not trust mirrors/forks. This is only native to HuggingFace.
-# This does not log or steal your HF token. I cannot guarantee that forks ot mirrors do not.
-# To combat this, I'm using the most restrictive license I know.
-# Licensed under Creative Commons Attribution Non Commercial No Derivatives 4.0

app.py

@@ -0,0 +1,924 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import inspect
+import threading
+import traceback
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from html import escape
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+
+import huggingface_hub as hf_hub
+
+if not hasattr(hf_hub, "HfFolder"):
+    class _CompatHfFolder:
+        @staticmethod
+        def get_token() -> str | None:
+            return None
+
+        @staticmethod
+        def save_token(token: str) -> None:
+            del token
+            return None
+
+        @staticmethod
+        def delete_token() -> None:
+            return None
+
+    hf_hub.HfFolder = _CompatHfFolder  # type: ignore[attr-defined]
+
+import gradio as gr
+
+from crawler import (
+    MAX_SHARD_ROWS,
+    NORMAL_TOTAL_WORKERS,
+    SUPER_TOTAL_WORKERS,
+    AsyncCrawler,
+    CrawlerConfig,
+)
+
+APP_CSS = """
+:root {
+  --bg-main: #0a0d12;
+  --bg-surface: #151a22;
+  --bg-panel: #1b2230;
+  --text-main: #f0f4fb;
+  --text-muted: #9aa4b6;
+  --accent: #3bd9ff;
+  --accent-2: #4cffb1;
+  --border: #2f3a50;
+  --shadow: 0 18px 36px rgba(0, 0, 0, 0.45);
+}
+
+:root[data-crawler-theme="red"] {
+  --bg-main: #17080c;
+  --bg-surface: #250d15;
+  --bg-panel: #341322;
+  --text-main: #f8e8ee;
+  --text-muted: #d5b0c0;
+  --accent: #7a0018;
+  --accent-2: #8e3ff5;
+  --border: #5a2035;
+}
+
+:root[data-crawler-theme="blue"] {
+  --bg-main: #021116;
+  --bg-surface: #08222c;
+  --bg-panel: #0e2f3b;
+  --text-main: #eaffff;
+  --text-muted: #8fbcc7;
+  --accent: #2fff9d;
+  --accent-2: #13e5ff;
+  --border: #1e5662;
+}
+
+:root[data-crawler-theme="light"] {
+  --bg-main: #f6f7f9;
+  --bg-surface: #ffffff;
+  --bg-panel: #eceff2;
+  --text-main: #111317;
+  --text-muted: #60666f;
+  --accent: #2a2f37;
+  --accent-2: #868b95;
+  --border: #d0d4db;
+  --shadow: 0 10px 25px rgba(35, 42, 52, 0.16);
+}
+
+:root[data-crawler-theme="dark"] {
+  --bg-main: #090909;
+  --bg-surface: #141414;
+  --bg-panel: #1d1d1d;
+  --text-main: #f0f0f0;
+  --text-muted: #a8a8a8;
+  --accent: #444444;
+  --accent-2: #686868;
+  --border: #2b2b2b;
+}
+
+:root[data-crawler-theme="green"] {
+  --bg-main: #08110b;
+  --bg-surface: #0f1d14;
+  --bg-panel: #17301e;
+  --text-main: #e8f8ed;
+  --text-muted: #97bc9f;
+  --accent: #2ea84b;
+  --accent-2: #185f2a;
+  --border: #2a5d36;
+}
+
+.gradio-container {
+  background:
+    radial-gradient(1200px 550px at 8% 0%, color-mix(in srgb, var(--accent) 18%, transparent), transparent),
+    radial-gradient(900px 600px at 100% 0%, color-mix(in srgb, var(--accent-2) 14%, transparent), transparent),
+    var(--bg-main);
+  color: var(--text-main);
+}
+
+.gradio-container .block,
+.gradio-container .form,
+.gradio-container .gr-box,
+.gradio-container .panel-wrap {
+  background: color-mix(in srgb, var(--bg-surface) 92%, transparent) !important;
+  border: 1px solid var(--border) !important;
+  box-shadow: var(--shadow);
+}
+
+.gradio-container h1,
+.gradio-container h2,
+.gradio-container h3,
+.gradio-container p,
+.gradio-container label,
+.gradio-container .prose,
+.gradio-container .prose * {
+  color: var(--text-main) !important;
+}
+
+.gradio-container input,
+.gradio-container textarea,
+.gradio-container select {
+  background: var(--bg-panel) !important;
+  color: var(--text-main) !important;
+  border: 1px solid var(--border) !important;
+}
+
+.gradio-container button {
+  border: 1px solid var(--border) !important;
+}
+
+.gradio-container button.primary {
+  background: linear-gradient(135deg, var(--accent), var(--accent-2)) !important;
+  color: #0b0e13 !important;
+  font-weight: 700;
+}
+
+.seed-widget,
+.token-widget {
+  display: flex;
+  flex-direction: column;
+  gap: 0.75rem;
+  border: 1px solid var(--border);
+  border-radius: 0.9rem;
+  padding: 0.85rem;
+  background: color-mix(in srgb, var(--bg-panel) 86%, transparent);
+}
+
+.seed-stats,
+.token-stats {
+  display: grid;
+  grid-template-columns: repeat(3, minmax(0, 1fr));
+  gap: 0.6rem;
+}
+
+.seed-stats > span,
+.token-stats > span {
+  display: block;
+  padding: 0.55rem;
+  border: 1px solid var(--border);
+  border-radius: 0.6rem;
+  background: color-mix(in srgb, var(--bg-surface) 90%, transparent);
+  color: var(--text-main);
+  font-size: 0.9rem;
+}
+
+.seed-chip-wrap {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 0.45rem;
+}
+
+.seed-chip {
+  border: 1px solid var(--border);
+  border-radius: 999px;
+  padding: 0.24rem 0.7rem;
+  color: var(--text-main);
+  background: linear-gradient(
+    145deg,
+    color-mix(in srgb, var(--accent) 20%, transparent),
+    color-mix(in srgb, var(--accent-2) 15%, transparent)
+  );
+  font-size: 0.83rem;
+}
+
+.seed-empty,
+.seed-overflow,
+.token-note {
+  color: var(--text-muted);
+  font-size: 0.83rem;
+  padding: 0.24rem 0.3rem;
+}
+"""
+
+THEME_JS = """
+(theme_name) => {
+  const theme = theme_name || "dark";
+  document.documentElement.setAttribute("data-crawler-theme", theme);
+  return [];
+}
+"""
+
+SEED_WIDGET_JS = """
+(seed_rows) => {
+  const parseRows = (rows) => {
+    if (!Array.isArray(rows)) return [];
+    const out = [];
+    for (const row of rows) {
+      let value = "";
+      if (Array.isArray(row)) {
+        value = String(row[0] ?? "").trim();
+      } else if (row && typeof row === "object") {
+        value = String(Object.values(row)[0] ?? "").trim();
+      } else if (row !== null && row !== undefined) {
+        value = String(row).trim();
+      }
+      if (value) out.push(value);
+    }
+    return out;
+  };
+
+  const dedupe = (values) => {
+    const seen = new Set();
+    const out = [];
+    for (const value of values) {
+      if (!seen.has(value)) {
+        seen.add(value);
+        out.push(value);
+      }
+    }
+    return out;
+  };
+
+  const domainOf = (value) => {
+    try {
+      return new URL(value).hostname || "";
+    } catch {
+      return "";
+    }
+  };
+
+  const escapeHtml = (value) => String(value)
+    .replaceAll("&", "&")
+    .replaceAll("<", "&lt;")
+    .replaceAll(">", "&gt;")
+    .replaceAll('"', "&quot;")
+    .replaceAll("'", "&#39;");
+
+  const seeds = dedupe(parseRows(seed_rows));
+  const domainSet = new Set(seeds.map(domainOf).filter(Boolean));
+  const chips = seeds.length
+    ? seeds.slice(0, 12).map((url) => `<span class=\"seed-chip\">${escapeHtml(url)}</span>`).join("")
+    : '<span class=\"seed-empty\">No seed URLs configured yet.</span>';
+  const overflow = seeds.length > 12
+    ? `<span class=\"seed-overflow\">+${seeds.length - 12} more</span>`
+    : "";
+
+  return `<div class=\"seed-widget\"><div class=\"seed-stats\"><span><strong>${seeds.length}</strong> seeds</span><span><strong>${domainSet.size}</strong> domains</span><span><strong>${seeds.slice(0, 1).join("").length || 0}</strong> first-url chars</span></div><div class=\"seed-chip-wrap\">${chips}${overflow}</div></div>`;
+}
+"""
+
+
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+def safe_queue_size(queue: Any) -> int:
+    try:
+        return int(queue.qsize())
+    except Exception:
+        return -1
+
+
+def parse_seed_url_rows(rows: Any) -> list[str]:
+    if rows is None:
+        return []
+
+    if isinstance(rows, (list, tuple)):
+        rows_iterable: list[Any] = list(rows)
+    elif hasattr(rows, "values"):
+        try:
+            rows_iterable = rows.values.tolist()  # pandas.DataFrame path
+        except Exception:
+            rows_iterable = []
+    else:
+        rows_iterable = [rows]
+
+    items: list[str] = []
+    for row in rows_iterable:
+        value = ""
+        if isinstance(row, dict):
+            value = str(next(iter(row.values()), "") or "").strip()
+        elif isinstance(row, (list, tuple)):
+            value = str(row[0] if row else "").strip()
+        elif row is not None:
+            value = str(row).strip()
+
+        if value:
+            items.append(value)
+    return items
+
+
+def unique_preserve_order(values: list[str]) -> list[str]:
+    seen: set[str] = set()
+    out: list[str] = []
+    for value in values:
+        if value in seen:
+            continue
+        seen.add(value)
+        out.append(value)
+    return out
+
+
+def collect_seed_urls(seed_urls_table: Any) -> list[str]:
+    return unique_preserve_order(parse_seed_url_rows(seed_urls_table))
+
+
+def render_seed_widget_html(seed_urls_table: Any) -> str:
+    seeds = collect_seed_urls(seed_urls_table)
+    domains = {(urlsplit(u).hostname or "").lower().strip(".") for u in seeds}
+    domains = {d for d in domains if d}
+
+    chips = [f'<span class="seed-chip">{escape(url)}</span>' for url in seeds[:12]]
+    chips_html = "".join(chips) if chips else '<span class="seed-empty">No seed URLs configured yet.</span>'
+    overflow_html = f'<span class="seed-overflow">+{len(seeds) - 12} more</span>' if len(seeds) > 12 else ""
+
+    return (
+        '<div class="seed-widget">'
+        '<div class="seed-stats">'
+        f"<span><strong>{len(seeds)}</strong> seeds</span>"
+        f"<span><strong>{len(domains)}</strong> domains</span>"
+        f"<span><strong>{len(seeds[0]) if seeds else 0}</strong> first-url chars</span>"
+        "</div>"
+        f'<div class="seed-chip-wrap">{chips_html}{overflow_html}</div>'
+        "</div>"
+    )
+
+
+def render_tokenization_widget_html(snapshot: dict[str, Any]) -> str:
+    tokenized_shards = int(snapshot.get("tokenized_shards", 0) or 0)
+    tokenized_rows = int(snapshot.get("tokenized_rows", 0) or 0)
+    tokenized_tokens = int(snapshot.get("tokenized_tokens", 0) or 0)
+    written_shards = int(snapshot.get("written_shards", 0) or 0)
+
+    return (
+        '<div class="token-widget">'
+        '<div class="token-stats">'
+        f"<span><strong>{tokenized_tokens}</strong> text tokens</span>"
+        f"<span><strong>{tokenized_rows}</strong> tokenized rows</span>"
+        f"<span><strong>{tokenized_shards}/{written_shards}</strong> tokenized shards</span>"
+        "</div>"
+        '<div class="token-note">Live shard tokenization uses tiktoken on the parquet <code>text</code> column.</div>'
+        "</div>"
+    )
+
+
+def render_qvp_widget_md(snapshot: dict[str, Any]) -> str:
+    queue_count = int(snapshot.get("fetch_queue", 0) or 0)
+    visited_count = int(snapshot.get("fetch_succeeded", 0) or 0)
+    parsed_count = int(snapshot.get("parsed_pages", 0) or 0)
+    return (
+        "### Live Metrics\n"
+        f"- Queue: `{queue_count}`\n"
+        f"- Visited: `{visited_count}`\n"
+        f"- Parsed: `{parsed_count}`"
+    )
+
+
+def validate_hf_requirements(enable_hf_upload: bool, hf_repo_id: str, hf_token: str) -> None:
+    if not enable_hf_upload:
+        return
+    if not hf_repo_id.strip():
+        raise ValueError("HF repo is required when upload is enabled.")
+    if not hf_token.strip():
+        raise ValueError("HF token is required when upload is enabled.")
+
+
+def build_crawler_config(
+    *,
+    seed_urls_table: Any,
+    max_links_per_page: int,
+    request_timeout_seconds: float,
+    max_response_bytes: int,
+    shard_size_rows: int,
+    enable_hf_upload: bool,
+    hf_repo_id: str,
+    hf_token: str,
+    hf_private_repo: bool,
+    hf_path_prefix: str,
+    total_workers: int,
+) -> CrawlerConfig:
+    validate_hf_requirements(enable_hf_upload, hf_repo_id, hf_token)
+    seed_urls = collect_seed_urls(seed_urls_table)
+
+    return CrawlerConfig(
+        seed_urls=seed_urls,
+        max_links_per_page=int(max_links_per_page),
+        request_timeout_seconds=float(request_timeout_seconds),
+        max_response_bytes=int(max_response_bytes),
+        shard_size_rows=int(shard_size_rows),
+        output_dir=Path(__file__).resolve().parent / "shards",
+        enable_hf_upload=bool(enable_hf_upload),
+        hf_repo_id=hf_repo_id.strip(),
+        hf_token=hf_token.strip(),
+        hf_private_repo=bool(hf_private_repo),
+        hf_path_prefix=hf_path_prefix.strip() or "crawl_shards",
+        total_workers=int(total_workers),
+    )
+
+
+@dataclass
+class RunState:
+    run_id: int = 0
+    running: bool = False
+    started_at: str = ""
+    finished_at: str = ""
+    stop_requested: bool = False
+    last_error: str = ""
+
+
+class CrawlerRunManager:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._crawler: AsyncCrawler | None = None
+        self._state = RunState()
+        self._logs: list[str] = []
+        self._last_snapshot: dict[str, Any] | None = None
+
+    def start(self, config: CrawlerConfig) -> str:
+        with self._lock:
+            if self._thread is not None and self._thread.is_alive():
+                return "A crawl is already running. Stop it before starting another one."
+
+            self._state.run_id += 1
+            self._state.running = True
+            self._state.started_at = utc_now_iso()
+            self._state.finished_at = ""
+            self._state.stop_requested = False
+            self._state.last_error = ""
+            self._last_snapshot = None
+            self._logs.clear()
+
+            run_id = self._state.run_id
+            self._logs.append(
+                f"[{utc_now_iso()}] Started run #{run_id} with {config.total_workers} workers "
+                f"({config.fetch_workers} fetch / {config.parser_workers} parser)."
+            )
+
+            self._thread = threading.Thread(
+                target=self._run_crawler,
+                args=(run_id, config),
+                daemon=True,
+                name=f"crawler-run-{run_id}",
+            )
+            self._thread.start()
+
+        return f"Run #{run_id} started."
+
+    def stop(self) -> str:
+        with self._lock:
+            if self._thread is None or not self._thread.is_alive():
+                return "No active crawl to stop."
+
+            self._state.stop_requested = True
+            crawler = self._crawler
+            loop = self._loop
+            run_id = self._state.run_id
+            self._logs.append(f"[{utc_now_iso()}] Stop requested for run #{run_id}")
+
+        if crawler is not None and loop is not None and loop.is_running():
+            loop.call_soon_threadsafe(crawler.request_stop, "user_requested_stop")
+        elif crawler is not None:
+            crawler.request_stop("user_requested_stop")
+
+        return f"Stop signal sent to run #{run_id}."
+
+    def _run_crawler(self, run_id: int, config: CrawlerConfig) -> None:
+        loop: asyncio.AbstractEventLoop | None = None
+        try:
+            crawler = AsyncCrawler(config)
+            if hasattr(asyncio, "Runner"):
+                with asyncio.Runner() as runner:  # type: ignore[attr-defined]
+                    loop = runner.get_loop()
+                    with self._lock:
+                        if self._state.run_id == run_id:
+                            self._crawler = crawler
+                            self._loop = loop
+                    runner.run(crawler.run())
+            else:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+                with self._lock:
+                    if self._state.run_id == run_id:
+                        self._crawler = crawler
+                        self._loop = loop
+                loop.run_until_complete(crawler.run())
+
+            final_snapshot = self._snapshot_from_crawler(crawler)
+            with self._lock:
+                if self._state.run_id == run_id:
+                    self._last_snapshot = final_snapshot
+                self._logs.append(f"[{utc_now_iso()}] Run #{run_id} completed")
+        except Exception:
+            error_text = traceback.format_exc(limit=20)
+            with self._lock:
+                self._state.last_error = error_text
+                self._logs.append(f"[{utc_now_iso()}] Run #{run_id} crashed")
+        finally:
+            with self._lock:
+                if self._state.run_id == run_id:
+                    self._state.running = False
+                    self._state.finished_at = utc_now_iso()
+                    self._crawler = None
+                    self._loop = None
+
+            if loop is not None and not loop.is_closed():
+                loop.close()
+            with contextlib.suppress(Exception):
+                asyncio.set_event_loop(None)
+
+    def _snapshot_from_crawler(self, crawler: AsyncCrawler) -> dict[str, Any]:
+        stats = crawler.stats
+        return {
+            "timestamp": utc_now_iso(),
+            "workers_total": crawler.config.total_workers,
+            "workers_split": f"{crawler.config.fetch_workers}/{crawler.config.parser_workers}",
+            "stop_reason": crawler.stop_reason or "-",
+            "fetch_succeeded": stats.fetch_succeeded,
+            "parsed_pages": stats.parsed_pages,
+            "written_shards": stats.written_shards,
+            "tokenized_shards": stats.tokenized_shards,
+            "tokenized_rows": stats.tokenized_rows,
+            "tokenized_tokens": stats.tokenized_tokens,
+            "fetch_queue": safe_queue_size(crawler.fetch_queue),
+            "parse_queue": safe_queue_size(crawler.parse_queue),
+            "record_queue": safe_queue_size(crawler.record_queue),
+            "stop_event": crawler.stop_event.is_set(),
+        }
+
+    def poll(self) -> tuple[str, dict[str, Any], str]:
+        with self._lock:
+            crawler = self._crawler
+            state = RunState(
+                run_id=self._state.run_id,
+                running=self._state.running,
+                started_at=self._state.started_at,
+                finished_at=self._state.finished_at,
+                stop_requested=self._state.stop_requested,
+                last_error=self._state.last_error,
+            )
+
+        if crawler is not None:
+            snapshot = self._snapshot_from_crawler(crawler)
+            with self._lock:
+                self._last_snapshot = snapshot
+
+        with self._lock:
+            latest = self._last_snapshot or {
+                "timestamp": utc_now_iso(),
+                "workers_total": 0,
+                "workers_split": "-",
+                "stop_reason": "-",
+                "fetch_succeeded": 0,
+                "parsed_pages": 0,
+                "written_shards": 0,
+                "tokenized_shards": 0,
+                "tokenized_rows": 0,
+                "tokenized_tokens": 0,
+                "fetch_queue": 0,
+                "parse_queue": 0,
+                "record_queue": 0,
+                "stop_event": False,
+            }
+            logs_text = "\n".join(self._logs[-500:])
+
+        status_lines = [
+            "### Crawler Status",
+            f"- Run ID: `{state.run_id}`",
+            f"- Running: `{state.running}`",
+            f"- Stop requested: `{state.stop_requested}`",
+            f"- Started at (UTC): `{state.started_at or '-'}`",
+            f"- Finished at (UTC): `{state.finished_at or '-'}`",
+        ]
+        if state.last_error:
+            status_lines.append("- Last error:")
+            status_lines.append("```text")
+            status_lines.append(state.last_error.strip())
+            status_lines.append("```")
+
+        return "\n".join(status_lines), latest, logs_text
+
+
+RUN_MANAGER = CrawlerRunManager()
+
+
+def _format_dashboard_response(
+    status: str,
+    snapshot: dict[str, Any],
+    logs: str,
+) -> tuple[str, str, str, str]:
+    return (
+        status,
+        render_qvp_widget_md(snapshot),
+        logs,
+        render_tokenization_widget_html(snapshot),
+    )
+
+
+def _start_crawl(
+    *,
+    total_workers: int,
+    seed_urls_table: Any,
+    max_links_per_page: int,
+    request_timeout_seconds: float,
+    max_response_bytes: int,
+    shard_size_rows: int,
+    enable_hf_upload: bool,
+    hf_repo_id: str,
+    hf_token: str,
+    hf_private_repo: bool,
+    hf_path_prefix: str,
+) -> tuple[str, str, str, str]:
+    try:
+        config = build_crawler_config(
+            seed_urls_table=seed_urls_table,
+            max_links_per_page=max_links_per_page,
+            request_timeout_seconds=request_timeout_seconds,
+            max_response_bytes=max_response_bytes,
+            shard_size_rows=shard_size_rows,
+            enable_hf_upload=enable_hf_upload,
+            hf_repo_id=hf_repo_id,
+            hf_token=hf_token,
+            hf_private_repo=hf_private_repo,
+            hf_path_prefix=hf_path_prefix,
+            total_workers=total_workers,
+        )
+    except ValueError as exc:
+        raise gr.Error(str(exc)) from exc
+
+    message = RUN_MANAGER.start(config)
+    status, snapshot, logs = RUN_MANAGER.poll()
+    return _format_dashboard_response(f"{status}\n\n{message}", snapshot, logs)
+
+
+def start_crawl_standard(
+    seed_urls_table: Any,
+    max_links_per_page: int,
+    request_timeout_seconds: float,
+    max_response_bytes: int,
+    shard_size_rows: int,
+    enable_hf_upload: bool,
+    hf_repo_id: str,
+    hf_token: str,
+    hf_private_repo: bool,
+    hf_path_prefix: str,
+) -> tuple[str, str, str, str]:
+    return _start_crawl(
+        total_workers=NORMAL_TOTAL_WORKERS,
+        seed_urls_table=seed_urls_table,
+        max_links_per_page=max_links_per_page,
+        request_timeout_seconds=request_timeout_seconds,
+        max_response_bytes=max_response_bytes,
+        shard_size_rows=shard_size_rows,
+        enable_hf_upload=enable_hf_upload,
+        hf_repo_id=hf_repo_id,
+        hf_token=hf_token,
+        hf_private_repo=hf_private_repo,
+        hf_path_prefix=hf_path_prefix,
+    )
+
+
+def start_crawl_super(
+    seed_urls_table: Any,
+    max_links_per_page: int,
+    request_timeout_seconds: float,
+    max_response_bytes: int,
+    shard_size_rows: int,
+    enable_hf_upload: bool,
+    hf_repo_id: str,
+    hf_token: str,
+    hf_private_repo: bool,
+    hf_path_prefix: str,
+) -> tuple[str, str, str, str]:
+    return _start_crawl(
+        total_workers=SUPER_TOTAL_WORKERS,
+        seed_urls_table=seed_urls_table,
+        max_links_per_page=max_links_per_page,
+        request_timeout_seconds=request_timeout_seconds,
+        max_response_bytes=max_response_bytes,
+        shard_size_rows=shard_size_rows,
+        enable_hf_upload=enable_hf_upload,
+        hf_repo_id=hf_repo_id,
+        hf_token=hf_token,
+        hf_private_repo=hf_private_repo,
+        hf_path_prefix=hf_path_prefix,
+    )
+
+
+def stop_crawl() -> tuple[str, str, str, str]:
+    message = RUN_MANAGER.stop()
+    status, snapshot, logs = RUN_MANAGER.poll()
+    return _format_dashboard_response(f"{status}\n\n{message}", snapshot, logs)
+
+
+def poll_dashboard() -> tuple[str, str, str, str]:
+    status, snapshot, logs = RUN_MANAGER.poll()
+    return _format_dashboard_response(status, snapshot, logs)
+
+
+def toggle_hf_fields(enable_hf_upload: bool) -> tuple[Any, Any, Any, Any]:
+    update = gr.update(visible=enable_hf_upload)
+    return update, update, update, update
+
+
+def build_ui() -> gr.Blocks:
+    defaults = CrawlerConfig(
+        seed_urls=[
+            "https://en.wikipedia.org/wiki/Main_Page",
+            "https://docs.python.org/3/",
+            "https://developer.mozilla.org/en-US/",
+            "https://www.nasa.gov/",
+        ]
+    )
+    default_seed_rows = [[url] for url in defaults.seed_urls]
+
+    with gr.Blocks(title="DataMuncherLabs AutoWS") as demo:
+        gr.Markdown("# DataMuncherLabs AutoWS")
+        gr.Markdown("Async web crawler dashboard with live parquet text tokenization.")
+
+        with gr.Row():
+            theme_name = gr.Dropdown(
+                choices=["red", "blue", "light", "dark", "green"],
+                value="dark",
+                label="Theme",
+                interactive=True,
+            )
+            gr.Markdown(
+                "- Standard mode: **12 threads** (`10 fetch`, `2 parse`)\n"
+                "- Super mode: **24 threads** (`20 fetch`, `4 parse`)"
+            )
+
+        with gr.Row():
+            with gr.Column(scale=2):
+                seed_urls_table = gr.Dataframe(
+                    headers=["seed_url"],
+                    datatype=["str"],
+                    type="array",
+                    row_count=(8, "dynamic"),
+                    value=default_seed_rows,
+                    interactive=True,
+                    label="Seed URL List (editable)",
+                )
+                seed_widget_html = gr.HTML(
+                    label="Seed URL Summary",
+                    value=render_seed_widget_html(default_seed_rows),
+                )
+                token_widget_html = gr.HTML(
+                    label="Live Tokenization",
+                    value=render_tokenization_widget_html({}),
+                )
+
+            with gr.Column(scale=1):
+                shard_size_rows = gr.Slider(
+                    label=f"Shard Size Rows (max {MAX_SHARD_ROWS})",
+                    minimum=100,
+                    maximum=MAX_SHARD_ROWS,
+                    step=100,
+                    value=min(defaults.shard_size_rows, MAX_SHARD_ROWS),
+                )
+                max_links_per_page = gr.Slider(
+                    label="Max Links Per Page",
+                    minimum=10,
+                    maximum=1000,
+                    step=10,
+                    value=defaults.max_links_per_page,
+                )
+                request_timeout_seconds = gr.Slider(
+                    label="Request Timeout (seconds)",
+                    minimum=3,
+                    maximum=60,
+                    step=1,
+                    value=defaults.request_timeout_seconds,
+                )
+                max_response_bytes = gr.Slider(
+                    label="Max Response Bytes",
+                    minimum=500_000,
+                    maximum=8_000_000,
+                    step=100_000,
+                    value=defaults.max_response_bytes,
+                )
+
+        with gr.Accordion("Hugging Face Upload", open=False):
+            enable_hf_upload = gr.Checkbox(
+                label="Upload shards to my HF repo",
+                value=False,
+            )
+            hf_repo_id = gr.Textbox(
+                label="HF Repo ID",
+                placeholder="username/dataset-name",
+                visible=False,
+            )
+            hf_token = gr.Textbox(
+                label="HF Token (write permissions)",
+                type="password",
+                placeholder="hf_xxx",
+                visible=False,
+            )
+            hf_private_repo = gr.Checkbox(
+                label="Private HF Repo",
+                value=False,
+                visible=False,
+            )
+            hf_path_prefix = gr.Textbox(
+                label="HF Path Prefix",
+                value="crawl_shards",
+                visible=False,
+            )
+
+        with gr.Row():
+            start_button = gr.Button("Start Crawl (12 Threads)", variant="primary")
+            super_button = gr.Button("Super Mode (24 Threads)", variant="primary")
+            stop_button = gr.Button("Stop Crawl", variant="stop")
+            refresh_button = gr.Button("Refresh")
+
+        status_md = gr.Markdown("### Crawler Status\n- Run ID: `0`\n- Running: `False`")
+        qvp_md = gr.Markdown("### Live Metrics\n- Queue: `0`\n- Visited: `0`\n- Parsed: `0`")
+        logs_box = gr.Textbox(label="Run Log", lines=12, interactive=False)
+
+        start_inputs = [
+            seed_urls_table,
+            max_links_per_page,
+            request_timeout_seconds,
+            max_response_bytes,
+            shard_size_rows,
+            enable_hf_upload,
+            hf_repo_id,
+            hf_token,
+            hf_private_repo,
+            hf_path_prefix,
+        ]
+        outputs = [status_md, qvp_md, logs_box, token_widget_html]
+
+        start_button.click(start_crawl_standard, inputs=start_inputs, outputs=outputs)
+        super_button.click(start_crawl_super, inputs=start_inputs, outputs=outputs)
+        stop_button.click(stop_crawl, inputs=[], outputs=outputs)
+        refresh_button.click(poll_dashboard, inputs=[], outputs=outputs)
+
+        enable_hf_upload.change(
+            toggle_hf_fields,
+            inputs=enable_hf_upload,
+            outputs=[hf_repo_id, hf_token, hf_private_repo, hf_path_prefix],
+        )
+
+        seed_urls_table.change(
+            fn=None,
+            inputs=[seed_urls_table],
+            outputs=[seed_widget_html],
+            js=SEED_WIDGET_JS,
+        )
+
+        theme_name.change(fn=None, inputs=theme_name, outputs=[], js=THEME_JS)
+        demo.load(
+            fn=None,
+            inputs=[],
+            outputs=[],
+            js='() => { document.documentElement.setAttribute("data-crawler-theme", "dark"); }',
+        )
+        demo.load(
+            fn=None,
+            inputs=[seed_urls_table],
+            outputs=[seed_widget_html],
+            js=SEED_WIDGET_JS,
+        )
+        demo.load(fn=poll_dashboard, inputs=[], outputs=outputs)
+
+        timer = gr.Timer(value=1.0)
+        timer.tick(fn=poll_dashboard, inputs=[], outputs=outputs)
+
+    return demo
+
+
+demo = build_ui()
+
+
+def main() -> None:
+    queued = demo.queue(default_concurrency_limit=32)
+    launch_sig = inspect.signature(queued.launch)
+    launch_kwargs: dict[str, Any] = {}
+
+    if "css" in launch_sig.parameters:
+        launch_kwargs["css"] = APP_CSS
+    if "theme" in launch_sig.parameters:
+        launch_kwargs["theme"] = gr.themes.Default(primary_hue="green")
+    if "ssr_mode" in launch_sig.parameters:
+        launch_kwargs["ssr_mode"] = False
+
+    queued.launch(**launch_kwargs)
+
+
+if __name__ == "__main__":
+    main()

crawler/__init__.py

@@ -0,0 +1,19 @@
+from .config import (
+    MAX_SHARD_ROWS,
+    MAX_SHARDS,
+    NORMAL_TOTAL_WORKERS,
+    SUPER_TOTAL_WORKERS,
+    CrawlerConfig,
+    compute_worker_split,
+)
+from .engine import AsyncCrawler
+
+__all__ = [
+    "AsyncCrawler",
+    "CrawlerConfig",
+    "MAX_SHARD_ROWS",
+    "MAX_SHARDS",
+    "NORMAL_TOTAL_WORKERS",
+    "SUPER_TOTAL_WORKERS",
+    "compute_worker_split",
+]

crawler/config.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+NORMAL_TOTAL_WORKERS = 12
+SUPER_TOTAL_WORKERS = 24
+MAX_SHARD_ROWS = 15_000
+MAX_SHARDS = 10
+
+
+def validate_total_workers(total_workers: int) -> int:
+    value = int(total_workers)
+    if value not in {NORMAL_TOTAL_WORKERS, SUPER_TOTAL_WORKERS}:
+        raise ValueError(
+            f"total_workers must be {NORMAL_TOTAL_WORKERS} or {SUPER_TOTAL_WORKERS}, got {value}."
+        )
+    return value
+
+
+def compute_worker_split(total_workers: int) -> tuple[int, int]:
+    total = validate_total_workers(total_workers)
+    fetch_workers = (total * 5) // 6
+    parser_workers = total - fetch_workers
+    if fetch_workers < 1 or parser_workers < 1:
+        raise ValueError(f"Invalid worker split for total_workers={total}.")
+    return fetch_workers, parser_workers
+
+
+@dataclass
+class CrawlerConfig:
+    seed_urls: list[str]
+
+    max_links_per_page: int = 250
+    request_timeout_seconds: float = 18.0
+    max_response_bytes: int = 3_000_000
+    user_agent: str = "HFDBContCrawler/1.0 (+https://huggingface.co/datasets)"
+    seen_url_cache_size: int = 2_000_000
+
+    fetch_queue_size: int = 100_000
+    parse_queue_size: int = 25_000
+    record_queue_size: int = 50_000
+    report_every_seconds: float = 5.0
+
+    output_dir: Path = field(
+        default_factory=lambda: Path(__file__).resolve().parents[1] / "shards"
+    )
+    shard_size_rows: int = 10_000
+    max_shards: int = MAX_SHARDS
+    parquet_compression: str = "zstd"
+    parquet_compression_level: int = 9
+
+    enable_hf_upload: bool = False
+    hf_repo_id: str = ""
+    hf_token: str = ""
+    hf_repo_type: str = "dataset"
+    hf_private_repo: bool = False
+    hf_path_prefix: str = "crawl_shards"
+
+    total_workers: int = NORMAL_TOTAL_WORKERS
+    request_delay_global_seconds: float = 0.02
+    request_delay_per_domain_seconds: float = 2.0
+
+    robots_cache_ttl_seconds: float = 3600.0
+    robots_fail_closed: bool = True
+    robots_max_bytes: int = 300_000
+
+    fetch_workers: int = field(init=False)
+    parser_workers: int = field(init=False)
+
+    def __post_init__(self) -> None:
+        self.seed_urls = [u.strip() for u in self.seed_urls if u and u.strip()]
+        if not self.seed_urls:
+            raise ValueError("At least one seed URL is required.")
+
+        self.total_workers = validate_total_workers(self.total_workers)
+        self.fetch_workers, self.parser_workers = compute_worker_split(self.total_workers)
+
+        self.shard_size_rows = int(self.shard_size_rows)
+        if self.shard_size_rows < 1 or self.shard_size_rows > MAX_SHARD_ROWS:
+            raise ValueError(f"shard_size_rows must be between 1 and {MAX_SHARD_ROWS}.")
+
+        self.max_shards = int(self.max_shards)
+        if self.max_shards < 1 or self.max_shards > MAX_SHARDS:
+            raise ValueError(f"max_shards must be between 1 and {MAX_SHARDS}.")
+
+        self.output_dir = Path(self.output_dir).expanduser()
+
+        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"
+
+        if self.enable_hf_upload:
+            if not self.hf_repo_id:
+                raise ValueError("hf_repo_id is required when enable_hf_upload=True.")
+            if not self.hf_token:
+                raise ValueError("hf_token is required when enable_hf_upload=True.")

crawler/engine.py

@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from collections import deque
+from typing import Any
+
+import aiohttp
+
+from .config import CrawlerConfig
+from .fetch import fetch_url
+from .models import CrawlStats, FetchResult
+from .parse import parse_page
+from .rate_limit import RequestRateLimiter
+from .robots import RobotsPolicy
+from .shards import ParquetShardWriter, ShardLimitReached
+from .utils import has_binary_extension, normalize_url
+
+
+class AsyncCrawler:
+    def __init__(self, config: CrawlerConfig):
+        self.config = config
+        self.stats = CrawlStats()
+        self.stop_event = asyncio.Event()
+        self.stop_reason = ""
+
+        self.fetch_queue: asyncio.Queue[str | None] = asyncio.Queue(
+            maxsize=config.fetch_queue_size
+        )
+        self.parse_queue: asyncio.Queue[FetchResult | None] = asyncio.Queue(
+            maxsize=config.parse_queue_size
+        )
+        self.record_queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue(
+            maxsize=config.record_queue_size
+        )
+
+        self.seen_urls: set[str] = set()
+        self.seen_order: deque[str] = deque()
+        self.seen_lock = asyncio.Lock()
+        self.counter_lock = asyncio.Lock()
+
+        self.active_fetchers = 0
+        self.active_parsers = 0
+
+        self.writer = ParquetShardWriter(config=config, stats=self.stats)
+        self.rate_limiter: RequestRateLimiter | None = None
+        self.robots_policy: RobotsPolicy | None = None
+
+    async def run(self) -> None:
+        await self.writer.initialize()
+
+        for seed in self.config.seed_urls:
+            await self.try_enqueue(seed)
+
+        connector = aiohttp.TCPConnector(
+            limit=max(200, self.config.fetch_workers * 4),
+            ttl_dns_cache=300,
+        )
+        timeout = aiohttp.ClientTimeout(total=self.config.request_timeout_seconds)
+
+        async with aiohttp.ClientSession(
+            connector=connector,
+            timeout=timeout,
+            headers={"User-Agent": self.config.user_agent},
+        ) as session:
+            self.rate_limiter = RequestRateLimiter(
+                global_interval_seconds=self.config.request_delay_global_seconds,
+                per_domain_interval_seconds=self.config.request_delay_per_domain_seconds,
+            )
+            self.robots_policy = RobotsPolicy(
+                session=session,
+                user_agent=self.config.user_agent,
+                cache_ttl_seconds=self.config.robots_cache_ttl_seconds,
+                fail_closed=self.config.robots_fail_closed,
+                max_bytes=self.config.robots_max_bytes,
+            )
+
+            fetchers = [
+                asyncio.create_task(self.fetcher_worker(worker_id=i, session=session))
+                for i in range(self.config.fetch_workers)
+            ]
+            parsers = [
+                asyncio.create_task(self.parser_worker(worker_id=i))
+                for i in range(self.config.parser_workers)
+            ]
+            writer_task = asyncio.create_task(self.writer.consume(self.record_queue))
+            reporter_task = asyncio.create_task(self.progress_reporter())
+
+            try:
+                await self.wait_until_complete(writer_task)
+                await self._graceful_shutdown(fetchers, parsers, writer_task)
+            except ShardLimitReached:
+                self.stop_reason = "shard_cap_reached"
+                self.stop_event.set()
+                await self._hard_shutdown(fetchers, parsers, writer_task)
+            finally:
+                reporter_task.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await reporter_task
+
+    def request_stop(self, reason: str = "user_requested_stop") -> None:
+        if not self.stop_reason:
+            self.stop_reason = reason
+        self.stop_event.set()
+
+    async def wait_until_complete(self, writer_task: asyncio.Task[None]) -> None:
+        while True:
+            if writer_task.done():
+                exc = writer_task.exception()
+                if exc is not None:
+                    raise exc
+                return
+
+            if self.stop_event.is_set():
+                if not self.stop_reason:
+                    self.stop_reason = "stop_event_set"
+                if self._is_pipeline_idle():
+                    return
+                await asyncio.sleep(0.2)
+                continue
+
+            if self._is_pipeline_idle():
+                self.stop_reason = "frontier_exhausted"
+                return
+
+            await asyncio.sleep(0.5)
+
+    async def _graceful_shutdown(
+        self,
+        fetchers: list[asyncio.Task[None]],
+        parsers: list[asyncio.Task[None]],
+        writer_task: asyncio.Task[None],
+    ) -> None:
+        for _ in fetchers:
+            await self.fetch_queue.put(None)
+        await asyncio.gather(*fetchers, return_exceptions=True)
+
+        for _ in parsers:
+            await self.parse_queue.put(None)
+        await asyncio.gather(*parsers, return_exceptions=True)
+
+        await self.record_queue.put(None)
+        await writer_task
+
+    async def _hard_shutdown(
+        self,
+        fetchers: list[asyncio.Task[None]],
+        parsers: list[asyncio.Task[None]],
+        writer_task: asyncio.Task[None],
+    ) -> None:
+        for task in fetchers + parsers:
+            task.cancel()
+        await asyncio.gather(*fetchers, *parsers, return_exceptions=True)
+
+        if not writer_task.done():
+            writer_task.cancel()
+        await asyncio.gather(writer_task, return_exceptions=True)
+
+    async def progress_reporter(self) -> None:
+        while True:
+            await asyncio.sleep(self.config.report_every_seconds)
+            print(
+                "[stats]"
+                f" workers={self.config.total_workers}"
+                f" split={self.config.fetch_workers}/{self.config.parser_workers}"
+                f" queued={self.stats.queued_urls}"
+                f" fetched={self.stats.fetch_reserved}"
+                f" fetch_ok={self.stats.fetch_succeeded}"
+                f" fetch_fail={self.stats.fetch_failed}"
+                f" parsed={self.stats.parsed_pages}"
+                f" parse_fail={self.stats.parse_failed}"
+                f" robots_blocked={self.stats.robots_blocked}"
+                f" rows={self.stats.stored_rows}"
+                f" shards={self.stats.written_shards}/{self.config.max_shards}"
+                f" tok_shards={self.stats.tokenized_shards}"
+                f" tok_rows={self.stats.tokenized_rows}"
+                f" tok_total={self.stats.tokenized_tokens}"
+                f" uploaded={self.stats.uploaded_shards}"
+                f" fetch_q={self.fetch_queue.qsize()}"
+                f" parse_q={self.parse_queue.qsize()}"
+                f" record_q={self.record_queue.qsize()}"
+            )
+
+    async def fetcher_worker(self, worker_id: int, session: aiohttp.ClientSession) -> None:
+        del worker_id
+        assert self.rate_limiter is not None
+        assert self.robots_policy is not None
+
+        while True:
+            url = await self.fetch_queue.get()
+            if url is None:
+                self.fetch_queue.task_done()
+                return
+
+            slot_reserved = await self.reserve_fetch_slot()
+            if not slot_reserved:
+                self.fetch_queue.task_done()
+                continue
+
+            self.active_fetchers += 1
+            try:
+                outcome = await fetch_url(
+                    session,
+                    url,
+                    config=self.config,
+                    mark_seen=self._mark_seen,
+                    rate_limiter=self.rate_limiter,
+                    robots_policy=self.robots_policy,
+                )
+                if outcome.robots_blocked:
+                    self.stats.robots_blocked += 1
+                if outcome.result is not None:
+                    self.stats.fetch_succeeded += 1
+                    if outcome.result.html:
+                        await self.parse_queue.put(outcome.result)
+                else:
+                    self.stats.fetch_failed += 1
+            finally:
+                self.active_fetchers -= 1
+                self.fetch_queue.task_done()
+
+    async def parser_worker(self, worker_id: int) -> None:
+        del worker_id
+
+        while True:
+            item = await self.parse_queue.get()
+            if item is None:
+                self.parse_queue.task_done()
+                return
+
+            self.active_parsers += 1
+            try:
+                record, links = parse_page(item)
+                if record is not None:
+                    await self.record_queue.put(record)
+                    self.stats.parsed_pages += 1
+
+                extracted = 0
+                for link in links:
+                    if extracted >= self.config.max_links_per_page:
+                        break
+                    if await self.try_enqueue(link):
+                        extracted += 1
+                self.stats.extracted_links += extracted
+            except Exception:
+                self.stats.parse_failed += 1
+            finally:
+                self.active_parsers -= 1
+                self.parse_queue.task_done()
+
+    async def reserve_fetch_slot(self) -> bool:
+        async with self.counter_lock:
+            if self.stop_event.is_set():
+                return False
+            self.stats.fetch_reserved += 1
+            return True
+
+    async def try_enqueue(self, raw_url: str) -> bool:
+        if self.stop_event.is_set():
+            return False
+
+        normalized = normalize_url(raw_url)
+        if not normalized:
+            self.stats.dropped_urls += 1
+            return False
+
+        if has_binary_extension(normalized):
+            self.stats.dropped_urls += 1
+            return False
+
+        async with self.seen_lock:
+            if self.config.seen_url_cache_size > 0 and normalized in self.seen_urls:
+                return False
+            self._remember_seen_locked(normalized)
+            self.stats.queued_urls += 1
+
+        await self.fetch_queue.put(normalized)
+        return True
+
+    async def _mark_seen(self, url: str) -> None:
+        async with self.seen_lock:
+            self._remember_seen_locked(url)
+
+    def _remember_seen_locked(self, url: str) -> None:
+        if self.config.seen_url_cache_size <= 0:
+            return
+        if url in self.seen_urls:
+            return
+
+        self.seen_urls.add(url)
+        self.seen_order.append(url)
+        while len(self.seen_order) > self.config.seen_url_cache_size:
+            expired = self.seen_order.popleft()
+            self.seen_urls.discard(expired)
+
+    def _is_pipeline_idle(self) -> bool:
+        return (
+            self.fetch_queue.empty()
+            and self.parse_queue.empty()
+            and self.active_fetchers == 0
+            and self.active_parsers == 0
+        )

crawler/fetch.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Awaitable, Callable
+from urllib.parse import urlsplit
+
+import aiohttp
+
+from .config import CrawlerConfig
+from .models import FetchResult
+from .rate_limit import RequestRateLimiter
+from .robots import RobotsPolicy
+from .utils import is_html_response, normalize_url
+
+
+@dataclass
+class FetchOutcome:
+    result: FetchResult | None
+    robots_blocked: bool = False
+
+
+async def fetch_url(
+    session: aiohttp.ClientSession,
+    url: str,
+    *,
+    config: CrawlerConfig,
+    mark_seen: Callable[[str], Awaitable[None]],
+    rate_limiter: RequestRateLimiter,
+    robots_policy: RobotsPolicy,
+) -> FetchOutcome:
+    fetched_at = datetime.now(timezone.utc).isoformat()
+    requested_domain = (urlsplit(url).hostname or "").lower().strip(".")
+    if not requested_domain:
+        return FetchOutcome(result=None)
+
+    if not await robots_policy.can_fetch(url):
+        return FetchOutcome(result=None, robots_blocked=True)
+
+    await rate_limiter.acquire(requested_domain)
+
+    try:
+        async with session.get(url, allow_redirects=True) as response:
+            content_type = response.headers.get("content-type", "").lower()
+            final_url = normalize_url(str(response.url))
+            if not final_url:
+                return FetchOutcome(result=None)
+
+            final_domain = (urlsplit(final_url).hostname or "").lower().strip(".")
+            if not final_domain:
+                return FetchOutcome(result=None)
+
+            if not await robots_policy.can_fetch(final_url):
+                return FetchOutcome(result=None, robots_blocked=True)
+
+            await mark_seen(final_url)
+
+            if response.status >= 400:
+                return FetchOutcome(result=None)
+
+            if not is_html_response(content_type, final_url):
+                return FetchOutcome(
+                    result=FetchResult(
+                        url=final_url,
+                        status=response.status,
+                        fetched_at=fetched_at,
+                        content_type=content_type,
+                        html="",
+                    )
+                )
+
+            raw = await response.content.read(config.max_response_bytes + 1)
+            if len(raw) > config.max_response_bytes:
+                raw = raw[: config.max_response_bytes]
+            html = raw.decode(response.charset or "utf-8", errors="ignore")
+
+            return FetchOutcome(
+                result=FetchResult(
+                    url=final_url,
+                    status=response.status,
+                    fetched_at=fetched_at,
+                    content_type=content_type,
+                    html=html,
+                )
+            )
+    except Exception:
+        return FetchOutcome(result=None)

crawler/models.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class FetchResult:
+    url: str
+    status: int
+    fetched_at: str
+    content_type: str
+    html: str
+
+
+@dataclass
+class CrawlStats:
+    queued_urls: int = 0
+    fetch_reserved: int = 0
+    fetch_succeeded: int = 0
+    fetch_failed: int = 0
+    parsed_pages: int = 0
+    parse_failed: int = 0
+    extracted_links: int = 0
+    dropped_urls: int = 0
+    robots_blocked: int = 0
+    stored_rows: int = 0
+    written_shards: int = 0
+    uploaded_shards: int = 0
+    tokenized_shards: int = 0
+    tokenized_rows: int = 0
+    tokenized_tokens: int = 0

crawler/parse.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urljoin, urlsplit
+
+from bs4 import BeautifulSoup
+
+from .models import FetchResult
+
+
+def parse_page(item: FetchResult) -> tuple[dict[str, Any] | None, list[str]]:
+    if not item.html:
+        return None, []
+
+    soup = BeautifulSoup(item.html, "lxml")
+
+    for tag in soup(["script", "style", "noscript", "svg", "iframe", "canvas"]):
+        tag.decompose()
+
+    text = soup.get_text(" ", strip=True)
+    if not text:
+        return None, []
+
+    links: list[str] = []
+    for anchor in soup.find_all("a", href=True):
+        href = anchor.get("href", "").strip()
+        if not href:
+            continue
+        links.append(urljoin(item.url, href))
+
+    domain = (urlsplit(item.url).hostname or "").lower().strip(".")
+    record = {
+        "text": text,
+        "url": item.url,
+        "domain": domain,
+        "timestamp": item.fetched_at,
+    }
+    return record, links

crawler/rate_limit.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Awaitable, Callable
+
+
+class RequestRateLimiter:
+    def __init__(
+        self,
+        global_interval_seconds: float,
+        per_domain_interval_seconds: float,
+        *,
+        clock: Callable[[], float] | None = None,
+        sleep: Callable[[float], Awaitable[None]] | None = None,
+    ) -> None:
+        self.global_interval_seconds = max(0.0, float(global_interval_seconds))
+        self.per_domain_interval_seconds = max(0.0, float(per_domain_interval_seconds))
+        self._clock = clock or time.monotonic
+        self._sleep = sleep or asyncio.sleep
+
+        self._global_lock = asyncio.Lock()
+        self._global_last: float | None = None
+
+        self._domain_guard = asyncio.Lock()
+        self._domain_locks: dict[str, asyncio.Lock] = {}
+        self._domain_last: dict[str, float] = {}
+
+    async def acquire(self, domain: str) -> None:
+        normalized = domain.lower().strip(".")
+        await self._acquire_global()
+        await self._acquire_domain(normalized)
+
+    async def _acquire_global(self) -> None:
+        if self.global_interval_seconds <= 0:
+            return
+
+        async with self._global_lock:
+            now = self._clock()
+            if self._global_last is not None:
+                wait = self.global_interval_seconds - (now - self._global_last)
+                if wait > 0:
+                    await self._sleep(wait)
+            self._global_last = self._clock()
+
+    async def _acquire_domain(self, domain: str) -> None:
+        if not domain or self.per_domain_interval_seconds <= 0:
+            return
+
+        lock = await self._get_domain_lock(domain)
+        async with lock:
+            now = self._clock()
+            last = self._domain_last.get(domain)
+            if last is not None:
+                wait = self.per_domain_interval_seconds - (now - last)
+                if wait > 0:
+                    await self._sleep(wait)
+            self._domain_last[domain] = self._clock()
+
+    async def _get_domain_lock(self, domain: str) -> asyncio.Lock:
+        async with self._domain_guard:
+            lock = self._domain_locks.get(domain)
+            if lock is None:
+                lock = asyncio.Lock()
+                self._domain_locks[domain] = lock
+            return lock

crawler/robots.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from dataclasses import dataclass
+from urllib.parse import urlsplit
+from urllib.robotparser import RobotFileParser
+
+import aiohttp
+
+
+@dataclass
+class _RobotsCacheEntry:
+    parser: RobotFileParser | None
+    expires_at: float
+
+
+class RobotsPolicy:
+    def __init__(
+        self,
+        session: aiohttp.ClientSession,
+        user_agent: str,
+        *,
+        cache_ttl_seconds: float = 3600.0,
+        fail_closed: bool = True,
+        max_bytes: int = 300_000,
+    ) -> None:
+        self.session = session
+        self.user_agent = user_agent
+        self.cache_ttl_seconds = max(1.0, float(cache_ttl_seconds))
+        self.fail_closed = bool(fail_closed)
+        self.max_bytes = int(max_bytes)
+
+        self._cache: dict[str, _RobotsCacheEntry] = {}
+        self._cache_lock = asyncio.Lock()
+        self._origin_locks: dict[str, asyncio.Lock] = {}
+
+    async def can_fetch(self, url: str) -> bool:
+        parts = urlsplit(url)
+        host = (parts.hostname or "").lower().strip(".")
+        scheme = parts.scheme.lower()
+        if scheme not in {"http", "https"} or not host:
+            return False
+
+        origin = f"{scheme}://{host}"
+        parser = await self._get_parser(origin)
+        if parser is None:
+            return not self.fail_closed
+        return parser.can_fetch(self.user_agent, url)
+
+    async def _get_parser(self, origin: str) -> RobotFileParser | None:
+        now = time.monotonic()
+        async with self._cache_lock:
+            cached = self._cache.get(origin)
+            if cached and cached.expires_at > now:
+                return cached.parser
+
+            lock = self._origin_locks.get(origin)
+            if lock is None:
+                lock = asyncio.Lock()
+                self._origin_locks[origin] = lock
+
+        async with lock:
+            now = time.monotonic()
+            async with self._cache_lock:
+                cached = self._cache.get(origin)
+                if cached and cached.expires_at > now:
+                    return cached.parser
+
+            parser = await self._download_and_parse(origin)
+            async with self._cache_lock:
+                self._cache[origin] = _RobotsCacheEntry(
+                    parser=parser,
+                    expires_at=time.monotonic() + self.cache_ttl_seconds,
+                )
+            return parser
+
+    async def _download_and_parse(self, origin: str) -> RobotFileParser | None:
+        robots_url = f"{origin}/robots.txt"
+        try:
+            async with self.session.get(robots_url, allow_redirects=True) as response:
+                if response.status >= 400:
+                    return None
+
+                raw = await response.content.read(self.max_bytes + 1)
+                if len(raw) > self.max_bytes:
+                    raw = raw[: self.max_bytes]
+                charset = response.charset or "utf-8"
+                text = raw.decode(charset, errors="ignore")
+        except Exception:
+            return None
+
+        parser = RobotFileParser()
+        parser.set_url(robots_url)
+        parser.parse(text.splitlines())
+        return parser

crawler/shards.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from .config import CrawlerConfig
+from .models import CrawlStats
+from .tokenizer import LiveShardTokenizer
+from .upload import HfShardUploader
+
+
+class ShardLimitReached(RuntimeError):
+    pass
+
+
+PARQUET_SCHEMA = pa.schema(
+    [
+        ("text", pa.string()),
+        ("url", pa.string()),
+        ("domain", pa.string()),
+        ("timestamp", pa.string()),
+    ]
+)
+
+
+class ParquetShardWriter:
+    def __init__(self, config: CrawlerConfig, stats: CrawlStats):
+        self.config = config
+        self.stats = stats
+        self.buffer: list[dict[str, Any]] = []
+        self.shard_index = 0
+        self.uploader: HfShardUploader | None = None
+        self.live_tokenizer = LiveShardTokenizer()
+
+    async def initialize(self) -> None:
+        self.config.output_dir.mkdir(parents=True, exist_ok=True)
+        if not self.config.enable_hf_upload:
+            return
+
+        self.uploader = HfShardUploader(
+            repo_id=self.config.hf_repo_id,
+            token=self.config.hf_token,
+            repo_type=self.config.hf_repo_type,
+            private_repo=self.config.hf_private_repo,
+            path_prefix=self.config.hf_path_prefix,
+        )
+        await self.uploader.initialize()
+
+    async def consume(self, record_queue: asyncio.Queue[dict[str, Any] | None]) -> None:
+        while True:
+            item = await record_queue.get()
+            if item is None:
+                record_queue.task_done()
+                break
+
+            try:
+                self.buffer.append(item)
+                if len(self.buffer) >= self.config.shard_size_rows:
+                    await self.flush()
+            finally:
+                record_queue.task_done()
+
+        if self.buffer:
+            await self.flush()
+
+    async def flush(self) -> None:
+        if not self.buffer:
+            return
+
+        if self.shard_index >= self.config.max_shards:
+            raise ShardLimitReached(f"Reached shard cap of {self.config.max_shards}.")
+
+        rows = self.buffer
+        self.buffer = []
+        normalized_rows = [
+            {
+                "text": str(row.get("text", "")),
+                "url": str(row.get("url", "")),
+                "domain": str(row.get("domain", "")),
+                "timestamp": str(row.get("timestamp", "")),
+            }
+            for row in rows
+            if row.get("text")
+        ]
+        if not normalized_rows:
+            return
+
+        timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+        shard_name = f"shard-{timestamp}-{self.shard_index:04d}.parquet"
+        shard_path = self.config.output_dir / shard_name
+
+        table = pa.Table.from_pylist(normalized_rows, schema=PARQUET_SCHEMA)
+        await asyncio.to_thread(
+            pq.write_table,
+            table,
+            shard_path,
+            compression=self.config.parquet_compression,
+            compression_level=self.config.parquet_compression_level,
+            use_dictionary=True,
+        )
+
+        self.shard_index += 1
+        self.stats.written_shards = self.shard_index
+        self.stats.stored_rows += len(normalized_rows)
+        token_rows, token_count = await asyncio.to_thread(
+            self.live_tokenizer.tokenize_shard_text, shard_path
+        )
+        self.stats.tokenized_shards += 1
+        self.stats.tokenized_rows += token_rows
+        self.stats.tokenized_tokens += token_count
+
+        if self.config.enable_hf_upload:
+            ok = await self._upload_and_delete(shard_path, rows=len(normalized_rows))
+            if ok:
+                self.stats.uploaded_shards += 1
+
+        if self.shard_index >= self.config.max_shards:
+            raise ShardLimitReached(f"Reached shard cap of {self.config.max_shards}.")
+
+    async def _upload_and_delete(self, shard_path: Path, rows: int) -> bool:
+        if self.uploader is None:
+            raise RuntimeError("Uploader not initialized.")
+        return await self.uploader.upload_and_delete(shard_path, rows)

crawler/tokenizer.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pyarrow.parquet as pq
+import tiktoken
+
+
+class LiveShardTokenizer:
+    def __init__(self, encoding_name: str = "cl100k_base") -> None:
+        self.encoding = tiktoken.get_encoding(encoding_name)
+
+    def tokenize_shard_text(self, shard_path: Path) -> tuple[int, int]:
+        table = pq.read_table(shard_path, columns=["text"])
+        if "text" not in table.column_names:
+            return 0, 0
+
+        rows = 0
+        token_count = 0
+        for value in table.column("text").to_pylist():
+            if value is None:
+                continue
+            text = str(value)
+            rows += 1
+            token_count += len(self.encoding.encode(text, disallowed_special=()))
+        return rows, token_count

crawler/upload.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+
+class HfShardUploader:
+    def __init__(
+        self,
+        *,
+        repo_id: str,
+        token: str,
+        repo_type: str = "dataset",
+        private_repo: bool = False,
+        path_prefix: str = "crawl_shards",
+    ) -> None:
+        self.repo_id = repo_id.strip()
+        self.token = token.strip()
+        self.repo_type = repo_type
+        self.private_repo = bool(private_repo)
+        self.path_prefix = path_prefix.strip("/")
+        self.api: HfApi | None = None
+
+    async def initialize(self) -> None:
+        self.api = HfApi(token=self.token or None)
+        await asyncio.to_thread(
+            self.api.create_repo,
+            repo_id=self.repo_id,
+            repo_type=self.repo_type,
+            private=self.private_repo,
+            exist_ok=True,
+        )
+
+    async def upload_and_delete(self, shard_path: Path, rows: int) -> bool:
+        if self.api is None:
+            raise RuntimeError("Uploader was not initialized.")
+
+        if self.path_prefix:
+            path_in_repo = f"{self.path_prefix}/{shard_path.name}"
+        else:
+            path_in_repo = shard_path.name
+
+        try:
+            await asyncio.to_thread(
+                self.api.upload_file,
+                path_or_fileobj=str(shard_path),
+                path_in_repo=path_in_repo,
+                repo_id=self.repo_id,
+                repo_type=self.repo_type,
+                commit_message=f"Add crawl shard {shard_path.name} ({rows} rows)",
+            )
+        except Exception:
+            return False
+
+        with contextlib.suppress(FileNotFoundError):
+            shard_path.unlink()
+        return True

crawler/utils.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import re
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+BINARY_EXTENSIONS = {
+    ".7z",
+    ".avi",
+    ".bin",
+    ".bz2",
+    ".csv",
+    ".doc",
+    ".docx",
+    ".epub",
+    ".gif",
+    ".gz",
+    ".ico",
+    ".jpeg",
+    ".jpg",
+    ".json",
+    ".m4a",
+    ".m4v",
+    ".mov",
+    ".mp3",
+    ".mp4",
+    ".mpeg",
+    ".ogg",
+    ".pdf",
+    ".png",
+    ".ppt",
+    ".pptx",
+    ".rar",
+    ".svg",
+    ".tar",
+    ".tgz",
+    ".tif",
+    ".tiff",
+    ".wav",
+    ".webm",
+    ".webp",
+    ".xls",
+    ".xlsx",
+    ".xml",
+    ".xz",
+    ".zip",
+}
+
+TRACKING_QUERY_KEYS = {
+    "fbclid",
+    "gclid",
+    "mc_cid",
+    "mc_eid",
+    "ref",
+    "source",
+    "spm",
+    "yclid",
+}
+
+
+def normalize_url(raw_url: str) -> str | None:
+    try:
+        parts = urlsplit(raw_url.strip())
+    except ValueError:
+        return None
+
+    scheme = parts.scheme.lower()
+    if scheme not in {"http", "https"}:
+        return None
+
+    host = (parts.hostname or "").lower().strip(".")
+    if not host:
+        return None
+
+    try:
+        port = parts.port
+    except ValueError:
+        return None
+
+    if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
+        netloc = host
+    elif port:
+        netloc = f"{host}:{port}"
+    else:
+        netloc = host
+
+    path = parts.path or "/"
+    path = re.sub(r"/{2,}", "/", path)
+
+    query_pairs: list[tuple[str, str]] = []
+    for key, value in parse_qsl(parts.query, keep_blank_values=True):
+        lowered = key.lower()
+        if lowered.startswith("utm_") or lowered in TRACKING_QUERY_KEYS:
+            continue
+        query_pairs.append((key, value))
+    query = urlencode(query_pairs, doseq=True)
+
+    return urlunsplit((scheme, netloc, path, query, ""))
+
+
+def has_binary_extension(url: str) -> bool:
+    path = urlsplit(url).path.lower()
+    if not path:
+        return False
+
+    dot_index = path.rfind(".")
+    if dot_index == -1:
+        return False
+
+    return path[dot_index:] in BINARY_EXTENSIONS
+
+
+def is_html_response(content_type: str, final_url: str) -> bool:
+    if has_binary_extension(final_url):
+        return False
+
+    if not content_type:
+        return True
+
+    lowered = content_type.lower()
+    return (
+        "text/html" in lowered
+        or "application/xhtml+xml" in lowered
+        or "text/plain" in lowered
+    )

requirements.txt

@@ -0,0 +1,7 @@
+aiohttp>=3.9.0
+beautifulsoup4>=4.12.0
+gradio>=4.44.0
+huggingface_hub>=0.24.0
+lxml>=4.9.0
+pyarrow>=16.1.0
+tiktoken>=0.7.0