CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Project Overview

PCrawler v2.0 - Production-grade web crawler với N8N + API orchestration

Purpose: Web crawling platform với visual workflow automation (N8N), API orchestration, và async batch processing
Main Use Cases:
- Crawl company directories (Zalo OA, 1900.com.vn)
- Two-phase crawling (list pages → detail pages)
- Large-scale batch crawling (20k+ URLs)
- LLM-powered data extraction (Crawl4AI)

Tech Stack

Core:

Python 3.12+ with uv package manager
FastAPI - REST API server
Celery - Async task queue (prefork pool, 4 workers)
Redis - Celery broker + result backend
PostgreSQL - Primary database (SQLAlchemy ORM)

Crawling:

Playwright - Headless browser (with rate limiting)
lxml - Fast CSS/XPath extraction
Crawl4AI - LLM-powered extraction (Llama 3.1 8B)
Regex - Email/phone pattern extraction

Orchestration:

N8N - Visual workflow automation (primary)
Airflow - Optional for heavy batch workflows

Infrastructure:

Docker Compose - All services containerized
Flower - Celery monitoring (http://localhost:5555)

Architecture

High-Level Flow

N8N Workflow (Visual) ─┐
                        ├─→ FastAPI (/n8n/crawl/dynamic) ─→ Celery Worker ─→ PostgreSQL
Direct API Call ───────┘    (app.api)                        (app.worker)

Key Components

1. API Layer (app/api/)

main.py - FastAPI app với CORS, Prometheus metrics
routers/n8n.py - N8N integration endpoints (/n8n/crawl/dynamic, /n8n/status/{id}, /n8n/results/{id})
routers/data.py - Data management endpoints (/tables, /tables/{name}/data)

2. Worker Layer (app/worker.py)

Celery tasks: run_dynamic_batch_crawl_task, run_batch_crawl_task, run_list_crawl_task
Batch processing: Split URLs into batches (default 10 URLs/batch)
Rate limiting: Exponential backoff per domain
CRITICAL: Always pass table_name parameter to crawl_url() to avoid UUID table creation

3. Crawl Engine (app/engine/)

crawl.py - Core crawl logic (fetch_page, crawl_url)
extractors.py - Data extraction strategies:
- LxmlExtractor - Fast CSS/XPath (production default)
- Crawl4AIExtractor - LLM-powered extraction
- RegexExtractor - Email/phone patterns
- SmartExtractor - Auto-combine all extractors
browser_service.py - Playwright browser pool management

4. Database Layer (app/database/)

models.py - SQLAlchemy models (Execution, DataItem)
data_service.py - CRUD operations, dynamic table creation

Data Flow for Dynamic Crawl

N8N calls → POST /n8n/crawl/dynamic with {table_name, schema_def, url_pattern, ids, selectors}
API creates execution → Saves to executions table with status="queued"
Celery task dispatched → run_dynamic_batch_crawl_task.delay(execution_id, crawl_config)
Worker processes batches:
- Split URLs into batches (batch_size=10)
- For each URL: fetch_page() → extract() → save_to_db()
- Rate limiting between requests (1-34s delay)
- Update execution logs with progress
Data saved → Dynamic table created based on schema_def, data inserted
N8N polls status → GET /n8n/status/{execution_id} until status="completed"
N8N fetches results → GET /n8n/results/{execution_id} or GET /tables/{table_name}/data

Common Commands

Development Setup

# Start all services (API, Worker, N8N, Redis, PostgreSQL)
docker compose -f docker-compose.n8n.yml up -d

# Check services
docker compose ps

# View logs
docker logs pcrawler-api --tail 100 -f
docker logs pcrawler-worker --tail 100 -f
docker logs pcrawler-n8n --tail 100 -f

# Restart services
docker restart pcrawler-worker  # Often needed after code changes
docker restart pcrawler-api

Access Services

API: http://localhost:8000
API Docs: http://localhost:8000/docs
N8N: http://localhost:5678
Flower: http://localhost:5555
Airflow (optional): http://localhost:8080 (admin/admin)

Testing

# Test dynamic crawl API
curl -X POST http://localhost:8000/n8n/crawl/dynamic \
  -H "Content-Type: application/json" \
  -d '{
    "table_name": "test",
    "url_pattern": "https://example.com/{id}",
    "ids": ["1"],
    "selectors": {"title": ["h1"]}
  }'

