Quick Start

1from crawlingo import Page
2
3# Fetch and parse in one call
4page = Page("https://example.com")
5
6print(page.title())           # "Example Domain"
7print(page.status)            # 200
8print(page.markdown()[:120])  # Clean GitHub-flavored markdown
9
10# CSS selectors
11h1 = page.css("h1").text()
12links = [el.attr("href") for el in page.css("a")]
13
14# XPath
15paragraphs = [p.text() for p in page.xpath("//p")]
16
17# Regex — returns raw matches
18prices = page.regex(r"\$[\d,.]+")
19
20# Text anchors (SIMD-accelerated)
21el = page.find_text("Price:")        # exact text lookup
22sibling = page.after_text("Price:")  # element after "Price:"

1from crawlingo import Session, Page
2
3with Session() as s:
4    # Self-healing selectors
5    s.auto_match(True)
6
7    # Stealth mode — rotate TLS fingerprints
8    s.fetcher_tier("stealthy")
9    s.browser_profile("chrome")  # or "firefox", "safari"
10
11    # Rate limiting (per-host token bucket)
12    s.rate_limit(5.0)   # 5 req/s per host
13
14    # Proxy rotation
15    s.proxy_pool([
16        "http://user:pass@proxy1:8080",
17        "http://user:pass@proxy2:8080",
18    ])
19
20    # Auth
21    s.bearer_auth("eyJhbGciOi...")
22
23    # Reuse session across multiple fetches
24    page1 = s.page("https://example.com")
25    page2 = s.page("https://other-site.com")
26    print(s.metrics())  # lock-free counters

1from crawlingo import Dataset, Session
2
3with Session() as s:
4    s.auto_match(True).rate_limit(5)
5
6    ds = (
7        Dataset("https://shop.example.com/products", session=s)
8        .field("title",    "h1")
9        .field("price",    ".price")
10        .field("rating",   ".star-score")
11        .field("url",      "")
12        .field("email",    r"[\w.+]+@[\w.]+", selector_type="regex")
13        .build()
14    )
15
16    print(ds.to_dict())        # {"title": "...", "price": "49.99"}
17    ds.to_json("data.json")    # JSON file
18    ds.to_csv("data.csv")      # CSV file
19    ds.to_parquet("data.parquet")  # Parquet for Spark/BigQuery

1from crawlingo import Crawl
2
3results = (
4    Crawl("https://docs.example.com")
5    .follow("nav a, main a")      # CSS for links to follow
6    .limit(500)                    # max pages
7    .depth(5)                      # max link depth
8    .concurrency(10)               # concurrent fetches
9    .delay(0.5)                    # polite delay
10    .field("title",   "h1")
11    .field("content", "article")
12    .field("url",     "")
13    .webhook("https://api.example.com/webhooks/crawl")
14    .build()
15)
16
17results.to_json("crawl.json")
18results.to_parquet("crawl.parquet")

1from crawlingo import Watch
2import threading
3
4def on_change(event):
5    print(f"Field '{event.field}' changed from {event.old_value} to {event.new_value}")
6
7w = (
8    Watch("https://shop.example.com/product/1")
9    .field("price", ".price")
10    .field("stock", ".stock-badge")
11    .interval(300)                    # poll every 5 minutes
12    .on_change(on_change)
13)
14
15# Run in background thread
16t = threading.Thread(target=w.run)
17t.start()
18# w.stop()           # signal stop