Advanced Features

1from crawlingo import Page
2from crawlingo.hooks import strip_whitespace, uppercase, log_request
3
4page = (
5    Page("https://example.com")
6    .before_fetch(log_request)                               # Before HTTP request
7    .before_parse(lambda html: html.replace("Old", "New"))   # Before parsing HTML
8    .after_extract(strip_whitespace)                         # After extraction
9    .after_extract(uppercase)                                # Chain multiple hooks
10)

1from crawlingo import Session, Dataset
2from crawlingo.transport import MockTransport
3
4mock = MockTransport()
5mock.with_html("https://example.com", "<h1>Hello World</h1>")
6
7session = Session()
8session.set_transport(mock)
9
10result = Dataset("https://example.com", session=session)\
11    .field("title", "h1")\
12    .build()
13
14print(result.to_dict())  # {"title": "Hello World"}

1dataset = Dataset("https://example.com")
2dataset.field("title", "h1")
3dataset.field("price", "span.price", extraction_type="price")
4
5stream = dataset.build_many_streamed(
6    urls=["https://example.com/a", "https://example.com/b"],
7    concurrency=10
8)
9
10for record in stream:
11    print(record.fields)

1session = Session()
2
3# 1. Static proxy
4session.proxy("http://user:pass@proxy:8080")
5
6# 2. Round-robin proxy pool
7session.proxy_pool([
8    "http://proxy1:8080",
9    "http://proxy2:8080",
10    "http://proxy3:8080",
11])
12
13# 3. Dynamic proxy provider endpoint
14session.proxy_provider("https://proxy-service.example.com/get")

1session = Session()
2
3# Retrieve lock-free atomic metrics
4metrics = session.metrics()
5print(f"Requests: {metrics.request_count}")
6print(f"Success:  {metrics.success_count}")
7print(f"Errors:   {metrics.error_count}")
8print(f"Cache Hits: {metrics.cache_hits}")

1crawlingo mcp --host 127.0.0.1 --port 8000