# Check execution status
curl http://localhost:8000/n8n/status/{execution_id}

# Get results
curl http://localhost:8000/tables/test/data?limit=10

Database

# Access PostgreSQL
docker exec -it pcrawler-postgres psql -U pcrawler -d pcrawler

# Common queries
\dt                                    # List tables
SELECT * FROM executions LIMIT 10;     # Recent executions
SELECT * FROM test LIMIT 10;           # Data from 'test' table

Worker Management

# View active tasks
docker logs pcrawler-worker --tail 50 | grep "Crawling"

# Check Celery status
docker exec pcrawler-worker celery -A app.tasks.celery_app status

# Inspect task
docker exec pcrawler-worker celery -A app.tasks.celery_app inspect active

# Restart worker (kills stuck tasks)
docker restart pcrawler-worker

Critical Implementation Details

1. Table Name Parameter Flow

MUST pass table_name through entire call chain:

# app/worker.py - CRITICAL FIX
table_name = crawl_config.get("table_name", "crawled_data")

data = await crawl_url(
    url=url,
    config=crawl_config,
    execution_id=f"{exec_id}_{idx}",
    table_name=table_name  # ← MUST pass this!
)

Why: Without this, crawl_url() falls back to execution_id (UUID) as table name, creating 200+ random tables.

2. Execution Logs Format

Two formats exist (backward compatibility required):

# OLD format (strings)
logs = ["Log message 1", "Log message 2"]

# NEW format (dicts with metadata)
logs = [
    {"message": "Log message", "metadata": {"table_name": "...", "total_urls": 100}},
    ...
]

# Always check format:
if isinstance(log, dict):
    msg = log.get("message")
    metadata = log.get("metadata", {})
else:
    msg = log

3. Rate Limiting

Exponential backoff per domain in app/engine/crawl.py:

# domain_request_tracker = {domain: last_request_time}
# Delay increases: 1s, 2s, 4s, 8s, ... up to 60s max

Why: Prevents Zalo/target sites from blocking. Can cause URL 1/100 to appear "stuck" for 30-60s.

4. LLM Prompt Strategy (Phase 2 Crawl)

Priority-based search for contact extraction:

Contact/Liên hệ section (highest priority)
Footer/cuối trang (fallback)
Header/top (supplementary for hotline)

Extract: phone (from website), email, facebook fanpage, address, description

5. Playwright Timeout

# app/engine/crawl.py
async def fetch_page(url: str, timeout: int = 60000):  # 60s default

Issue: Task can still hang if Playwright process gets stuck. Solution: docker restart pcrawler-worker

N8N Workflow Patterns

Two-Phase Crawl with LLM

File: docs/n8n-workflows/two-phase-crawl-with-llm.json

Flow:

Phase 1: Crawl Zalo OA pages (CSS selectors) → Extract phone, website, address
Phase 2: Crawl websites from Phase 1 (LLM extraction) → Extract email, phone, facebook, description
Format Data: Map to Google Sheets columns (business_type, company_name, website_url, phone_zalo, phone_website, email, facebook, description)
Append to Google Sheets: Skip sensitive fields (source_url, oa_id)

Usage:

Import JSON to N8N
Configure Google Sheets OAuth2 credential
Create spreadsheet with header row
Execute workflow

Dynamic Crawl Workflow

Flow:

Read IDs from file (/home/node/.n8n-files/zalo_oa_ids.txt)
Parse IDs (limit to 100 for testing)
Call /n8n/crawl/dynamic with selectors
Poll /n8n/status/{id} every 10s
When complete, fetch results from /tables/{name}/data

Code Style and Conventions

Python Style

Type hints required for function signatures
Docstrings for public functions/classes (Google style)
Async/await for I/O operations
FastAPI dependency injection for database sessions
SQLAlchemy ORM for database operations

