Delete crawler

crawler/__init__.py

@@ -1,19 +0,0 @@
-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

@@ -1,97 +0,0 @@
-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

@@ -1,302 +0,0 @@
-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

@@ -1,87 +0,0 @@
-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

@@ -1,31 +0,0 @@
-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

@@ -1,38 +0,0 @@
-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

@@ -1,66 +0,0 @@
-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

@@ -1,96 +0,0 @@
-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

@@ -1,128 +0,0 @@
-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

@@ -1,26 +0,0 @@
-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

@@ -1,60 +0,0 @@
-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

@@ -1,124 +0,0 @@
-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
-    )