Exported on 5/31/2025 at 19:45:18 PDT from Cursor (0.50.5)
User
We continue to run into DB contention when adding to and retrieving URLs from the frontier of this crawler that I've built with you and Gemini in the crawler folder. The fundamental issue seems to be that SQLite is not designed for high write concurrency, but we need high concurrency to scale our crawler to hit the desired throughput of 10s of millions of pages per day (see PLAN.MD for the original plan, which is now outdated in a few places).
You proposed Redis in our last chat. I wonder if PostgreSQL could also be a good choice. My goal is to minimize code change and preserve the goals of high performance and durability, but I feel that staying with SQLite is no longer feasible. Note that in addition to the modules in PLAN.MD there are several tests which mock sqlite that we will need to update, such as crawler/tests/test_frontier.py.
Let's figure out a new datastore design to help scale the crawler, adjust the PLAN.MD accordingly, and implement it together.
Cursor
Looking at your crawler's architecture and the contention issues, I believe PostgreSQL would be the better choice over Redis for several reasons:
- Minimal code changes: PostgreSQL supports the same SQL queries including
RETURNINGclause - Durability: Built-in ACID guarantees without extra configuration
- Concurrency: MVCC allows many concurrent writers without locking issues
- Familiar operations: Your team already knows SQL
- Single system: Can store frontier, visited URLs, and robots data in one place
Let me first examine the current architecture to understand what needs to change:
Based on your crawler's architecture and the contention issues with 500 workers, PostgreSQL is indeed the best choice. Here's why:
- Minimal code changes: PostgreSQL supports all the SQL features you're using (including
RETURNING) - Superior concurrency: MVCC allows many concurrent writers without locking
- Built-in connection pooling: psycopg3 has excellent async support and pooling
- Same durability guarantees: Full ACID compliance like SQLite
Here's my proposed approach to minimize code changes while enabling PostgreSQL:
# Web Crawler Project Plan
## 1. Introduction
This document outlines the plan for building an experimental/educational web crawler. The primary goals are:
1. Run on a single machine, crawling up to 50 million pages in 24 hours from a seed list of domains.
2. Store text content of crawled pages.
3. Adhere to politeness standards (`robots.txt`, crawl delays, informative User-Agent).
4. Be fault-tolerant, allowing stopping and resuming.
5. Provide a way to inspect progress and crawled data.
The project prioritizes simplicity and educational value, using Python as the core language.
## 2. Architecture
The crawler will be an asynchronous, single-process application with support for high concurrency through PostgreSQL.
### 2.1. Core Components
* **Main Controller:** Initializes and orchestrates the crawl. Manages configuration.
* **Seed Loader:** Reads the initial list of domains/URLs from the seed file.
* **Frontier Manager:**
* Manages the queue of URLs to be crawled (the "frontier").
* Prioritizes URLs based on domain politeness rules (e.g., 70-second delay per domain).
* Persists the frontier for fault tolerance.
* Uses atomic claim-and-read pattern to prevent worker contention.
* **Fetcher (`asyncio`/`aiohttp`):**
* Downloads web page content for a given URL.
* Sets the custom User-Agent string.
* **Robots.txt Manager:**
* Fetches and parses `robots.txt` for each domain.
* Provides an interface to check if a URL is allowed for crawling.
* Caches `robots.txt` rules.
* **Parser (`BeautifulSoup4` or `lxml`):**
* Extracts new links (URLs) from downloaded HTML content.
* Extracts the main text content from the page.
* **Storage Manager:**
* **State Storage (PostgreSQL/SQLite):** Persists the frontier, set of visited URLs, domain last-crawl timestamps, `robots.txt` cache, and manually excluded domains.
- **PostgreSQL (recommended for production)**: Handles high concurrency with MVCC, supports hundreds of concurrent workers
- **SQLite (for testing/small scale)**: Simple setup, limited to ~50-100 concurrent workers
* **Content Storage (File System):** Stores extracted text content in files.
* **Politeness Enforcer:**
* Ensures `robots.txt` rules are followed.
* Enforces the 70-second delay between crawls of the same domain.
* Manages the user-provided exclusion list.
* **Database Abstraction Layer:**
* Provides a common interface for both PostgreSQL and SQLite backends
* Handles connection pooling and query translation if needed
// ... existing code ...
### 3.2. State Storage (Database Schema)
The schema is designed to work with both PostgreSQL and SQLite:
* **`frontier` table:**
* `id` SERIAL/INTEGER PRIMARY KEY
* `url` TEXT UNIQUE NOT NULL
* `domain` TEXT NOT NULL
* `depth` INTEGER DEFAULT 0
* `added_timestamp` BIGINT NOT NULL
* `priority_score` REAL DEFAULT 0
* `claimed_at` BIGINT DEFAULT NULL (for atomic claim-and-read pattern)
* *Indexes: `(domain)`, `(priority_score, added_timestamp)`, `(claimed_at)`*
// ... existing code ...
## 8. Dependencies
* **Python 3.8+**
* **`aiohttp`:** For asynchronous HTTP requests.
* **`beautifulsoup4` or `lxml`:** For HTML parsing. (lxml is generally faster)
* **Database drivers:**
* **`psycopg3[binary]`:** PostgreSQL async driver (recommended for production)
* **`sqlite3`:** (Python standard library) For small-scale testing
* **`robotexclusionrulesparser` or `reppy`:** For `robots.txt` parsing. `reppy` is often well-regarded.
* **`aiofiles`:** For asynchronous file operations (writing content).
* *(Optional)* `cchardet` or `charset_normalizer`: For robust character encoding detection. `aiohttp` handles basic cases.
// ... existing code ...
### 9.2. Running the Crawler
The crawler will be run via `main.py` with command-line arguments:
```bash
python main.py --seed-file path/to/seeds.txt --email your.email@example.com [options]Required Arguments:
--seed-file <path>: Path to the seed file (newline-separated domains/URLs).--email <email_address>: Your contact email for the User-Agent string.
Optional Arguments:
--data-dir <path>: Directory to store database and crawled content (default:./crawler_data).--exclude-file <path>: Path to a file of domains to exclude.--max-workers <int>: Number of concurrent fetcher tasks (default: e.g., 10 or 50).--max-pages <int>: Maximum number of pages to crawl.--max-duration <seconds>: Maximum duration for the crawl.--log-level <level>: Logging level (e.g., INFO, DEBUG).--resume: Flag to attempt resuming from existing data directory.--db-type <sqlite|postgresql>: Database backend to use (default: sqlite for compatibility, postgresql recommended for scale).--db-url <url>: PostgreSQL connection URL (required if db-type is postgresql).
// ... existing code ...
// ... existing code ...
- Database Setup:
- For SQLite: No additional setup required, limited to ~50-100 workers
- For PostgreSQL:
- Install PostgreSQL 14+ for better performance
- Configure
max_connectionsto handle worker count + overhead (e.g., 600 for 500 workers) - Tune
shared_buffers,work_mem, andmaintenance_work_memfor performance - Use connection pooling (built into psycopg3)
// ... existing code ...
For scaling beyond ~100 workers, migration to PostgreSQL is recommended:
- Export SQLite data: Use provided migration scripts
- Create PostgreSQL database:
CREATE DATABASE web_crawler;
- Run schema creation: The crawler will auto-create tables on first run
- Import data: Use provided import scripts
- Update configuration: Switch
--db-typetopostgresqland provide--db-url
The crawler uses a database abstraction layer that:
- Provides async/await interface for both backends
- Handles connection pooling appropriately for each database
- Translates queries when needed (most SQL is compatible)
- Allows easy switching between backends for testing vs. production
Now let's implement the database abstraction layer. I'll create a new module for database backends:
```python
"""
Database backend abstraction layer for the web crawler.
Supports both SQLite and PostgreSQL with a common interface.
"""
import abc
import asyncio
import logging
import sqlite3
import time
from typing import Any, List, Tuple, Optional, Dict
from pathlib import Path
from contextlib import asynccontextmanager
import queue
import threading
logger = logging.getLogger(__name__)
class DatabaseBackend(abc.ABC):
"""Abstract base class for database backends."""
@abc.abstractmethod
async def initialize(self) -> None:
"""Initialize the database connection pool."""
pass
@abc.abstractmethod
async def close(self) -> None:
"""Close all database connections."""
pass
@abc.abstractmethod
async def execute(self, query: str, params: Optional[Tuple] = None) -> None:
"""Execute a query without returning results."""
pass
@abc.abstractmethod
async def execute_many(self, query: str, params_list: List[Tuple]) -> None:
"""Execute a query multiple times with different parameters."""
pass
@abc.abstractmethod
async def fetch_one(self, query: str, params: Optional[Tuple] = None) -> Optional[Tuple]:
"""Execute a query and return a single row."""
pass
@abc.abstractmethod
async def fetch_all(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query and return all rows."""
pass
@abc.abstractmethod
async def execute_returning(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query with RETURNING clause and return results."""
pass
@abc.abstractmethod
async def begin_transaction(self) -> Any:
"""Begin a transaction. Returns a transaction context."""
pass
@abc.abstractmethod
async def commit_transaction(self, tx: Any) -> None:
"""Commit a transaction."""
pass
@abc.abstractmethod
async def rollback_transaction(self, tx: Any) -> None:
"""Rollback a transaction."""
pass
class SQLiteBackend(DatabaseBackend):
"""SQLite implementation of the database backend."""
def __init__(self, db_path: str | Path, pool_size: int = 10, timeout: int = 30):
self.db_path = str(db_path)
self.pool_size = pool_size
self.timeout = timeout
self._pool: queue.Queue = queue.Queue(maxsize=pool_size)
self._lock = threading.Lock()
self._connections_created = 0
self._initialized = False
async def initialize(self) -> None:
"""Initialize the SQLite connection pool."""
if self._initialized:
return
# Create connections in a thread to avoid blocking
await asyncio.to_thread(self._initialize_pool)
self._initialized = True
def _initialize_pool(self):
"""Initialize the connection pool (runs in thread)."""
for _ in range(self.pool_size):
try:
conn = self._create_connection()
self._pool.put_nowait(conn)
except Exception as e:
logger.error(f"Error creating initial connection for pool: {e}")
logger.info(f"SQLite pool initialized with {self._pool.qsize()} connections.")
def _create_connection(self):
"""Create a new SQLite connection."""
try:
conn = sqlite3.connect(self.db_path, timeout=self.timeout, check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA cache_size=-64000") # 64MB cache
conn.execute("PRAGMA temp_store=MEMORY")
conn.execute("PRAGMA mmap_size=268435456") # 256MB memory-mapped I/O
self._connections_created += 1
return conn
except sqlite3.Error as e:
logger.error(f"Failed to create SQLite connection: {e}")
raise
def _get_connection(self, timeout: float = 5.0):
"""Get a connection from the pool."""
try:
return self._pool.get(timeout=timeout)
except queue.Empty:
raise Exception(f"Connection pool exhausted (timeout={timeout}s)")
def _return_connection(self, conn):
"""Return a connection to the pool."""
try:
self._pool.put_nowait(conn)
except queue.Full:
logger.warning("Connection pool is full, closing connection")
conn.close()
async def close(self) -> None:
"""Close all SQLite connections."""
def _close_all():
closed_count = 0
while not self._pool.empty():
try:
conn = self._pool.get_nowait()
conn.close()
closed_count += 1
except queue.Empty:
break
logger.info(f"Closed {closed_count} SQLite connections")
await asyncio.to_thread(_close_all)
async def execute(self, query: str, params: Optional[Tuple] = None) -> None:
"""Execute a query without returning results."""
def _execute(conn):
cursor = conn.cursor()
try:
cursor.execute(query, params or ())
conn.commit()
finally:
cursor.close()
conn = None
try:
conn = await asyncio.to_thread(self._get_connection)
await asyncio.to_thread(_execute, conn)
finally:
if conn:
await asyncio.to_thread(self._return_connection, conn)
async def execute_many(self, query: str, params_list: List[Tuple]) -> None:
"""Execute a query multiple times with different parameters."""
def _execute_many(conn):
cursor = conn.cursor()
try:
cursor.executemany(query, params_list)
conn.commit()
finally:
cursor.close()
conn = None
try:
conn = await asyncio.to_thread(self._get_connection)
await asyncio.to_thread(_execute_many, conn)
finally:
if conn:
await asyncio.to_thread(self._return_connection, conn)
async def fetch_one(self, query: str, params: Optional[Tuple] = None) -> Optional[Tuple]:
"""Execute a query and return a single row."""
def _fetch_one(conn):
cursor = conn.cursor()
try:
cursor.execute(query, params or ())
return cursor.fetchone()
finally:
cursor.close()
conn = None
try:
conn = await asyncio.to_thread(self._get_connection)
return await asyncio.to_thread(_fetch_one, conn)
finally:
if conn:
await asyncio.to_thread(self._return_connection, conn)
async def fetch_all(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query and return all rows."""
def _fetch_all(conn):
cursor = conn.cursor()
try:
cursor.execute(query, params or ())
return cursor.fetchall()
finally:
cursor.close()
conn = None
try:
conn = await asyncio.to_thread(self._get_connection)
return await asyncio.to_thread(_fetch_all, conn)
finally:
if conn:
await asyncio.to_thread(self._return_connection, conn)
async def execute_returning(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query with RETURNING clause and return results."""
def _execute_returning(conn):
cursor = conn.cursor()
try:
cursor.execute(query, params or ())
results = cursor.fetchall()
conn.commit()
return results
finally:
cursor.close()
conn = None
try:
conn = await asyncio.to_thread(self._get_connection)
return await asyncio.to_thread(_execute_returning, conn)
finally:
if conn:
await asyncio.to_thread(self._return_connection, conn)
async def begin_transaction(self) -> sqlite3.Connection:
"""Begin a transaction by getting a connection."""
return await asyncio.to_thread(self._get_connection)
async def commit_transaction(self, conn: sqlite3.Connection) -> None:
"""Commit a transaction and return connection to pool."""
def _commit_and_return():
conn.commit()
self._return_connection(conn)
await asyncio.to_thread(_commit_and_return)
async def rollback_transaction(self, conn: sqlite3.Connection) -> None:
"""Rollback a transaction and return connection to pool."""
def _rollback_and_return():
conn.rollback()
self._return_connection(conn)
await asyncio.to_thread(_rollback_and_return)
class PostgreSQLBackend(DatabaseBackend):
"""PostgreSQL implementation of the database backend using psycopg3."""
def __init__(self, db_url: str, min_size: int = 10, max_size: int = 50):
self.db_url = db_url
self.min_size = min_size
self.max_size = max_size
self._pool = None
self._initialized = False
async def initialize(self) -> None:
"""Initialize the PostgreSQL connection pool."""
if self._initialized:
return
try:
import psycopg_pool
import psycopg
# Create async connection pool
self._pool = psycopg_pool.AsyncConnectionPool(
self.db_url,
min_size=self.min_size,
max_size=self.max_size,
timeout=30,
# Configure for high concurrency
configure=self._configure_connection
)
await self._pool.wait()
self._initialized = True
logger.info(f"PostgreSQL pool initialized with min={self.min_size}, max={self.max_size}")
except ImportError:
raise ImportError("psycopg3 is required for PostgreSQL support. Install with: pip install psycopg[binary,pool]")
except Exception as e:
logger.error(f"Failed to initialize PostgreSQL pool: {e}")
raise
def _configure_connection(self, conn):
"""Configure each connection for optimal performance."""
conn.execute("SET synchronous_commit = OFF") # Faster writes, still durable
conn.execute("SET work_mem = '16MB'")
conn.execute("SET maintenance_work_mem = '64MB'")
async def close(self) -> None:
"""Close the PostgreSQL connection pool."""
if self._pool:
await self._pool.close()
logger.info("PostgreSQL connection pool closed")
@asynccontextmanager
async def _get_connection(self):
"""Get a connection from the pool (context manager)."""
async with self._pool.connection() as conn:
yield conn
async def execute(self, query: str, params: Optional[Tuple] = None) -> None:
"""Execute a query without returning results."""
async with self._get_connection() as conn:
await conn.execute(query, params or ())
async def execute_many(self, query: str, params_list: List[Tuple]) -> None:
"""Execute a query multiple times with different parameters."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
for params in params_list:
await cur.execute(query, params)
async def fetch_one(self, query: str, params: Optional[Tuple] = None) -> Optional[Tuple]:
"""Execute a query and return a single row."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
await cur.execute(query, params or ())
return await cur.fetchone()
async def fetch_all(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query and return all rows."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
await cur.execute(query, params or ())
return await cur.fetchall()
async def execute_returning(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query with RETURNING clause and return results."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
await cur.execute(query, params or ())
return await cur.fetchall()
async def begin_transaction(self) -> Any:
"""Begin a transaction. Returns a connection from the pool."""
# In psycopg3, we need to manage the connection lifecycle manually for transactions
conn = await self._pool.getconn()
return conn
async def commit_transaction(self, conn: Any) -> None:
"""Commit a transaction and return connection to pool."""
await conn.commit()
await self._pool.putconn(conn)
async def rollback_transaction(self, conn: Any) -> None:
"""Rollback a transaction and return connection to pool."""
await conn.rollback()
await self._pool.putconn(conn)
def create_backend(db_type: str, **kwargs) -> DatabaseBackend:
"""
Factory function to create a database backend.
Args:
db_type: Either 'sqlite' or 'postgresql'
**kwargs: Backend-specific arguments
For SQLite: db_path, pool_size, timeout
For PostgreSQL: db_url, min_size, max_size
Returns:
DatabaseBackend instance
"""
if db_type.lower() == 'sqlite':
return SQLiteBackend(
db_path=kwargs['db_path'],
pool_size=kwargs.get('pool_size', 10),
timeout=kwargs.get('timeout', 30)
)
elif db_type.lower() in ('postgresql', 'postgres', 'pg'):
return PostgreSQLBackend(
db_url=kwargs['db_url'],
min_size=kwargs.get('min_size', 10),
max_size=kwargs.get('max_size', 50)
)
else:
raise ValueError(f"Unsupported database type: {db_type}")
"""
Database backend abstraction layer for the web crawler.
Supports both SQLite and PostgreSQL with a common interface.
"""
import abc
import asyncio
import logging
import sqlite3
import time
from typing import Any, List, Tuple, Optional, Dict, TYPE_CHECKING
from pathlib import Path
from contextlib import asynccontextmanager
import queue
import threading
# Type checking imports
if TYPE_CHECKING:
try:
import psycopg_pool
import psycopg
except ImportError:
psycopg_pool = None
psycopg = None
logger = logging.getLogger(__name__)
// ... existing code ...
class PostgreSQLBackend(DatabaseBackend):
"""PostgreSQL implementation of the database backend using psycopg3."""
def __init__(self, db_url: str, min_size: int = 10, max_size: int = 50):
self.db_url = db_url
self.min_size = min_size
self.max_size = max_size
self._pool: Optional[Any] = None # Will be psycopg_pool.AsyncConnectionPool
self._initialized = False
async def initialize(self) -> None:
"""Initialize the PostgreSQL connection pool."""
if self._initialized:
return
try:
import psycopg_pool
import psycopg
# Create async connection pool
self._pool = psycopg_pool.AsyncConnectionPool(
self.db_url,
min_size=self.min_size,
max_size=self.max_size,
timeout=30,
# Configure for high concurrency
configure=self._configure_connection
)
await self._pool.wait()
self._initialized = True
logger.info(f"PostgreSQL pool initialized with min={self.min_size}, max={self.max_size}")
except ImportError:
raise ImportError("psycopg3 is required for PostgreSQL support. Install with: pip install 'psycopg[binary,pool]'")
except Exception as e:
logger.error(f"Failed to initialize PostgreSQL pool: {e}")
raise
def _configure_connection(self, conn) -> None:
"""Configure each connection for optimal performance."""
conn.execute("SET synchronous_commit = OFF") # Faster writes, still durable
conn.execute("SET work_mem = '16MB'")
conn.execute("SET maintenance_work_mem = '64MB'")
async def close(self) -> None:
"""Close the PostgreSQL connection pool."""
if self._pool:
await self._pool.close()
logger.info("PostgreSQL connection pool closed")
@asynccontextmanager
async def _get_connection(self):
"""Get a connection from the pool (context manager)."""
if not self._pool:
raise RuntimeError("PostgreSQL pool not initialized")
async with self._pool.connection() as conn:
yield conn
async def execute(self, query: str, params: Optional[Tuple] = None) -> None:
"""Execute a query without returning results."""
async with self._get_connection() as conn:
await conn.execute(query, params or ())
async def execute_many(self, query: str, params_list: List[Tuple]) -> None:
"""Execute a query multiple times with different parameters."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
for params in params_list:
await cur.execute(query, params)
async def fetch_one(self, query: str, params: Optional[Tuple] = None) -> Optional[Tuple]:
"""Execute a query and return a single row."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
await cur.execute(query, params or ())
return await cur.fetchone()
async def fetch_all(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query and return all rows."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
await cur.execute(query, params or ())
return await cur.fetchall()
async def execute_returning(self, query: str, params: Optional[Tuple] = None) -> List[Tuple]:
"""Execute a query with RETURNING clause and return results."""
async with self._get_connection() as conn:
async with conn.cursor() as cur:
await cur.execute(query, params or ())
return await cur.fetchall()
async def begin_transaction(self) -> Any:
"""Begin a transaction. Returns a connection from the pool."""
if not self._pool:
raise RuntimeError("PostgreSQL pool not initialized")
# In psycopg3, we need to manage the connection lifecycle manually for transactions
conn = await self._pool.getconn()
return conn
async def commit_transaction(self, conn: Any) -> None:
"""Commit a transaction and return connection to pool."""
if not self._pool:
raise RuntimeError("PostgreSQL pool not initialized")
await conn.commit()
await self._pool.putconn(conn)
async def rollback_transaction(self, conn: Any) -> None:
"""Rollback a transaction and return connection to pool."""
if not self._pool:
raise RuntimeError("PostgreSQL pool not initialized")
await conn.rollback()
await self._pool.putconn(conn)
// ... existing code ...Now let's update the config.py to add the new database configuration options:
import argparse
import logging
from dataclasses import dataclass
from pathlib import Path
DEFAULT_DATA_DIR = "./crawler_data"
DEFAULT_MAX_WORKERS = 20
DEFAULT_LOG_LEVEL = "INFO"
DEFAULT_DB_TYPE = "sqlite"
@dataclass
class CrawlerConfig:
seed_file: Path
email: str
data_dir: Path
exclude_file: Path | None
max_workers: int
max_pages: int | None
max_duration: int | None # in seconds
log_level: str
resume: bool
user_agent: str # Will be constructed
seeded_urls_only: bool
db_type: str # sqlite or postgresql
db_url: str | None # PostgreSQL connection URL
def parse_args() -> CrawlerConfig:
parser = argparse.ArgumentParser(description="An experimental web crawler.")
parser.add_argument(
"--seed-file",
type=Path,
required=True,
help="Path to the seed file (newline-separated domains/URLs)."
)
parser.add_argument(
"--email",
type=str,
required=True,
help="Contact email for the User-Agent string."
)
parser.add_argument(
"--data-dir",
type=Path,
default=Path(DEFAULT_DATA_DIR),
help=f"Directory to store database and crawled content (default: {DEFAULT_DATA_DIR})"
)
parser.add_argument(
"--exclude-file",
type=Path,
default=None,
help="Optional path to a file of domains to exclude (newline-separated)."
)
parser.add_argument(
"--max-workers",
type=int,
default=DEFAULT_MAX_WORKERS,
help=f"Number of concurrent fetcher tasks (default: {DEFAULT_MAX_WORKERS})"
)
parser.add_argument(
"--max-pages",
type=int,
default=None,
help="Maximum number of pages to crawl."
)
parser.add_argument(
"--max-duration",
type=int,
default=None,
help="Maximum duration for the crawl in seconds."
)
parser.add_argument(
"--log-level",
type=str,
default=DEFAULT_LOG_LEVEL,
choices=["DEBUG", "INFO", "WARNING", "ERROR"],
help=f"Logging level (default: {DEFAULT_LOG_LEVEL})"
)
parser.add_argument(
"--resume",
action="store_true",
help="Attempt to resume from existing data in data-dir. If not set and data-dir exists, crawler may exit."
)
parser.add_argument(
"--seeded-urls-only",
action="store_true",
help="Only crawl seeded URLs."
)
parser.add_argument(
"--db-type",
type=str,
default=DEFAULT_DB_TYPE,
choices=["sqlite", "postgresql"],
help=f"Database backend to use (default: {DEFAULT_DB_TYPE}). Use 'postgresql' for high concurrency."
)
parser.add_argument(
"--db-url",
type=str,
default=None,
help="PostgreSQL connection URL (required if db-type is postgresql). Example: postgresql://user:pass@localhost/dbname"
)
args = parser.parse_args()
# Validate database configuration
if args.db_type == "postgresql" and not args.db_url:
parser.error("--db-url is required when using PostgreSQL (--db-type=postgresql)")
# Construct User-Agent
# Example: MyEducationalCrawler/1.0 (+http://example.com/crawler-info; mailto:user@example.com)
# For now, a simpler version. We can make the URL part configurable later if needed.
user_agent = f"MyEducationalCrawler/1.0 (mailto:{args.email})"
return CrawlerConfig(
seed_file=args.seed_file,
email=args.email,
data_dir=args.data_dir,
exclude_file=args.exclude_file,
max_workers=args.max_workers,
max_pages=args.max_pages,
max_duration=args.max_duration,
log_level=args.log_level.upper(), # Ensure log level is uppercase for logging module
resume=args.resume,
user_agent=user_agent,
seeded_urls_only=args.seeded_urls_only,
db_type=args.db_type,
db_url=args.db_url
)import sqlite3
import logging
from pathlib import Path
import hashlib # For URL hashing
import time
import asyncio # For to_thread if needed for DB, and for async file saving
import aiofiles # For async file operations
from .config import CrawlerConfig # Assuming config.py is in the same directory
from .db_backends import DatabaseBackend, create_backend
logger = logging.getLogger(__name__)
DB_SCHEMA_VERSION = 2
class StorageManager:
def __init__(self, config: CrawlerConfig, db_backend: DatabaseBackend):
self.config = config
self.data_dir = Path(config.data_dir)
self.db_path = self.data_dir / "crawler_state.db" if config.db_type == "sqlite" else None
self.content_dir = self.data_dir / "content"
self.db = db_backend
self._init_storage()
def _init_storage(self):
"""Initializes the data directory, content directory, and database schema."""
try:
self.data_dir.mkdir(parents=True, exist_ok=True)
self.content_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Storage directories initialized: {self.data_dir}")
except OSError as e:
logger.error(f"Error creating storage directories: {e}")
raise
async def init_db_schema(self):
"""Initializes the database schema if it doesn't exist or needs upgrade."""
try:
logger.info(f"Initializing database schema for {self.config.db_type}")
await self._create_tables()
except Exception as e:
logger.error(f"Database error during schema initialization: {e}")
raise
async def _create_tables(self):
"""Creates the necessary tables in the database if they don't already exist."""
try:
# Check current schema version
await self.db.execute("""
CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)
""")
row = await self.db.fetch_one("SELECT version FROM schema_version ORDER BY version DESC LIMIT 1")
current_version = row[0] if row else 0
if current_version < DB_SCHEMA_VERSION:
logger.info(f"Database schema version is {current_version}. Upgrading to {DB_SCHEMA_VERSION}.")
# Frontier Table - adjust for PostgreSQL/SQLite differences
if self.config.db_type == "postgresql":
await self.db.execute("""
CREATE TABLE IF NOT EXISTS frontier (
id SERIAL PRIMARY KEY,
url TEXT UNIQUE NOT NULL,
domain TEXT NOT NULL,
depth INTEGER DEFAULT 0,
added_timestamp BIGINT NOT NULL,
priority_score REAL DEFAULT 0,
claimed_at BIGINT DEFAULT NULL
)
""")
else: # SQLite
await self.db.execute("""
CREATE TABLE IF NOT EXISTS frontier (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT UNIQUE NOT NULL,
domain TEXT NOT NULL,
depth INTEGER DEFAULT 0,
added_timestamp INTEGER NOT NULL,
priority_score REAL DEFAULT 0,
claimed_at INTEGER DEFAULT NULL
)
""")
# Create indexes
await self.db.execute("CREATE INDEX IF NOT EXISTS idx_frontier_domain ON frontier (domain)")
await self.db.execute("CREATE INDEX IF NOT EXISTS idx_frontier_priority ON frontier (priority_score, added_timestamp)")
await self.db.execute("CREATE INDEX IF NOT EXISTS idx_frontier_claimed ON frontier (claimed_at)")
# Add claimed_at column if upgrading from version 1 (SQLite only)
if current_version == 1 and self.config.db_type == "sqlite":
try:
await self.db.execute("ALTER TABLE frontier ADD COLUMN claimed_at INTEGER DEFAULT NULL")
logger.info("Added claimed_at column to frontier table")
except:
# Column might already exist
pass
# Visited URLs Table
await self.db.execute("""
CREATE TABLE IF NOT EXISTS visited_urls (
url_sha256 TEXT PRIMARY KEY,
url TEXT NOT NULL,
domain TEXT NOT NULL,
crawled_timestamp BIGINT NOT NULL,
http_status_code INTEGER,
content_type TEXT,
content_hash TEXT,
content_storage_path TEXT,
redirected_to_url TEXT
)
""")
await self.db.execute("CREATE INDEX IF NOT EXISTS idx_visited_domain ON visited_urls (domain)")
# Domain Metadata Table
await self.db.execute("""
CREATE TABLE IF NOT EXISTS domain_metadata (
domain TEXT PRIMARY KEY,
last_scheduled_fetch_timestamp BIGINT DEFAULT 0,
robots_txt_content TEXT,
robots_txt_fetched_timestamp BIGINT,
robots_txt_expires_timestamp BIGINT,
is_manually_excluded INTEGER DEFAULT 0,
is_seeded INTEGER DEFAULT 0
)
""")
# Update schema version
await self.db.execute("INSERT INTO schema_version (version) VALUES (?) ON CONFLICT DO NOTHING", (DB_SCHEMA_VERSION,))
logger.info("Database tables created/updated successfully.")
else:
logger.info(f"Database schema is up to date (version {current_version}).")
except Exception as e:
logger.error(f"Error creating database tables: {e}")
raise
def get_url_sha256(self, url: str) -> str:
"""Generates a SHA256 hash for a given URL."""
return hashlib.sha256(url.encode('utf-8')).hexdigest()
async def save_content_to_file(self, url_hash: str, text_content: str) -> Path | None:
"""Saves extracted text content to a file asynchronously."""
if not text_content: # Do not save if no text content
return None
file_path = self.content_dir / f"{url_hash}.txt"
try:
async with aiofiles.open(file_path, mode='w', encoding='utf-8') as f:
await f.write(text_content)
logger.debug(f"Saved content for {url_hash} to {file_path}")
return file_path
except IOError as e:
logger.error(f"IOError saving content file {file_path}: {e}")
return None
except Exception as e:
logger.error(f"Unexpected error saving content file {file_path}: {e}", exc_info=True)
return None
async def add_visited_page(
self,
url: str,
domain: str,
status_code: int,
crawled_timestamp: int,
content_type: str | None = None,
content_text: str | None = None,
content_storage_path_str: str | None = None,
redirected_to_url: str | None = None
):
"""Adds a record of a visited page to the database."""
url_sha256 = self.get_url_sha256(url)
content_hash: str | None = None
if content_text:
content_hash = hashlib.sha256(content_text.encode('utf-8')).hexdigest()
try:
await self.db.execute("""
INSERT INTO visited_urls
(url_sha256, url, domain, crawled_timestamp, http_status_code, content_type, content_hash, content_storage_path, redirected_to_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (url_sha256) DO UPDATE SET
crawled_timestamp = excluded.crawled_timestamp,
http_status_code = excluded.http_status_code,
content_type = excluded.content_type,
content_hash = excluded.content_hash,
content_storage_path = excluded.content_storage_path,
redirected_to_url = excluded.redirected_to_url
""", (
url_sha256, url, domain, crawled_timestamp, status_code,
content_type, content_hash, content_storage_path_str, redirected_to_url
))
logger.info(f"Recorded visited URL: {url} (Status: {status_code})")
except Exception as e:
logger.error(f"DB error adding visited page {url}: {e}")
raise
async def close(self):
"""Close database connections."""
await self.db.close()Now let's update the frontier.py to use the new database backend:
import logging
import time
from pathlib import Path
from typing import Set
import asyncio
from .config import CrawlerConfig
from .storage import StorageManager
from .utils import normalize_url, extract_domain
from .politeness import PolitenessEnforcer
logger = logging.getLogger(__name__)
class FrontierManager:
def __init__(self, config: CrawlerConfig, storage: StorageManager, politeness: PolitenessEnforcer):
self.config = config
self.storage = storage
self.politeness = politeness
self.seen_urls: Set[str] = set() # In-memory set for quick checks during a session
async def _populate_seen_urls_from_db(self) -> Set[str]:
"""Loads all URLs from visited_urls and frontier tables."""
seen = set()
try:
# Load from visited_urls
rows = await self.storage.db.fetch_all("SELECT url FROM visited_urls")
for row in rows:
seen.add(row[0])
# Load from current frontier (in case of resume)
rows = await self.storage.db.fetch_all("SELECT url FROM frontier")
for row in rows:
seen.add(row[0])
logger.info(f"Loaded {len(seen)} URLs for seen_urls set.")
except Exception as e:
logger.error(f"Error loading for seen_urls set: {e}")
return seen
async def initialize_frontier(self):
"""Loads seed URLs and previously saved frontier URLs if resuming."""
# Populate seen_urls from DB
self.seen_urls = await self._populate_seen_urls_from_db()
logger.info(f"Initialized seen_urls with {len(self.seen_urls)} URLs from DB.")
if self.config.resume:
count = await self.count_frontier()
logger.info(f"Resuming crawl. Frontier has {count} URLs.")
if count == 0:
logger.warning("Resuming with an empty frontier. Attempting to load seeds.")
await self._load_seeds()
else:
logger.info("Starting new crawl. Clearing any existing frontier and loading seeds.")
await self._clear_frontier_db()
await self._load_seeds()
async def _mark_domain_as_seeded(self, domain: str):
"""Marks a domain as seeded in the domain_metadata table."""
try:
await self.storage.db.execute("INSERT INTO domain_metadata (domain) VALUES (?) ON CONFLICT DO NOTHING", (domain,))
await self.storage.db.execute("UPDATE domain_metadata SET is_seeded = 1 WHERE domain = ?", (domain,))
logger.debug(f"Marked domain {domain} as seeded in DB.")
except Exception as e:
logger.error(f"DB error marking domain {domain} as seeded: {e}")
raise
async def _load_seeds(self):
"""Reads seed URLs from the seed file and adds them to the frontier."""
if not self.config.seed_file.exists():
logger.error(f"Seed file not found: {self.config.seed_file}")
return
added_count = 0
urls = []
async def _seed_worker(start, end):
nonlocal added_count
for url in urls[start:end]:
if await self.add_url(url):
added_count += 1
# Also mark domain as is_seeded = 1 in domain_metadata table
domain = extract_domain(url)
if domain:
await self._mark_domain_as_seeded(domain)
try:
with open(self.config.seed_file, 'r') as f:
for line in f:
url = line.strip()
if url and not url.startswith("#"):
urls.append(url)
num_per_worker = (len(urls) + self.config.max_workers - 1) // self.config.max_workers
seed_workers = []
for i in range(self.config.max_workers):
start = i * num_per_worker
end = min((i + 1) * num_per_worker, len(urls))
if start < len(urls):
seed_workers.append(_seed_worker(start, end))
await asyncio.gather(*seed_workers)
logger.info(f"Loaded {added_count} URLs from seed file: {self.config.seed_file}")
except IOError as e:
logger.error(f"Error reading seed file {self.config.seed_file}: {e}")
async def _clear_frontier_db(self):
"""Clears the frontier table."""
try:
await self.storage.db.execute("DELETE FROM frontier")
logger.info("Cleared frontier table in database.")
self.seen_urls.clear()
except Exception as e:
logger.error(f"DB error clearing frontier: {e}")
raise
async def add_url(self, url: str, depth: int = 0) -> bool:
"""Adds a normalized URL to the frontier if not already seen or invalid domain."""
normalized_url = normalize_url(url)
if not normalized_url:
return False
if normalized_url in self.seen_urls:
return False # Already processed or in frontier (in-memory check)
domain = extract_domain(normalized_url)
if not domain:
logger.warning(f"Could not extract domain from URL: {normalized_url}, skipping.")
return False
if not await self.politeness.is_url_allowed(normalized_url):
logger.debug(f"URL {normalized_url} disallowed by politeness, not adding.")
self.seen_urls.add(normalized_url) # Mark as seen to avoid re-checking politeness
return False
# Check if already visited
row = await self.storage.db.fetch_one(
"SELECT 1 FROM visited_urls WHERE url_sha256 = ?",
(self.storage.get_url_sha256(normalized_url),)
)
if row:
self.seen_urls.add(normalized_url)
return False
# Try to add to frontier
try:
await self.storage.db.execute(
"INSERT INTO frontier (url, domain, depth, added_timestamp, priority_score, claimed_at) VALUES (?, ?, ?, ?, ?, ?)",
(normalized_url, domain, depth, int(time.time()), 0, None)
)
self.seen_urls.add(normalized_url)
return True
except Exception as e:
# Likely a unique constraint violation (URL already in frontier)
logger.debug(f"Could not add URL {normalized_url} to frontier: {e}")
self.seen_urls.add(normalized_url)
return False
async def count_frontier(self) -> int:
"""Counts the number of URLs currently in the frontier table."""
try:
row = await self.storage.db.fetch_one("SELECT COUNT(*) FROM frontier")
return row[0] if row else 0
except Exception as e:
logger.error(f"Error counting frontier: {e}")
return 0
async def get_next_url(self) -> tuple[str, str, int] | None:
"""Gets the next URL to crawl from the frontier, respecting politeness rules.
Uses atomic claim-and-read to minimize contention.
"""
batch_size = 10 # Claim a small batch to amortize DB overhead
claim_expiry_seconds = 300 # URLs claimed but not processed expire after 5 minutes
# Get worker identifier for logging
worker_task = asyncio.current_task()
worker_name = worker_task.get_name() if worker_task else "unknown"
try:
# Atomically claim a batch of URLs
claimed_urls = await self._atomic_claim_urls(batch_size, claim_expiry_seconds)
if not claimed_urls:
return None
logger.debug(f"{worker_name} claimed {len(claimed_urls)} URLs for processing")
# Track which URLs we need to unclaim if we return early
urls_to_unclaim = []
selected_url_info = None
# Process claimed URLs with politeness checks
for i, (url_id, url, domain) in enumerate(claimed_urls):
# Check if URL is allowed by robots.txt and manual exclusions
if not await self.politeness.is_url_allowed(url):
logger.debug(f"URL {url} (ID: {url_id}) disallowed by politeness rules. Removing.")
await self._delete_url(url_id)
self.seen_urls.add(url)
continue
# Check if we can fetch from this domain now
if await self.politeness.can_fetch_domain_now(domain):
await self.politeness.record_domain_fetch_attempt(domain)
logger.debug(f"{worker_name} processing URL ID {url_id} ({url}) from domain {domain}")
# Delete the URL since we're processing it
await self._delete_url(url_id)
selected_url_info = (url, domain, url_id)
# Mark remaining URLs to be unclaimed
urls_to_unclaim = [(uid, u, d) for uid, u, d in claimed_urls[i+1:]]
break
else:
# Domain not ready - unclaim this URL so another worker can try later
await self._unclaim_url(url_id)
logger.debug(f"Domain {domain} not ready for fetch, unclaimed URL ID {url_id}")
# Unclaim any remaining URLs if we found one to process
for url_id, url, domain in urls_to_unclaim:
await self._unclaim_url(url_id)
logger.debug(f"Unclaimed remaining URL ID {url_id} from batch")
return selected_url_info
except Exception as e:
logger.error(f"Error in get_next_url: {e}", exc_info=True)
return None
async def _atomic_claim_urls(self, batch_size: int, claim_expiry_seconds: int) -> list[tuple[int, str, str]]:
"""Atomically claims a batch of URLs using UPDATE...RETURNING.
Returns list of (id, url, domain) tuples for claimed URLs."""
claim_timestamp = int(time.time() * 1000000) # Microsecond precision
expired_timestamp = claim_timestamp - (claim_expiry_seconds * 1000000)
try:
# First, release any expired claims
await self.storage.db.execute("""
UPDATE frontier
SET claimed_at = NULL
WHERE claimed_at IS NOT NULL AND claimed_at < ?
""", (expired_timestamp,))
# Atomically claim unclaimed URLs using RETURNING
claimed_urls = await self.storage.db.execute_returning("""
UPDATE frontier
SET claimed_at = ?
WHERE id IN (
SELECT id FROM frontier
WHERE claimed_at IS NULL
ORDER BY added_timestamp ASC
LIMIT ?
)
RETURNING id, url, domain
""", (claim_timestamp, batch_size))
return claimed_urls
except Exception as e:
logger.error(f"DB error in atomic claim: {e}")
raise
async def _delete_url(self, url_id: int):
"""Deletes a URL from the frontier after processing or if disallowed."""
try:
await self.storage.db.execute("DELETE FROM frontier WHERE id = ?", (url_id,))
except Exception as e:
logger.error(f"DB error deleting URL ID {url_id}: {e}")
async def _unclaim_url(self, url_id: int):
"""Releases a claim on a URL so other workers can process it."""
try:
await self.storage.db.execute("UPDATE frontier SET claimed_at = NULL WHERE id = ?", (url_id,))
except Exception as e:
logger.error(f"DB error unclaiming URL ID {url_id}: {e}")aiohttp
lxml[html_clean]
robotexclusionrulesparser
aiofiles
cchardet==2.2.0a2
tldextract
psutil
pytest
pytest-asyncio
httpx
# Optional dependencies for PostgreSQL support
# Uncomment the following line to install PostgreSQL driver:
# psycopg[binary,pool]>=3.1#!/usr/bin/env python3
"""
Migration script to transfer data from SQLite to PostgreSQL.
Usage: python migrate_to_postgresql.py --sqlite-db path/to/crawler_state.db --pg-url postgresql://user:pass@host/dbname
"""
import argparse
import sqlite3
import asyncio
import logging
from pathlib import Path
import sys
# Add the crawler module to path
sys.path.insert(0, str(Path(__file__).parent))
from crawler_module.db_backends import create_backend
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
async def migrate_table(sqlite_conn, pg_backend, table_name, batch_size=1000):
"""Migrate a single table from SQLite to PostgreSQL."""
cursor = sqlite_conn.cursor()
try:
# Get total count
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
total_rows = cursor.fetchone()[0]
logger.info(f"Migrating {total_rows} rows from {table_name}")
# Get column names
cursor.execute(f"PRAGMA table_info({table_name})")
columns = [row[1] for row in cursor.fetchall()]
placeholders = ', '.join(['?' for _ in columns])
column_list = ', '.join(columns)
# Migrate in batches
offset = 0
migrated = 0
while offset < total_rows:
cursor.execute(f"SELECT {column_list} FROM {table_name} LIMIT ? OFFSET ?", (batch_size, offset))
rows = cursor.fetchall()
if not rows:
break
# Insert into PostgreSQL
insert_query = f"INSERT INTO {table_name} ({column_list}) VALUES ({placeholders})"
await pg_backend.execute_many(insert_query, rows)
migrated += len(rows)
offset += batch_size
logger.info(f" Migrated {migrated}/{total_rows} rows from {table_name}")
logger.info(f"Successfully migrated {table_name}")
except Exception as e:
logger.error(f"Error migrating {table_name}: {e}")
raise
finally:
cursor.close()
async def main():
parser = argparse.ArgumentParser(description="Migrate crawler data from SQLite to PostgreSQL")
parser.add_argument("--sqlite-db", required=True, help="Path to SQLite database file")
parser.add_argument("--pg-url", required=True, help="PostgreSQL connection URL")
parser.add_argument("--batch-size", type=int, default=1000, help="Batch size for migration (default: 1000)")
args = parser.parse_args()
if not Path(args.sqlite_db).exists():
logger.error(f"SQLite database not found: {args.sqlite_db}")
return 1
# Connect to SQLite
sqlite_conn = sqlite3.connect(args.sqlite_db)
sqlite_conn.row_factory = sqlite3.Row
# Create PostgreSQL backend
pg_backend = create_backend('postgresql', db_url=args.pg_url)
try:
# Initialize PostgreSQL connection
await pg_backend.initialize()
logger.info("Connected to PostgreSQL")
# Create schema in PostgreSQL
logger.info("Creating schema in PostgreSQL...")
# Create tables (PostgreSQL version)
await pg_backend.execute("""
CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)
""")
await pg_backend.execute("""
CREATE TABLE IF NOT EXISTS frontier (
id SERIAL PRIMARY KEY,
url TEXT UNIQUE NOT NULL,
domain TEXT NOT NULL,
depth INTEGER DEFAULT 0,
added_timestamp BIGINT NOT NULL,
priority_score REAL DEFAULT 0,
claimed_at BIGINT DEFAULT NULL
)
""")
await pg_backend.execute("""
CREATE TABLE IF NOT EXISTS visited_urls (
url_sha256 TEXT PRIMARY KEY,
url TEXT NOT NULL,
domain TEXT NOT NULL,
crawled_timestamp BIGINT NOT NULL,
http_status_code INTEGER,
content_type TEXT,
content_hash TEXT,
content_storage_path TEXT,
redirected_to_url TEXT
)
""")
await pg_backend.execute("""
CREATE TABLE IF NOT EXISTS domain_metadata (
domain TEXT PRIMARY KEY,
last_scheduled_fetch_timestamp BIGINT DEFAULT 0,
robots_txt_content TEXT,
robots_txt_fetched_timestamp BIGINT,
robots_txt_expires_timestamp BIGINT,
is_manually_excluded INTEGER DEFAULT 0,
is_seeded INTEGER DEFAULT 0
)
""")
# Create indexes
logger.info("Creating indexes...")
await pg_backend.execute("CREATE INDEX IF NOT EXISTS idx_frontier_domain ON frontier (domain)")
await pg_backend.execute("CREATE INDEX IF NOT EXISTS idx_frontier_priority ON frontier (priority_score, added_timestamp)")
await pg_backend.execute("CREATE INDEX IF NOT EXISTS idx_frontier_claimed ON frontier (claimed_at)")
await pg_backend.execute("CREATE INDEX IF NOT EXISTS idx_visited_domain ON visited_urls (domain)")
# Migrate data
logger.info("Starting data migration...")
# Order matters due to foreign keys
tables = ['schema_version', 'domain_metadata', 'visited_urls', 'frontier']
for table in tables:
# Check if table exists in SQLite
cursor = sqlite_conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,))
if cursor.fetchone():
await migrate_table(sqlite_conn, pg_backend, table, args.batch_size)
else:
logger.warning(f"Table {table} not found in SQLite database, skipping")
cursor.close()
logger.info("Migration completed successfully!")
# Show summary
cursor = sqlite_conn.cursor()
logger.info("\nMigration Summary:")
for table in tables:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
count = cursor.fetchone()[0] if cursor.fetchone() else 0
logger.info(f" {table}: {count} rows")
cursor.close()
except Exception as e:
logger.error(f"Migration failed: {e}")
return 1
finally:
sqlite_conn.close()
await pg_backend.close()
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))import asyncio
import logging
import time
from pathlib import Path
from typing import Set, Optional
import os # For process ID
import psutil
from .config import CrawlerConfig
from .storage import StorageManager
from .fetcher import Fetcher, FetchResult
from .politeness import PolitenessEnforcer
from .frontier import FrontierManager
from .parser import PageParser, ParseResult
from .utils import extract_domain # For getting domain for storage
from .db_backends import create_backend
logger = logging.getLogger(__name__)
# How long a worker should sleep if the frontier is temporarily empty
# or all domains are on cooldown, before trying again.
EMPTY_FRONTIER_SLEEP_SECONDS = 10
# How often the main orchestrator loop checks status and stopping conditions
ORCHESTRATOR_STATUS_INTERVAL_SECONDS = 5
class CrawlerOrchestrator:
def __init__(self, config: CrawlerConfig):
self.config = config
data_dir_path = Path(config.data_dir)
try:
data_dir_path.mkdir(parents=True, exist_ok=True)
logger.info(f"Ensured data directory exists: {data_dir_path}")
except OSError as e:
logger.error(f"Critical error creating data directory {data_dir_path}: {e}")
raise # Stop if we can't create data directory
# Initialize database backend
if config.db_type == 'sqlite':
self.db_backend = create_backend(
'sqlite',
db_path=data_dir_path / "crawler_state.db",
pool_size=config.max_workers,
timeout=30
)
else: # PostgreSQL
# For PostgreSQL, scale pool size with workers but cap it
min_pool = min(10, config.max_workers)
max_pool = min(config.max_workers + 10, 100) # Cap at 100 connections
self.db_backend = create_backend(
'postgresql',
db_url=config.db_url,
min_size=min_pool,
max_size=max_pool
)
self.storage: StorageManager = StorageManager(config, self.db_backend)
self.fetcher: Fetcher = Fetcher(config)
self.politeness: PolitenessEnforcer = PolitenessEnforcer(config, self.storage, self.fetcher)
self.frontier: FrontierManager = FrontierManager(config, self.storage, self.politeness)
self.parser: PageParser = PageParser()
self.pages_crawled_count: int = 0
self.start_time: float = 0.0
self.worker_tasks: Set[asyncio.Task] = set()
self._shutdown_event: asyncio.Event = asyncio.Event()
self.total_urls_added_to_frontier: int = 0
async def _worker(self, worker_id: int):
"""Core logic for a single crawl worker."""
logger.info(f"Worker-{worker_id}: Starting.")
try:
while not self._shutdown_event.is_set():
next_url_info = await self.frontier.get_next_url()
if next_url_info is None:
# Check if the frontier is truly empty (not just all domains on cooldown)
current_frontier_size_for_worker = await self.frontier.count_frontier()
if current_frontier_size_for_worker == 0:
logger.info(f"Worker-{worker_id}: Frontier is confirmed empty by count. Waiting...")
else:
# logger.debug(f"Worker-{worker_id}: No suitable URL available (cooldowns?). Waiting... Frontier size: {current_frontier_size_for_worker}")
pass # Still URLs, but none fetchable now
await asyncio.sleep(EMPTY_FRONTIER_SLEEP_SECONDS)
continue
url_to_crawl, domain, frontier_id = next_url_info
logger.info(f"Worker-{worker_id}: Processing URL ID {frontier_id}: {url_to_crawl} from domain {domain}")
fetch_result: FetchResult = await self.fetcher.fetch_url(url_to_crawl)
crawled_timestamp = int(time.time())
if fetch_result.error_message or fetch_result.status_code >= 400:
logger.warning(f"Worker-{worker_id}: Fetch failed for {url_to_crawl}. Status: {fetch_result.status_code}, Error: {fetch_result.error_message}")
# Record failed attempt
await self.storage.add_visited_page(
url=fetch_result.final_url,
domain=extract_domain(fetch_result.final_url) or domain, # Use original domain as fallback
status_code=fetch_result.status_code,
crawled_timestamp=crawled_timestamp,
content_type=fetch_result.content_type,
redirected_to_url=fetch_result.final_url if fetch_result.is_redirect and fetch_result.initial_url != fetch_result.final_url else None
)
else: # Successful fetch (2xx or 3xx that was followed)
self.pages_crawled_count += 1
logger.info(f"Worker-{worker_id}: Successfully fetched {fetch_result.final_url} (Status: {fetch_result.status_code}). Total crawled: {self.pages_crawled_count}")
content_storage_path_str: str | None = None
parse_data: Optional[ParseResult] = None
if fetch_result.text_content and fetch_result.content_type and "html" in fetch_result.content_type.lower():
logger.debug(f"Worker-{worker_id}: Parsing HTML content for {fetch_result.final_url}")
parse_data = self.parser.parse_html_content(fetch_result.text_content, fetch_result.final_url)
if parse_data.text_content:
url_hash = self.storage.get_url_sha256(fetch_result.final_url)
saved_path = await self.storage.save_content_to_file(url_hash, parse_data.text_content)
if saved_path:
# Store relative path from data_dir for portability
try:
content_storage_path_str = str(saved_path.relative_to(self.config.data_dir))
except ValueError: # Should not happen if content_dir is child of data_dir
content_storage_path_str = str(saved_path)
logger.debug(f"Worker-{worker_id}: Saved content for {fetch_result.final_url} to {content_storage_path_str}")
if parse_data.extracted_links:
logger.debug(f"Worker-{worker_id}: Found {len(parse_data.extracted_links)} links on {fetch_result.final_url}")
links_added_this_page = 0
for link in parse_data.extracted_links:
if await self.frontier.add_url(link):
links_added_this_page +=1
if links_added_this_page > 0:
self.total_urls_added_to_frontier += links_added_this_page
logger.info(f"Worker-{worker_id}: Added {links_added_this_page} new URLs to frontier from {fetch_result.final_url}. Total added: {self.total_urls_added_to_frontier}")
else:
logger.debug(f"Worker-{worker_id}: Not HTML or no text content for {fetch_result.final_url} (Type: {fetch_result.content_type})")
# Record success
await self.storage.add_visited_page(
url=fetch_result.final_url,
domain=extract_domain(fetch_result.final_url) or domain, # Use original domain as fallback
status_code=fetch_result.status_code,
crawled_timestamp=crawled_timestamp,
content_type=fetch_result.content_type,
content_text=parse_data.text_content if parse_data else None, # For hashing
content_storage_path_str=content_storage_path_str,
redirected_to_url=fetch_result.final_url if fetch_result.is_redirect and fetch_result.initial_url != fetch_result.final_url else None
)
# Check stopping conditions after processing a page
if self._check_stopping_conditions():
self._shutdown_event.set()
logger.info(f"Worker-{worker_id}: Stopping condition met, signaling shutdown.")
break
await asyncio.sleep(0) # Yield control to event loop
except asyncio.CancelledError:
logger.info(f"Worker-{worker_id}: Cancelled.")
except Exception as e:
logger.error(f"Worker-{worker_id}: Unhandled exception: {e}", exc_info=True)
finally:
logger.info(f"Worker-{worker_id}: Shutting down.")
def _check_stopping_conditions(self) -> bool:
if self.config.max_pages and self.pages_crawled_count >= self.config.max_pages:
logger.info(f"Stopping: Max pages reached ({self.pages_crawled_count}/{self.config.max_pages})")
return True
if self.config.max_duration and (time.time() - self.start_time) >= self.config.max_duration:
logger.info(f"Stopping: Max duration reached ({time.time() - self.start_time:.0f}s / {self.config.max_duration}s)")
return True
# More complex: check if frontier is empty AND all workers are idle (e.g. waiting on frontier.get_next_url)
# This is handled by the orchestrator's main loop watching worker states and frontier count.
return False
async def run_crawl(self):
self.start_time = time.time()
logger.info(f"Crawler starting with config: {self.config}")
current_process = psutil.Process(os.getpid())
try:
# Initialize database
await self.db_backend.initialize()
await self.storage.init_db_schema()
# Initialize frontier (loads seeds etc.)
await self.frontier.initialize_frontier()
initial_frontier_size = await self.frontier.count_frontier()
self.total_urls_added_to_frontier = initial_frontier_size
logger.info(f"Frontier initialized. Size: {initial_frontier_size}")
if initial_frontier_size == 0 and not self.config.resume:
logger.warning("No URLs in frontier after seed loading. Nothing to crawl.")
self._shutdown_event.set() # Ensure a clean shutdown path
elif initial_frontier_size == 0 and self.config.resume:
logger.warning("Resuming with an empty frontier. Crawler will wait for new URLs or shutdown if none appear.")
# Start worker tasks
for i in range(self.config.max_workers):
task = asyncio.create_task(self._worker(i + 1))
self.worker_tasks.add(task)
logger.info(f"Started {len(self.worker_tasks)} worker tasks.")
# Main monitoring loop
while not self._shutdown_event.is_set():
if not any(not task.done() for task in self.worker_tasks):
logger.info("All worker tasks have completed.")
self._shutdown_event.set()
break
current_frontier_size = await self.frontier.count_frontier()
active_workers = sum(1 for task in self.worker_tasks if not task.done())
# Resource monitoring
rss_mem_mb = "N/A"
fds_count = "N/A"
if current_process:
try:
rss_mem_bytes = current_process.memory_info().rss
rss_mem_mb = f"{rss_mem_bytes / (1024 * 1024):.2f} MB"
fds_count = current_process.num_fds()
except psutil.Error as e:
logger.warning(f"psutil error during resource monitoring: {e}")
# Fallback or disable further psutil calls for this iteration if needed
# Connection pool info - different for each backend
pool_info = "N/A"
if self.config.db_type == 'sqlite' and hasattr(self.db_backend, '_pool'):
pool_info = f"available={self.db_backend._pool.qsize()}"
elif self.config.db_type == 'postgresql':
pool_info = f"min={self.db_backend.min_size},max={self.db_backend.max_size}"
status_parts = [
f"Crawled={self.pages_crawled_count}",
f"Frontier={current_frontier_size}",
f"AddedToFrontier={self.total_urls_added_to_frontier}",
f"ActiveWorkers={active_workers}/{len(self.worker_tasks)}",
f"DBPool({self.config.db_type})={pool_info}",
f"MemRSS={rss_mem_mb}",
f"OpenFDs={fds_count}",
f"Runtime={(time.time() - self.start_time):.0f}s"
]
logger.info(f"Status: {', '.join(status_parts)}")
# If frontier is empty and workers might be idle, it could be a natural end
# This condition needs to be careful not to prematurely shut down if workers are just between tasks.
# A more robust check would involve seeing if workers are truly blocked on an empty frontier for a while.
if current_frontier_size == 0 and self.pages_crawled_count > 0: # Check if any crawling happened
# Wait a bit to see if workers add more URLs or finish
logger.info(f"Frontier is empty. Pages crawled: {self.pages_crawled_count}. Monitoring workers...")
await asyncio.sleep(EMPTY_FRONTIER_SLEEP_SECONDS * 2) # Wait longer than worker sleep
current_frontier_size = await self.frontier.count_frontier()
if current_frontier_size == 0 and not any(not task.done() for task in self.worker_tasks):
logger.info("Frontier empty and all workers appear done. Signaling shutdown.")
self._shutdown_event.set()
break
if self._check_stopping_conditions(): # Check global conditions again
self._shutdown_event.set()
try:
await asyncio.wait_for(self._shutdown_event.wait(), timeout=ORCHESTRATOR_STATUS_INTERVAL_SECONDS)
except asyncio.TimeoutError:
pass # Continue loop for status update and checks
logger.info("Shutdown signaled. Waiting for workers to finish...")
if self.worker_tasks:
# Give workers a chance to finish up gracefully
done, pending = await asyncio.wait(self.worker_tasks, timeout=EMPTY_FRONTIER_SLEEP_SECONDS * 2)
for task in pending:
logger.warning(f"Worker task {task.get_name()} did not finish in time, cancelling.")
task.cancel()
if pending:
await asyncio.gather(*pending, return_exceptions=True) # Await cancellations
logger.info("All worker tasks are finalized.")
except Exception as e:
logger.critical(f"Orchestrator critical error: {e}", exc_info=True)
self._shutdown_event.set() # Ensure shutdown on critical error
# Propagate worker cancellations if any are still running
for task in self.worker_tasks:
if not task.done():
task.cancel()
if self.worker_tasks:
await asyncio.gather(*[task for task in self.worker_tasks if not task.done()], return_exceptions=True)
finally:
logger.info("Performing final cleanup...")
await self.fetcher.close_session()
# Close the database
await self.storage.close()
logger.info(f"Crawl finished. Total pages crawled: {self.pages_crawled_count}")
logger.info(f"Total runtime: {(time.time() - self.start_time):.2f} seconds.")import logging
import time
from pathlib import Path
from urllib.parse import urljoin
from robotexclusionrulesparser import RobotExclusionRulesParser # type: ignore
import asyncio
from .config import CrawlerConfig
from .storage import StorageManager
from .utils import extract_domain
from .fetcher import Fetcher, FetchResult # Added Fetcher import
logger = logging.getLogger(__name__)
DEFAULT_ROBOTS_TXT_TTL = 24 * 60 * 60 # 24 hours in seconds
MIN_CRAWL_DELAY_SECONDS = 70 # Our project's default minimum
class PolitenessEnforcer:
def __init__(self, config: CrawlerConfig, storage: StorageManager, fetcher: Fetcher):
self.config = config
self.storage = storage
self.fetcher = fetcher # Added fetcher instance
self.robots_parsers: dict[str, RobotExclusionRulesParser] = {} # Cache for parsed robots.txt
# Load manual exclusions will be async now
self._manual_exclusions_loaded = False
async def _load_manual_exclusions(self):
"""Loads manually excluded domains from the config file into the DB."""
if self._manual_exclusions_loaded:
return
if self.config.exclude_file and self.config.exclude_file.exists():
try:
with open(self.config.exclude_file, 'r') as f:
count = 0
for line in f:
domain_to_exclude = line.strip().lower()
if domain_to_exclude and not domain_to_exclude.startswith("#"):
await self.storage.db.execute(
"INSERT INTO domain_metadata (domain) VALUES (?) ON CONFLICT DO NOTHING",
(domain_to_exclude,)
)
await self.storage.db.execute(
"UPDATE domain_metadata SET is_manually_excluded = 1 WHERE domain = ?",
(domain_to_exclude,)
)
count += 1
logger.info(f"Loaded and marked {count} domains as manually excluded from {self.config.exclude_file}.")
except IOError as e:
logger.error(f"Error reading exclude file {self.config.exclude_file}: {e}")
except Exception as e:
logger.error(f"DB error loading manual exclusions: {e}")
raise
else:
logger.info("No manual exclude file specified or found.")
self._manual_exclusions_loaded = True
async def _get_cached_robots(self, domain: str) -> tuple | None:
try:
row = await self.storage.db.fetch_one(
"SELECT robots_txt_content, robots_txt_expires_timestamp FROM domain_metadata WHERE domain = ?",
(domain,)
)
return row
except Exception as e:
logger.warning(f"DB error fetching cached robots.txt for {domain}: {e}")
return None
async def _update_robots_cache(self, domain: str, robots_content: str, fetched_timestamp: int, expires_timestamp: int):
try:
await self.storage.db.execute(
"INSERT INTO domain_metadata (domain) VALUES (?) ON CONFLICT DO NOTHING",
(domain,)
)
await self.storage.db.execute(
"UPDATE domain_metadata SET robots_txt_content = ?, robots_txt_fetched_timestamp = ?, robots_txt_expires_timestamp = ? WHERE domain = ?",
(robots_content, fetched_timestamp, expires_timestamp, domain)
)
logger.debug(f"Cached robots.txt for {domain} in DB.")
except Exception as e:
logger.error(f"DB error caching robots.txt for {domain}: {e}")
raise
async def _get_robots_for_domain(self, domain: str) -> RobotExclusionRulesParser | None:
"""Fetches (async), parses, and caches robots.txt for a domain."""
current_time = int(time.time())
rerp = RobotExclusionRulesParser()
robots_content: str | None = None
# Check DB cache
db_row = await self._get_cached_robots(domain)
if db_row and db_row[0] is not None and db_row[1] is not None and db_row[1] > current_time:
logger.debug(f"Using fresh robots.txt for {domain} from DB (expires: {db_row[1]}).")
robots_content = db_row[0]
# If this fresh DB content matches an existing in-memory parser (by source), reuse parser object
if domain in self.robots_parsers and hasattr(self.robots_parsers[domain], 'source_content') and \
self.robots_parsers[domain].source_content == robots_content:
logger.debug(f"In-memory robots_parser for {domain} matches fresh DB. Reusing parser object.")
return self.robots_parsers[domain]
elif domain in self.robots_parsers:
logger.debug(f"DB cache miss/stale for {domain}. Using pre-existing in-memory robots_parser.")
return self.robots_parsers[domain]
# If robots_content is still None here, it means DB was not fresh/valid and we need to fetch
if robots_content is None:
fetched_timestamp = int(time.time())
expires_timestamp = fetched_timestamp + DEFAULT_ROBOTS_TXT_TTL
robots_url_http = f"http://{domain}/robots.txt"
robots_url_https = f"https://{domain}/robots.txt"
logger.info(f"Attempting to fetch robots.txt for {domain} (HTTP first)")
fetch_result_http: FetchResult = await self.fetcher.fetch_url(robots_url_http, is_robots_txt=True)
robots_content_found = False
if fetch_result_http.status_code == 200 and fetch_result_http.text_content is not None:
robots_content = fetch_result_http.text_content
robots_content_found = True
if not robots_content_found: # If HTTP didn't succeed with 200
if fetch_result_http.status_code == 404:
logger.debug(f"robots.txt not found (404) via HTTP for {domain}. Trying HTTPS.")
else:
logger.info(f"HTTP fetch for robots.txt did not yield content for {domain} (status: {fetch_result_http.status_code}). Trying HTTPS.")
fetch_result_https: FetchResult = await self.fetcher.fetch_url(robots_url_https, is_robots_txt=True)
if fetch_result_https.status_code == 200 and fetch_result_https.text_content is not None:
robots_content = fetch_result_https.text_content
elif fetch_result_https.status_code == 404:
logger.debug(f"robots.txt not found (404) via HTTPS for {domain}. Assuming allow all.")
robots_content = ""
else:
logger.warning(f"Failed to fetch robots.txt for {domain} via HTTPS as well (status: {fetch_result_https.status_code}). Assuming allow all.")
robots_content = ""
# Ensure robots_content has a default value if all attempts failed to set it
if robots_content is None:
logger.warning(f"Robots content was unexpectedly None for {domain} after fetch attempts. Defaulting to allow (empty rules).")
robots_content = ""
if robots_content is not None: # robots_content will always be a string here
try:
await self._update_robots_cache(domain, robots_content, fetched_timestamp, expires_timestamp)
except Exception as e:
logger.error(f"Unexpected error caching robots.txt for {domain}: {e}")
if robots_content is not None:
rerp.parse(robots_content)
rerp.source_content = robots_content
else:
rerp.parse("")
rerp.source_content = ""
self.robots_parsers[domain] = rerp # Update/add to in-memory cache
return rerp
async def _check_manual_exclusion(self, domain: str) -> bool:
"""Returns True if domain is manually excluded or (if seeded_urls_only) not seeded."""
try:
if self.config.seeded_urls_only:
row = await self.storage.db.fetch_one(
"SELECT is_manually_excluded, is_seeded FROM domain_metadata WHERE domain = ?",
(domain,)
)
if row:
is_excluded = row[0] == 1
is_seeded = row[1] == 1
return is_excluded or not is_seeded
else:
# Domain not in metadata - it's not seeded
return True
else:
row = await self.storage.db.fetch_one(
"SELECT is_manually_excluded FROM domain_metadata WHERE domain = ?",
(domain,)
)
return row and row[0] == 1
except Exception as e:
logger.warning(f"DB error checking manual exclusion for {domain}: {e}")
return False # Default to not excluded on error
async def is_url_allowed(self, url: str) -> bool:
"""Checks if a URL is allowed by robots.txt and not manually excluded."""
# Ensure manual exclusions are loaded
await self._load_manual_exclusions()
domain = extract_domain(url)
if not domain:
logger.warning(f"Could not extract domain for URL: {url} for robots check. Allowing.")
return True
# Check manual exclusion
if await self._check_manual_exclusion(domain):
logger.debug(f"URL {url} from manually excluded or non-seeded domain: {domain}")
return False
# If not manually excluded, check robots.txt
rerp = await self._get_robots_for_domain(domain)
if rerp:
is_allowed_by_robots = rerp.is_allowed(self.config.user_agent, url)
if not is_allowed_by_robots:
logger.debug(f"URL disallowed by robots.txt for {domain}: {url}")
return is_allowed_by_robots
logger.warning(f"No robots.txt parser available for {domain} after checks. Allowing URL: {url}")
return True # Default to allow if RERP is None
async def get_crawl_delay(self, domain: str) -> float:
"""Gets the crawl delay for a domain."""
rerp = await self._get_robots_for_domain(domain)
delay = None
if rerp:
agent_delay = rerp.get_crawl_delay(self.config.user_agent)
if agent_delay is not None:
delay = float(agent_delay)
logger.debug(f"Using {delay}s crawl delay from robots.txt for {domain} (agent: {self.config.user_agent})")
if delay is None:
if rerp and rerp.get_crawl_delay("*") is not None:
wildcard_delay = rerp.get_crawl_delay("*")
if wildcard_delay is not None:
delay = float(wildcard_delay)
logger.debug(f"Using {delay}s crawl delay from robots.txt for {domain} (wildcard agent)")
if delay is None:
logger.debug(f"No crawl delay specified in robots.txt for {domain}. Using default: {MIN_CRAWL_DELAY_SECONDS}s")
return float(MIN_CRAWL_DELAY_SECONDS)
return max(float(delay), float(MIN_CRAWL_DELAY_SECONDS))
async def can_fetch_domain_now(self, domain: str) -> bool:
"""Checks if the domain can be fetched now based on crawl delay."""
last_fetch_time = 0
try:
row = await self.storage.db.fetch_one(
"SELECT last_scheduled_fetch_timestamp FROM domain_metadata WHERE domain = ?",
(domain,)
)
if row and row[0] is not None:
last_fetch_time = row[0]
except Exception as e:
logger.error(f"Error checking last fetch time for {domain}: {e}. Assuming cannot fetch.")
return False
crawl_delay = await self.get_crawl_delay(domain)
current_time = int(time.time())
return current_time >= last_fetch_time + crawl_delay
async def record_domain_fetch_attempt(self, domain: str):
"""Records that we are about to fetch from a domain."""
try:
current_time = int(time.time())
await self.storage.db.execute(
"INSERT INTO domain_metadata (domain, last_scheduled_fetch_timestamp) VALUES (?, ?) ON CONFLICT (domain) DO UPDATE SET last_scheduled_fetch_timestamp = ?",
(domain, current_time, current_time)
)
except Exception as e:
logger.error(f"Error recording fetch attempt for {domain}: {e}") async def _check_manual_exclusion(self, domain: str) -> bool:
"""Returns True if domain is manually excluded or (if seeded_urls_only) not seeded."""
try:
if self.config.seeded_urls_only:
row = await self.storage.db.fetch_one(
"SELECT is_manually_excluded, is_seeded FROM domain_metadata WHERE domain = ?",
(domain,)
)
if row:
is_excluded = bool(row[0] == 1)
is_seeded = bool(row[1] == 1)
return is_excluded or not is_seeded
else:
# Domain not in metadata - it's not seeded
return True
else:
row = await self.storage.db.fetch_one(
"SELECT is_manually_excluded FROM domain_metadata WHERE domain = ?",
(domain,)
)
return bool(row and row[0] == 1)
except Exception as e:
logger.warning(f"DB error checking manual exclusion for {domain}: {e}")
return False # Default to not excluded on error#!/bin/bash
# Script to set up system limits for high-concurrency web crawler
echo "Setting up system limits for web crawler..."
# Check current limits
echo "Current file descriptor limit: $(ulimit -n)"
echo "Current max user processes: $(ulimit -u)"
# Set limits for current session
ulimit -n 65536
ulimit -u 32768
echo "Updated limits for current session:"
echo " File descriptors: $(ulimit -n)"
echo " Max processes: $(ulimit -u)"
# Create systemd service override if running as a service
if [ -d "/etc/systemd" ]; then
echo ""
echo "To make these limits permanent for systemd services, create:"
echo "/etc/systemd/system/crawler.service.d/limits.conf"
echo "with content:"
echo "[Service]"
echo "LimitNOFILE=65536"
echo "LimitNPROC=32768"
fi
# Suggest permanent changes
echo ""
echo "For permanent system-wide changes, add to /etc/security/limits.conf:"
echo "* soft nofile 65536"
echo "* hard nofile 65536"
echo "* soft nproc 32768"
echo "* hard nproc 32768"
echo ""
echo "And to /etc/sysctl.conf:"
echo "fs.file-max = 2097152"
echo "fs.nr_open = 1048576"
echo ""
echo "Then run: sudo sysctl -p"
# Check if running with sufficient limits
if [ $(ulimit -n) -lt 10000 ]; then
echo ""
echo "WARNING: File descriptor limit is still low for high concurrency!"
echo "Consider reducing workers or increasing limits."
fi
# PostgreSQL Migration Guide
This guide explains how to migrate your web crawler from SQLite to PostgreSQL for better concurrency and scaling to hundreds of workers.
## Why PostgreSQL?
- **Better concurrency**: PostgreSQL's MVCC allows many concurrent writers without locking
- **Scales to 500+ workers**: No "database is locked" errors
- **Same SQL queries**: Minimal code changes required (we use the same RETURNING clause)
- **Full ACID compliance**: Same durability guarantees as SQLite
## Prerequisites
1. PostgreSQL 14+ installed and running
2. A database created for the crawler
3. Python PostgreSQL driver: `pip install 'psycopg[binary,pool]'`
## Setup Instructions
### 1. Install PostgreSQL (Ubuntu/Debian)
```bash
# Install PostgreSQL
sudo apt update
sudo apt install postgresql postgresql-contrib
# Start PostgreSQL
sudo systemctl start postgresql
sudo systemctl enable postgresql# Switch to postgres user
sudo -u postgres psql
# In PostgreSQL prompt:
CREATE DATABASE web_crawler;
CREATE USER crawler_user WITH PASSWORD 'your_secure_password';
GRANT ALL PRIVILEGES ON DATABASE web_crawler TO crawler_user;
\qEdit /etc/postgresql/14/main/postgresql.conf:
# Connection settings
max_connections = 600 # Support 500 workers + overhead
shared_buffers = 256MB # 25% of RAM for dedicated server
effective_cache_size = 1GB # 50-75% of RAM
work_mem = 16MB # Per sort/hash operation
maintenance_work_mem = 64MB # For vacuum, index creation
# Write performance
synchronous_commit = off # Faster writes, still durable
wal_buffers = 16MB
checkpoint_completion_target = 0.9Restart PostgreSQL:
sudo systemctl restart postgresql# Uncomment the psycopg line in requirements.txt, then:
pip install -r requirements.txt
# Or install directly:
pip install 'psycopg[binary,pool]>=3.1'Simply run the crawler with PostgreSQL parameters:
python main.py \
--seed-file seeds.txt \
--email your@email.com \
--db-type postgresql \
--db-url "postgresql://crawler_user:your_secure_password@localhost/web_crawler" \
--max-workers 500Use the provided migration script:
# Basic migration
python migrate_to_postgresql.py \
--sqlite-db crawler_data/crawler_state.db \
--pg-url "postgresql://crawler_user:your_secure_password@localhost/web_crawler"
# With custom batch size for large databases
python migrate_to_postgresql.py \
--sqlite-db crawler_data/crawler_state.db \
--pg-url "postgresql://crawler_user:your_secure_password@localhost/web_crawler" \
--batch-size 5000python main.py \
--seed-file top-1M-domains.txt \
--email your@email.com \
--db-type postgresql \
--db-url "postgresql://crawler_user:password@localhost/web_crawler" \
--max-workers 500postgresql://[user[:password]@][host][:port][/dbname][?param1=value1&...]
Examples:
- Local:
postgresql://crawler_user:pass@localhost/web_crawler - Remote:
postgresql://crawler_user:pass@db.example.com:5432/web_crawler - With SSL:
postgresql://user:pass@host/db?sslmode=require
Before running with many workers, increase system limits:
# Run the setup script
chmod +x setup_system_limits.sh
./setup_system_limits.sh
# Or manually:
ulimit -n 65536 # File descriptors
ulimit -u 32768 # Max processesThe crawler automatically configures connection pooling based on worker count:
- Min pool size:
min(10, max_workers) - Max pool size:
min(max_workers + 10, 100)
You can monitor connections:
-- Active connections
SELECT count(*) FROM pg_stat_activity WHERE datname = 'web_crawler';
-- Connection details
SELECT pid, usename, application_name, state, query_start
FROM pg_stat_activity
WHERE datname = 'web_crawler';The crawler logs database pool information:
Status: ... DBPool(postgresql)=min=10,max=60 ...
-- Table sizes
SELECT
schemaname,
tablename,
pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS size
FROM pg_tables
WHERE schemaname = 'public'
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC;
-- Slow queries
SELECT query, mean_exec_time, calls
FROM pg_stat_statements
ORDER BY mean_exec_time DESC
LIMIT 10;
-- Lock monitoring
SELECT * FROM pg_locks WHERE NOT granted;- Check PostgreSQL is running:
sudo systemctl status postgresql - Verify connection settings in
pg_hba.conf - Check firewall rules if remote
- Increase
max_connectionsinpostgresql.conf - Reduce
--max-workers - Check for connection leaks
- Run
ANALYZEon tables after migration - Check indexes exist (migration script creates them)
- Monitor slow query log
- Consider partitioning frontier table for very large crawls
| Feature | SQLite | PostgreSQL |
|---|---|---|
| Max concurrent workers | ~50-100 | 500+ |
| Write concurrency | Single writer | Multiple writers (MVCC) |
| "Database locked" errors | Common at scale | None |
| Setup complexity | None | Moderate |
| Memory usage | Low | Higher (configurable) |
| Best for | Testing, small crawls | Production, large scale |
If needed, you can always switch back to SQLite:
python main.py \
--seed-file seeds.txt \
--email your@email.com \
--db-type sqlite \
--max-workers 50 # Reduce workers for SQLiteThe crawler maintains compatibility with both databases.
```python
import pytest
import pytest_asyncio # For async fixtures and tests
import asyncio
from pathlib import Path
import shutil
import logging
import time
from dataclasses import dataclass
from unittest.mock import MagicMock, AsyncMock # Added AsyncMock
from crawler_module.frontier import FrontierManager
from crawler_module.storage import StorageManager
from crawler_module.config import CrawlerConfig
from crawler_module.politeness import PolitenessEnforcer # Added
from crawler_module.fetcher import Fetcher # Added for PolitenessEnforcer mock typing
from crawler_module.db_backends import create_backend # Use new database abstraction
# Configure basic logging for tests
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class FrontierTestConfig:
data_dir: Path
seed_file: Path
email: str = "frontier_test@example.com"
exclude_file: Path | None = None
max_workers: int = 1 # Keep low for predictable test flow
max_pages: int | None = None
max_duration: int | None = None
log_level: str = "DEBUG"
resume: bool = False
user_agent: str = "FrontierTestCrawler/1.0"
seeded_urls_only: bool = False
db_type: str = "sqlite" # Use SQLite for tests
db_url: str | None = None
@pytest_asyncio.fixture
async def temp_test_frontier_dir(tmp_path: Path) -> Path:
test_data_dir = tmp_path / "test_crawler_data_frontier"
test_data_dir.mkdir(parents=True, exist_ok=True)
logger.debug(f"Created temp test dir for frontier: {test_data_dir}")
return test_data_dir
@pytest_asyncio.fixture
async def frontier_test_config_obj(temp_test_frontier_dir: Path) -> FrontierTestConfig:
"""Provides the FrontierTestConfig object, seeds file created here."""
seed_file_path = temp_test_frontier_dir / "test_seeds.txt"
with open(seed_file_path, 'w') as sf:
sf.write("http://example.com/seed1\n")
sf.write("http://example.org/seed2\n")
sf.write("http://example.com/seed1\n") # Duplicate to test seen
return FrontierTestConfig(data_dir=temp_test_frontier_dir, seed_file=seed_file_path)
@pytest_asyncio.fixture
async def actual_config_for_frontier(frontier_test_config_obj: FrontierTestConfig) -> CrawlerConfig:
"""Provides the actual CrawlerConfig based on FrontierTestConfig."""
return CrawlerConfig(**vars(frontier_test_config_obj))
@pytest_asyncio.fixture
async def db_backend(actual_config_for_frontier: CrawlerConfig):
"""Provides a database backend for tests."""
backend = create_backend(
'sqlite',
db_path=actual_config_for_frontier.data_dir / "test_crawler_state.db",
pool_size=1,
timeout=10
)
await backend.initialize()
yield backend
await backend.close()
@pytest_asyncio.fixture
async def storage_manager_for_frontier(actual_config_for_frontier: CrawlerConfig, db_backend) -> StorageManager:
sm = StorageManager(config=actual_config_for_frontier, db_backend=db_backend)
await sm.init_db_schema()
yield sm
@pytest_asyncio.fixture
def mock_politeness_enforcer_for_frontier(actual_config_for_frontier: CrawlerConfig,
mock_storage_manager: MagicMock,
mock_fetcher: MagicMock) -> MagicMock:
"""Provides a mocked PolitenessEnforcer for FrontierManager tests."""
mock_pe = AsyncMock(spec=PolitenessEnforcer)
# Default mock behaviors for permissive testing of FrontierManager
mock_pe.is_url_allowed = AsyncMock(return_value=True)
mock_pe.can_fetch_domain_now = AsyncMock(return_value=True)
mock_pe.record_domain_fetch_attempt = AsyncMock()
mock_pe.get_crawl_delay = AsyncMock(return_value=0.0) # So it doesn't delay tests
mock_pe._load_manual_exclusions = AsyncMock() # Mock the async initialization
return mock_pe
# Fixture for a mock fetcher (can be shared or defined per test file)
@pytest.fixture
def mock_fetcher() -> MagicMock:
return AsyncMock(spec=Fetcher)
@pytest.fixture
def mock_storage_manager() -> MagicMock: # This mock is for PolitenessEnforcer, may not need db if methods are mocked
mock = MagicMock(spec=StorageManager)
# Mock the db attribute with async methods
mock.db = AsyncMock()
return mock
@pytest_asyncio.fixture
async def frontier_manager(
actual_config_for_frontier: CrawlerConfig,
storage_manager_for_frontier: StorageManager, # This SM instance now uses the db_backend fixture
mock_politeness_enforcer_for_frontier: MagicMock
) -> FrontierManager:
fm = FrontierManager(
config=actual_config_for_frontier,
storage=storage_manager_for_frontier,
politeness=mock_politeness_enforcer_for_frontier
)
return fm
@pytest.mark.asyncio
async def test_frontier_initialization_new(frontier_manager: FrontierManager, mock_politeness_enforcer_for_frontier: MagicMock):
logger.info("Testing Frontier Initialization (New Crawl)")
# Ensure is_url_allowed is permissive during seed loading
mock_politeness_enforcer_for_frontier.is_url_allowed.return_value = True
await frontier_manager.initialize_frontier()
# count_frontier is now async
assert await frontier_manager.count_frontier() == 2
logger.info("Frontier initialization (new) test passed.")
# Check that is_url_allowed was called for seeds
assert mock_politeness_enforcer_for_frontier.is_url_allowed.call_count >= 2
@pytest.mark.asyncio
async def test_add_and_get_urls(frontier_manager: FrontierManager, mock_politeness_enforcer_for_frontier: MagicMock):
logger.info("Testing Add and Get URLs from Frontier")
# Permissive politeness for initialization and adding
mock_politeness_enforcer_for_frontier.is_url_allowed.return_value = True
mock_politeness_enforcer_for_frontier.can_fetch_domain_now.return_value = True
await frontier_manager.initialize_frontier()
initial_count = await frontier_manager.count_frontier()
assert initial_count == 2
# Test adding a new URL
await frontier_manager.add_url("http://test.com/page1")
assert await frontier_manager.count_frontier() == 3
# is_url_allowed should be called by add_url
# Initial calls for seeds + 1 for this add_url
assert mock_politeness_enforcer_for_frontier.is_url_allowed.call_count >= 3
# Try adding a duplicate of a seed - should not increase count
current_is_allowed_calls = mock_politeness_enforcer_for_frontier.is_url_allowed.call_count
await frontier_manager.add_url("http://example.com/seed1")
assert await frontier_manager.count_frontier() == 3
# Get URLs
# For get_next_url, ensure politeness checks are permissive for testing FIFO logic here
mock_politeness_enforcer_for_frontier.is_url_allowed.reset_mock(return_value=True) # Reset and keep permissive
mock_politeness_enforcer_for_frontier.can_fetch_domain_now.reset_mock(return_value=True)
mock_politeness_enforcer_for_frontier.record_domain_fetch_attempt.reset_mock()
expected_order = ["http://example.com/seed1", "http://example.org/seed2", "http://test.com/page1"]
retrieved_urls = []
for i in range(len(expected_order)):
next_url_info = await frontier_manager.get_next_url()
assert next_url_info is not None, f"Expected URL, got None at iteration {i}"
retrieved_urls.append(next_url_info[0])
# Politeness checks for get_next_url
assert mock_politeness_enforcer_for_frontier.is_url_allowed.called
assert mock_politeness_enforcer_for_frontier.can_fetch_domain_now.called
assert mock_politeness_enforcer_for_frontier.record_domain_fetch_attempt.called
# Reset mocks for next iteration if asserting per-iteration calls
mock_politeness_enforcer_for_frontier.is_url_allowed.reset_mock(return_value=True)
mock_politeness_enforcer_for_frontier.can_fetch_domain_now.reset_mock(return_value=True)
mock_politeness_enforcer_for_frontier.record_domain_fetch_attempt.reset_mock()
assert retrieved_urls == expected_order
assert await frontier_manager.count_frontier() == 0
next_url_info = await frontier_manager.get_next_url()
assert next_url_info is None, "Frontier should be empty"
logger.info("Add and get URLs test passed with politeness mocks.")
@pytest.mark.asyncio
async def test_frontier_resume_with_politeness(
temp_test_frontier_dir: Path,
frontier_test_config_obj: FrontierTestConfig,
mock_politeness_enforcer_for_frontier: MagicMock,
# tmp_path is needed to create distinct DBs for each run if not using a shared fixture pool
tmp_path: Path
):
logger.info("Testing Frontier Resume Functionality with Politeness Mocks")
# Permissive politeness for all operations in this test
mock_politeness_enforcer_for_frontier.is_url_allowed.return_value = True
mock_politeness_enforcer_for_frontier.can_fetch_domain_now.return_value = True
mock_politeness_enforcer_for_frontier.record_domain_fetch_attempt.return_value = None # AsyncMock should return None by default
# Define a unique DB file for this multi-run test
db_file_for_resume_test = temp_test_frontier_dir / "resume_test_frontier.db"
# --- First run: populate and close ---
cfg_run1_dict = vars(frontier_test_config_obj).copy()
cfg_run1_dict['resume'] = False # Ensure it's a new run
cfg_run1 = CrawlerConfig(**cfg_run1_dict)
backend_run1 = create_backend('sqlite', db_path=db_file_for_resume_test, pool_size=1)
await backend_run1.initialize()
storage_run1 = StorageManager(config=cfg_run1, db_backend=backend_run1)
await storage_run1.init_db_schema()
frontier_run1 = FrontierManager(config=cfg_run1, storage=storage_run1, politeness=mock_politeness_enforcer_for_frontier)
await frontier_run1.initialize_frontier()
await frontier_run1.add_url("http://persistent.com/page_from_run1")
assert await frontier_run1.count_frontier() == 3
url_to_retrieve = await frontier_run1.get_next_url()
assert url_to_retrieve is not None
assert await frontier_run1.count_frontier() == 2
await backend_run1.close()
# --- Second run: resume ---
cfg_run2_dict = vars(frontier_test_config_obj).copy()
cfg_run2_dict['resume'] = True
cfg_run2 = CrawlerConfig(**cfg_run2_dict)
backend_run2 = create_backend('sqlite', db_path=db_file_for_resume_test, pool_size=1)
await backend_run2.initialize()
storage_run2 = StorageManager(config=cfg_run2, db_backend=backend_run2)
await storage_run2.init_db_schema()
frontier_run2 = FrontierManager(config=cfg_run2, storage=storage_run2, politeness=mock_politeness_enforcer_for_frontier)
await frontier_run2.initialize_frontier()
assert await frontier_run2.count_frontier() == 2
# Try to get the remaining URLs
next_url = await frontier_run2.get_next_url()
assert next_url is not None
assert next_url[0] == "http://example.org/seed2"
next_url = await frontier_run2.get_next_url()
assert next_url is not None
assert next_url[0] == "http://persistent.com/page_from_run1"
assert await frontier_run2.count_frontier() == 0
await backend_run2.close()
logger.info("Frontier resume test passed with politeness mocks.")
# Update schema version
if self.config.db_type == "postgresql":
await self.db.execute("INSERT INTO schema_version (version) VALUES (%s) ON CONFLICT DO NOTHING", (DB_SCHEMA_VERSION,))
else: # SQLite
await self.db.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", (DB_SCHEMA_VERSION,)) try:
if self.config.db_type == "postgresql":
await self.db.execute("""
INSERT INTO visited_urls
(url_sha256, url, domain, crawled_timestamp, http_status_code, content_type, content_hash, content_storage_path, redirected_to_url)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (url_sha256) DO UPDATE SET
crawled_timestamp = excluded.crawled_timestamp,
http_status_code = excluded.http_status_code,
content_type = excluded.content_type,
content_hash = excluded.content_hash,
content_storage_path = excluded.content_storage_path,
redirected_to_url = excluded.redirected_to_url
""", (
url_sha256, url, domain, crawled_timestamp, status_code,
content_type, content_hash, content_storage_path_str, redirected_to_url
))
else: # SQLite
await self.db.execute("""
INSERT OR REPLACE INTO visited_urls
(url_sha256, url, domain, crawled_timestamp, http_status_code, content_type, content_hash, content_storage_path, redirected_to_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
url_sha256, url, domain, crawled_timestamp, status_code,
content_type, content_hash, content_storage_path_str, redirected_to_url
))
logger.info(f"Recorded visited URL: {url} (Status: {status_code})") if domain_to_exclude and not domain_to_exclude.startswith("#"):
if self.storage.config.db_type == "postgresql":
await self.storage.db.execute(
"INSERT INTO domain_metadata (domain) VALUES (%s) ON CONFLICT DO NOTHING",
(domain_to_exclude,)
)
else: # SQLite
await self.storage.db.execute(
"INSERT OR IGNORE INTO domain_metadata (domain) VALUES (?)",
(domain_to_exclude,)
)
await self.storage.db.execute(
"UPDATE domain_metadata SET is_manually_excluded = 1 WHERE domain = %s" if self.storage.config.db_type == "postgresql" else "UPDATE domain_metadata SET is_manually_excluded = 1 WHERE domain = ?",
(domain_to_exclude,)
) async def _update_robots_cache(self, domain: str, robots_content: str, fetched_timestamp: int, expires_timestamp: int):
try:
if self.storage.config.db_type == "postgresql":
await self.storage.db.execute(
"INSERT INTO domain_metadata (domain) VALUES (%s) ON CONFLICT DO NOTHING",
(domain,)
)
await self.storage.db.execute(
"UPDATE domain_metadata SET robots_txt_content = %s, robots_txt_fetched_timestamp = %s, robots_txt_expires_timestamp = %s WHERE domain = %s",
(robots_content, fetched_timestamp, expires_timestamp, domain)
)
else: # SQLite
await self.storage.db.execute(
"INSERT OR IGNORE INTO domain_metadata (domain) VALUES (?)",
(domain,)
)
await self.storage.db.execute(
"UPDATE domain_metadata SET robots_txt_content = ?, robots_txt_fetched_timestamp = ?, robots_txt_expires_timestamp = ? WHERE domain = ?",
(robots_content, fetched_timestamp, expires_timestamp, domain)
)
logger.debug(f"Cached robots.txt for {domain} in DB.")
except Exception as e:
logger.error(f"DB error caching robots.txt for {domain}: {e}")
raise async def record_domain_fetch_attempt(self, domain: str):
"""Records that we are about to fetch from a domain."""
try:
current_time = int(time.time())
if self.storage.config.db_type == "postgresql":
await self.storage.db.execute(
"INSERT INTO domain_metadata (domain, last_scheduled_fetch_timestamp) VALUES (%s, %s) ON CONFLICT (domain) DO UPDATE SET last_scheduled_fetch_timestamp = %s",
(domain, current_time, current_time)
)
else: # SQLite
await self.storage.db.execute(
"INSERT OR REPLACE INTO domain_metadata (domain, last_scheduled_fetch_timestamp) VALUES (?, ?)",
(domain, current_time)
)
except Exception as e:
logger.error(f"Error recording fetch attempt for {domain}: {e}") async def _mark_domain_as_seeded(self, domain: str):
"""Marks a domain as seeded in the domain_metadata table."""
try:
if self.storage.config.db_type == "postgresql":
await self.storage.db.execute("INSERT INTO domain_metadata (domain) VALUES (%s) ON CONFLICT DO NOTHING", (domain,))
await self.storage.db.execute("UPDATE domain_metadata SET is_seeded = 1 WHERE domain = %s", (domain,))
else: # SQLite
await self.storage.db.execute("INSERT OR IGNORE INTO domain_metadata (domain) VALUES (?)", (domain,))
await self.storage.db.execute("UPDATE domain_metadata SET is_seeded = 1 WHERE domain = ?", (domain,))
logger.debug(f"Marked domain {domain} as seeded in DB.")
except Exception as e:
logger.error(f"DB error marking domain {domain} as seeded: {e}")
raise async def execute(self, query: str, params: Optional[Tuple] = None) -> None:
"""Execute a query without returning results."""
async with self._get_connection() as conn:
# psycopg3 automatically handles parameter placeholders
await conn.execute(query, params or ())# Database Migration Summary
This document summarizes the changes made to support PostgreSQL as a database backend for the web crawler, enabling scaling to 500+ concurrent workers.
## Overview
The crawler has been refactored to support both SQLite (for small-scale testing) and PostgreSQL (for production use with high concurrency). The migration addresses the "database is locked" errors that occurred with SQLite when running hundreds of workers.
## Key Changes
### 1. Database Abstraction Layer (`db_backends.py`)
- Created abstract `DatabaseBackend` interface
- Implemented `SQLiteBackend` with connection pooling
- Implemented `PostgreSQLBackend` using psycopg3 with async support
- Factory function `create_backend()` to instantiate the appropriate backend
### 2. Configuration Updates (`config.py`)
- Added `--db-type` parameter (sqlite/postgresql)
- Added `--db-url` parameter for PostgreSQL connection string
- Defaults to SQLite for backward compatibility
### 3. Storage Manager Changes (`storage.py`)
- Refactored to use `DatabaseBackend` instead of direct SQLite connections
- Updated schema creation to handle both database types
- Converted all database operations to async
- Fixed SQL syntax differences (e.g., `INSERT OR IGNORE` vs `ON CONFLICT DO NOTHING`)
### 4. Frontier Manager Updates (`frontier.py`)
- Removed direct SQLite connection handling
- All database operations now use the storage manager's backend
- Maintained atomic claim-and-read pattern for URL processing
### 5. Politeness Enforcer Updates (`politeness.py`)
- Removed `conn_in` parameters from all methods
- Converted to fully async database operations
- Fixed SQL parameter placeholders (`?` for SQLite, `%s` for PostgreSQL)
### 6. Orchestrator Changes (`orchestrator.py`)
- Initialize database backend based on configuration
- Improved connection pool configuration for PostgreSQL
- Added database-specific monitoring in status logs
### 7. Test Updates (`test_frontier.py`)
- Tests now use the database abstraction layer
- Maintain SQLite for fast testing
- All tests passing with the new architecture
### 8. Migration Tools
- Created `migrate_to_postgresql.py` script for data migration
- Created `setup_system_limits.sh` for OS configuration
- Created `PostgreSQL_Migration.md` with detailed setup instructions
## SQL Compatibility
The code handles SQL dialect differences between SQLite and PostgreSQL:
| Operation | SQLite | PostgreSQL |
|-----------|--------|------------|
| Upsert | `INSERT OR IGNORE/REPLACE` | `ON CONFLICT DO ...` |
| Auto-increment | `INTEGER PRIMARY KEY AUTOINCREMENT` | `SERIAL PRIMARY KEY` |
| Parameter placeholder | `?` | `%s` |
| Integer types | `INTEGER` | `BIGINT` for timestamps |
## Performance Benefits
With PostgreSQL:
- **No more "database is locked" errors** at high concurrency
- **Scales to 500+ workers** without contention
- **MVCC** allows multiple concurrent writers
- **Built-in connection pooling** with psycopg3
- **Better monitoring** through pg_stat views
## Usage
### SQLite (default, for testing)
```bash
python main.py --seed-file seeds.txt --email user@example.com --max-workers 50python main.py \
--seed-file seeds.txt \
--email user@example.com \
--db-type postgresql \
--db-url "postgresql://user:pass@localhost/crawler_db" \
--max-workers 500- SQLite remains the default database
- No changes required for existing SQLite deployments
- Migration script available for moving data to PostgreSQL
- All tests continue to use SQLite for speed
-
Install PostgreSQL dependencies:
pip install 'psycopg[binary,pool]>=3.1' -
Set up PostgreSQL (see PostgreSQL_Migration.md)
-
Migrate existing data (optional):
python migrate_to_postgresql.py \ --sqlite-db crawler_data/crawler_state.db \ --pg-url "postgresql://user:pass@localhost/crawler_db" -
Configure system limits:
./setup_system_limits.sh
-
Run with PostgreSQL and enjoy 500+ concurrent workers!
crawler_module/db_backends.py(new)crawler_module/config.pycrawler_module/storage.pycrawler_module/frontier.pycrawler_module/politeness.pycrawler_module/orchestrator.pytests/test_frontier.pyrequirements.txtmigrate_to_postgresql.py(new)setup_system_limits.sh(new)PostgreSQL_Migration.md(new)PLAN.MD(updated)
```python
#!/usr/bin/env python3
"""
Simple test script to verify PostgreSQL backend functionality.
Run this after setting up PostgreSQL to ensure everything works.
"""
import asyncio
import sys
from crawler_module.db_backends import create_backend
async def test_postgresql():
if len(sys.argv) < 2:
print("Usage: python test_postgresql_backend.py <postgresql_url>")
print("Example: python test_postgresql_backend.py 'postgresql://user:pass@localhost/crawler_db'")
return 1
db_url = sys.argv[1]
try:
# Create PostgreSQL backend
print("Creating PostgreSQL backend...")
backend = create_backend('postgresql', db_url=db_url)
# Initialize connection pool
print("Initializing connection pool...")
await backend.initialize()
# Test basic operations
print("Testing basic operations...")
# Create a test table
await backend.execute("""
CREATE TABLE IF NOT EXISTS test_table (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
value INTEGER
)
""")
print("✓ Created test table")
# Insert data
await backend.execute(
"INSERT INTO test_table (name, value) VALUES (%s, %s)",
("test_item", 42)
)
print("✓ Inserted data")
# Query data
row = await backend.fetch_one(
"SELECT name, value FROM test_table WHERE name = %s",
("test_item",)
)
if row and row[0] == "test_item" and row[1] == 42:
print("✓ Queried data successfully")
else:
print("✗ Query returned unexpected data")
# Test RETURNING clause
results = await backend.execute_returning(
"INSERT INTO test_table (name, value) VALUES (%s, %s) RETURNING id, name",
("returning_test", 100)
)
if results and len(results) > 0:
print(f"✓ RETURNING clause works: id={results[0][0]}, name={results[0][1]}")
# Clean up
await backend.execute("DROP TABLE test_table")
print("✓ Cleaned up test table")
# Close connections
await backend.close()
print("\n✅ All tests passed! PostgreSQL backend is working correctly.")
return 0
except ImportError as e:
print(f"\n❌ Error: {e}")
print("Please install psycopg3: pip install 'psycopg[binary,pool]>=3.1'")
return 1
except Exception as e:
print(f"\n❌ Error: {e}")
print("Make sure PostgreSQL is running and the connection URL is correct.")
return 1
if __name__ == "__main__":
sys.exit(asyncio.run(test_postgresql()))
I've successfully migrated your web crawler to support PostgreSQL as a database backend, addressing the database contention issues you were experiencing with SQLite at high concurrency levels. Here's what I've done:
-
Database Abstraction Layer - Created a new
db_backends.pymodule with:- Abstract
DatabaseBackendinterface SQLiteBackendimplementation (preserving existing functionality)PostgreSQLBackendimplementation using psycopg3 with async support- Factory function to create the appropriate backend based on configuration
- Abstract
-
Configuration Updates - Added command-line options:
--db-typeto choose between sqlite (default) and postgresql--db-urlfor PostgreSQL connection string
-
Code Refactoring - Updated all modules to use the database abstraction:
storage.py: Now uses the database backend instead of direct SQLite connectionsfrontier.py: Converted to fully async database operationspoliteness.py: Removed connection passing, simplified to async operationsorchestrator.py: Initializes the appropriate database backend
-
SQL Compatibility - Handled differences between SQLite and PostgreSQL:
INSERT OR IGNORE/REPLACEvsON CONFLICT DO ...- Parameter placeholders (
?vs%s) - Data types (INTEGER vs BIGINT for timestamps)
-
Migration Tools - Created helpful scripts:
migrate_to_postgresql.py: Migrates data from SQLite to PostgreSQLsetup_system_limits.sh: Configures OS limits for high concurrencytest_postgresql_backend.py: Verifies PostgreSQL setup
-
Documentation:
- Updated
PLAN.MDwith PostgreSQL architecture - Created
PostgreSQL_Migration.mdwith detailed setup instructions - Created
README_DATABASE_MIGRATION.mdsummarizing all changes
- Updated
With PostgreSQL, you can now:
- Scale to 500+ concurrent workers without "database is locked" errors
- Take advantage of MVCC for true concurrent writes
- Use built-in connection pooling with psycopg3
- Monitor performance with PostgreSQL's rich tooling
- SQLite remains the default database
- All existing SQLite-based deployments continue to work unchanged
- Tests still use SQLite for speed
- Easy migration path when you're ready to scale
-
For testing with SQLite (unchanged):
python main.py --seed-file seeds.txt --email user@example.com --max-workers 50
-
For production with PostgreSQL:
# Install PostgreSQL driver pip install 'psycopg[binary,pool]>=3.1' # Set up PostgreSQL (see PostgreSQL_Migration.md) # Run with PostgreSQL python main.py \ --seed-file top-1M-domains.txt \ --email user@example.com \ --db-type postgresql \ --db-url "postgresql://user:pass@localhost/crawler_db" \ --max-workers 500
The solution maintains the same atomic claim-and-read pattern you implemented but now leverages PostgreSQL's superior concurrency handling to eliminate the bottleneck. The code changes are minimal, and the architecture remains clean and maintainable.
python main.py --seed-file seeds.txt --email user@example.com --max-workers 50
# Install PostgreSQL driver
pip install 'psycopg[binary,pool]>=3.1'
# Set up PostgreSQL (see PostgreSQL_Migration.md)
# Run with PostgreSQL
python main.py \
--seed-file top-1M-domains.txt \
--email user@example.com \
--db-type postgresql \
--db-url "postgresql://user:pass@localhost/crawler_db" \
--max-workers 500
User
Do you want to add these instructions to crawler/README.md?
Cursor
I'll check if there's an existing README.md file and then suggest appropriate additions for documenting the new PostgreSQL support.
Yes, I think it would be very helpful to add information about the PostgreSQL support to the main README.md. Here's what I suggest adding:
# Experimental Web Crawler
This project is an experimental/educational web crawler built in Python, designed to crawl web pages, extract text content, and manage politeness and fault tolerance.
Refer to `PLAN.MD` for a detailed project plan and architecture.
## Core Components
* `crawler_module/config.py`: Handles command-line argument parsing and configuration management.
* `crawler_module/utils.py`: Provides utility functions, especially for URL normalization and domain extraction.
* `crawler_module/storage.py`: Manages data persistence using SQLite for state (frontier, visited URLs, domain metadata) and the file system for crawled text content.
* `crawler_module/fetcher.py`: Asynchronously fetches web content (HTML, robots.txt) using `aiohttp`.
* `crawler_module/politeness.py`: Enforces politeness rules, including `robots.txt` parsing, crawl delays, and manual exclusion lists.
* `crawler_module/frontier.py`: Manages the queue of URLs to be crawled, interacting with storage and politeness rules.
* `crawler_module/parser.py`: Parses HTML content using `lxml` to extract links and text.
* `crawler_module/orchestrator.py`: Coordinates all components, manages concurrent worker tasks, and controls the overall crawl lifecycle.
* `crawler_module/db_backends.py`: Database abstraction layer supporting both SQLite and PostgreSQL.
* `main.py`: The main entry point to run the crawler.
## Database Backend Support
The crawler supports two database backends:
### SQLite (Default)
- **Best for**: Testing, development, and small-scale crawls
- **Max concurrent workers**: ~50-100
- **Setup**: No additional setup required
- **Limitations**: Single writer limitation can cause "database is locked" errors at high concurrency
### PostgreSQL
- **Best for**: Production use and large-scale crawls
- **Max concurrent workers**: 500+
- **Setup**: Requires PostgreSQL installation and configuration
- **Benefits**: True concurrent writes, better performance at scale
For detailed PostgreSQL setup and migration instructions, see [`PostgreSQL_Migration.md`](PostgreSQL_Migration.md).
## Setup
1. **Clone/Create Project**: Get the project files into a local directory.
2. **Navigate to `crawler/` directory**.
3. **Create a Python Virtual Environment**:
```bash
python3 -m venv .venv
```
4. **Activate the Virtual Environment**:
* On macOS/Linux:
```bash
source .venv/bin/activate
```
* On Windows:
```bash
.venv\Scripts\activate
```
5. **Install Dependencies**:
```bash
pip install -r requirements.txt
```
Key dependencies include: `aiohttp`, `lxml`, `robotexclusionrulesparser`, `aiofiles`, `cchardet`, `tldextract`, `pytest`, `pytest-asyncio`, `httpx`, and optionally `psycopg[binary,pool]` for PostgreSQL support.
## Running the Crawler
The crawler is run via `main.py` from the project root directory (e.g., `vibe-coding-experiments/` if `crawler` is inside it, or directly from `crawler/` if you `cd` into it first):
### Using SQLite (Default)
```bash
python crawler/main.py --seed-file path/to/your/seeds.txt --email your.email@example.com [options]Or if you are inside the crawler directory:
python main.py --seed-file path/to/your/seeds.txt --email your.email@example.com [options]python main.py --seed-file path/to/your/seeds.txt --email your.email@example.com \
--db-type postgresql \
--db-url "postgresql://user:password@localhost/crawler_db" \
--max-workers 500Required Arguments:
--seed-file <path>: Path to the seed file (newline-separated domains/URLs).--email <email_address>: Your contact email for the User-Agent string.
Optional Arguments (see crawler_module/config.py or run with --help for full list):
--data-dir <path>: Directory for database and content (default:./crawler_data).--exclude-file <path>: File of domains to exclude.--max-workers <int>: Number of concurrent fetchers (default: 20).--max-pages <int>: Max pages to crawl.--max-duration <seconds>: Max crawl duration.--log-level <LEVEL>: DEBUG, INFO, WARNING, ERROR (default: INFO).--resume: Attempt to resume from existing data.--seeded-urls-only: Only crawl seeded URLs.--db-type <sqlite|postgresql>: Database backend to use (default: sqlite).--db-url <url>: PostgreSQL connection URL (required if db-type is postgresql).
When running with many workers (especially with PostgreSQL), you may need to increase system limits:
# Run the provided setup script
chmod +x setup_system_limits.sh
./setup_system_limits.sh
# Or manually set limits
ulimit -n 65536 # Increase file descriptor limitThe project uses pytest for testing. To run the tests:
- Ensure you have activated your virtual environment and installed dependencies (including
pytestandpytest-asynciofromrequirements.txt). - Navigate to the
crawler/directory (the root of the crawler project wheretests/is located). - Run pytest:
Or, for more verbose output:
pytest
pytest -v
- Logs: Check console output or redirect to a file. Log level is configurable.
- Database (
crawler_data/crawler_state.dbwithin your specified data directory): Use any SQLite client (e.g.,sqlite3CLI, DB Browser for SQLite) to inspect tables likefrontier,visited_urls,domain_metadata. - Content Files (
crawler_data/content/within your specified data directory): Extracted text content is stored in.txtfiles named by URL hash.
If you have an existing SQLite crawler database and want to migrate to PostgreSQL:
python migrate_to_postgresql.py \
--sqlite-db crawler_data/crawler_state.db \
--pg-url "postgresql://user:password@localhost/crawler_db"See PostgreSQL_Migration.md for detailed instructions.
I've successfully updated the README.md file to include comprehensive information about the new PostgreSQL support. The additions include:
1. **Database Backend Support section**: Clearly explains the two database options (SQLite and PostgreSQL), their use cases, limitations, and benefits.
2. **Updated setup instructions**: Added the optional PostgreSQL dependency to the list of packages.
3. **Enhanced running instructions**:
- Separated examples for SQLite (default) and PostgreSQL usage
- Added the new `--db-type` and `--db-url` command-line options
4. **System requirements section**: Instructions for setting up system limits when running with high concurrency.
5. **Migration section**: Quick example of how to migrate from SQLite to PostgreSQL using the provided script.
The updates maintain backward compatibility by keeping SQLite as the default while clearly documenting how to leverage PostgreSQL for production-scale crawling with 500+ workers. The documentation now provides a complete picture of the database options available and guides users to the detailed migration documentation when needed.