Naming Conventions

Functions: snake_case (e.g., fetch_page, crawl_url)
Classes: PascalCase (e.g., LxmlExtractor, WorkflowEngine)
Constants: UPPER_SNAKE_CASE (e.g., REDIS_URL, MAX_RETRIES)
Private: Prefix with _ (e.g., _internal_helper)

Error Handling

# Log errors but don't crash entire batch
try:
    result = await crawl_url(url)
except Exception as e:
    logger.error(f"Failed to crawl {url}: {e}")
    errors.append({"url": url, "error": str(e)})
    continue  # Continue with next URL

Logging

import logging
logger = logging.getLogger(__name__)

# Use WARNING for business logic events (visible in Celery logs)
logger.warning(f"[{exec_id}] [1/100] Crawling: {url}")
logger.warning(f"[{exec_id}] ✅ Saved to {table_name}: {url}")

# Use INFO for debug details
logger.info(f"Successfully fetched {url}")

Common Issues and Solutions

Task Stuck at "STARTED"

Symptom: Flower shows STARTED but no logs, task hangs on first URL

Causes:

Playwright browser stuck/timeout
URL blocking/unresponsive
No error handling for timeout

Fix:

docker restart pcrawler-worker  # Kills stuck process

200+ Random UUID Tables

Symptom: Database has tables like f66b32e9_3210_4083_... instead of zalo_oa_companies

Cause: table_name not passed to crawl_url() in app/worker.py

Fix: See "Table Name Parameter Flow" above

Status Endpoint 500 Error

Symptom: N8N fails to check status with AttributeError

Causes:

Code accessing non-existent Execution model fields
Mixing old/new log formats

Fix: Extract metadata from logs[0].metadata instead of direct field access, handle both log formats

Rate Limit Issues

Symptom: "Rate limit reached" in logs, slow crawling

Not a bug: Exponential backoff working as designed to avoid bans

To speed up: Implement proxy rotation (see docs/PROXY-ROTATION-PLAN.md)

When Making Changes

After Modifying Worker Code

docker restart pcrawler-worker  # Worker doesn't auto-reload

After Modifying API Code

API auto-reloads with --reload flag (in development)

After Changing N8N Workflows

Export workflow to JSON
Save to docs/n8n-workflows/
Document changes in workflow-specific docs

Before Deploying

Test with small batch (10-20 URLs)
Check logs for errors
Verify data in database
Scale up gradually

Documentation Files

README.md - Quick start, architecture overview
AIRFLOW.md - Airflow setup, DAG configuration (optional feature)
N8N.md - N8N integration, workflow patterns
REFERENCE.md - Command cheatsheet, troubleshooting
docs/QUICK-START-DYNAMIC-CRAWL.md - Dynamic crawl API guide
docs/DYNAMIC-CRAWL-WORKFLOW.md - Complete workflow documentation
docs/GOOGLE-SHEETS-INTEGRATION.md - Google Sheets export setup
docs/PROXY-ROTATION-PLAN.md - Proxy implementation plan

FilesExpand file tree

CLAUDE.md

Latest commit

History

CLAUDE.md

File metadata and controls

CLAUDE.md

Project Overview

Tech Stack

Architecture

High-Level Flow

Key Components

Data Flow for Dynamic Crawl

Common Commands

Development Setup

Access Services

Testing

Database

Worker Management

Critical Implementation Details

1. Table Name Parameter Flow

2. Execution Logs Format

3. Rate Limiting

4. LLM Prompt Strategy (Phase 2 Crawl)

5. Playwright Timeout

N8N Workflow Patterns

Two-Phase Crawl with LLM

Dynamic Crawl Workflow

Code Style and Conventions

Python Style

Naming Conventions

Error Handling

Logging

Common Issues and Solutions

Task Stuck at "STARTED"

200+ Random UUID Tables

Status Endpoint 500 Error

Rate Limit Issues

When Making Changes

After Modifying Worker Code

After Modifying API Code

After Changing N8N Workflows

Before Deploying

Documentation Files